Nick
ca4f7593d7
CI Pipeline / best-practices (push) Successful in 8s
Auto-version and Release / release (push) Failing after 14s
CI Pipeline / documentation (push) Successful in 5s
CI Pipeline / lint (push) Failing after 34s
CI Pipeline / canary (push) Successful in 37s
CI Pipeline / type-check (push) Successful in 31s
CI Pipeline / gitea-offline (push) Successful in 16s
CI Pipeline / unit-tests (push) Successful in 1m4s
CI Pipeline / required-checks (push) Successful in 1s
deploy / deploy (push) Successful in 48s
CORRECTION 1: #314 reopened — recall_with_associations mode field defect CONFIRMED in current source and FIXED (mode: association_merged). CORRECTION 3: Gitea credential removed from remote URL (credential helper). TOKEN EXPOSED: yes (in diagnostic output). TOKEN ROTATED: not yet (requires Gitea web UI access). Remote URL now shows username only, no secret. CORRECTION 4: duplicate-release-commit defect ROOT CAUSE: convergence computed warranted bump from commits-since-tag without checking if master version already satisfies the bump. FIX: already-converged guard in version_convergence.py (master version > last tag → no commit). Repo-aware (args.repo passed to read_version/latest_tag). CORRECTION 6: lifecycle terminalization — execution exception now guarantees task terminalizes as failed with evidence (no immortal RUNNING). CORRECTION 7: route instruction task-scoped on RuneTask.route_instruction (NOT shared operator state). Read-only leading-verb override. Isolation tests prove no cross-contamination between interleaved tasks. PL1-01/02/03/05: pl1_records.py (route resolution, proof level, failure class, candidate status), metrics with honest NOT_YET_MEASURABLE semantics, route-instruction learning loop (PE-1789102658-0269 SEED). 1253 tests green; ratchet 352/80 (shrink); henxels clean.
117 lines
4.3 KiB
Python
117 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
|