exitdrill/tests/test_directus_demo.py at main · ghzhost/exitdrill · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
import json
import shutil
import subprocess
import sys
from pathlib import Path


def test_directus_api_capture_canary_acceptance() -> None:
    project = Path(__file__).parents[1]
    completed = subprocess.run(  # noqa: S603 - fixed interpreter and repository script
        [sys.executable, str(project / "scripts" / "check_directus_canary_demo.py")],
        cwd=project,
        check=False,
        capture_output=True,
        text=True,
    )

    assert completed.returncode == 0, completed.stderr
    assert completed.stderr == ""
    summary = json.loads(completed.stdout)
    assert summary == {
        "clean_overall_status": "structurally_restorable",
        "lossy_observed_remediation_signals": 6,
        "lossy_overall_status": "not_structurally_restorable",
        "row_and_file_counts_preserved": True,
        "source_profile": "directus-11.17.4-civic-case/v0.1",
        "status": "directus_api_capture_canary_verified",
    }


def test_lossy_builder_rejects_overlapping_output_paths(tmp_path: Path) -> None:
    project = Path(__file__).parents[1]
    committed = project / "examples" / "directus-11.17.4-civic-case" / "native"
    source = tmp_path / "native"
    shutil.copytree(committed, source)
    script = project / "scripts" / "build_directus_lossy_canary.py"

    cases = (
        (source / "nested-destination", tmp_path / "statement.json"),
        (tmp_path / "lossy-native", tmp_path / "lossy-native" / "statement.json"),
    )
    for destination, statement in cases:
        completed = subprocess.run(  # noqa: S603 - fixed interpreter and repository script
            [
                sys.executable,
                str(script),
                str(source),
                str(destination),
                "--statement",
                str(statement),
            ],
            cwd=project,
            check=False,
            capture_output=True,
            text=True,
        )
        assert completed.returncode == 2
        assert completed.stdout == ""
        assert not destination.exists()
        assert not statement.exists()

    assert not list(source.glob(".directus-lossy-canary.*"))
    assert set(path.name for path in source.iterdir()) == {
        "activity.json",
        "assets",
        "capture-manifest.json",
        "case-people.json",
        "cases.json",
        "files.json",
        "people.json",
        "permissions.json",
        "policies.json",
        "schema.json",
    }


def test_lossy_builder_rejects_unverified_source_without_outputs(tmp_path: Path) -> None:
    project = Path(__file__).parents[1]
    committed = project / "examples" / "directus-11.17.4-civic-case" / "native"
    source = tmp_path / "native"
    shutil.copytree(committed, source)
    (source / "people.json").write_bytes(b' {"data":[]}')
    destination = tmp_path / "lossy-native"
    statement = tmp_path / "statement.json"

    completed = subprocess.run(  # noqa: S603 - fixed interpreter and repository script
        [
            sys.executable,
            str(project / "scripts" / "build_directus_lossy_canary.py"),
            str(source),
            str(destination),
            "--statement",
            str(statement),
        ],
        cwd=project,
        check=False,
        capture_output=True,
        text=True,
    )

    assert completed.returncode == 2
    assert completed.stdout == ""
    assert str(source) not in completed.stderr
    assert str(destination) not in completed.stderr
    assert not destination.exists()
    assert not statement.exists()


def test_lossy_builder_is_pinned_to_the_committed_clean_manifest(tmp_path: Path) -> None:
    project = Path(__file__).parents[1]
    committed = project / "examples" / "directus-11.17.4-civic-case" / "native"
    source = tmp_path / "native"
    shutil.copytree(committed, source)
    manifest_path = source / "capture-manifest.json"
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    manifest["exported_at"] = "2026-08-02T02:38:29.000Z"
    manifest_path.write_text(
        json.dumps(manifest, allow_nan=False, sort_keys=True, separators=(",", ":")) + "\n",
        encoding="utf-8",
    )
    destination = tmp_path / "lossy-native"
    statement = tmp_path / "statement.json"

    completed = subprocess.run(  # noqa: S603 - fixed interpreter and repository script
        [
            sys.executable,
            str(project / "scripts" / "build_directus_lossy_canary.py"),
            str(source),
            str(destination),
            "--statement",
            str(statement),
        ],
        cwd=project,
        check=False,
        capture_output=True,
        text=True,
    )

    assert completed.returncode == 2
    assert "not the committed clean Directus canary" in completed.stderr
    assert str(source) not in completed.stderr
    assert not destination.exists()
    assert not statement.exists()