[SQG] Do not follow symlinks (#41778)

### What does this PR do?
This is a minor update in experimental SQGs to stop following symlinks and reflecting them in report as such to get precise size estimation.

Author: JSGette

tasks/quality_gates.py

@@ -424,7 +424,6 @@ def measure_package_local(
     config_path="test/static/static_quality_gates.yml",
     output_path=None,
     build_job_name="local_test",
-    max_files=20000,
     no_checksums=False,
     debug=False,
 ):
@@ -454,7 +453,6 @@ def measure_package_local(
         config_path=config_path,
         output_path=output_path,
         build_job_name=build_job_name,
-        max_files=max_files,
         no_checksums=no_checksums,
         debug=debug,
     )

tasks/static_quality_gates/experimental_gates.py

@@ -27,18 +27,27 @@
 class FileInfo:
     """
     Information about a single file within an artifact.
+
+    For regular files, only relative_path, size_bytes, and optionally checksum are set.
+    For symlinks, is_symlink is True and symlink_target contains the link target.
+    For broken symlinks, is_broken is True.
     """
 
     relative_path: str
     size_bytes: int
     checksum: str | None = None
+    is_symlink: bool | None = None
+    symlink_target: str | None = None
+    is_broken: bool | None = None
 
     def __post_init__(self):
         """Validate file info data"""
         if not self.relative_path:
             raise ValueError("relative_path cannot be empty")
         if self.size_bytes < 0:
             raise ValueError("size_bytes must be non-negative")
+        if self.is_symlink and not self.symlink_target:
+            raise ValueError("symlink_target must be provided when is_symlink is True")
 
     @property
     def size_mb(self) -> float:
@@ -124,7 +133,6 @@ def measure_package(
         package_path: str,
         gate_name: str,
         build_job_name: str,
-        max_files: int = 20000,
         generate_checksums: bool = True,
         debug: bool = False,
     ) -> InPlaceArtifactReport:
@@ -153,7 +161,7 @@ def measure_package(
         gate_config = create_quality_gate_config(gate_name, self.config[gate_name])
 
         measurement, file_inventory = self._extract_and_analyze_package(
-            ctx, package_path, gate_config, max_files, generate_checksums, debug
+            ctx, package_path, gate_config, generate_checksums, debug
         )
 
         return InPlaceArtifactReport(
@@ -177,7 +185,6 @@ def _extract_and_analyze_package(
         ctx: Context,
         package_path: str,
         config: QualityGateConfig,
-        max_files: int = 10000,
         generate_checksums: bool = True,
         debug: bool = False,
     ) -> tuple[ArtifactMeasurement, list[FileInfo]]:
@@ -188,7 +195,6 @@ def _extract_and_analyze_package(
             ctx: Invoke context for running commands
             package_path: Path to the package file
             config: Quality gate configuration
-            max_files: Maximum number of files to process in inventory
             generate_checksums: Whether to generate checksums for files
             debug: Enable debug logging
 
@@ -213,7 +219,7 @@ def _extract_and_analyze_package(
                     artifact_path=package_path, on_wire_size=wire_size, on_disk_size=disk_size
                 )
 
-                file_inventory = self._walk_extracted_files(extract_dir, max_files, generate_checksums, debug)
+                file_inventory = self._walk_extracted_files(extract_dir, generate_checksums, debug)
 
                 if debug:
                     print("✅ Single extraction completed:")
@@ -226,15 +232,12 @@ def _extract_and_analyze_package(
         except Exception as e:
             raise RuntimeError(f"Failed to extract and analyze package {package_path}: {e}") from e
 
-    def _walk_extracted_files(
-        self, extract_dir: str, max_files: int, generate_checksums: bool, debug: bool
-    ) -> list[FileInfo]:
+    def _walk_extracted_files(self, extract_dir: str, generate_checksums: bool, debug: bool) -> list[FileInfo]:
         """
         Walk through extracted files and create file inventory.
 
         Args:
             extract_dir: Directory containing extracted package files
-            max_files: Maximum number of files to process
             generate_checksums: Whether to generate checksums for files
             debug: Enable debug logging
 
@@ -260,19 +263,55 @@ def _walk_extracted_files(
         total_size = 0
 
         for file_path in extract_path.rglob('*'):
-            if file_path.is_file():
-                # Respect max_files limit
-                if files_processed >= max_files:
-                    if debug:
-                        print(f"⚠️  Reached max files limit ({max_files}), stopping inventory")
-                    break
-
-                try:
-                    relative_path = str(file_path.relative_to(extract_path))
-                    file_stat = file_path.stat()
+            # Skip directories
+            if file_path.is_dir():
+                continue
+
+            try:
+                relative_path = str(file_path.relative_to(extract_path))
+
+                if file_path.is_symlink():
+                    try:
+                        symlink_target = os.readlink(file_path)
+                        logical_size = len(symlink_target)
+                        is_broken = False
+
+                        try:
+                            resolved_target = file_path.resolve(strict=True)
+                            if resolved_target.is_relative_to(extract_path):
+                                symlink_target_rel = str(resolved_target.relative_to(extract_path))
+                            else:
+                                symlink_target_rel = symlink_target
+                        except (OSError, RuntimeError):
+                            symlink_target_rel = symlink_target
+                            is_broken = True
+
+                        file_inventory.append(
+                            FileInfo(
+                                relative_path=relative_path,
+                                size_bytes=logical_size,
+                                checksum=None,
+                                is_symlink=True,
+                                symlink_target=symlink_target_rel,
+                                is_broken=is_broken if is_broken else None,
+                            )
+                        )
+
+                        if debug and files_processed % 1000 == 0:
+                            broken_marker = " [BROKEN]" if is_broken else ""
+                            print(f"🔗 Symlink: {relative_path} -> {symlink_target_rel}{broken_marker}")
+
+                    except OSError as e:
+                        if debug:
+                            print(f"⚠️  Could not read symlink {file_path}: {e}")
+                        continue
+
+                elif file_path.is_file():
+                    # Regular file - use lstat to not follow symlinks
+                    file_stat = file_path.lstat()
                     size_bytes = file_stat.st_size
 
-                    checksum = self._generate_checksum(file_path)
+                    checksum = self._generate_checksum(file_path) if generate_checksums else None
 
                     file_inventory.append(
                         FileInfo(
@@ -282,16 +321,17 @@ def _walk_extracted_files(
                         )
                     )
 
-                    files_processed += 1
                     total_size += size_bytes
 
                     if debug and files_processed % 1000 == 0:
                         print(f"📋 Processed {files_processed} files...")
 
-                except (OSError, PermissionError) as e:
-                    if debug:
-                        print(f"⚠️  Skipping file {file_path}: {e}")
-                    continue
+                files_processed += 1
+
+            except (OSError, PermissionError) as e:
+                if debug:
+                    print(f"⚠️  Skipping file {file_path}: {e}")
+                continue
 
         # Sort by size (descending) for easier analysis
         file_inventory.sort(key=lambda f: f.size_bytes, reverse=True)
@@ -349,14 +389,7 @@ def save_report_to_yaml(self, report: InPlaceArtifactReport, output_path: str) -
                 "arch": report.arch,
                 "os": report.os,
                 "build_job_name": report.build_job_name,
-                "file_inventory": [
-                    {
-                        "relative_path": file_info.relative_path,
-                        "size_bytes": file_info.size_bytes,
-                        "checksum": file_info.checksum,
-                    }
-                    for file_info in report.file_inventory
-                ],
+                "file_inventory": [self._serialize_file_info(file_info) for file_info in report.file_inventory],
             }
 
             with open(output_path, 'w') as f:
@@ -365,6 +398,32 @@ def save_report_to_yaml(self, report: InPlaceArtifactReport, output_path: str) -
         except Exception as e:
             raise RuntimeError(f"Failed to save report to {output_path}: {e}") from e
 
+    def _serialize_file_info(self, file_info: FileInfo) -> dict[str, Any]:
+        """
+        Serialize a FileInfo object to a dictionary, excluding None/False fields for regular files.
+
+        Args:
+            file_info: The FileInfo object to serialize
+
+        Returns:
+            Dictionary with only relevant fields
+        """
+        result = {
+            "relative_path": file_info.relative_path,
+            "size_bytes": file_info.size_bytes,
+        }
+
+        if file_info.checksum is not None:
+            result["checksum"] = file_info.checksum
+
+        if file_info.is_symlink:
+            result["is_symlink"] = True
+            result["symlink_target"] = file_info.symlink_target
+            if file_info.is_broken:
+                result["is_broken"] = True
+
+        return result
+
 
 def measure_package_local(
     ctx,
@@ -373,7 +432,6 @@ def measure_package_local(
     config_path="test/static/static_quality_gates.yml",
     output_path=None,
     build_job_name="local_test",
-    max_files=20000,
     no_checksums=False,
     debug=False,
 ):
@@ -389,7 +447,6 @@ def measure_package_local(
         config_path: Path to quality gates configuration (default: test/static/static_quality_gates.yml)
         output_path: Path to save the measurement report (default: {gate_name}_report.yml)
         build_job_name: Simulated build job name (default: local_test)
-        max_files: Maximum number of files to process in inventory (default: 20000)
         no_checksums: Skip checksum generation for faster processing (default: false)
         debug: Enable debug logging for troubleshooting (default: false)
 
@@ -437,7 +494,6 @@ def measure_package_local(
             package_path=package_path,
             gate_name=gate_name,
             build_job_name=build_job_name,
-            max_files=max_files,
             generate_checksums=not no_checksums,
             debug=debug,
         )

tasks/unit_tests/experimental_gates_tests.py

@@ -221,9 +221,7 @@ def test_measure_package_success(self, mock_exists):
 
         # Verify key methods were called
         mock_exists.assert_called_once_with("/path/to/package.deb")
-        mock_extract_analyze.assert_called_once_with(
-            mock_ctx, "/path/to/package.deb", unittest.mock.ANY, 20000, True, False
-        )
+        mock_extract_analyze.assert_called_once_with(mock_ctx, "/path/to/package.deb", unittest.mock.ANY, True, False)
 
     def test_measure_package_missing_file(self):
         """Test measuring package with missing file."""
@@ -279,7 +277,7 @@ def test_generate_checksum_failure(self, mock_file):
 
     def test_save_report_to_yaml(self):
         """Test saving report to YAML file."""
-        # Create a sample report
+        # Create a sample report with regular files and symlinks
         report = InPlaceArtifactReport(
             artifact_path="/path/to/package.deb",
             gate_name="static_quality_gate_agent_deb_amd64",
@@ -290,6 +288,8 @@ def test_save_report_to_yaml(self):
             file_inventory=[
                 FileInfo("opt/agent/bin/agent", 400000, "sha256:abc123"),
                 FileInfo("etc/config.yaml", 100000, None),
+                FileInfo("opt/bin/python3", 9, None, True, "python3.13", None),
+                FileInfo("opt/bin/broken_link", 15, None, True, "/nonexistent/path", True),
             ],
             measurement_timestamp="2024-01-15T14:30:22.123456Z",
             pipeline_id="12345",
@@ -316,13 +316,96 @@ def test_save_report_to_yaml(self):
             self.assertEqual(saved_data['gate_name'], "static_quality_gate_agent_deb_amd64")
             self.assertEqual(saved_data['on_wire_size'], 100000)
             self.assertEqual(saved_data['on_disk_size'], 500000)
-            self.assertEqual(len(saved_data['file_inventory']), 2)
-            self.assertEqual(saved_data['file_inventory'][0]['relative_path'], "opt/agent/bin/agent")
-            self.assertEqual(saved_data['file_inventory'][0]['checksum'], "sha256:abc123")
+            self.assertEqual(len(saved_data['file_inventory']), 4)
+
+            # Verify regular file doesn't have symlink fields
+            regular_file = saved_data['file_inventory'][0]
+            self.assertEqual(regular_file['relative_path'], "opt/agent/bin/agent")
+            self.assertEqual(regular_file['checksum'], "sha256:abc123")
+            self.assertNotIn('is_symlink', regular_file)
+            self.assertNotIn('symlink_target', regular_file)
+
+            # Verify file without checksum
+            no_checksum_file = saved_data['file_inventory'][1]
+            self.assertEqual(no_checksum_file['relative_path'], "etc/config.yaml")
+            self.assertNotIn('checksum', no_checksum_file)
+            self.assertNotIn('is_symlink', no_checksum_file)
+
+            # Verify symlink has symlink fields
+            symlink_file = saved_data['file_inventory'][2]
+            self.assertEqual(symlink_file['relative_path'], "opt/bin/python3")
+            self.assertTrue(symlink_file['is_symlink'])
+            self.assertEqual(symlink_file['symlink_target'], "python3.13")
+            self.assertNotIn('is_broken', symlink_file)
+
+            # Verify broken symlink has is_broken field
+            broken_symlink = saved_data['file_inventory'][3]
+            self.assertEqual(broken_symlink['relative_path'], "opt/bin/broken_link")
+            self.assertTrue(broken_symlink['is_symlink'])
+            self.assertEqual(broken_symlink['symlink_target'], "/nonexistent/path")
+            self.assertTrue(broken_symlink['is_broken'])
 
         finally:
             os.unlink(output_path)
 
+    def test_serialize_file_info_regular_file_with_checksum(self):
+        """Test serializing regular file with checksum."""
+        file_info = FileInfo("opt/agent/bin/agent", 400000, "sha256:abc123")
+        result = self.measurer._serialize_file_info(file_info)
+
+        # Should have relative_path, size_bytes, and checksum
+        self.assertEqual(result['relative_path'], "opt/agent/bin/agent")
+        self.assertEqual(result['size_bytes'], 400000)
+        self.assertEqual(result['checksum'], "sha256:abc123")
+
+        # Should NOT have symlink fields
+        self.assertNotIn('is_symlink', result)
+        self.assertNotIn('symlink_target', result)
+
+    def test_serialize_file_info_regular_file_without_checksum(self):
+        """Test serializing regular file without checksum."""
+        file_info = FileInfo("etc/config.yaml", 1024)
+        result = self.measurer._serialize_file_info(file_info)
+
+        # Should have relative_path and size_bytes only
+        self.assertEqual(result['relative_path'], "etc/config.yaml")
+        self.assertEqual(result['size_bytes'], 1024)
+
+        # Should NOT have checksum or symlink fields
+        self.assertNotIn('checksum', result)
+        self.assertNotIn('is_symlink', result)
+        self.assertNotIn('symlink_target', result)
+
+    def test_serialize_file_info_symlink(self):
+        """Test serializing symlink."""
+        file_info = FileInfo("opt/bin/python3", 9, None, True, "python3.13", None)
+        result = self.measurer._serialize_file_info(file_info)
+
+        # Should have relative_path, size_bytes, is_symlink, and symlink_target
+        self.assertEqual(result['relative_path'], "opt/bin/python3")
+        self.assertEqual(result['size_bytes'], 9)
+        self.assertTrue(result['is_symlink'])
+        self.assertEqual(result['symlink_target'], "python3.13")
+
+        # Should NOT have checksum or is_broken
+        self.assertNotIn('checksum', result)
+        self.assertNotIn('is_broken', result)
+
+    def test_serialize_file_info_broken_symlink(self):
+        """Test serializing broken symlink."""
+        file_info = FileInfo("opt/bin/broken_link", 15, None, True, "/nonexistent/path", True)
+        result = self.measurer._serialize_file_info(file_info)
+
+        # Should have all symlink fields plus is_broken
+        self.assertEqual(result['relative_path'], "opt/bin/broken_link")
+        self.assertEqual(result['size_bytes'], 15)
+        self.assertTrue(result['is_symlink'])
+        self.assertEqual(result['symlink_target'], "/nonexistent/path")
+        self.assertTrue(result['is_broken'])
+
+        # Should NOT have checksum
+        self.assertNotIn('checksum', result)
+
     @patch('builtins.open', side_effect=OSError("Permission denied"))
     def test_save_report_to_yaml_failure(self, mock_file):
         """Test handling of YAML save failures."""