import hashlib import os import zipfile import pytest from viano_bug_analysis.package import ( PackageLimits, PackageReader, PackageValidator, PackageWriteError, PackageWriter, ) def assert_rejected(writer, source, target, code=None, **kwargs): before = source.read_bytes() target_before = target.read_bytes() if target.exists() else None with pytest.raises(PackageWriteError) as error: writer.write(source, target, **kwargs) if code: assert error.value.result.issues[0].code == code assert source.read_bytes() == before assert (target.read_bytes() if target.exists() else None) == target_before assert not list(target.parent.glob(".vbcase-*.tmp")) def test_full_roundtrip(manifest, make_package, tmp_path): manifest["extension"] = {"unknown": [1, 2]} manifest["routing"]["custom"] = "preserve" manifest["artifacts"]["vendor"] = "unknown/binary" originals = {"original/report.json": b'{"report":"generic"}', "unknown/binary": bytes(range(256)), "empty/": b"", "analysis/revisions/0001/data.json": b"old revision", "analysis/current.json": b'{"revision":1}'} source = make_package(entries=originals) before = source.read_bytes() target = tmp_path / "renamed.anything" result = PackageWriter().write( source, target, additions={"analysis/revisions/0002/data.json": b"new revision"}, index_updates={"analysis/current.json": b'{"revision":2}'}, manifest_updates={"state": "VALIDATED", "routing": {"destination": "cases"}, "analysis_revision": 2, "history": {"analysis_revisions": 2}, "artifacts": {"case": "analysis/revisions/0002/data.json"}}, ) assert result.valid and PackageValidator().validate(target).valid assert source.read_bytes() == before with PackageReader(target) as reader: updated = reader.manifest for field in ("package_id", "case_id", "created_at", "project", "extension"): assert updated[field] == manifest[field] assert updated["package_generation"] == 1 assert updated["analysis_revision"] == 2 assert updated["routing"]["custom"] == "preserve" assert updated["artifacts"]["vendor"] == "unknown/binary" assert updated["updated_at"] >= manifest["updated_at"] for name, data in originals.items(): if name != "analysis/current.json" and not name.endswith("/"): assert reader.read_bytes(name) == data assert updated["integrity"][name]["sha256"] == hashlib.sha256(data).hexdigest() with zipfile.ZipFile(target) as archive: assert "empty/" in archive.namelist() next_target = tmp_path / "generation2" PackageWriter().write(target, next_target) with PackageReader(next_target) as reader: assert reader.manifest["package_generation"] == 2 assert reader.manifest["analysis_revision"] == 2 @pytest.mark.parametrize("name", ["original/report", "unknown/file", "analysis/revisions/0001/data"]) def test_immutable_existing_files(name, make_package, tmp_path): source = make_package(entries={name: b"old"}) assert_rejected(PackageWriter(), source, tmp_path / "out", "ARTIFACT_EXISTS", additions={name: b"new"}) def test_frozen_revision_namespace(make_package, tmp_path): source = make_package(entries={"analysis/revisions/0001/data": b"old"}) assert_rejected(PackageWriter(), source, tmp_path / "out", "REVISION_IMMUTABLE", additions={"analysis/revisions/0001/new": b"new"}) @pytest.mark.parametrize("name", ["original/report", "unknown/file", "analysis/revisions/0001/data", "manifest.json"]) def test_index_allowlist(name, make_package, tmp_path): source = make_package(entries={name: b"old"} if name != "manifest.json" else {}) assert_rejected(PackageWriter(), source, tmp_path / "out", index_updates={name: b"new"}) @pytest.mark.parametrize("name", ["../escape", "/abs", "x\\y", "manifest.json", "a:stream"]) def test_unsafe_additions(name, make_package, tmp_path): assert_rejected(PackageWriter(), make_package(), tmp_path / "out", "UNSAFE_PATH", additions={name: b"new"}) @pytest.mark.parametrize("entries,addition", [({"a": b"x"}, "a/b"), ({"a/b": b"x"}, "a"), ({"A": b"x"}, "a"), ({"dir/": b""}, "dir")]) def test_colliding_paths(entries, addition, make_package, tmp_path): assert_rejected(PackageWriter(), make_package(entries=entries), tmp_path / "out", "ENTRY_COLLISION", additions={addition: b"new"}) @pytest.mark.parametrize("field,value", [("package_id", "new"), ("case_id", "new"), ("package_generation", 99), ("created_at", "2026-01-01T00:00:00Z"), ("integrity", {}), ("project", {}), ("format_version", 2)]) def test_reserved_manifest_fields(field, value, make_package, tmp_path): assert_rejected(PackageWriter(), make_package(), tmp_path / "out", "MANIFEST_UPDATE_FORBIDDEN", manifest_updates={field: value}) @pytest.mark.parametrize("updates", [{"state": "WRONG"}, {"artifacts": {"case": "absent"}}, {"routing": {"destination": "wrong"}}, {"analysis_revision": -1}]) def test_invalid_manifest_update(updates, make_package, tmp_path): assert_rejected(PackageWriter(), make_package(), tmp_path / "out", manifest_updates=updates) def test_revision_and_history_cannot_decrease(manifest, make_package, tmp_path): manifest["analysis_revision"] = 2 manifest["history"]["evidence"] = 3 source = make_package() assert_rejected(PackageWriter(), source, tmp_path / "out", "REVISION_REGRESSION", manifest_updates={"analysis_revision": 1}) assert_rejected(PackageWriter(), source, tmp_path / "out", "HISTORY_REGRESSION", manifest_updates={"history": {"evidence": 2}}) def test_invalid_input(make_package, tmp_path): assert_rejected(PackageWriter(), make_package(raw=b"{}"), tmp_path / "out") def test_input_alias_and_hardlink(make_package, tmp_path): source = make_package() assert_rejected(PackageWriter(), source, source, "INPUT_TARGET_ALIAS", overwrite=True) alias = tmp_path / "alias" os.link(source, alias) assert_rejected(PackageWriter(), source, alias, "INPUT_TARGET_ALIAS", overwrite=True) def test_explicit_target_replacement(make_package, tmp_path): source, target = make_package(), tmp_path / "out" target.write_bytes(b"old target") assert_rejected(PackageWriter(), source, target, "TARGET_EXISTS") assert PackageWriter().write(source, target, overwrite=True).valid def test_failure_preserves_existing_target(make_package, tmp_path): source, target = make_package(), tmp_path / "out" target.write_bytes(b"old target") assert_rejected(PackageWriter(), source, target, overwrite=True, manifest_updates={"artifacts": {"case": "absent"}}) @pytest.mark.parametrize("field,value", [("max_file_bytes", 1024), ("max_total_bytes", 2000), ("max_entries", 1), ("max_archive_bytes", 2000), ("max_manifest_bytes", 800)]) def test_output_limits(field, value, make_package, tmp_path): limits = PackageLimits(**{field: value}) assert_rejected(PackageWriter(limits), make_package(), tmp_path / "out", additions={"new": b"x" * 3000}) def test_compressible_artifact_is_valid_with_default_policy(make_package, tmp_path): source, target = make_package(), tmp_path / "out" assert PackageWriter().write(source, target, additions={"new": b"x" * 100_000}).valid def test_integrity_extensions_retained(manifest, make_package, tmp_path): manifest["integrity"]["analysis/current.json"] = {"size": 3, "sha256": hashlib.sha256(b"old").hexdigest(), "vendor": "preserve"} source = make_package(entries={"analysis/current.json": b"old"}) result = PackageWriter().write(source, tmp_path / "out", index_updates={"analysis/current.json": b"new"}) assert result.manifest["integrity"]["analysis/current.json"]["vendor"] == "preserve" def test_atomic_replace_failure(make_package, tmp_path, monkeypatch): source, target = make_package(), tmp_path / "out" target.write_bytes(b"old") def fail(*args): raise OSError("simulated rename failure") monkeypatch.setattr(os, "replace", fail) assert_rejected(PackageWriter(), source, target, "WRITE_ERROR", overwrite=True) def test_competing_target_not_clobbered(make_package, tmp_path, monkeypatch): source, target = make_package(), tmp_path / "out" original_link = os.link def race(src, dst): target.write_bytes(b"competitor") original_link(src, dst) monkeypatch.setattr(os, "link", race) with pytest.raises(PackageWriteError): PackageWriter().write(source, target) assert target.read_bytes() == b"competitor" assert not list(tmp_path.glob(".vbcase-*.tmp")) def test_preservation_check_detects_valid_tampered_output(make_package, tmp_path, monkeypatch): source = make_package(entries={"unknown": b"original"}) writer = PackageWriter() original_validate = writer.validator.validate def tamper(path): with zipfile.ZipFile(path) as archive: entries = {name: archive.read(name) for name in archive.namelist()} import json doc = json.loads(entries["manifest.json"]) entries["unknown"] = b"tampered" doc["integrity"]["unknown"].update(size=8, sha256=hashlib.sha256(b"tampered").hexdigest()) entries["manifest.json"] = json.dumps(doc).encode() with zipfile.ZipFile(path, "w") as archive: for name, data in entries.items(): archive.writestr(name, data) return original_validate(path) monkeypatch.setattr(writer.validator, "validate", tamper) assert_rejected(writer, source, tmp_path / "out", "MANIFEST_CHANGED") def test_preservation_check_detects_unlisted_file_drop(make_package, tmp_path, monkeypatch): source = make_package(entries={"unknown": b"original"}) writer = PackageWriter() original_verify = writer._verify_preservation def drop(reader, path, indexes): with zipfile.ZipFile(path) as archive: entries = {name: archive.read(name) for name in archive.namelist() if name != "unknown"} with zipfile.ZipFile(path, "w") as archive: for name, data in entries.items(): archive.writestr(name, data) original_verify(reader, path, indexes) monkeypatch.setattr(writer, "_verify_preservation", drop) assert_rejected(writer, source, tmp_path / "out", "PRESERVATION_FAILED") def test_frozen_empty_revision_directory(make_package, tmp_path): source = make_package(entries={"analysis/revisions/0001/": b""}) assert_rejected(PackageWriter(), source, tmp_path / "out", "REVISION_IMMUTABLE", additions={"analysis/revisions/0001/new": b"new"}) def test_missing_target_parent(make_package, tmp_path): source = make_package() with pytest.raises(PackageWriteError): PackageWriter().write(source, tmp_path / "absent" / "out") def test_new_index_and_type_validation(make_package, tmp_path): source = make_package() assert PackageWriter().write(source, tmp_path / "out", index_updates={"analysis/current.json": b"{}"}).valid assert_rejected(PackageWriter(), source, tmp_path / "other", "ARTIFACT_TYPE", additions={"new": "string"}) def test_future_timestamp_does_not_regress(manifest, make_package, tmp_path): manifest["updated_at"] = "2099-01-01T00:00:00Z" result = PackageWriter().write(make_package(), tmp_path / "out") from datetime import datetime assert datetime.fromisoformat(result.manifest["updated_at"]) == datetime.fromisoformat("2099-01-01T00:00:00+00:00") def test_fsync_failure(make_package, tmp_path, monkeypatch): source, target = make_package(), tmp_path / "out" target.write_bytes(b"old") def fail(*args): raise OSError("simulated disk failure") monkeypatch.setattr(os, "fsync", fail) assert_rejected(PackageWriter(), source, target, "WRITE_ERROR", overwrite=True) def test_byte_comparison_detects_changed_file(make_package, tmp_path, monkeypatch): source = make_package(entries={"unknown": b"original"}) writer = PackageWriter() original_verify = writer._verify_preservation def change(reader, path, indexes): with zipfile.ZipFile(path) as archive: entries = {name: archive.read(name) for name in archive.namelist()} entries["unknown"] = b"changed" with zipfile.ZipFile(path, "w") as archive: for name, data in entries.items(): archive.writestr(name, data) original_verify(reader, path, indexes) monkeypatch.setattr(writer, "_verify_preservation", change) assert_rejected(writer, source, tmp_path / "out", "PRESERVATION_FAILED") def test_zip_comments_preserved(make_package, tmp_path): source = make_package(entries={"unknown": b"data"}) with zipfile.ZipFile(source, "a") as archive: archive.comment = b"vendor archive comment" PackageWriter().write(source, tmp_path / "out") with zipfile.ZipFile(tmp_path / "out") as archive: assert archive.comment == b"vendor archive comment" def test_cleanup_failure_after_publication_reports_success(make_package, tmp_path, monkeypatch): from pathlib import Path source = make_package() original_unlink = Path.unlink def fail(path, *args, **kwargs): if path.name.startswith(".vbcase-"): raise OSError("simulated cleanup failure") return original_unlink(path, *args, **kwargs) monkeypatch.setattr(Path, "unlink", fail) with pytest.warns(ResourceWarning, match="Package published"): assert PackageWriter().write(source, tmp_path / "out").valid assert PackageValidator().validate(tmp_path / "out").valid for path in tmp_path.glob(".vbcase-*.tmp"): original_unlink(path)