gizmoduck/tests/test_guided_import_discovery.py

"""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() == ()