"""TDD — IdentityPlugin records layer. Covers: - IdentityRecord, RelationshipRecord, SignatureRecord typed shapes - identity_path / relationship_path encoding and round-trip parsing - Record serialisation to / from JSON bytes - parse_identity_path / parse_relationship_path (inverse of encode) """ from __future__ import annotations import json import pytest from muse.plugins.identity.records import ( IdentityRecord, RelationshipRecord, SignatureRecord, identity_path, parse_identity_path, parse_relationship_path, record_from_bytes, record_to_bytes, relationship_path, ) # ── path encoding ───────────────────────────────────────────────────────────── class TestIdentityPath: def test_human_handle(self) -> None: assert identity_path("gabriel") == "identities/gabriel.json" def test_agent_handle(self) -> None: assert identity_path("claude-code") == "identities/claude-code.json" def test_org_handle(self) -> None: assert identity_path("musehub-org") == "identities/musehub-org.json" def test_path_is_posix(self) -> None: p = identity_path("alice") assert "/" in p assert "\\" not in p def test_parse_round_trips(self) -> None: for handle in ("gabriel", "claude-code", "acme-org", "a1"): encoded = identity_path(handle) assert parse_identity_path(encoded) == handle def test_non_identity_path_returns_none(self) -> None: assert parse_identity_path("relationships/x--spawns--y.json") is None assert parse_identity_path("other/file.json") is None class TestRelationshipPath: def test_spawns_edge(self) -> None: p = relationship_path("gabriel", "spawns", "claude-code") assert p == "relationships/gabriel--spawns--claude-code.json" def test_member_of_edge(self) -> None: p = relationship_path("gabriel", "member_of", "musehub-org") assert p == "relationships/gabriel--member_of--musehub-org.json" def test_path_is_posix(self) -> None: p = relationship_path("a", "spawns", "b") assert "/" in p assert "\\" not in p def test_parse_round_trips(self) -> None: cases = [ ("gabriel", "spawns", "claude-code"), ("gabriel", "member_of", "musehub-org"), ("acme", "member_of", "lab"), ] for frm, edge, to in cases: encoded = relationship_path(frm, edge, to) assert parse_relationship_path(encoded) == (frm, edge, to) def test_non_relationship_path_returns_none(self) -> None: assert parse_relationship_path("identities/gabriel.json") is None assert parse_relationship_path("other/file.json") is None # ── record serialisation ────────────────────────────────────────────────────── class TestIdentityRecordSerialisation: def _human_record(self) -> IdentityRecord: return IdentityRecord( handle="gabriel", type="human", pubkey="ed25519:AAAA", quorum=None, registered_at="2026-04-21T22:00:00Z", metadata={}, ) def test_round_trip_human(self) -> None: rec = self._human_record() raw = record_to_bytes(rec) parsed = record_from_bytes(raw) assert parsed == rec def test_serialised_is_valid_json(self) -> None: raw = record_to_bytes(self._human_record()) obj = json.loads(raw) assert obj["handle"] == "gabriel" assert obj["type"] == "human" def test_round_trip_org(self) -> None: rec = IdentityRecord( handle="acme", type="org", pubkey=None, quorum=2, registered_at="2026-04-21T22:00:00Z", metadata={}, ) assert record_from_bytes(record_to_bytes(rec)) == rec def test_round_trip_agent(self) -> None: rec = IdentityRecord( handle="claude-code", type="agent", pubkey="ed25519:BBBB", quorum=None, registered_at="2026-04-21T22:00:00Z", metadata={"model_id": "claude-sonnet-4-6"}, ) assert record_from_bytes(record_to_bytes(rec)) == rec class TestRelationshipRecordSerialisation: def _spawns_record(self) -> RelationshipRecord: return RelationshipRecord( from_handle="gabriel", to_handle="claude-code", edge_type="spawns", weight=None, authorized_by=[ SignatureRecord( signer="gabriel", signature="ed25519:CCCC", signed_at="2026-04-21T22:00:00Z", ) ], ) def test_round_trip_spawns(self) -> None: rec = self._spawns_record() assert record_from_bytes(record_to_bytes(rec)) == rec def test_round_trip_member_of(self) -> None: rec = RelationshipRecord( from_handle="gabriel", to_handle="musehub-org", edge_type="member_of", weight="1", authorized_by=[ SignatureRecord( signer="gabriel", signature="ed25519:DDDD", signed_at="2026-04-21T22:00:00Z", ) ], ) assert record_from_bytes(record_to_bytes(rec)) == rec def test_serialised_preserves_weight(self) -> None: rec = RelationshipRecord( from_handle="alice", to_handle="acme", edge_type="member_of", weight="0.5", authorized_by=[], ) raw = record_to_bytes(rec) parsed = record_from_bytes(raw) assert parsed["weight"] == "0.5"