Nick
84eb94185c
Discharges S3 quality finding: _load wraps JSONDecodeError / non-list root / bad item in CorruptMappingError (house convention: G1 CorruptOutboxError, CorruptEventStoreError). Mapping tests 4->7; suite 216.
101 lines
2.9 KiB
Python
101 lines
2.9 KiB
Python
"""Mapping store: durable bidirectional conversation binds."""
|
|
|
|
import pytest
|
|
|
|
from rune.orchestration.mapping import (
|
|
ConversationMapping,
|
|
CorruptMappingError,
|
|
MappingStore,
|
|
)
|
|
|
|
|
|
def test_link_and_lookup_both_directions(tmp_path):
|
|
store = MappingStore(tmp_path / "m")
|
|
m = ConversationMapping(
|
|
external_id="room_42",
|
|
external_thread="t1",
|
|
surface_id="sfc_1",
|
|
internal_conversation_id="conv_9",
|
|
)
|
|
store.link(m)
|
|
assert store.by_external("room_42", "t1").internal_conversation_id == "conv_9"
|
|
assert store.by_internal("conv_9")[0].external_id == "room_42"
|
|
|
|
|
|
def test_re_link_same_internal_is_idempotent(tmp_path):
|
|
store = MappingStore(tmp_path / "m")
|
|
m = ConversationMapping(
|
|
external_id="room_42",
|
|
external_thread="t1",
|
|
surface_id="sfc_1",
|
|
internal_conversation_id="conv_9",
|
|
)
|
|
store.link(m)
|
|
store.link(m)
|
|
assert len(store.all()) == 1
|
|
|
|
|
|
def test_re_link_different_internal_raises(tmp_path):
|
|
store = MappingStore(tmp_path / "m")
|
|
store.link(
|
|
ConversationMapping(
|
|
external_id="e",
|
|
external_thread="t",
|
|
surface_id="s",
|
|
internal_conversation_id="c1",
|
|
)
|
|
)
|
|
with pytest.raises(ValueError):
|
|
store.link(
|
|
ConversationMapping(
|
|
external_id="e",
|
|
external_thread="t",
|
|
surface_id="s",
|
|
internal_conversation_id="c2",
|
|
)
|
|
)
|
|
|
|
|
|
def test_persists_across_reopen(tmp_path):
|
|
store = MappingStore(tmp_path / "m")
|
|
store.link(
|
|
ConversationMapping(
|
|
external_id="room_1",
|
|
external_thread="t",
|
|
surface_id="sfc_1",
|
|
internal_conversation_id="c1",
|
|
)
|
|
)
|
|
store2 = MappingStore(tmp_path / "m")
|
|
assert store2.by_external("room_1", "t").internal_conversation_id == "c1"
|
|
|
|
|
|
def test_corrupt_json_raises_typed_error(tmp_path):
|
|
# S3 quality discharge: corruption never silently skips, and the error
|
|
# is TYPED (house convention: CorruptMappingError, matching G1
|
|
# CorruptOutboxError / CorruptEventStoreError).
|
|
from rune.orchestration.atomic import AtomicStore
|
|
|
|
AtomicStore(tmp_path / "m" / "state").write("mappings.json", "{bad json")
|
|
with pytest.raises(CorruptMappingError):
|
|
MappingStore(tmp_path / "m")
|
|
|
|
|
|
def test_non_list_json_raises_typed_error(tmp_path):
|
|
from rune.orchestration.atomic import AtomicStore
|
|
|
|
AtomicStore(tmp_path / "m" / "state").write("mappings.json", '{"a": 1}')
|
|
with pytest.raises(CorruptMappingError):
|
|
MappingStore(tmp_path / "m")
|
|
|
|
|
|
def test_bad_item_raises_typed_error(tmp_path):
|
|
from rune.orchestration.atomic import AtomicStore
|
|
|
|
AtomicStore(tmp_path / "m" / "state").write(
|
|
"mappings.json",
|
|
'[{"external_id": "x"}]', # missing required fields
|
|
)
|
|
with pytest.raises(CorruptMappingError):
|
|
MappingStore(tmp_path / "m")
|