gizmoduck/tests/test_export_command.py

import json
import threading
from datetime import datetime, timezone as datetime_timezone
from io import StringIO

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

from finance.exports import ExportArtifact, ExportError, write_export


def synthetic_artifact(filename="gizmoduck-export-20260907T120000Z.zip"):
    return ExportArtifact(
        filename=filename,
        content=b"synthetic zip bytes",
        version=1,
        row_counts={"accounts.csv": 2, "transactions.csv": 4},
    )


@pytest.fixture
def export_root(tmp_path, settings):
    settings.EXPORT_ROOT = tmp_path
    return tmp_path


def test_write_export_generates_a_plain_name_and_publishes_complete_content(
    export_root, monkeypatch
):
    now = datetime(2026, 9, 7, 12, 0, tzinfo=datetime_timezone.utc)
    monkeypatch.setattr("finance.exports.timezone.now", lambda: now)
    monkeypatch.setattr("finance.exports.build_export", synthetic_artifact)

    result = write_export()

    assert result.name == "gizmoduck-export-20260907T120000Z.zip"
    assert result.path == export_root / result.name
    assert result.path.read_bytes() == b"synthetic zip bytes"
    assert list(export_root.iterdir()) == [result.path]


@pytest.mark.parametrize(
    "name",
    [
        "../outside.zip",
        "nested/outside.zip",
        "nested\\outside.zip",
        "/tmp/outside.zip",
        "C:\\outside.zip",
        "export.txt",
        ".zip",
    ],
)
def test_write_export_rejects_non_plain_zip_names(export_root, name):
    with pytest.raises(ExportError, match="invalid_name"):
        write_export(name)

    assert list(export_root.iterdir()) == []


def test_write_export_rejects_existing_target_before_snapshot(export_root, monkeypatch):
    existing = export_root / "existing.zip"
    existing.write_bytes(b"original")

    def unexpected_snapshot():
        raise AssertionError("snapshot must not be built")

    monkeypatch.setattr("finance.exports.build_export", unexpected_snapshot)

    with pytest.raises(ExportError, match="existing_destination"):
        write_export(existing.name)

    assert existing.read_bytes() == b"original"


def test_write_export_does_not_expose_final_name_while_snapshot_is_built(
    export_root, monkeypatch
):
    destination = export_root / "private-until-complete.zip"

    def inspect_during_snapshot():
        assert not destination.exists()
        return synthetic_artifact()

    monkeypatch.setattr("finance.exports.build_export", inspect_during_snapshot)

    write_export(destination.name)

    assert destination.read_bytes() == b"synthetic zip bytes"


def test_write_export_preserves_final_file_created_concurrently(
    export_root, monkeypatch
):
    destination = export_root / "concurrent.zip"
    snapshot_started = threading.Event()
    competing_file_created = threading.Event()
    writer_errors = []

    def concurrent_snapshot():
        snapshot_started.set()
        assert competing_file_created.wait(5), "competing writer did not finish"
        return synthetic_artifact()

    def publish_competing_file():
        try:
            assert snapshot_started.wait(5), "export did not start its snapshot"
            destination.write_bytes(b"concurrent writer")
        except BaseException as exc:
            writer_errors.append(exc)
        finally:
            competing_file_created.set()

    monkeypatch.setattr("finance.exports.build_export", concurrent_snapshot)
    writer = threading.Thread(target=publish_competing_file)
    writer.start()

    try:
        with pytest.raises(ExportError, match="existing_destination"):
            write_export(destination.name)
    finally:
        writer.join(5)

    assert not writer.is_alive()
    assert writer_errors == []
    assert destination.read_bytes() == b"concurrent writer"
    assert list(export_root.iterdir()) == [destination]


def test_write_export_rejects_destination_symlink(export_root, monkeypatch):
    outside = export_root.parent / "outside-export.zip"
    outside.write_bytes(b"outside original")
    link = export_root / "linked.zip"
    try:
        link.symlink_to(outside)
    except OSError:
        pytest.skip("Symlink creation unavailable")
    monkeypatch.setattr("finance.exports.build_export", synthetic_artifact)

    with pytest.raises(ExportError, match="invalid_name"):
        write_export(link.name)

    assert outside.read_bytes() == b"outside original"
    assert link.is_symlink()


def test_write_export_removes_reservation_and_temporary_file_on_snapshot_failure(
    export_root, monkeypatch
):
    def failed_snapshot():
        raise RuntimeError("SYNTHETIC PRIVATE DATABASE DETAIL")

    monkeypatch.setattr("finance.exports.build_export", failed_snapshot)

    with pytest.raises(ExportError, match="export_failed") as caught:
        write_export("failed.zip")

    assert "PRIVATE" not in str(caught.value)
    assert list(export_root.iterdir()) == []


def test_write_export_cleans_up_if_atomic_publication_fails(export_root, monkeypatch):
    monkeypatch.setattr("finance.exports.build_export", synthetic_artifact)

    def failed_link(_source, _destination):
        raise OSError("SYNTHETIC PRIVATE FILESYSTEM DETAIL")

    monkeypatch.setattr("finance.exports.os.link", failed_link)

    with pytest.raises(ExportError, match="publication_failed") as caught:
        write_export("failed.zip")

    assert "PRIVATE" not in str(caught.value)
    assert list(export_root.iterdir()) == []


def test_export_command_prints_only_safe_package_metadata(export_root, monkeypatch):
    monkeypatch.setattr("finance.exports.build_export", synthetic_artifact)
    output = StringIO()

    call_command("export_finances", "--output", "synthetic.zip", stdout=output)

    payload = json.loads(output.getvalue())
    assert payload == {
        "name": "synthetic.zip",
        "version": 1,
        "rows": {"accounts.csv": 2, "transactions.csv": 4},
    }
    assert str(export_root) not in output.getvalue()
    assert "synthetic zip bytes" not in output.getvalue()


def test_export_command_converts_private_failure_to_safe_code(export_root, monkeypatch):
    def failed_snapshot():
        raise RuntimeError("SYNTHETIC PRIVATE DATABASE DETAIL")

    monkeypatch.setattr("finance.exports.build_export", failed_snapshot)

    with pytest.raises(CommandError, match="export_failed") as caught:
        call_command("export_finances", "--output", "failed.zip")

    assert "PRIVATE" not in str(caught.value)
    assert list(export_root.iterdir()) == []