56 lines
1.7 KiB
Python
56 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
INVENTORY = ROOT / "docs" / "recovery-operation-inventory.json"
|
|
MODES = {
|
|
"atomic",
|
|
"compensation",
|
|
"snapshot_restore",
|
|
"forward_recovery",
|
|
"irreversible",
|
|
}
|
|
ADOPTION_STATES = {"planned", "reference-implementation", "adopted"}
|
|
REQUIRED_PREFIXES = {
|
|
"campaign.",
|
|
"files.",
|
|
"mail.",
|
|
"connectors.",
|
|
"dataflow.",
|
|
"workflow-engine.",
|
|
"core.module-lifecycle.",
|
|
}
|
|
|
|
|
|
def test_recovery_operation_inventory_is_complete_and_actionable() -> None:
|
|
payload = json.loads(INVENTORY.read_text(encoding="utf-8"))
|
|
assert payload["schema_version"] == 1
|
|
operations = payload["operations"]
|
|
ids = [item["id"] for item in operations]
|
|
assert len(ids) == len(set(ids))
|
|
assert all(any(item.startswith(prefix) for item in ids) for prefix in REQUIRED_PREFIXES)
|
|
|
|
for item in operations:
|
|
assert item["mode"] in MODES
|
|
assert item["adoption"] in ADOPTION_STATES
|
|
assert item["repository"].startswith("govoplan-")
|
|
assert item["resources"]
|
|
assert item["fenced"] is True
|
|
issue = urlparse(item["issue"])
|
|
assert issue.scheme == "https"
|
|
assert issue.netloc == "git.add-ideas.de"
|
|
assert issue.path.startswith(f"/GovOPlaN/{item['repository']}/issues/")
|
|
|
|
|
|
def test_non_atomic_operations_do_not_claim_plain_database_rollback() -> None:
|
|
operations = json.loads(INVENTORY.read_text(encoding="utf-8"))["operations"]
|
|
for item in operations:
|
|
if item["mode"] == "atomic":
|
|
assert item["resources"] == ["postgresql"] or item["id"] == (
|
|
"connectors.sync.read-snapshot"
|
|
)
|