tegwick
3e2726f5f1
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a15374-f59f-7a31-86d2-d23b5d2ed181
92 lines
3.4 KiB
Python
92 lines
3.4 KiB
Python
import hashlib
|
|
import json
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
from repo_manager.fleet_snapshot import export_snapshot
|
|
|
|
|
|
def git(repo, *args):
|
|
return subprocess.check_output(["git", "-C", str(repo), *args]).decode().strip()
|
|
|
|
|
|
@pytest.fixture
|
|
def source(tmp_path):
|
|
repo = tmp_path / "source"
|
|
repo.mkdir()
|
|
git(repo, "init", "-q")
|
|
git(repo, "config", "user.name", "Fixture")
|
|
git(repo, "config", "user.email", "fixture@example.invalid")
|
|
for path, content in {"workplans/active.md": "committed work", "decisions/one.md": "accepted",
|
|
"private.txt": "excluded", "workplans/archived/old.md": "done"}.items():
|
|
file = repo / path
|
|
file.parent.mkdir(parents=True, exist_ok=True)
|
|
file.write_text(content)
|
|
git(repo, "add", ".")
|
|
git(repo, "commit", "-qm", "fixture")
|
|
return repo
|
|
|
|
|
|
def request(source):
|
|
return {"observed_at": "2026-10-19T10:00:00Z", "repositories": [
|
|
{"slug": "example", "path": str(source), "commit": git(source, "rev-parse", "HEAD")}]}
|
|
|
|
|
|
def test_exact_committed_bytes_allowlist_and_manifest(source, tmp_path):
|
|
req = request(source)
|
|
(source / "workplans/active.md").write_text("dirty excluded")
|
|
out = tmp_path / "snapshot"
|
|
manifest = export_snapshot(req, out)
|
|
assert (out / "example/workplans/active.md").read_text() == "committed work"
|
|
assert not (out / "example/private.txt").exists()
|
|
assert not (out / "example/.git").exists()
|
|
assert len(manifest["repositories"][0]["files"]) == 3
|
|
assert manifest["freshness_verified"] is False
|
|
assert (out / "COMPLETE").read_text().strip() == hashlib.sha256((out / "manifest.json").read_bytes()).hexdigest()
|
|
assert json.loads((out / "manifest.json").read_text()) == manifest
|
|
|
|
|
|
def test_no_overwrite(source, tmp_path):
|
|
out = tmp_path / "snapshot"
|
|
out.mkdir()
|
|
with pytest.raises(ValueError, match="exists"):
|
|
export_snapshot(request(source), out)
|
|
assert not list(out.iterdir())
|
|
|
|
|
|
@pytest.mark.parametrize("change", ["slug", "commit", "timestamp", "duplicate"])
|
|
def test_invalid_request_is_not_published(source, tmp_path, change):
|
|
req = request(source)
|
|
if change == "slug":
|
|
req["repositories"][0]["slug"] = "../outside"
|
|
elif change == "commit":
|
|
req["repositories"][0]["commit"] = "HEAD"
|
|
elif change == "timestamp":
|
|
req["observed_at"] = "2026-10-19T10:00:00"
|
|
else:
|
|
req["repositories"].append(dict(req["repositories"][0]))
|
|
with pytest.raises(ValueError):
|
|
export_snapshot(req, tmp_path / "snapshot")
|
|
assert not (tmp_path / "snapshot").exists()
|
|
|
|
|
|
def test_record_symlink_refused(source, tmp_path):
|
|
(source / "workplans/link.md").symlink_to("../private.txt")
|
|
git(source, "add", ".")
|
|
git(source, "commit", "-qm", "symlink")
|
|
with pytest.raises(ValueError, match="unsafe"):
|
|
export_snapshot(request(source), tmp_path / "snapshot")
|
|
assert not (tmp_path / "snapshot").exists()
|
|
|
|
|
|
def test_symlink_parent_refused(source, tmp_path):
|
|
(tmp_path / "alias").symlink_to(source, target_is_directory=True)
|
|
with pytest.raises(ValueError, match="symlinks"):
|
|
export_snapshot(request(source), tmp_path / "alias/snapshot")
|
|
|
|
|
|
def test_size_limit_refused(source, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("repo_manager.fleet_snapshot.MAX_FILE_BYTES", 1)
|
|
with pytest.raises(ValueError, match="size"):
|
|
export_snapshot(request(source), tmp_path / "snapshot")
|