a8fc1a3438
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a11538-05ea-7b63-b3c7-a2df5f38898d
90 lines
3.4 KiB
Python
90 lines
3.4 KiB
Python
"""Retention proposals preserve known recovered points and never own expiration."""
|
|
|
|
import importlib.util
|
|
import sys
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def planner(monkeypatch):
|
|
scripts = Path(__file__).resolve().parents[1] / "scripts"
|
|
monkeypatch.syspath_prepend(str(scripts))
|
|
spec = importlib.util.spec_from_file_location(
|
|
"application_archive_retention_plan", scripts / "application_archive_retention_plan.py"
|
|
)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
sys.modules.pop("primary_application_archive", None)
|
|
return module
|
|
|
|
|
|
def inventory(planner):
|
|
now = datetime(2027, 1, 12, 3, 15, tzinfo=UTC)
|
|
result = []
|
|
for day in range(70):
|
|
stamp = now - timedelta(days=day)
|
|
result.append(
|
|
{
|
|
"application": "coulomb-social",
|
|
"status": "offsite_fetched_verified",
|
|
"remote_retrieval_verified": True,
|
|
"captured_at": stamp.isoformat(),
|
|
"object_key": planner.PROFILES["coulomb-social"]
|
|
+ stamp.strftime("%Y/%m/%d/%Y%m%dT%H%M%SZ-")
|
|
+ f"{day:024x}"
|
|
+ ".tar.age",
|
|
"version_id": "version-" + str(day),
|
|
"ciphertext_sha256": "a" * 64,
|
|
"ciphertext_bytes": 5000,
|
|
}
|
|
)
|
|
recovery = dict(
|
|
result[-1], status="application_restored_verified", application_restore_verified=True
|
|
)
|
|
return now, result, recovery
|
|
|
|
|
|
def test_protects_old_recovered_point_and_deduplicates_daily_weekly(planner):
|
|
now, records, recovery = inventory(planner)
|
|
records.append(dict(records[0], application="email-connect", object_key="foreign"))
|
|
records.append(dict(records[0], status="failed", object_key="failed"))
|
|
result = planner.plan("coulomb-social", records, recovery, now)
|
|
keys = [item["object_key"] for item in result["keep"]]
|
|
assert len(keys) == len(set(keys)) <= 19
|
|
assert records[0]["object_key"] in keys and recovery["object_key"] in keys
|
|
assert all(item["object_key"] in keys for item in records[:14])
|
|
assert result["ignored"] == ["foreign", "failed"]
|
|
assert result["expiration_candidates"]
|
|
assert not result["provider_mutations"] and not result["local_mutations"]
|
|
assert result["protected"]["version_id"] == recovery["version_id"]
|
|
|
|
|
|
@pytest.mark.parametrize("change", ["missing", "version", "hash", "unproved"])
|
|
def test_refuses_missing_changed_or_unproved_application_recovery(planner, change):
|
|
now, records, recovery = inventory(planner)
|
|
if change == "missing":
|
|
records.pop()
|
|
elif change == "version":
|
|
recovery["version_id"] = "different-version"
|
|
elif change == "hash":
|
|
recovery["ciphertext_sha256"] = "b" * 64
|
|
else:
|
|
recovery["application_restore_verified"] = False
|
|
with pytest.raises(ValueError):
|
|
planner.plan("coulomb-social", records, recovery, now)
|
|
|
|
|
|
@pytest.mark.parametrize("change", ["duplicate", "future", "unversioned"])
|
|
def test_refuses_ambiguous_inventory(planner, change):
|
|
now, records, recovery = inventory(planner)
|
|
if change == "duplicate":
|
|
records.append(dict(records[0]))
|
|
elif change == "future":
|
|
records[0]["captured_at"] = (now + timedelta(days=1)).isoformat()
|
|
else:
|
|
records[0]["version_id"] = "null"
|
|
with pytest.raises(ValueError):
|
|
planner.plan("coulomb-social", records, recovery, now)
|