File size: 5,469 Bytes
d2d1903
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
"""
Tests for Stage 68 β€” bundle export with manifest.

bundle_pilot_output zips a pilot output dir + writes a manifest.json
(sha256 per file). verify_bundle re-hashes and compares. The headline
test is a round-trip: bundle β†’ unzip β†’ verify checksums match.
"""
import hashlib
import json
import zipfile

import pytest

from delivery.bundle import bundle_pilot_output, verify_bundle


def _populate(tmp_path):
    """Build a tiny pilot-shaped dir with predictable content."""
    src = tmp_path / "pilot_out"
    src.mkdir()
    (src / "customer_report.html").write_text("<h1>Report</h1>",
                                                encoding="utf-8")
    (src / "customer_report.pdf").write_bytes(b"%PDF-stub\nfake")
    (src / "decision_queue.html").write_text("<h1>Queue</h1>",
                                                encoding="utf-8")
    (src / "evidence").mkdir()
    (src / "evidence" / "run_xyz.html").write_text("<h1>Evidence</h1>",
                                                      encoding="utf-8")
    return src


def test_bundle_creates_zip_next_to_source(tmp_path):
    src = _populate(tmp_path)
    out = bundle_pilot_output(src)
    assert out.exists()
    assert out.suffix == ".zip"
    assert out.parent == src.parent
    assert out.stem == src.name


def test_bundle_explicit_path(tmp_path):
    src = _populate(tmp_path)
    target = tmp_path / "custom_name.zip"
    out = bundle_pilot_output(src, bundle_path=target)
    assert out == target.resolve()
    assert out.exists()


def test_bundle_includes_every_source_file(tmp_path):
    src = _populate(tmp_path)
    out = bundle_pilot_output(src)
    with zipfile.ZipFile(out) as zf:
        names = set(zf.namelist())
    # manifest + the four files we wrote, with relative paths
    assert "manifest.json" in names
    assert "customer_report.html" in names
    assert "customer_report.pdf" in names
    assert "decision_queue.html" in names
    assert "evidence/run_xyz.html" in names


def test_bundle_manifest_carries_sha256_per_file(tmp_path):
    src = _populate(tmp_path)
    out = bundle_pilot_output(src)
    with zipfile.ZipFile(out) as zf:
        manifest = json.loads(zf.read("manifest.json"))
    assert manifest["source_dir_name"] == src.name
    assert manifest["n_files"] == 4
    assert manifest["total_bytes"] > 0
    # spot-check one file's hash
    by_path = {f["path"]: f for f in manifest["files"]}
    expected = hashlib.sha256(b"<h1>Report</h1>").hexdigest()
    assert by_path["customer_report.html"]["sha256"] == expected
    assert by_path["customer_report.html"]["bytes"] == len(b"<h1>Report</h1>")


def test_bundle_manifest_files_sorted(tmp_path):
    """Deterministic manifest β€” same input gives same file order, so
    diffing two bundles is meaningful."""
    src = _populate(tmp_path)
    out = bundle_pilot_output(src)
    with zipfile.ZipFile(out) as zf:
        manifest = json.loads(zf.read("manifest.json"))
    paths = [f["path"] for f in manifest["files"]]
    assert paths == sorted(paths)


def test_bundle_missing_source_dir_raises(tmp_path):
    with pytest.raises(FileNotFoundError):
        bundle_pilot_output(tmp_path / "nope")


# --- verify_bundle ----------------------------------------------------

def test_verify_bundle_returns_ok_on_intact_zip(tmp_path):
    src = _populate(tmp_path)
    out = bundle_pilot_output(src)
    report = verify_bundle(out)
    assert report["ok"] is True
    assert report["n_files"] == 4
    assert report["n_mismatches"] == 0
    assert report["mismatches"] == []


def test_verify_bundle_detects_tampered_file(tmp_path):
    """The whole point of the manifest β€” if a file in the bundle was
    modified, verify must spot it."""
    src = _populate(tmp_path)
    out = bundle_pilot_output(src)
    # rewrite the zip, replacing one of the files with different bytes
    tampered = tmp_path / "tampered.zip"
    with zipfile.ZipFile(out, "r") as zin, \
            zipfile.ZipFile(tampered, "w", zipfile.ZIP_DEFLATED) as zout:
        for name in zin.namelist():
            data = zin.read(name)
            if name == "customer_report.html":
                data = b"<h1>MALICIOUS</h1>"
            zout.writestr(name, data)
    report = verify_bundle(tampered)
    assert report["ok"] is False
    assert report["n_mismatches"] == 1
    assert (report["mismatches"][0]["path"]
             == "customer_report.html")


# --- end-to-end through the pilot CLI -------------------------------

def test_pilot_cli_bundle_flag_creates_zip(tmp_path):
    """The pilot's --bundle flag end-to-end: real pilot run + bundle
    write. Doesn't exhaustively check the bundle (the helper tests
    do that) β€” just confirms the CLI wiring works."""
    from delivery.pilot import _cli
    out_dir = tmp_path / "pilot_out"
    rc = _cli([
        "--seed", "7",
        "--out", str(out_dir),
        "--bundle",
    ])
    assert rc == 0
    zip_path = out_dir.with_suffix(".zip")
    assert zip_path.exists()
    # the zip is verifiable
    assert verify_bundle(zip_path)["ok"]


def test_pilot_cli_no_bundle_flag_skips_zip(tmp_path):
    """Backward compat β€” pre-Stage-68 cron lines that don't know
    about --bundle still work as before."""
    from delivery.pilot import _cli
    out_dir = tmp_path / "pilot_out"
    rc = _cli([
        "--seed", "7",
        "--out", str(out_dir),
    ])
    assert rc == 0
    zip_path = out_dir.with_suffix(".zip")
    assert not zip_path.exists()