diff --git a/src/twyn/dependency_parser/parsers/requirements_txt_parser.py b/src/twyn/dependency_parser/parsers/requirements_txt_parser.py index 1fad7cb..db7b362 100644 --- a/src/twyn/dependency_parser/parsers/requirements_txt_parser.py +++ b/src/twyn/dependency_parser/parsers/requirements_txt_parser.py @@ -1,5 +1,6 @@ """Parser for requirements.txt dependencies.""" +import logging import re from pathlib import Path @@ -7,6 +8,10 @@ from twyn.dependency_parser.parsers.abstract_parser import AbstractParser from twyn.dependency_parser.parsers.constants import REQUIREMENTS_TXT +from twyn.file_handler.exceptions import EmptyFileError, PathIsNotFileError, PathNotFoundError +from twyn.file_handler.file_handler import FileHandler + +logger = logging.getLogger("twyn") class RequirementsTxtParser(AbstractParser): @@ -37,11 +42,17 @@ def parse(self) -> set[str]: return self._parse_internal(self.file_path, seen_files=set()) def _parse_internal(self, source: str | Path, seen_files: set[Path]) -> set[str]: - """Parse requirements file and handle includes recursively.""" + """Parse a requirements file and handle ``-r`` includes recursively.""" packages: set[str] = set() - base_dir = Path(source).parent if isinstance(source, Path) else Path(".") + source = Path(source) + base_dir = source.parent + + # The entry file reuses the parser's own handler (keeping its + # existence/emptiness errors); every ``-r`` include is read from its + # own path instead of re-reading the entry file. + handler = self.file_handler if source == self.file_path else FileHandler(str(source)) - with self.file_handler.open("r") as fp: + with handler.open("r") as fp: for raw_line in fp: line = raw_line.strip() @@ -49,11 +60,7 @@ def _parse_internal(self, source: str | Path, seen_files: set[Path]) -> set[str] continue if line.startswith("-r "): - ref = line[3:].strip() - ref_path = (base_dir / ref).resolve() - if ref_path not in seen_files: - seen_files.add(ref_path) - packages.update(self._parse_internal(ref_path, seen_files)) + packages.update(self._parse_include(line[3:].strip(), base_dir, seen_files)) continue if line.startswith("-e "): @@ -73,6 +80,18 @@ def _parse_internal(self, source: str | Path, seen_files: set[Path]) -> set[str] return packages + def _parse_include(self, ref: str, base_dir: Path, seen_files: set[Path]) -> set[str]: + """Parse a file referenced with ``-r``, skipping it if already seen, missing, or empty.""" + ref_path = (base_dir / ref).resolve() + if ref_path in seen_files: + return set() + seen_files.add(ref_path) + try: + return self._parse_internal(ref_path, seen_files) + except (PathNotFoundError, PathIsNotFileError, EmptyFileError): + logger.warning("Skipping referenced requirements file (missing or empty): %s", ref_path) + return set() + @staticmethod def _is_valid_line(line: str) -> bool: """Check if line is valid for parsing.""" diff --git a/tests/dependency_parser/test_dependency_parser.py b/tests/dependency_parser/test_dependency_parser.py index 5b723a4..13a6f93 100644 --- a/tests/dependency_parser/test_dependency_parser.py +++ b/tests/dependency_parser/test_dependency_parser.py @@ -73,6 +73,24 @@ def test_parse_requirements_txt_file(self, requirements_txt_file: Path) -> None: parser = RequirementsTxtParser(file_path=requirements_txt_file) assert parser.parse() == {"South", "pycrypto", "Flask", "django", "requests", "urllib3"} + def test_parse_requirements_txt_file_with_includes(self, tmp_path: Path) -> None: + """Packages from files pulled in with ``-r`` are included, recursively.""" + (tmp_path / "deep.txt").write_text("rich\n") + (tmp_path / "extra.txt").write_text("requests==2.31\ncolorama>=0.4\n-r deep.txt\n") + main = tmp_path / "requirements.txt" + main.write_text("flask==2.0\n-r extra.txt\n") + + parser = RequirementsTxtParser(file_path=str(main)) + assert parser.parse() == {"flask", "requests", "colorama", "rich"} + + def test_parse_requirements_txt_file_skips_missing_include(self, tmp_path: Path) -> None: + """A missing ``-r`` target is skipped without aborting the rest of the parse.""" + main = tmp_path / "requirements.txt" + main.write_text("flask==2.0\n-r does-not-exist.txt\n") + + parser = RequirementsTxtParser(file_path=str(main)) + assert parser.parse() == {"flask"} + class TestLockParser: def test_parse_poetry_lock_file_lt_1_5(self, poetry_lock_file_lt_1_5: Path) -> None: