95 lines
3.1 KiB
Python
95 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from govoplan_core.core.object_storage import LocalFilesystemStorageBackend
|
|
from govoplan_core.core.recovery import RecoveryMode
|
|
from govoplan_campaign.backend.persistence.campaigns import (
|
|
CampaignPersistenceError,
|
|
_StoredEmlArtifact,
|
|
_build_storage_expectations,
|
|
_delete_storage_keys,
|
|
_verify_build_storage_manifest,
|
|
_verify_storage_keys_absent,
|
|
)
|
|
from govoplan_campaign.backend.routes.versions import _campaign_build_recovery_plan
|
|
|
|
|
|
def test_object_only_and_managed_output_builds_use_distinct_recovery_modes() -> None:
|
|
assert _campaign_build_recovery_plan({}).mode == RecoveryMode.COMPENSATION
|
|
assert (
|
|
_campaign_build_recovery_plan(
|
|
{"delivery": {"print": {"persist_to_files": True}}}
|
|
).mode
|
|
== RecoveryMode.FORWARD_RECOVERY
|
|
)
|
|
assert (
|
|
_campaign_build_recovery_plan(
|
|
{"delivery": {"print": {"persist_to_files": False}}}
|
|
).mode
|
|
== RecoveryMode.COMPENSATION
|
|
)
|
|
|
|
|
|
def test_generated_object_manifest_verifies_exact_bytes(tmp_path: Path) -> None:
|
|
storage = LocalFilesystemStorageBackend(tmp_path)
|
|
payload = b"Message-ID: <build@example.test>\r\n\r\nbody"
|
|
key = "campaign-artifacts/tenant/campaign/version/build/00000001.eml"
|
|
storage.put_bytes(key, payload)
|
|
artifact = _StoredEmlArtifact(
|
|
storage_key=key,
|
|
size_bytes=len(payload),
|
|
sha256=hashlib.sha256(payload).hexdigest(),
|
|
message_id_header="<build@example.test>",
|
|
)
|
|
|
|
evidence = _verify_build_storage_manifest(
|
|
storage,
|
|
_build_storage_expectations(
|
|
stored_eml_by_index={1: artifact},
|
|
print_outputs_by_index={},
|
|
),
|
|
)
|
|
|
|
assert evidence["object_count"] == 1
|
|
assert evidence["total_bytes"] == len(payload)
|
|
storage.put_bytes(key, b"tampered")
|
|
with pytest.raises(CampaignPersistenceError, match="does not match"):
|
|
_verify_build_storage_manifest(
|
|
storage,
|
|
_build_storage_expectations(
|
|
stored_eml_by_index={1: artifact},
|
|
print_outputs_by_index={},
|
|
),
|
|
)
|
|
|
|
|
|
def test_compensation_requires_verified_object_absence(tmp_path: Path) -> None:
|
|
storage = LocalFilesystemStorageBackend(tmp_path)
|
|
key = "campaign-artifacts/tenant/campaign/version/build/object"
|
|
storage.put_bytes(key, b"payload")
|
|
assert _delete_storage_keys(storage, [key]) == []
|
|
assert _verify_storage_keys_absent(storage, [key]) == (True, 0)
|
|
|
|
|
|
class _UnremovableStorage:
|
|
name = "unremovable"
|
|
|
|
def delete(self, _key: str) -> None:
|
|
from govoplan_core.core.object_storage import StorageBackendError
|
|
|
|
raise StorageBackendError("unavailable")
|
|
|
|
def exists(self, _key: str) -> bool:
|
|
return True
|
|
|
|
|
|
def test_failed_compensation_remains_observable() -> None:
|
|
storage = _UnremovableStorage()
|
|
key = "campaign-artifacts/tenant/campaign/version/build/object"
|
|
assert _delete_storage_keys(storage, [key]) == [key] # type: ignore[arg-type]
|
|
assert _verify_storage_keys_absent(storage, [key]) == (False, 1) # type: ignore[arg-type]
|