Nick
ecdb990c2d
Discharges all S1 quality-gate findings: external-id collision now raises (F1), re-register is an explicit capability refresh (F2), registers store a defensive copy (F3), Capability is canonical in records.py and SurfaceRecord round-trips Capability objects (F4). Surface tests 5→9; suite 205.
123 lines
4.3 KiB
Python
123 lines
4.3 KiB
Python
"""Surface records: bidirectional IDs, stable registry."""
|
|
|
|
import pytest
|
|
from pathlib import Path
|
|
|
|
from rune.orchestration.surface import Capability, SurfaceRegistry
|
|
from rune.orchestration.records import SurfaceRecord
|
|
|
|
|
|
def test_capability_is_frozen_and_maps_to_dict():
|
|
c = Capability("images")
|
|
assert c.available is True
|
|
assert Capability.from_dict(c.to_dict()) == c
|
|
|
|
|
|
def test_register_and_lookup_both_directions():
|
|
reg = SurfaceRegistry()
|
|
s = SurfaceRecord(
|
|
surface_id="sfc_1",
|
|
external_id="room_42",
|
|
kind="remote",
|
|
capabilities=(Capability("images"), Capability("history")),
|
|
)
|
|
reg.register(s)
|
|
assert reg.get("sfc_1").external_id == "room_42"
|
|
assert reg.by_external("room_42").surface_id == "sfc_1"
|
|
|
|
|
|
def test_re_register_same_id_same_external_is_idempotent():
|
|
reg = SurfaceRegistry()
|
|
s = SurfaceRecord(surface_id="sfc_1", external_id="room_42", kind="remote")
|
|
reg.register(s)
|
|
reg.register(
|
|
s
|
|
) # same id + same external: re-register replaces (refresh), still one record
|
|
assert len(reg.all()) == 1
|
|
|
|
|
|
def test_re_register_same_id_different_external_raises():
|
|
reg = SurfaceRegistry()
|
|
s1 = SurfaceRecord(surface_id="sfc_1", external_id="room_42", kind="remote")
|
|
reg.register(s1)
|
|
s2 = SurfaceRecord(surface_id="sfc_1", external_id="room_99", kind="remote")
|
|
with pytest.raises(ValueError):
|
|
reg.register(s2)
|
|
|
|
|
|
def test_all_returns_insertion_order():
|
|
reg = SurfaceRegistry()
|
|
reg.register(SurfaceRecord(surface_id="sfc_a", external_id="ea", kind="local"))
|
|
reg.register(SurfaceRecord(surface_id="sfc_b", external_id="eb", kind="tui"))
|
|
assert [s.surface_id for s in reg.all()] == ["sfc_a", "sfc_b"]
|
|
|
|
|
|
def test_external_id_collision_raises():
|
|
# S1 quality F1 discharge: a DIFFERENT surface_id claiming an external
|
|
# id that is already bound must fail loud — never silently overwrite
|
|
# the external-side mapping (the bidirectional identity stays stable).
|
|
reg = SurfaceRegistry()
|
|
reg.register(
|
|
SurfaceRecord(surface_id="sfc_1", external_id="room_42", kind="remote")
|
|
)
|
|
with pytest.raises(ValueError):
|
|
reg.register(
|
|
SurfaceRecord(surface_id="sfc_2", external_id="room_42", kind="remote")
|
|
)
|
|
assert reg.by_external("room_42").surface_id == "sfc_1"
|
|
|
|
|
|
def test_re_register_same_binding_refreshes_capabilities():
|
|
# S1 quality F2 discharge: re-registering the SAME surface_id + external
|
|
# REPLACES the record (capability refresh), it is not a strict no-op —
|
|
# a surface that gains capabilities can renegotiate under the same id.
|
|
reg = SurfaceRegistry()
|
|
reg.register(
|
|
SurfaceRecord(
|
|
surface_id="sfc_1",
|
|
external_id="room_42",
|
|
kind="remote",
|
|
capabilities=(Capability("images"),),
|
|
)
|
|
)
|
|
reg.register(
|
|
SurfaceRecord(
|
|
surface_id="sfc_1",
|
|
external_id="room_42",
|
|
kind="remote",
|
|
capabilities=(Capability("images"), Capability("history")),
|
|
)
|
|
)
|
|
assert len(reg.all()) == 1
|
|
assert [c.name for c in reg.get("sfc_1").capabilities] == ["images", "history"]
|
|
|
|
|
|
def test_register_stores_defensive_copy():
|
|
# S1 quality F3 discharge: caller mutation of the registered record must
|
|
# never corrupt the registry (records are treated as immutable here).
|
|
reg = SurfaceRegistry()
|
|
s = SurfaceRecord(surface_id="sfc_1", external_id="room_42", kind="remote")
|
|
reg.register(s)
|
|
s.external_id = "room_99" # caller mutates the original
|
|
assert reg.get("sfc_1").external_id == "room_42"
|
|
assert reg.by_external("room_42") is not None
|
|
assert reg.by_external("room_99") is None
|
|
|
|
|
|
def test_surface_record_roundtrips_capability_objects(tmp_path: Path):
|
|
# S1 quality F4 discharge: SurfaceRecord.to_dict/from_dict must restore
|
|
# Capability OBJECTS (not dicts) — Task 2's negotiator reads .available.
|
|
from rune.orchestration.records import SurfaceRecord as SR
|
|
|
|
s = SR(
|
|
surface_id="sfc_1",
|
|
external_id="room_42",
|
|
kind="remote",
|
|
capabilities=(Capability("images"), Capability("history", available=False)),
|
|
)
|
|
s2 = SR.from_dict(s.to_dict())
|
|
assert s2 == s
|
|
assert all(isinstance(c, Capability) for c in s2.capabilities)
|
|
assert s2.capabilities[0].name == "images"
|
|
assert s2.capabilities[1].available is False
|