119 lines
4.3 KiB
Python
119 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
|
|
|
|
import release_metadata_adapter as adapter
|
|
|
|
|
|
TARGET = "c7adece1cc8beffe7c718d53c775898b45d6c7f8"
|
|
EVIDENCE = "8c2898baf07195b9861ed2ff74f03a71674ec9e9"
|
|
ARTIFACT = "ce09a46f28b453634784abceda4597fd51c3aa0838e72a5ad5fefd03d58b949e"
|
|
PLAN = "sha256:c8199d638c2bb2e4896ca9b219ca52db79713bee007df05653ab348905d6cf28"
|
|
TAG = "1" * 40
|
|
|
|
|
|
def value():
|
|
return {
|
|
"version": "V2R1C7",
|
|
"target_oid": TARGET,
|
|
"evidence_closure_oid": EVIDENCE,
|
|
"artifact_sha256": ARTIFACT,
|
|
"approval_plan_id": PLAN,
|
|
"production_service": "aios-v2r1c7-production.service",
|
|
"production_port": 18927,
|
|
}
|
|
|
|
|
|
def old_baseline(path: Path) -> str:
|
|
data = {
|
|
"current_baseline": "V2R1C5",
|
|
"version": "V2R1C5",
|
|
"target_oid": "a" * 40,
|
|
"tag_oid": "b" * 40,
|
|
"generation": 2,
|
|
"authority_generation": 2,
|
|
"status": "PRODUCTION_DEPLOYED",
|
|
}
|
|
path.write_text(json.dumps(data, sort_keys=True) + "\n", encoding="utf-8")
|
|
return adapter.sha256_file(path)
|
|
|
|
|
|
def test_baseline_record_uses_real_previous_production_baseline(tmp_path, monkeypatch):
|
|
# This test isolates previous-baseline semantics; convergence is tested separately.
|
|
monkeypatch.setattr(adapter, "assert_source_registry_convergence", lambda *a, **kw: None)
|
|
baseline = tmp_path / "baseline.json"
|
|
previous_sha = old_baseline(baseline)
|
|
registry_root = tmp_path / "versions"
|
|
registry_root.mkdir()
|
|
args = type("A", (), {
|
|
"repository": tmp_path / "repo.git",
|
|
"registry_root": registry_root,
|
|
"baseline_path": baseline,
|
|
"expected_previous_baseline": "V2R1C5",
|
|
"expected_previous_baseline_sha256": previous_sha,
|
|
"health_url": "http://127.0.0.1:18927/health",
|
|
})()
|
|
monkeypatch.setattr(adapter, "verify_local_tag", lambda *a: (TAG, TARGET))
|
|
monkeypatch.setattr(adapter, "secure_baseline_authority_file", lambda path: None)
|
|
monkeypatch.setattr(adapter, "health", lambda *a: {
|
|
"deployment_state": "ACTIVE",
|
|
"production_deployed": True,
|
|
})
|
|
registry = adapter.registry_record(value(), TAG, {"deployment_state": "ACTIVE"})
|
|
adapter.atomic_json(registry_root / "V2R1C7.json", registry)
|
|
result = adapter.baseline_apply(args, value())
|
|
assert result["status"] == "VERIFIED"
|
|
current = json.loads(baseline.read_text())
|
|
assert current["previous_baseline"] == "V2R1C5"
|
|
assert current["rollback_version"] == "V2R1C5"
|
|
assert current["rollback_target_oid"] == "a" * 40
|
|
assert current["status"] == "PRODUCTION_DEPLOYED"
|
|
|
|
|
|
def test_baseline_cas_conflict_fails_closed(tmp_path, monkeypatch):
|
|
baseline = tmp_path / "baseline.json"
|
|
previous_sha = old_baseline(baseline)
|
|
baseline.write_text('{"current_baseline":"OTHER"}\n', encoding="utf-8")
|
|
args = type("A", (), {
|
|
"repository": tmp_path / "repo.git",
|
|
"registry_root": tmp_path / "versions",
|
|
"baseline_path": baseline,
|
|
"expected_previous_baseline": "V2R1C5",
|
|
"expected_previous_baseline_sha256": previous_sha,
|
|
"health_url": "http://127.0.0.1:18927/health",
|
|
})()
|
|
monkeypatch.setattr(adapter, "verify_local_tag", lambda *a: (TAG, TARGET))
|
|
monkeypatch.setattr(adapter, "secure_baseline_authority_file", lambda path: None)
|
|
result = adapter.baseline_probe(args, value())
|
|
assert result["status"] == "CONFLICT"
|
|
|
|
|
|
def test_registry_record_binds_all_exact_release_fields():
|
|
record = adapter.registry_record(value(), TAG, {"deployment_state": "ACTIVE"})
|
|
assert record["target_oid"] == TARGET
|
|
assert record["artifact_sha256"] == ARTIFACT
|
|
assert record["evidence_closure_oid"] == EVIDENCE
|
|
assert record["approval_plan_id"] == PLAN
|
|
assert record["tag_oid"] == TAG
|
|
assert record["production_deployed"] is True
|
|
|
|
|
|
def test_atomic_json_preserves_existing_mode(tmp_path):
|
|
target = tmp_path / "current.json"
|
|
target.write_text('{"old":true}\n', encoding="utf-8")
|
|
target.chmod(0o440)
|
|
before = target.stat()
|
|
adapter.atomic_json(target, {"new": True})
|
|
after = target.stat()
|
|
assert after.st_mode & 0o777 == 0o440
|
|
assert after.st_uid == before.st_uid
|
|
assert after.st_gid == before.st_gid
|