""" Tests for services/matrix-bridge-dagi/app/room_mapping.py """ import sys from pathlib import Path import pytest _BRIDGE = Path(__file__).parent.parent / "services" / "matrix-bridge-dagi" if str(_BRIDGE) not in sys.path: sys.path.insert(0, str(_BRIDGE)) from app.room_mapping import RoomMappingConfig, parse_room_map, RoomMapping # noqa: E402 ALLOWED = frozenset({"sofiia", "druid"}) ROOM1 = "!QwHczWXgefDHBEVkTH:daarion.space" ROOM2 = "!AnotherRoom123:daarion.space" # ── Parsing — valid ──────────────────────────────────────────────────────────── def test_parse_single_mapping(): cfg = parse_room_map(f"sofiia:{ROOM1}", ALLOWED) assert cfg.total_mappings == 1 assert cfg.mappings[0].agent_id == "sofiia" assert cfg.mappings[0].room_id == ROOM1 def test_parse_multiple_mappings(): raw = f"sofiia:{ROOM1},druid:{ROOM2}" cfg = parse_room_map(raw, ALLOWED) assert cfg.total_mappings == 2 def test_parse_empty_string(): cfg = parse_room_map("", ALLOWED) assert cfg.total_mappings == 0 def test_parse_whitespace_only(): cfg = parse_room_map(" ", ALLOWED) assert cfg.total_mappings == 0 def test_parse_trailing_comma(): cfg = parse_room_map(f"sofiia:{ROOM1},", ALLOWED) assert cfg.total_mappings == 1 def test_parse_spaces_around_entries(): cfg = parse_room_map(f" sofiia:{ROOM1} , druid:{ROOM2} ", ALLOWED) assert cfg.total_mappings == 2 # ── Parsing — invalid ───────────────────────────────────────────────────────── def test_parse_missing_colon_raises(): with pytest.raises(ValueError, match="parse errors"): parse_room_map("sofiia_no_colon", ALLOWED) def test_parse_invalid_room_id_format_raises(): with pytest.raises(ValueError, match="invalid room_id format"): # Room ID must start with ! parse_room_map(f"sofiia:#badroom:server", ALLOWED) def test_parse_absolute_path_as_room_raises(): with pytest.raises(ValueError, match="invalid room_id format"): parse_room_map("sofiia:/etc/passwd", ALLOWED) def test_parse_empty_agent_id_raises(): with pytest.raises(ValueError, match="parse errors"): parse_room_map(f":{ROOM1}", ALLOWED) def test_parse_empty_room_id_raises(): with pytest.raises(ValueError, match="parse errors"): parse_room_map("sofiia:", ALLOWED) # ── agent_for_room ──────────────────────────────────────────────────────────── def test_agent_for_room_found(): cfg = parse_room_map(f"sofiia:{ROOM1}", ALLOWED) assert cfg.agent_for_room(ROOM1) == "sofiia" def test_agent_for_room_not_found(): cfg = parse_room_map(f"sofiia:{ROOM1}", ALLOWED) assert cfg.agent_for_room("!unknownroom:server") is None def test_agent_for_room_not_allowed(): """Agent in mapping but not in allowed_agents → None.""" cfg = parse_room_map(f"druid:{ROOM1}", frozenset({"sofiia"})) # druid not allowed # mapping is accepted but agent_for_room returns None assert cfg.agent_for_room(ROOM1) is None def test_agent_for_room_allowed_when_in_set(): cfg = parse_room_map(f"druid:{ROOM1}", frozenset({"druid"})) assert cfg.agent_for_room(ROOM1) == "druid" # ── rooms_for_agent ─────────────────────────────────────────────────────────── def test_rooms_for_agent_single(): cfg = parse_room_map(f"sofiia:{ROOM1}", ALLOWED) assert cfg.rooms_for_agent("sofiia") == [ROOM1] def test_rooms_for_agent_multiple(): cfg = parse_room_map(f"sofiia:{ROOM1},sofiia:{ROOM2}", ALLOWED) rooms = cfg.rooms_for_agent("sofiia") assert ROOM1 in rooms assert ROOM2 in rooms def test_rooms_for_agent_unknown(): cfg = parse_room_map(f"sofiia:{ROOM1}", ALLOWED) assert cfg.rooms_for_agent("nonexistent") == [] # ── as_summary ──────────────────────────────────────────────────────────────── def test_summary_contains_expected_fields(): cfg = parse_room_map(f"sofiia:{ROOM1}", ALLOWED) summary = cfg.as_summary() assert len(summary) == 1 entry = summary[0] assert entry["room_id"] == ROOM1 assert entry["agent_id"] == "sofiia" assert entry["allowed"] is True def test_summary_allowed_false_for_unknown_agent(): cfg = parse_room_map(f"druid:{ROOM1}", frozenset({"sofiia"})) summary = cfg.as_summary() assert summary[0]["allowed"] is False def test_summary_no_tokens_in_output(): """Access tokens must never appear in summary.""" cfg = parse_room_map(f"sofiia:{ROOM1}", ALLOWED) summary = cfg.as_summary() for entry in summary: assert "token" not in str(entry).lower() assert "secret" not in str(entry).lower() # ── M2.0: N rooms, duplicate validation, multi-room lookup ──────────────────── ROOM3 = "!ThirdRoom456:daarion.space" ROOM4 = "!FourthRoom789:daarion.space" ROOM5 = "!FifthRoom000:daarion.space" ALLOWED_MULTI = frozenset({"sofiia", "druid", "helion", "nutra", "alateya"}) def test_parse_five_rooms(): """N rooms (up to 5) should all parse correctly.""" raw = ( f"sofiia:{ROOM1},druid:{ROOM2},helion:{ROOM3}," f"nutra:{ROOM4},alateya:{ROOM5}" ) cfg = parse_room_map(raw, ALLOWED_MULTI) assert cfg.total_mappings == 5 assert cfg.agent_for_room(ROOM1) == "sofiia" assert cfg.agent_for_room(ROOM2) == "druid" assert cfg.agent_for_room(ROOM3) == "helion" assert cfg.agent_for_room(ROOM4) == "nutra" assert cfg.agent_for_room(ROOM5) == "alateya" def test_duplicate_room_id_raises(): """Same room_id bound to two agents must raise ValueError (M2.0 fail-fast).""" raw = f"sofiia:{ROOM1},druid:{ROOM1}" with pytest.raises(ValueError, match="Duplicate room_id"): parse_room_map(raw, ALLOWED_MULTI) def test_duplicate_room_id_same_agent_raises(): """Even same agent repeated for same room must raise — 1 room = 1 agent.""" raw = f"sofiia:{ROOM1},sofiia:{ROOM1}" with pytest.raises(ValueError, match="Duplicate room_id"): parse_room_map(raw, ALLOWED_MULTI) def test_multi_room_o1_lookup(): """agent_for_room must return correct agent for each of N rooms (O(1) index).""" raw = f"sofiia:{ROOM1},druid:{ROOM2},helion:{ROOM3}" cfg = parse_room_map(raw, ALLOWED_MULTI) assert cfg.agent_for_room(ROOM1) == "sofiia" assert cfg.agent_for_room(ROOM2) == "druid" assert cfg.agent_for_room(ROOM3) == "helion" assert cfg.agent_for_room("!unknown:server") is None def test_rooms_for_agent_multi_room(): """rooms_for_agent returns all rooms bound to a given agent.""" raw = f"sofiia:{ROOM1},sofiia:{ROOM2},druid:{ROOM3}" allowed = frozenset({"sofiia", "druid"}) cfg = parse_room_map(raw, allowed) sofiia_rooms = cfg.rooms_for_agent("sofiia") assert set(sofiia_rooms) == {ROOM1, ROOM2} assert cfg.rooms_for_agent("druid") == [ROOM3] def test_multi_room_summary_count(): """as_summary() must return one entry per mapping.""" raw = f"sofiia:{ROOM1},druid:{ROOM2},helion:{ROOM3}" cfg = parse_room_map(raw, ALLOWED_MULTI) summary = cfg.as_summary() assert len(summary) == 3 room_ids = {s["room_id"] for s in summary} assert room_ids == {ROOM1, ROOM2, ROOM3} def test_multi_room_unknown_agent_filtered(): """agent_for_room returns None if agent not in allowed_agents (even if mapping exists).""" raw = f"sofiia:{ROOM1},unknown_bot:{ROOM2}" allowed = frozenset({"sofiia"}) # unknown_bot not allowed cfg = parse_room_map(raw, allowed) assert cfg.agent_for_room(ROOM1) == "sofiia" assert cfg.agent_for_room(ROOM2) is None # not in allowed_agents