b16f8bab8c
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b
82 lines
3.8 KiB
Python
82 lines
3.8 KiB
Python
"""Exercise the real multipart engine through the exact recovery remapping."""
|
|
import copy
|
|
import hashlib
|
|
import importlib.util
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
import application_recovery_mirror as mirror
|
|
|
|
spec = importlib.util.spec_from_file_location("primary_tests", Path(__file__).with_name("test_primary_application_archive.py"))
|
|
provider = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(provider)
|
|
|
|
|
|
def fixture(tmp_path, failure=None):
|
|
tmp_path.chmod(0o700)
|
|
source = tmp_path / "ciphertext.age"
|
|
data = b"age-encryption.org/v1\nsynthetic-encrypted-pair"
|
|
source.write_bytes(data)
|
|
source.chmod(0o600)
|
|
key = mirror.primary.key_for("coulomb-social", "2026-10-29T00:00:00Z")
|
|
receipt = {"schema": "platform.application-primary-archive.v1", "application": "coulomb-social",
|
|
"status": "offsite_fetched_verified", "object_key": key, "version_id": "primary-version",
|
|
"destination": "s3://" + mirror.primary.BUCKET + "/" + key,
|
|
"ciphertext_sha256": hashlib.sha256(data).hexdigest(), "ciphertext_bytes": len(data),
|
|
"remote_retrieval_verified": True, "immutable_create": True,
|
|
"captured_at": "2026-10-29T00:00:00Z", "image": "synthetic-image",
|
|
"age_recipient": "synthetic-recipient", "database_sha256": "a" * 64, "content_sha256": "b" * 64}
|
|
return source, receipt, provider.Provider(data, failure)
|
|
|
|
|
|
def test_multipart_mirror_and_independent_version_retrieval(tmp_path):
|
|
source, original, client = fixture(tmp_path)
|
|
before = copy.deepcopy(original)
|
|
journal = {}
|
|
receipt = mirror.mirror(client, source, tmp_path / "readback.age", original, journal, lambda: None)
|
|
assert original == before
|
|
assert receipt["version_id"] == "primary-version"
|
|
assert receipt["recovery_mirror"]["version_id"] == client.version
|
|
for _, args in client.calls:
|
|
assert args["Bucket"] == mirror.BUCKET
|
|
assert args["Key"] == "archives/" + original["object_key"]
|
|
client.calls = []
|
|
result = mirror.retrieve(client, tmp_path / "independent.age", receipt)
|
|
assert result["provider_mutations"] is False
|
|
assert [name for name, _ in client.calls] == ["get"]
|
|
|
|
|
|
def test_uncertain_completion_reuses_same_mirror_without_reupload(tmp_path):
|
|
source, receipt, client = fixture(tmp_path, "lost-response")
|
|
journal = {}
|
|
with pytest.raises(TimeoutError):
|
|
mirror.mirror(client, source, tmp_path / "readback.age", receipt, journal, lambda: None)
|
|
client.failure = None
|
|
mirror.mirror(client, source, tmp_path / "retry.age", receipt, journal, lambda: None)
|
|
assert len([name for name, _ in client.calls if name == "start"]) == 1
|
|
|
|
|
|
@pytest.mark.parametrize("field,value", [("bucket", "foreign"), ("project_id", "foreign"),
|
|
("object_key", "archives/foreign"), ("ciphertext_sha256", "c" * 64)])
|
|
def test_foreign_mirror_reference_refuses_before_contact(tmp_path, field, value):
|
|
source, receipt, client = fixture(tmp_path)
|
|
result = mirror.mirror(client, source, tmp_path / "readback.age", receipt, {}, lambda: None)
|
|
client.calls = []
|
|
result["recovery_mirror"][field] = value
|
|
with pytest.raises(mirror.Refused):
|
|
mirror.retrieve(client, tmp_path / "independent.age", result)
|
|
assert client.calls == []
|
|
|
|
|
|
def test_adapter_cannot_address_siblings_or_delete(tmp_path):
|
|
_, receipt, client = fixture(tmp_path)
|
|
adapter = mirror.Mirror(client)
|
|
with pytest.raises(mirror.Refused):
|
|
adapter.delete_object(Bucket=mirror.primary.BUCKET, Key=receipt["object_key"])
|
|
with pytest.raises(ValueError):
|
|
adapter.get_object(Bucket=mirror.primary.BUCKET, Key="platform-pg/application-archives/email-connect/foreign")
|
|
assert client.calls == []
|