teleo-infrastructure/tests/test_attest_gcp_reasoning_compute.py

128 lines
4.4 KiB
Python

import json
from pathlib import Path
import pytest
from ops import attest_gcp_reasoning_compute as attestation
def reasoning_receipt(path: Path) -> Path:
path.write_text(
json.dumps(
{
"schema": attestation.REASONING_SCHEMA,
"status": "pass",
"generated_at_utc": "2026-07-15T01:00:00+00:00",
"completed_at_utc": "2026-07-15T01:01:00+00:00",
"target_database": "teleo_clone_current_test",
"source_compute": "teleo-prod-1",
"errors": [],
"posted_to_telegram": False,
"database_write_attempted": False,
},
sort_keys=True,
)
+ "\n",
encoding="utf-8",
)
return path
def compute_identity() -> dict:
return {
"project": "teleo-501523",
"project_number": "785938879453",
"instance": "teleo-prod-1",
"instance_id": "123456789",
"zone": "europe-west6-a",
"zone_resource": "projects/123456789/zones/europe-west6-a",
"network_resource": "projects/123456789/networks/teleo-staging-net",
"private_ip": "10.61.0.2",
"service_account": "teleo-runtime@teleo-501523.iam.gserviceaccount.com",
"expected_private_network": "projects/teleo-501523/global/networks/teleo-staging-net",
"checks": {
"project": True,
"project_number": True,
"expected_network_project": True,
"instance": True,
"instance_id": True,
"zone": True,
"network": True,
"private_ip": True,
"service_account": True,
},
}
def test_attestation_binds_reasoning_hash_to_live_metadata(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
receipt_path = reasoning_receipt(tmp_path / "reasoning.json")
observed = {}
def fake_identity(project, instance, zone, network, *, timeout):
observed.update(
{
"project": project,
"instance": instance,
"zone": zone,
"network": network,
"timeout": timeout,
}
)
return compute_identity()
monkeypatch.setattr(attestation, "gce_compute_identity", fake_identity)
result = attestation.attest_reasoning_compute(
reasoning_receipt_path=receipt_path,
restore_run_id="gcp-restore-current-test123",
target_database="teleo_clone_current_test",
generated_at_utc="2026-07-15T01:01:05+00:00",
)
assert result["status"] == "pass"
assert result["checks"]["gce_metadata_identity"] is True
assert result["reasoning_receipt"]["sha256"] == attestation.sha256_file(receipt_path)
assert result["compute"]["instance_id"] == "123456789"
assert observed == {
"project": "teleo-501523",
"instance": "teleo-prod-1",
"zone": "europe-west6-a",
"network": "projects/teleo-501523/global/networks/teleo-staging-net",
"timeout": 10.0,
}
def test_attestation_rejects_claimed_wrong_compute_before_metadata(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
receipt_path = reasoning_receipt(tmp_path / "reasoning.json")
payload = json.loads(receipt_path.read_text())
payload["source_compute"] = "untrusted-vm"
receipt_path.write_text(json.dumps(payload) + "\n", encoding="utf-8")
monkeypatch.setattr(
attestation,
"gce_compute_identity",
lambda *args, **kwargs: pytest.fail("metadata must not be queried"),
)
with pytest.raises(ValueError, match="source_compute"):
attestation.attest_reasoning_compute(
reasoning_receipt_path=receipt_path,
restore_run_id="gcp-restore-current-test123",
target_database="teleo_clone_current_test",
)
def test_attestation_rejects_send_or_write_receipt(tmp_path: Path) -> None:
receipt_path = reasoning_receipt(tmp_path / "reasoning.json")
payload = json.loads(receipt_path.read_text())
payload["posted_to_telegram"] = True
payload["database_write_attempted"] = True
receipt_path.write_text(json.dumps(payload) + "\n", encoding="utf-8")
with pytest.raises(ValueError, match=r"no_database_write.*no_telegram_send"):
attestation.attest_reasoning_compute(
reasoning_receipt_path=receipt_path,
restore_run_id="gcp-restore-current-test123",
target_database="teleo_clone_current_test",
)