"""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 HEADER = b"Data,Dependencia,Historico,Documento,Tipo,Valor\n" def test_discovery_returns_only_safe_metadata_for_direct_supported_files(tmp_path): bank = HEADER + b"01/02/2026,,SYNTHETIC BANK ROW,,,12.34\n" full = b"1 - SYNTHETIC OWNER\n" + b"15.02.2026" + b"SYNTHETIC CARD ROW".ljust(37) + b"BR 12,34 0,00\n" nested = tmp_path / "nested" nested.mkdir() (tmp_path / "bank.csv").write_bytes(bank) (tmp_path / "card.txt").write_bytes(full) (nested / "hidden.csv").write_bytes(bank) with override_settings(IMPORT_ROOT=tmp_path): rows = discover_imports() assert [(row.name, row.profile, row.sections, row.code) for row in rows] == [ ("bank.csv", "BANK_CSV_V1", ("1",), None), ("card.txt", "OUROCARD_FULL_V1", ("1",), None), ] assert [row.document_hash for row in rows] == [sha256(bank).hexdigest(), sha256(full).hexdigest()] assert all(not hasattr(row, "source_text") for row in rows) def test_discovery_reports_safe_codes_for_direct_unsupported_or_invalid_files(tmp_path): (tmp_path / "unknown.ofx").write_bytes(b"SYNTHETIC OFX") (tmp_path / "statement.pdf").write_bytes(b"%PDF-SYNTHETIC") (tmp_path / "invalid.txt").write_bytes(b"SYNTHETIC UNRECOGNIZED TEXT") (tmp_path / "oversize.csv").write_bytes(b"x" * (10 * 1024 * 1024 + 1)) with override_settings(IMPORT_ROOT=tmp_path): rows = discover_imports() assert [(row.name, row.profile, row.sections, row.code) for row in rows] == [ ("invalid.txt", None, (), "UNRECOGNIZED_LAYOUT"), ("oversize.csv", None, (), "INPUT_TOO_LARGE"), ("statement.pdf", None, (), "UNSUPPORTED_FILE"), ("unknown.ofx", None, (), "UNSUPPORTED_FILE"), ] assert all(not hasattr(row, "source_text") for row in rows) def test_discovery_ignores_symlinks_and_non_files(tmp_path): bank = HEADER + b"01/02/2026,,SYNTHETIC BANK ROW,,,12.34\n" target = tmp_path / "target.csv" target.write_bytes(bank) (tmp_path / "linked.csv").symlink_to(target) (tmp_path / "directory.csv").mkdir() with override_settings(IMPORT_ROOT=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 (value == candidate.name or 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() bank = HEADER + b"01/02/2026,,SYNTHETIC BANK ROW,,,12.34\n" (root / "bank.csv").write_bytes(bank) original_listdir = os.listdir def delayed_failure(value): if isinstance(value, int): raise OSError("SYNTHETIC LISTING FAILURE") return original_listdir(value) monkeypatch.setattr(guided_imports.os, "listdir", delayed_failure) with override_settings(IMPORT_ROOT=root): assert discover_imports() == () def test_discovery_returns_empty_results_when_opening_the_import_root_fails(tmp_path, monkeypatch): root = tmp_path / "unopenable" root.mkdir() (root / "bank.csv").write_bytes(HEADER + b"01/02/2026,,SYNTHETIC BANK ROW,,,12.34\n") original_open = os.open def root_open_failure(value, flags, *args, **kwargs): if value == os.path.sep: raise OSError("SYNTHETIC ROOT OPEN FAILURE") return original_open(value, flags, *args, **kwargs) monkeypatch.setattr(guided_imports.os, "open", root_open_failure) with override_settings(IMPORT_ROOT=root): assert discover_imports() == () def test_discovery_keeps_using_the_opened_root_after_the_root_path_is_replaced(tmp_path, monkeypatch): root = tmp_path / "imports" outside = tmp_path / "outside" parked_root = tmp_path / "parked-imports" root.mkdir() outside.mkdir() inside = HEADER + b"01/02/2026,,SYNTHETIC INSIDE ROW,,,12.34\n" external = HEADER + b"01/02/2026,,SYNTHETIC EXTERNAL ROW,,,98.76\n" (root / "bank.csv").write_bytes(inside) (outside / "bank.csv").write_bytes(external) original_iterdir = Path.iterdir original_listdir = os.listdir replaced = False def replace_root(): nonlocal replaced if not replaced: replaced = True root.rename(parked_root) root.symlink_to(outside, target_is_directory=True) def replace_then_iterdir(self): if self == root.resolve(): replace_root() return original_iterdir(self) def replace_then_listdir(value): if isinstance(value, int): replace_root() return original_listdir(value) monkeypatch.setattr(Path, "iterdir", replace_then_iterdir) monkeypatch.setattr(guided_imports.os, "listdir", replace_then_listdir) with override_settings(IMPORT_ROOT=root): rows = discover_imports() assert [(row.name, row.document_hash, row.profile, row.code) for row in rows] == [ ("bank.csv", sha256(inside).hexdigest(), "BANK_CSV_V1", None), ] def test_discovery_fails_closed_for_a_symlinked_import_root_parent(tmp_path): actual_parent = tmp_path / "actual-parent" root = actual_parent / "imports" linked_parent = tmp_path / "linked-parent" root.mkdir(parents=True) (root / "bank.csv").write_bytes(HEADER + b"01/02/2026,,SYNTHETIC BANK ROW,,,12.34\n") linked_parent.symlink_to(actual_parent, target_is_directory=True) with override_settings(IMPORT_ROOT=linked_parent / "imports"): assert discover_imports() == ()