teleo-infrastructure/tests/test_leo_identity_version_manifest.py
twentyOne2x dfd73b1cfa
Some checks are pending
CI / lint-and-test (push) Waiting to run
feat: emit deterministic Leo identity version manifest (#169)
2026-07-15 22:45:49 +02:00

231 lines
8.1 KiB
Python

from __future__ import annotations
import json
import os
import subprocess
import sys
from pathlib import Path
import pytest
from scripts import leo_behavior_manifest as behavior
from scripts import leo_identity_version_manifest as version
from tests.test_leo_identity_manifest import _fixture, _write
ROOT = Path(__file__).resolve().parents[1]
def _build(fixture: dict[str, Path | dict]) -> dict:
return version.build_version_manifest(
profile=fixture["profile"],
hermes_root=fixture["hermes"],
deployment_root=fixture["repo"],
source_root=fixture["repo"],
database_fingerprint=fixture["fingerprint"],
constitution=fixture["constitution"],
database_identity=fixture["database_identity"],
)
def _rehash(value: dict) -> None:
stable = {key: item for key, item in value.items() if key != "manifest_sha256"}
value["manifest_sha256"] = behavior.canonical_sha256(stable)
def _input_args(fixture: dict[str, Path | dict]) -> list[str]:
return [
"--profile",
str(fixture["profile"]),
"--hermes-root",
str(fixture["hermes"]),
"--deployment-root",
str(fixture["repo"]),
"--source-root",
str(fixture["repo"]),
"--database-fingerprint",
str(fixture["fingerprint"]),
"--constitution",
str(fixture["constitution"]),
"--database-identity",
str(fixture["database_identity"]),
]
def test_version_manifest_names_every_required_identity_boundary(tmp_path: Path) -> None:
manifest = _build(_fixture(tmp_path))
assert manifest["schema"] == version.SCHEMA
assert (
manifest["runtime_version"]["source_commit"] == manifest["runtime_version"]["teleo_infrastructure"]["git_head"]
)
assert [item["name"] for item in manifest["skill_versions"]["skills"]] == ["identity"]
assert manifest["canonical_database_version"] == {
"authority": "synthetic_noncanonical_fixture",
"canonical_authority_granted": False,
"database": "fixture",
"database_user": "reader",
"system_identifier": "12345",
"fingerprint_sha256": "1" * 64,
"table_rows_sha256": "2" * 64,
"structure_sha256": "3" * 64,
"table_count": 2,
"total_rows": 2,
"identity_records_sha256": manifest["identity_manifest"]["identity_inputs"]["identity_sources"][
"database_derived_identity"
]["semantic_sha256"],
"identity_record_count": 2,
"source_mode": "synthetic_noncanonical_fixture",
"same_transaction_as_fingerprint": False,
"export_receipt_sha256": None,
}
assert manifest["compiled_identity_version"]["role"] == "derived_view_from_versioned_inputs_not_authority"
assert manifest["session_boundary"]["included_in_identity_inputs"] is False
assert manifest["session_boundary"]["may_be_used_as_evidence"] is False
assert manifest["authority_contract"] == version.expected_authority_contract()
version.validate_version_manifest(manifest)
def test_version_manifest_is_identical_across_checkout_paths(tmp_path: Path) -> None:
first = _build(_fixture(tmp_path / "checkout-a"))
second = _build(_fixture(tmp_path / "checkout-b"))
assert first == second
assert str(tmp_path) not in json.dumps(first, sort_keys=True)
@pytest.mark.parametrize(
("field", "replacement", "message"),
[
(
"authority_contract",
{
"durable_knowledge_authority": "session_history",
"compiled_identity_role": "authority",
"session_history_role": "provenance",
},
"authority contract drift",
),
(
"session_boundary",
{
"classification": "canonical",
"included_in_identity_inputs": True,
"may_be_used_as_evidence": True,
"restart_policy": "reuse_history_as_identity",
},
"session boundary projection drift",
),
],
)
def test_rehashed_authority_or_session_tampering_fails_closed(
tmp_path: Path, field: str, replacement: dict, message: str
) -> None:
manifest = _build(_fixture(tmp_path))
manifest[field] = replacement
_rehash(manifest)
with pytest.raises(version.IdentityVersionManifestError, match=message):
version.validate_version_manifest(manifest)
def test_rehashed_database_authority_tampering_fails_closed(tmp_path: Path) -> None:
manifest = _build(_fixture(tmp_path))
manifest["canonical_database_version"]["canonical_authority_granted"] = True
manifest["canonical_database_version"]["authority"] = "canonical_database"
_rehash(manifest)
with pytest.raises(version.IdentityVersionManifestError, match="canonical database version projection drift"):
version.validate_version_manifest(manifest)
def test_named_skill_manifest_rejects_fully_rehashed_path_or_hash_tampering(tmp_path: Path) -> None:
manifest = _build(_fixture(tmp_path))
skill = manifest["skill_versions"]["skills"][0]
skill["files"][0]["path"] = "../MEMORY.md"
skill["content_sha256"] = behavior.canonical_sha256(skill["files"])
_rehash(manifest)
with pytest.raises(version.IdentityVersionManifestError, match="skill identity path is invalid"):
version.validate_version_manifest(manifest)
def test_named_skill_manifest_rejects_a_rehashed_parent_name(tmp_path: Path) -> None:
manifest = _build(_fixture(tmp_path))
manifest["skill_versions"]["skills"][0]["name"] = ".."
_rehash(manifest)
with pytest.raises(version.IdentityVersionManifestError, match="skill name is invalid"):
version.validate_version_manifest(manifest)
def test_rebuilder_rejects_a_different_clean_input_commit(tmp_path: Path) -> None:
fixture = _fixture(tmp_path)
manifest = _build(fixture)
_write(fixture["profile"] / "skills" / "identity" / "SKILL.md", "# changed identity readback\n")
subprocess.run(["git", "-C", str(fixture["repo"]), "add", "."], check=True)
subprocess.run(
["git", "-C", str(fixture["repo"]), "commit", "-qm", "change identity input"],
check=True,
env={
**dict(os.environ),
"GIT_AUTHOR_DATE": "2026-01-02T00:00:00Z",
"GIT_COMMITTER_DATE": "2026-01-02T00:00:00Z",
},
)
with pytest.raises(version.IdentityVersionManifestError, match="does not match the supplied versioned inputs"):
version.verify_against_inputs(
manifest,
profile=fixture["profile"],
hermes_root=fixture["hermes"],
deployment_root=fixture["repo"],
source_root=fixture["repo"],
database_fingerprint=fixture["fingerprint"],
constitution=fixture["constitution"],
database_identity=fixture["database_identity"],
)
def test_cli_generate_is_deterministic_and_verify_rebuilds_inputs(tmp_path: Path) -> None:
fixture = _fixture(tmp_path)
first = tmp_path / "identity-version-first.json"
second = tmp_path / "identity-version-second.json"
common = _input_args(fixture)
for output in (first, second):
result = subprocess.run(
[
sys.executable,
"-m",
"scripts.leo_identity_version_manifest",
"generate",
*common,
"--output",
str(output),
],
cwd=ROOT,
check=True,
capture_output=True,
text=True,
)
summary = json.loads(result.stdout)
assert summary["status"] == "ok"
assert summary["skills"] == ["identity"]
assert first.read_bytes() == second.read_bytes()
verify = subprocess.run(
[
sys.executable,
"-m",
"scripts.leo_identity_version_manifest",
"verify",
"--manifest",
str(first),
*common,
],
cwd=ROOT,
check=True,
capture_output=True,
text=True,
)
assert json.loads(verify.stdout)["manifest_sha256"] == json.loads(first.read_text())["manifest_sha256"]