gizmoduck/tests/test_import_command.py

"""Synthetic command-boundary tests; lifecycle persistence has separate coverage."""
import copy
import importlib
import json
from io import StringIO
from types import SimpleNamespace
from uuid import UUID

import pytest
from django.core.management import call_command
from django.core.management.base import CommandError


BATCH = "11111111-1111-4111-8111-111111111111"
ACTOR = ["--actor-kind", "HUMAN", "--actor-identifier", "synthetic-owner"]


@pytest.fixture
def command(tmp_path, settings, monkeypatch):
    settings.IMPORT_ROOT = tmp_path
    module = importlib.import_module("ingestion.management.commands.import_review")
    review = {
        "batch_id": BATCH, "revision": 1, "state": "PREVIEW",
        "profile": "BANK_CSV_V1", "encoding": "utf-8",
        "summary": {"items": 1},
        "items": [{"locator": "line:2", "candidate": {"description": "SYNTHETIC PRIVATE LABEL"},
                   "issues": [], "duplicates": [], "decision": None}],
        "decisions": [{"locator": "line:2", "action": "CREATE"}],
    }
    batch = SimpleNamespace(pk=UUID(BATCH), state="PREVIEW", revision=1)
    monkeypatch.setattr(module.services, "export_review", lambda *a, **kw: copy.deepcopy(review))
    monkeypatch.setattr(module.services, "apply_decisions", lambda *a, **kw: batch)
    return module, review


def run(command, operation, *args):
    output = StringIO()
    call_command(command[0].Command(), operation, *args, *ACTOR, stdout=output)
    return output.getvalue()


def test_export_keeps_candidate_text_in_local_file_only(command, tmp_path):
    output = run(command, "export", "--batch", BATCH, "--output", "review.json")
    assert "SYNTHETIC PRIVATE LABEL" not in output
    assert json.loads((tmp_path / "review.json").read_text())["items"][0]["candidate"]["description"] == "SYNTHETIC PRIVATE LABEL"
    assert json.loads(output)["batch_id"] == BATCH


@pytest.mark.parametrize("path", ["../outside.json", "nested/../../outside.json"])
def test_output_rejects_traversal(command, path):
    with pytest.raises(CommandError, match="INVALID_PATH"):
        run(command, "export", "--batch", BATCH, "--output", path)


def test_output_never_overwrites_input_or_existing_review(command, tmp_path):
    destination = tmp_path / "source.json"
    destination.write_text("synthetic original")
    with pytest.raises(CommandError):
        run(command, "export", "--batch", BATCH, "--output", "source.json")
    assert destination.read_text() == "synthetic original"


def test_output_rejects_symlink_escape(command, tmp_path):
    link = tmp_path / "escape"
    try:
        link.symlink_to(tmp_path.parent, target_is_directory=True)
    except OSError:
        pytest.skip("Symlink creation unavailable")
    with pytest.raises(CommandError, match="INVALID_PATH"):
        run(command, "export", "--batch", BATCH, "--output", "escape/review.json")


@pytest.mark.parametrize("payload", [
    '{"batch_id":"' + BATCH + '","revision":1,"revision":2,"decisions":[]}',
    '{"batch_id":"' + BATCH + '","revision":1,"decisions":[{"locator":"2","action":"CREATE","action":"SKIP"}]}',
    '{"batch_id":"' + BATCH + '","revision":NaN,"decisions":[]}',
])
def test_apply_rejects_ambiguous_json(command, tmp_path, payload):
    (tmp_path / "decisions.json").write_text(payload)
    with pytest.raises(CommandError, match="INVALID_JSON"):
        run(command, "apply", "--batch", BATCH, "--input", "decisions.json")


@pytest.mark.parametrize("change", ["candidate", "metadata", "unknown", "revision", "batch"])
def test_apply_rejects_modified_evidence_or_wrong_envelope(command, tmp_path, change):
    payload = copy.deepcopy(command[1])
    if change == "candidate":
        payload["items"][0]["candidate"]["description"] = "tampered"
    elif change == "metadata":
        payload["encoding"] = "latin-1"
    elif change == "unknown":
        payload["extra"] = True
    elif change == "revision":
        payload["revision"] = True
    else:
        payload["batch_id"] = "22222222-2222-4222-8222-222222222222"
    (tmp_path / "decisions.json").write_text(json.dumps(payload))
    with pytest.raises(CommandError):
        run(command, "apply", "--batch", BATCH, "--input", "decisions.json")


@pytest.mark.parametrize("full", [True, False])
def test_apply_accepts_decisions_only_or_unchanged_full_export(command, tmp_path, full):
    payload = copy.deepcopy(command[1])
    payload["decisions"] = [{"locator": "line:2", "action": "SKIP", "reason": "USER_EXCLUDED"}]
    if not full:
        payload = {key: payload[key] for key in ("batch_id", "revision", "decisions")}
    (tmp_path / "decisions.json").write_text(json.dumps(payload))
    output = run(command, "apply", "--batch", BATCH, "--input", "decisions.json")
    assert "SYNTHETIC PRIVATE LABEL" not in output
    assert json.loads(output)["state"] == "PREVIEW"


def test_file_read_is_bounded(command, tmp_path):
    with (tmp_path / "large.json").open("wb") as stream:
        stream.truncate(10 * 1024 * 1024 + 1)
    with pytest.raises(CommandError, match="INPUT_TOO_LARGE"):
        run(command, "apply", "--batch", BATCH, "--input", "large.json")


def test_service_exception_does_not_disclose_source(command, monkeypatch):
    def fail(*args, **kwargs):
        raise RuntimeError("SYNTHETIC PRIVATE LABEL")
    monkeypatch.setattr(command[0].services, "export_review", fail)
    with pytest.raises(CommandError) as caught:
        run(command, "export", "--batch", BATCH, "--output", "review.json")
    assert "SYNTHETIC" not in str(caught.value)


def test_all_operations_require_explicit_actor(command):
    with pytest.raises(CommandError):
        call_command(command[0].Command(), "export", "--batch", BATCH, "--output", "review.json")