diff --git a/ingestion/guided_imports.py b/ingestion/guided_imports.py index e1a29be..4fc3db5 100644 --- a/ingestion/guided_imports.py +++ b/ingestion/guided_imports.py @@ -1,5 +1,7 @@ """Safe discovery of local source documents for the guided import workflow.""" +import os +import stat from dataclasses import dataclass from hashlib import sha256 from pathlib import Path @@ -36,12 +38,27 @@ def _profile_and_sections(path: Path, data: bytes) -> tuple[str | None, tuple[st return None, () -def _read_discovered_file(path: Path) -> DiscoveredFile: +def _unreadable(path: Path) -> DiscoveredFile: + return DiscoveredFile(path.name, "", None, (), "UNREADABLE_FILE") + + +def _read_discovered_file(path: Path, expected: os.stat_result) -> DiscoveredFile: + descriptor = None try: - with path.open("rb") as stream: + flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) + if no_follow := getattr(os, "O_NOFOLLOW", 0): + flags |= no_follow + descriptor = os.open(path, flags) + opened = os.fstat(descriptor) + if not stat.S_ISREG(opened.st_mode) or not os.path.samestat(expected, opened): + return _unreadable(path) + with os.fdopen(descriptor, "rb", closefd=False) as stream: data = stream.read(MAX_BYTES + 1) except OSError: - return DiscoveredFile(path.name, "", None, (), "UNREADABLE_FILE") + return _unreadable(path) + finally: + if descriptor is not None: + os.close(descriptor) if len(data) > MAX_BYTES: return DiscoveredFile(path.name, "", None, (), "INPUT_TOO_LARGE") document_hash = sha256(data).hexdigest() @@ -55,17 +72,19 @@ def _read_discovered_file(path: Path) -> DiscoveredFile: def discover_imports() -> tuple[DiscoveredFile, ...]: """Return safe metadata for direct regular files under ``IMPORT_ROOT``.""" - root = Path(settings.IMPORT_ROOT).resolve() try: - entries = root.iterdir() - except OSError: + root = Path(settings.IMPORT_ROOT).resolve() + entries = tuple(root.iterdir()) + except (OSError, RuntimeError): return () discovered = [] for path in entries: try: - if path.is_symlink() or not path.is_file(): + expected = path.lstat() + if not stat.S_ISREG(expected.st_mode): continue except OSError: + discovered.append(_unreadable(path)) continue - discovered.append(_read_discovered_file(path)) + discovered.append(_read_discovered_file(path, expected)) return tuple(sorted(discovered, key=lambda row: row.name)) diff --git a/tests/test_guided_import_discovery.py b/tests/test_guided_import_discovery.py index 64feb11..d382c33 100644 --- a/tests/test_guided_import_discovery.py +++ b/tests/test_guided_import_discovery.py @@ -1,9 +1,12 @@ """Synthetic checks for source-safe local import discovery.""" +import os from hashlib import sha256 +from pathlib import Path from django.test import override_settings +from ingestion import guided_imports from ingestion.guided_imports import discover_imports @@ -59,3 +62,66 @@ def test_discovery_ignores_symlinks_and_non_files(tmp_path): rows = discover_imports() assert [row.name for row in rows] == ["target.csv"] + + +def test_discovery_marks_a_post_listing_symlink_replacement_unreadable(tmp_path, monkeypatch): + bank = HEADER + b"01/02/2026,,SYNTHETIC BANK ROW,,,12.34\n" + candidate = tmp_path / "candidate.csv" + replacement = tmp_path / "replacement.csv" + candidate.write_bytes(bank) + replacement.write_bytes(bank) + original_open = os.open + replaced = False + + def replace_then_open(value, flags, *args, **kwargs): + nonlocal replaced + if not replaced and Path(value) == candidate: + replaced = True + candidate.unlink() + candidate.symlink_to(replacement) + return original_open(value, flags, *args, **kwargs) + + monkeypatch.setattr(guided_imports, "os", os, raising=False) + monkeypatch.setattr(guided_imports.os, "open", replace_then_open) + + with override_settings(IMPORT_ROOT=tmp_path): + rows = discover_imports() + + assert [(row.name, row.profile, row.sections, row.code) for row in rows] == [ + ("candidate.csv", None, (), "UNREADABLE_FILE"), + ("replacement.csv", "BANK_CSV_V1", ("1",), None), + ] + + +def test_discovery_returns_empty_results_when_listing_the_import_root_fails(tmp_path, monkeypatch): + root = tmp_path / "unavailable" + root.mkdir() + original_iterdir = Path.iterdir + + def delayed_failure(self): + if self == root.resolve(): + def entries(): + raise OSError("SYNTHETIC LISTING FAILURE") + yield + return entries() + return original_iterdir(self) + + monkeypatch.setattr(Path, "iterdir", delayed_failure) + + with override_settings(IMPORT_ROOT=root): + assert discover_imports() == () + + +def test_discovery_returns_empty_results_when_resolving_the_import_root_fails(tmp_path, monkeypatch): + root = tmp_path / "unresolvable" + original_resolve = Path.resolve + + def resolution_failure(self, *args, **kwargs): + if self == root: + raise RuntimeError("SYNTHETIC RESOLUTION FAILURE") + return original_resolve(self, *args, **kwargs) + + monkeypatch.setattr(Path, "resolve", resolution_failure) + + with override_settings(IMPORT_ROOT=root): + assert discover_imports() == ()