3c98ca96bb
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b
91 lines
3.3 KiB
Python
91 lines
3.3 KiB
Python
import hashlib
|
|
import io
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
|
import application_recovery_kit as kit
|
|
|
|
|
|
class Missing(Exception):
|
|
def __init__(self):
|
|
self.response = {"ResponseMetadata": {"HTTPStatusCode": 404}}
|
|
|
|
|
|
class Provider:
|
|
def __init__(self):
|
|
self.data, self.calls = None, []
|
|
self.version = "synthetic-kit-version"
|
|
self.corrupt = False
|
|
|
|
def head_object(self, **args):
|
|
self.calls.append("head")
|
|
if self.data is None:
|
|
raise Missing()
|
|
return {"ContentLength": len(self.data), "VersionId": self.version,
|
|
"Metadata": {"sha256": hashlib.sha256(self.data).hexdigest()}}
|
|
|
|
def put_object(self, **args):
|
|
self.calls.append("put")
|
|
assert args["Bucket"] == kit.BUCKET and args["IfNoneMatch"] == "*"
|
|
self.data = args["Body"].read()
|
|
return {"VersionId": self.version}
|
|
|
|
def get_object(self, **args):
|
|
self.calls.append("get")
|
|
assert args["Bucket"] == kit.BUCKET and args["VersionId"] == self.version
|
|
body = b"X" * len(self.data) if self.corrupt else self.data
|
|
return {"VersionId": self.version, "ContentLength": len(body), "Body": io.BytesIO(body)}
|
|
|
|
|
|
@pytest.fixture
|
|
def fixture(tmp_path, monkeypatch):
|
|
tmp_path.chmod(0o700)
|
|
# Real outside-Git enforcement is covered in identity tests; isolate the
|
|
# synthetic provider paths from a workstation's possible /tmp Git ancestor.
|
|
monkeypatch.setattr(kit, "private_directory", kit.archive.private_directory)
|
|
source = tmp_path / "kit.tar"
|
|
source.write_bytes(b"synthetic-public-kit-no-private-keys")
|
|
source.chmod(0o600)
|
|
return source, Provider()
|
|
|
|
|
|
def test_immutable_versioned_kit_readback_and_idempotent_publication(fixture, tmp_path):
|
|
source, client = fixture
|
|
ref = kit.publish(client, source)
|
|
assert ref["automatic_expiry"] is False
|
|
assert ref["object_key"] == "kits/" + hashlib.sha256(source.read_bytes()).hexdigest() + "/recovery-kit.tar"
|
|
assert kit.download(client, ref, tmp_path / "fetched.tar")["independently_retrieved_hash_verified"]
|
|
assert kit.publish(client, source) == ref
|
|
assert client.calls.count("put") == 1
|
|
|
|
|
|
def test_corruption_is_not_accepted_as_independent_kit(fixture, tmp_path):
|
|
source, client = fixture
|
|
ref = kit.publish(client, source)
|
|
client.corrupt = True
|
|
with pytest.raises(kit.Refused, match="kit_readback_hash_differs"):
|
|
kit.download(client, ref, tmp_path / "fetched.tar")
|
|
|
|
|
|
@pytest.mark.parametrize("field,value", [("project_id", "foreign"), ("bucket", "foreign"),
|
|
("object_key", "kits/foreign.tar")])
|
|
def test_foreign_reference_refused_before_get(fixture, tmp_path, field, value):
|
|
source, client = fixture
|
|
ref = kit.publish(client, source)
|
|
ref[field] = value
|
|
with pytest.raises(kit.Refused):
|
|
kit.download(client, ref, tmp_path / "fetched.tar")
|
|
assert "get" not in client.calls
|
|
|
|
|
|
def test_existing_output_is_never_overwritten(fixture, tmp_path):
|
|
source, client = fixture
|
|
ref = kit.publish(client, source)
|
|
output = tmp_path / "fetched.tar"
|
|
output.write_bytes(b"keep")
|
|
with pytest.raises(kit.Refused):
|
|
kit.download(client, ref, output)
|
|
assert output.read_bytes() == b"keep"
|