133 lines
5.1 KiB
Python
133 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO_ROOT / "scripts"))
|
|
|
|
import compile_kb_source_packet as compiler # noqa: E402
|
|
import run_leo_integrated_learning_lifecycle as lifecycle # noqa: E402
|
|
|
|
|
|
def _write_source_packet(tmp_path: Path) -> Path:
|
|
packet = tmp_path / "source-packet"
|
|
prepared = packet / "prepared"
|
|
prepared.mkdir(parents=True)
|
|
artifact = REPO_ROOT / "fixtures" / "working-leo" / "document-ingestion-v1.json"
|
|
manifest = REPO_ROOT / "fixtures" / "working-leo" / "source-compiler-manifest-v1.json"
|
|
extracted = prepared / "extracted-text.txt"
|
|
manifest_target = prepared / "source-manifest.json"
|
|
shutil.copyfile(artifact, extracted)
|
|
shutil.copyfile(manifest, manifest_target)
|
|
expected = compiler.compile_source_packet(extracted, extracted, manifest_target)
|
|
(packet / "proposal-packet.json").write_text(
|
|
json.dumps(expected, indent=2, sort_keys=True) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
return packet
|
|
|
|
|
|
def _docker_available() -> bool:
|
|
if not shutil.which("docker"):
|
|
return False
|
|
result = subprocess.run(
|
|
["docker", "info", "--format", "{{.ServerVersion}}"],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
return result.returncode == 0
|
|
|
|
|
|
def test_source_packet_recompiles_to_retained_expectation(tmp_path: Path) -> None:
|
|
packet = _write_source_packet(tmp_path)
|
|
|
|
compiled, expected, paths = lifecycle.compile_source_packet(packet)
|
|
|
|
assert compiled == expected
|
|
assert lifecycle.sha256_file(paths["artifact"]) == expected["validated_manifest"]["artifact_sha256"]
|
|
assert compiled["status"] == "pending_review"
|
|
assert compiled["strict_child_proposal"]["status"] == "pending_review"
|
|
|
|
|
|
def test_source_packet_refuses_retained_packet_drift(tmp_path: Path) -> None:
|
|
packet = _write_source_packet(tmp_path)
|
|
expected_path = packet / "proposal-packet.json"
|
|
expected = json.loads(expected_path.read_text(encoding="utf-8"))
|
|
expected["strict_child_proposal"]["status"] = "approved"
|
|
expected_path.write_text(json.dumps(expected), encoding="utf-8")
|
|
|
|
with pytest.raises(lifecycle.LifecycleError, match="does not match retained"):
|
|
lifecycle.compile_source_packet(packet)
|
|
|
|
|
|
def test_exact_identities_adds_observed_evidence_ids(tmp_path: Path) -> None:
|
|
packet = _write_source_packet(tmp_path)
|
|
compiled, _expected, _paths = lifecycle.compile_source_packet(packet)
|
|
child = compiled["strict_child_proposal"]
|
|
apply_payload = child["payload"]["apply_payload"]
|
|
planned = apply_payload["evidence"][0]
|
|
evidence_id = "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee"
|
|
|
|
identities = lifecycle.exact_identities(
|
|
child,
|
|
{
|
|
"claim_evidence": [{"id": evidence_id, **planned}],
|
|
"claim_edges": [],
|
|
},
|
|
)
|
|
|
|
assert identities["proposal_id"] == child["id"]
|
|
assert identities["claim_ids"] == [row["id"] for row in apply_payload["claims"]]
|
|
assert identities["source_ids"] == [row["id"] for row in apply_payload["sources"]]
|
|
assert identities["planned_evidence_keys"] == [
|
|
{
|
|
"claim_id": row["claim_id"],
|
|
"source_id": row["source_id"],
|
|
"role": row["role"],
|
|
}
|
|
for row in apply_payload["evidence"]
|
|
]
|
|
assert identities["evidence_ids"] == [evidence_id]
|
|
|
|
|
|
def test_lifecycle_requires_explicit_approve_action(tmp_path: Path) -> None:
|
|
with pytest.raises(lifecycle.LifecycleError, match="only supported explicit review action"):
|
|
lifecycle.run_lifecycle(
|
|
tmp_path / "unused",
|
|
tmp_path / "receipt.json",
|
|
review_action="inspect",
|
|
)
|
|
|
|
|
|
@pytest.mark.skipif(not _docker_available(), reason="Docker daemon is required for the isolated lifecycle")
|
|
def test_disposable_postgres_lifecycle_end_to_end(tmp_path: Path) -> None:
|
|
packet = _write_source_packet(tmp_path)
|
|
output = tmp_path / "integrated-learning-receipt.json"
|
|
|
|
receipt = lifecycle.run_lifecycle(
|
|
packet,
|
|
output,
|
|
review_action="approve",
|
|
review_note="Approved exact fixture payload for isolated lifecycle regression coverage.",
|
|
)
|
|
|
|
assert receipt["status"] == "pass", receipt.get("error")
|
|
assert receipt["current_tier"] == "realistic_isolated_local_container_end_to_end"
|
|
assert receipt["phases"]["staged_pending_review"]["proposal"]["status"] == "pending_review"
|
|
assert receipt["phases"]["approved_unapplied"]["proposal"]["status"] == "approved"
|
|
assert receipt["phases"]["approved_unapplied"]["proposal"]["applied_at"] is None
|
|
assert receipt["phases"]["applied"]["proposal"]["status"] == "applied"
|
|
assert receipt["phases"]["applied"]["apply_receipt"]["replay_ready"] is True
|
|
assert receipt["phases"]["recompiled_readback"]["matches_initial_compile"] is True
|
|
assert receipt["identities"]["evidence_ids"]
|
|
assert len(receipt["checks"]) >= 20
|
|
assert all(receipt["checks"].values())
|
|
assert receipt["cleanup"]["all"] is True
|
|
assert json.loads(output.read_text(encoding="utf-8")) == receipt
|