Merge branch 'main' into chore/update-plan-code-smells

Author: jimisola

src/reqstool/command.py

@@ -17,8 +17,12 @@
 
 from reqstool_python_decorators.decorators.decorators import Requirements
 
-from reqstool.commands.exit_codes import EXIT_CODE_ALL_REQS_NOT_IMPLEMENTED, EXIT_CODE_MISSING_REQUIREMENTS_FILE
-from reqstool.common.exceptions import MissingRequirementsFileError
+from reqstool.commands.exit_codes import (
+    EXIT_CODE_ALL_REQS_NOT_IMPLEMENTED,
+    EXIT_CODE_ARTIFACT_ERROR,
+    EXIT_CODE_MISSING_REQUIREMENTS_FILE,
+)
+from reqstool.common.exceptions import ArtifactDownloadError, ArtifactExtractionError, MissingRequirementsFileError
 from reqstool.commands.generate_json.generate_json import GenerateJsonCommand
 from reqstool.commands.report import report
 from reqstool.commands.report.criterias.group_by import GroupbyOptions
@@ -405,6 +409,9 @@ def main():
     except MissingRequirementsFileError as exc:
         logging.fatal(str(exc))
         sys.exit(EXIT_CODE_MISSING_REQUIREMENTS_FILE)
+    except (ArtifactDownloadError, ArtifactExtractionError) as exc:
+        logging.fatal(str(exc))
+        sys.exit(EXIT_CODE_ARTIFACT_ERROR)
 
     sys.exit(exit_code)
 

src/reqstool/commands/exit_codes.py

@@ -1,3 +1,4 @@
 EXIT_CODE_MISSING_REQUIREMENTS_FILE = 1
+EXIT_CODE_ARTIFACT_ERROR = 2
 EXIT_CODE_SYNTAX_VALIDATION_ERROR = 128
 EXIT_CODE_ALL_REQS_NOT_IMPLEMENTED = 200

src/reqstool/common/exceptions.py

@@ -23,3 +23,17 @@ class CircularImplementationError(Exception):
     def __init__(self, urn: str, chain: list[str]):
         self.urn = urn
         super().__init__(f"Circular implementation detected: {' -> '.join(chain)} -> {urn}")
+
+
+class ArtifactDownloadError(Exception):
+    """Raised when a remote artifact cannot be downloaded."""
+
+    def __init__(self, message: str):
+        super().__init__(message)
+
+
+class ArtifactExtractionError(Exception):
+    """Raised when a downloaded artifact cannot be extracted."""
+
+    def __init__(self, message: str):
+        super().__init__(message)

src/reqstool/common/utils.py

@@ -201,30 +201,27 @@ def parse_version(version_str: str, urn_id: UrnId) -> PkgVersion:
             raise TypeError(f"Invalid version: {e} for: {urn_id}")
 
 
-class TempDirectoryUtil:
-    tmpdir: tempfile.TemporaryDirectory = None
-    count: int = 0
+class TempDirectoryManager:
+    """Instance-based temporary directory manager with guaranteed cleanup."""
 
-    @staticmethod
-    def _get_tmpdir() -> tempfile.TemporaryDirectory:
-        if TempDirectoryUtil.tmpdir is None:
-            TempDirectoryUtil.tmpdir = tempfile.TemporaryDirectory()
-        return TempDirectoryUtil.tmpdir
+    def __init__(self):
+        self._tmpdir = tempfile.TemporaryDirectory()
+        self._count = 0
 
-    @staticmethod
-    def get_path() -> Path:
-        return Path(TempDirectoryUtil._get_tmpdir().name)
+    def get_path(self) -> Path:
+        return Path(self._tmpdir.name)
 
-    @staticmethod
-    def get_suffix_path(suffix: str) -> Path:
-        new_path = Path(
-            os.path.join(
-                TempDirectoryUtil._get_tmpdir().name,
-                str(TempDirectoryUtil.count),
-                suffix,
-            )
-        )
+    def get_suffix_path(self, suffix: str) -> Path:
+        new_path = Path(self._tmpdir.name) / str(self._count) / suffix
         new_path.mkdir(parents=True, exist_ok=True)
-        TempDirectoryUtil.count += 1
-
+        self._count += 1
         return new_path
+
+    def cleanup(self):
+        self._tmpdir.cleanup()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        self.cleanup()

src/reqstool/locations/maven_location.py

@@ -2,12 +2,12 @@
 
 import logging
 import os
-import sys
 from typing import Optional
 
 from maven_artifact import Artifact, Downloader, RequestException
 from reqstool_python_decorators.decorators.decorators import Requirements
 
+from reqstool.common.exceptions import ArtifactDownloadError, ArtifactExtractionError
 from reqstool.common.utils import Utils
 from reqstool.locations.location import LocationInterface
 
@@ -41,13 +41,11 @@ def _make_available_on_localdisk(self, dst_path: str):
             if not downloader.download(artifact, filename=dst_path):
                 raise RequestException(f"Error downloading artifact {artifact} from: {self.url}")
         except RequestException as e:
-            logging.fatal(e.msg)
-            sys.exit(1)
+            raise ArtifactDownloadError(f"Error downloading artifact {artifact} from {self.url}: {e.msg}") from e
 
         logging.debug(f"Unzipping {artifact.get_filename(dst_path)} to {dst_path}\n")
 
         try:
             return Utils.extract_zip(artifact.get_filename(dst_path), dst_path)
         except ValueError as e:
-            logging.fatal(f"Maven artifact {artifact} from {self.url}: {e}")
-            sys.exit(1)
+            raise ArtifactExtractionError(f"Maven artifact {artifact} from {self.url}: {e}") from e

src/reqstool/locations/pypi_location.py

@@ -1,13 +1,13 @@
 import logging
 import os
 import re
-import sys
 import tarfile
 from typing import Optional
 from urllib.parse import urljoin
 
 import requests
 from bs4 import BeautifulSoup
+from reqstool.common.exceptions import ArtifactDownloadError, ArtifactExtractionError
 from reqstool.common.utils import Utils
 from reqstool.locations.location import LocationInterface
 
@@ -47,15 +47,12 @@ def _make_available_on_localdisk(self, dst_path: str):
             logging.debug(f"Extracting {downloaded_file} to {dst_path}\n")
             return Utils.extract_targz(downloaded_file, dst_path)
         except (ValueError, tarfile.TarError) as e:
-            logging.fatal(str(e))
-            sys.exit(1)
+            raise ArtifactExtractionError(str(e)) from e
         except Exception as e:
-            logging.fatal(
+            raise ArtifactDownloadError(
                 f"Error when downloading etc sdist pypi package for {self.package}=={self.version}"
-                f" in repo {self.url} {'with token' if token else ''}",
-                e,
-            )
-            sys.exit(1)
+                f" in repo {self.url} {'with token' if token else ''}: {e}"
+            ) from e
 
     @staticmethod
     def get_package_url(package, version, base_url, token) -> str:

src/reqstool/model_generators/combined_raw_datasets_generator.py

@@ -7,7 +7,7 @@
 from reqstool_python_decorators.decorators.decorators import Requirements
 
 from reqstool.common.exceptions import CircularImplementationError, CircularImportError, MissingRequirementsFileError
-from reqstool.common.utils import TempDirectoryUtil, Utils
+from reqstool.common.utils import TempDirectoryManager, Utils
 from reqstool.common.validators.semantic_validator import SemanticValidator
 from reqstool.location_resolver.location_resolver import LocationResolver
 from reqstool.locations.location import LocationInterface
@@ -34,6 +34,7 @@ def __init__(
         initial_location: LocationInterface,
         semantic_validator: SemanticValidator,
         database: Optional[RequirementsDatabase] = None,
+        tmpdir_manager: Optional[TempDirectoryManager] = None,
     ):
         self.__level: int = 0
         self.__initial_location_handler: LocationResolver = LocationResolver(
@@ -43,11 +44,12 @@ def __init__(
         self._parsing_order: List[str] = []
         self._parsing_graph: Dict[str, List[Tuple[str, str]]] = defaultdict(list)
         self._database = database
+        self._tmpdir_manager = tmpdir_manager if tmpdir_manager is not None else TempDirectoryManager()
         self.combined_raw_datasets = self.__generate()
 
     def __generate(self) -> CombinedRawDataset:
         # handle initial source
-        logging.debug(f"Using temporary path: {TempDirectoryUtil.get_path()}\n")
+        logging.debug(f"Using temporary path: {self._tmpdir_manager.get_path()}\n")
 
         raw_datasets: Dict[str, RawDataset] = {}
 
@@ -235,7 +237,7 @@ def __parse_source(self, current_location_handler: LocationResolver) -> RawDatas
         mvrs_data = None
         automated_tests = None
 
-        tmp_path = TempDirectoryUtil.get_suffix_path("can_we_use_urn_here").absolute()
+        tmp_path = self._tmpdir_manager.get_suffix_path("can_we_use_urn_here").absolute()
 
         actual_tmp_path = current_location_handler.make_available_on_localdisk(dst_path=tmp_path)
 

src/reqstool/storage/pipeline.py

@@ -5,6 +5,7 @@
 from contextlib import contextmanager
 from typing import Generator
 
+from reqstool.common.utils import TempDirectoryManager
 from reqstool.common.validators.lifecycle_validator import LifecycleValidator
 from reqstool.common.validators.semantic_validator import SemanticValidator
 from reqstool.locations.location import LocationInterface
@@ -20,13 +21,18 @@ def build_database(
     location: LocationInterface,
     semantic_validator: SemanticValidator,
     filter_data: bool = True,
+    tmpdir_manager: TempDirectoryManager = None,
 ) -> Generator[tuple[RequirementsDatabase, CombinedRawDataset], None, None]:
+    _owns_tmpdir = tmpdir_manager is None
+    if _owns_tmpdir:
+        tmpdir_manager = TempDirectoryManager()
     db = RequirementsDatabase()
     try:
         crdg = CombinedRawDatasetsGenerator(
             initial_location=location,
             semantic_validator=semantic_validator,
             database=db,
+            tmpdir_manager=tmpdir_manager,
         )
         crd = crdg.combined_raw_datasets
 
@@ -38,3 +44,5 @@ def build_database(
         yield db, crd
     finally:
         db.close()
+        if _owns_tmpdir:
+            tmpdir_manager.cleanup()

tests/unit/reqstool/common/test_temp_directory_manager.py

@@ -0,0 +1,31 @@
+from reqstool.common.utils import TempDirectoryManager
+
+
+def test_get_suffix_path_returns_unique_paths():
+    with TempDirectoryManager() as mgr:
+        p1 = mgr.get_suffix_path("test")
+        p2 = mgr.get_suffix_path("test")
+        assert p1 != p2
+        assert p1.exists()
+        assert p2.exists()
+
+
+def test_cleanup_removes_directory():
+    mgr = TempDirectoryManager()
+    path = mgr.get_path()
+    assert path.exists()
+    mgr.cleanup()
+    assert not path.exists()
+
+
+def test_context_manager_cleans_up():
+    with TempDirectoryManager() as mgr:
+        path = mgr.get_path()
+        assert path.exists()
+    assert not path.exists()
+
+
+def test_two_instances_are_independent():
+    with TempDirectoryManager() as mgr1:
+        with TempDirectoryManager() as mgr2:
+            assert mgr1.get_path() != mgr2.get_path()