tegwick
51a527128a
Some checks failed
Use-case regression gate / usecase-regressions (push) Failing after 6m47s
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
217 lines
8 KiB
Python
217 lines
8 KiB
Python
"""Provider observation must distinguish absence, uncertainty and exact versions."""
|
|
|
|
import copy
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from tests.test_backup_health import NOW
|
|
from tests.test_backup_health import health_state as health_state
|
|
from tests.test_backup_health import monitor as monitor
|
|
from tests.test_backup_retention_review import evidence as evidence
|
|
from tests.test_backup_retention_review import reviewer as reviewer
|
|
|
|
|
|
@pytest.fixture
|
|
def observer(monkeypatch):
|
|
scripts = Path(__file__).resolve().parents[1] / "scripts"
|
|
monkeypatch.syspath_prepend(str(scripts))
|
|
spec = importlib.util.spec_from_file_location(
|
|
"retention_observer", scripts / "observe-backup-retention.py"
|
|
)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class ProviderError(Exception):
|
|
def __init__(self, status, code):
|
|
self.response = {"ResponseMetadata": {"HTTPStatusCode": status}, "Error": {"Code": code}}
|
|
|
|
|
|
class ReadOnlyStore:
|
|
def __init__(self, evidence):
|
|
self.buckets = {b["bucket"]: copy.deepcopy(b) for b in evidence[2]["observations"]}
|
|
self.heads = {
|
|
(h["bucket"], h["object_key"], h["version_id"]): {
|
|
"VersionId": h["version_id"],
|
|
"ContentLength": h["bytes"],
|
|
"Metadata": {"sha256": h["sha256"]},
|
|
}
|
|
for h in evidence[2]["version_heads"]
|
|
}
|
|
self.payloads = {}
|
|
self.calls = []
|
|
|
|
def get_bucket_versioning(self, Bucket):
|
|
self.calls.append(("versioning", Bucket))
|
|
return self.buckets[Bucket]["versioning"]
|
|
|
|
def get_bucket_lifecycle_configuration(self, Bucket):
|
|
self.calls.append(("lifecycle", Bucket))
|
|
return self.buckets[Bucket]["lifecycle"]
|
|
|
|
def list_object_versions(self, Bucket, Prefix, MaxKeys, **kwargs):
|
|
self.calls.append(("versions", Bucket, Prefix))
|
|
assert Prefix == self.buckets[Bucket]["prefix"] and MaxKeys == 1000
|
|
return {"Versions": copy.deepcopy(self.buckets[Bucket]["versions"]), "IsTruncated": False}
|
|
|
|
def head_object(self, Bucket, Key, VersionId):
|
|
self.calls.append(("head", Bucket, Key, VersionId))
|
|
return self.heads[Bucket, Key, VersionId]
|
|
|
|
def get_object(self, Bucket, Key, VersionId):
|
|
self.calls.append(("get", Bucket, Key, VersionId))
|
|
return {
|
|
"VersionId": VersionId,
|
|
"ContentLength": self.heads[Bucket, Key, VersionId]["ContentLength"],
|
|
"Body": io.BytesIO(self.payloads[Bucket, Key, VersionId]),
|
|
}
|
|
|
|
|
|
def without_sidecar(evidence):
|
|
ref = evidence[0][0]["recovery_mirror"]
|
|
for key in list(ref):
|
|
if key.startswith("receipt_"):
|
|
del ref[key]
|
|
|
|
|
|
def test_observation_rechecks_current_versions_and_passes_offline_review(
|
|
observer, reviewer, evidence
|
|
):
|
|
without_sidecar(evidence)
|
|
store = ReadOnlyStore(evidence)
|
|
before = copy.deepcopy(evidence)
|
|
result = observer.observe(store, store, evidence[0], evidence[1], NOW)
|
|
result["head_observed_at"] = NOW.isoformat()
|
|
plan = reviewer.review(evidence[0], evidence[1], result, evidence[3], NOW)
|
|
assert len(result["version_heads"]) == 4 and plan["generation_selection_valid"]
|
|
assert not result["provider_mutations"] and not result["archive_bodies_refetched"]
|
|
assert not result["protected_kit_body_refetched"]
|
|
assert evidence == before
|
|
|
|
|
|
def test_missing_primary_is_allowed_but_missing_mirror_is_not(observer, evidence):
|
|
without_sidecar(evidence)
|
|
store = ReadOnlyStore(evidence)
|
|
store.buckets[observer.PRIMARY]["versions"] = []
|
|
result = observer.observe(store, store, evidence[0], evidence[1], NOW)
|
|
assert len(result["version_heads"]) == 3
|
|
store.buckets[observer.health.MIRROR_BUCKET]["versions"] = []
|
|
with pytest.raises(ValueError, match="retained_version_missing"):
|
|
observer.observe(store, store, evidence[0], evidence[1], NOW)
|
|
|
|
|
|
@pytest.mark.parametrize("change", ["version", "hash", "size", "delete_marker"])
|
|
def test_changed_exact_head_refuses_observation(observer, evidence, change):
|
|
without_sidecar(evidence)
|
|
store = ReadOnlyStore(evidence)
|
|
head = next(iter(store.heads.values()))
|
|
if change == "version":
|
|
head["VersionId"] = "another-version"
|
|
elif change == "hash":
|
|
head["Metadata"]["sha256"] = "0" * 64
|
|
elif change == "size":
|
|
head["ContentLength"] += 1
|
|
else:
|
|
head["DeleteMarker"] = True
|
|
with pytest.raises(ValueError, match="exact_version_head_differs"):
|
|
observer.observe(store, store, evidence[0], evidence[1], NOW)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"status,code,accepted",
|
|
[
|
|
(404, "NoSuchLifecycleConfiguration", True),
|
|
(404, "NoSuchBucket", False),
|
|
(403, "AccessDenied", False),
|
|
(500, "InternalError", False),
|
|
],
|
|
)
|
|
def test_only_explicit_lifecycle_absence_is_accepted(observer, evidence, status, code, accepted):
|
|
store = ReadOnlyStore(evidence)
|
|
|
|
def fail(**kwargs):
|
|
raise ProviderError(status, code)
|
|
|
|
store.get_bucket_lifecycle_configuration = fail
|
|
if accepted:
|
|
assert observer.observe_bucket(store, observer.KIT)["lifecycle"]["Rules"] == []
|
|
else:
|
|
with pytest.raises(ProviderError):
|
|
observer.observe_bucket(store, observer.KIT)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"failure",
|
|
["foreign", "duplicate", "null_version", "boolean_size", "no_completion", "cursor_cycle"],
|
|
)
|
|
def test_ambiguous_or_incomplete_inventory_refuses(observer, evidence, failure):
|
|
store = ReadOnlyStore(evidence)
|
|
bucket = observer.health.MIRROR_BUCKET
|
|
rows = store.buckets[bucket]["versions"]
|
|
if failure == "foreign":
|
|
rows[0]["Key"] = "foreign/object"
|
|
elif failure == "duplicate":
|
|
rows.append(dict(rows[0]))
|
|
elif failure == "null_version":
|
|
rows[0]["VersionId"] = "null"
|
|
elif failure == "boolean_size":
|
|
rows[0]["Size"] = True
|
|
elif failure == "no_completion":
|
|
store.list_object_versions = lambda **kwargs: {"Versions": []}
|
|
else:
|
|
store.list_object_versions = lambda **kwargs: {
|
|
"Versions": [],
|
|
"IsTruncated": True,
|
|
"NextKeyMarker": "archives/synthetic",
|
|
"NextVersionIdMarker": "cursor",
|
|
}
|
|
with pytest.raises(ValueError):
|
|
observer.observe_bucket(store, bucket)
|
|
|
|
|
|
@pytest.mark.parametrize("corrupt", [False, True])
|
|
def test_retained_receipt_is_fetched_and_compared_with_exact_public_filter(
|
|
observer, evidence, corrupt
|
|
):
|
|
receipt = evidence[0][0]
|
|
ref = receipt["recovery_mirror"]
|
|
payload = observer.worker.canonical(observer.worker.public_receipt(receipt, receipt))
|
|
digest = hashlib.sha256(payload).hexdigest()
|
|
ref.update(
|
|
receipt_object_key=ref["object_key"] + "." + digest + ".receipt.json",
|
|
receipt_sha256=digest,
|
|
receipt_bytes=len(payload),
|
|
receipt_version_id="sidecar",
|
|
)
|
|
store = ReadOnlyStore(evidence)
|
|
bucket = observer.health.MIRROR_BUCKET
|
|
store.buckets[bucket]["versions"].append(
|
|
{"Key": ref["receipt_object_key"], "VersionId": "sidecar", "Size": len(payload)}
|
|
)
|
|
key = bucket, ref["receipt_object_key"], "sidecar"
|
|
store.heads[key] = {
|
|
"VersionId": "sidecar",
|
|
"ContentLength": len(payload),
|
|
"Metadata": {"sha256": digest},
|
|
}
|
|
store.payloads[key] = b"bad" if corrupt else payload
|
|
if corrupt:
|
|
with pytest.raises(ValueError, match="independent_public_receipt_readback_differs"):
|
|
observer.observe(store, store, evidence[0], evidence[1], NOW)
|
|
else:
|
|
result = observer.observe(store, store, evidence[0], evidence[1], NOW)
|
|
assert result["independent_public_receipts_fetched_verified"] == 1
|
|
|
|
|
|
def test_foreign_protected_reference_refuses_before_provider_contact(observer, evidence):
|
|
without_sidecar(evidence)
|
|
store = ReadOnlyStore(evidence)
|
|
evidence[1]["protected_kit"]["bucket"] = "foreign"
|
|
with pytest.raises(ValueError, match="protected_custody_binding"):
|
|
observer.observe(store, store, evidence[0], evidence[1], NOW)
|
|
assert store.calls == []
|