| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122 |
- """Unit tests for the Chatto platform adapter.
- Covers:
- - Protobuf codec (varint, tag, fields, client/server frames, projections)
- - Emoji shortcode conversion
- - Adapter instantiation and properties
- - Registration and requirements
- - Send / reactions / edit / delete (mocked RPC)
- - Typing indicator lifecycle
- - Read state and notifications
- - DM initiation and room creation
- - User lookup (with caching)
- - Presence and custom status
- - Message dispatch (self-echo suppression, handler invocation)
- - Attachment upload (chunked)
- All network calls are mocked — no real HTTP or WebSocket connections.
- """
- import asyncio
- import hashlib
- import os
- import sys
- import tempfile
- from unittest.mock import AsyncMock, MagicMock, patch, call
- from collections import OrderedDict
- import pytest
- import pytest_asyncio
- # ── Path setup ────────────────────────────────────────────────────────────
- sys.path.insert(0, "/opt/hermes")
- sys.path.insert(0, "/root/.hermes/plugins/platforms/chatto")
- import adapter as chatto_adapter
- from adapter import (
- _encode_varint,
- _decode_varint,
- _encode_tag,
- _encode_field_varint,
- _encode_field_bytes,
- _encode_field_string,
- _encode_submessage,
- _decode_fields,
- _get_first,
- _get_all,
- _encode_client_hello,
- _encode_subscribe_events,
- _encode_ping,
- _encode_client_frame_hello,
- _encode_client_frame_subscribe,
- _encode_client_frame_ping,
- _decode_server_frame,
- _decode_projection_event,
- _decode_projection_operation,
- _decode_room_timeline_event,
- _decode_room_timeline_event_upsert,
- _decode_message_posted,
- _decode_message,
- _decode_thread,
- _decode_timestamp,
- _decode_event_envelope,
- _decode_mention_notification,
- _decode_dm_notification,
- _decode_server_hello,
- _decode_error,
- _EMOJI_TO_SHORTCODE,
- _REALTIME_PROTOCOL_VERSION,
- _MAX_MESSAGE_LENGTH,
- _SEEN_CAP,
- ChattoAdapter,
- check_requirements,
- validate_config,
- register,
- )
- from gateway.config import Platform, PlatformConfig
- from gateway.platforms.base import SendResult, MessageEvent, MessageType
- # ── Helpers ───────────────────────────────────────────────────────────────
- class _MockPluginContext:
- """Minimal mock for the plugin registration context."""
- def __init__(self):
- self.registered_names = []
- self.registered_kwargs = None
- def register_platform(self, **kwargs):
- from gateway.platform_registry import platform_registry, PlatformEntry
- entry = PlatformEntry(
- name=kwargs["name"],
- label=kwargs.get("label", kwargs["name"]),
- adapter_factory=kwargs.get("adapter_factory"),
- check_fn=kwargs.get("check_fn"),
- validate_config=kwargs.get("validate_config"),
- is_connected=kwargs.get("is_connected"),
- required_env=kwargs.get("required_env", []),
- source="plugin",
- )
- platform_registry.register(entry)
- self.registered_names.append(kwargs["name"])
- self.registered_kwargs = kwargs
- def _ensure_chatto_registered():
- """Register chatto in the platform registry so Platform('chatto') works."""
- from gateway.platform_registry import platform_registry
- if not platform_registry.is_registered("chatto"):
- ctx = _MockPluginContext()
- register(ctx)
- _CHATTO_ENV_KEYS = [
- "CHATTO_URL",
- "CHATTO_LOGIN",
- "CHATTO_PASSWORD",
- "CHATTO_CHANNELS",
- "CHATTO_HOME_CHANNEL",
- "CHATTO_REQUIRE_MENTION",
- "CHATTO_ALLOWED_USERS",
- "CHATTO_ALLOW_ALL_USERS",
- ]
- def _clear_chatto_env(monkeypatch=None):
- """Remove all CHATTO_* env vars so tests start from a clean slate."""
- for key in _CHATTO_ENV_KEYS:
- if monkeypatch is not None:
- monkeypatch.delenv(key, raising=False)
- else:
- os.environ.pop(key, None)
- def _make_config(**extra_overrides):
- """Create a minimal PlatformConfig for testing."""
- _ensure_chatto_registered()
- extra = {"url": "https://chat.example.com", "channels": ["room1"]}
- extra.update(extra_overrides)
- return PlatformConfig(enabled=True, extra=extra)
- def _make_adapter(**extra_overrides):
- """Create a ChattoAdapter with mocked config. Token is pre-set to avoid
- _ensure_token triggering a real login. All CHATTO_* env vars are cleared
- first so the config.extra values are not overridden by the environment."""
- _clear_chatto_env()
- cfg = _make_config(**extra_overrides)
- adapter = ChattoAdapter(cfg)
- adapter._token = "test-token"
- adapter._user_id = "bot-user-id"
- adapter._user_login = "hermes_bot"
- adapter._user_display = "Hermes Bot"
- return adapter
- # ── Protobuf codec: varint ────────────────────────────────────────────────
- class TestVarint:
- """Test _encode_varint / _decode_varint roundtrips."""
- @pytest.mark.parametrize(
- "value",
- [0, 1, 127, 128, 16384, 2**32, 2**63 - 1],
- )
- def test_varint_roundtrip(self, value):
- encoded = _encode_varint(value)
- decoded, offset = _decode_varint(encoded, 0)
- assert decoded == value
- assert offset == len(encoded)
- def test_varint_zero(self):
- assert _encode_varint(0) == b"\x00"
- def test_varint_one(self):
- assert _encode_varint(1) == b"\x01"
- def test_varint_127(self):
- assert _encode_varint(127) == b"\x7f"
- def test_varint_128(self):
- assert _encode_varint(128) == b"\x80\x01"
- def test_varint_300(self):
- # 300 = 0b100101100 → 0xAC 0x02
- assert _encode_varint(300) == b"\xac\x02"
- def test_varint_16384(self):
- # 16384 = 0x4000 → 0x80 0x80 0x01
- assert _encode_varint(16384) == b"\x80\x80\x01"
- def test_decode_varint_truncated(self):
- with pytest.raises(ValueError, match="Truncated"):
- _decode_varint(b"\x80", 0)
- def test_decode_varint_too_long(self):
- # 10 continuation bytes — exceeds 64-bit
- with pytest.raises(ValueError, match="too long"):
- _decode_varint(b"\x80" * 10, 0)
- # ── Protobuf codec: tag and field encoders ────────────────────────────────
- class TestTagAndFields:
- """Test tag encoding and field-level helpers."""
- def test_encode_tag_field1_varint(self):
- # field 1, wire type 0 → (1<<3)|0 = 8 → 0x08
- assert _encode_tag(1, 0) == b"\x08"
- def test_encode_tag_field2_length_delimited(self):
- # field 2, wire type 2 → (2<<3)|2 = 18 → 0x12
- assert _encode_tag(2, 2) == b"\x12"
- def test_encode_tag_field15_varint(self):
- # field 15, wire type 0 → (15<<3)|0 = 120 → 0x78
- assert _encode_tag(15, 0) == b"\x78"
- def test_encode_field_varint(self):
- result = _encode_field_varint(1, 150)
- # tag(1,0)=0x08 + varint(150)=0x96 0x01
- assert result == b"\x08\x96\x01"
- def test_encode_field_bytes(self):
- result = _encode_field_bytes(2, b"hello")
- # tag(2,2)=0x12 + len(5)=0x05 + "hello"
- assert result == b"\x12\x05hello"
- def test_encode_field_string(self):
- result = _encode_field_string(3, "hi")
- # tag(3,2)=0x1a + len(2)=0x02 + "hi"
- assert result == b"\x1a\x02hi"
- def test_encode_submessage(self):
- inner = _encode_field_varint(1, 42)
- result = _encode_submessage(5, inner)
- # The submessage is length-delimited
- fields = _decode_fields(result)
- assert 5 in fields
- assert isinstance(fields[5][0], bytes)
- def test_decode_fields_varint(self):
- data = _encode_field_varint(1, 42)
- fields = _decode_fields(data)
- assert fields[1] == [42]
- def test_decode_fields_bytes(self):
- data = _encode_field_bytes(2, b"test")
- fields = _decode_fields(data)
- assert fields[2] == [b"test"]
- def test_decode_fields_string(self):
- data = _encode_field_string(3, "hello")
- fields = _decode_fields(data)
- assert fields[3] == [b"hello"]
- def test_decode_fields_multiple(self):
- data = _encode_field_varint(1, 10) + _encode_field_string(2, "abc")
- fields = _decode_fields(data)
- assert fields[1] == [10]
- assert fields[2] == [b"abc"]
- def test_decode_fields_repeated(self):
- data = _encode_field_string(2, "a") + _encode_field_string(2, "b")
- fields = _decode_fields(data)
- assert fields[2] == [b"a", b"b"]
- def test_decode_fields_empty(self):
- fields = _decode_fields(b"")
- assert fields == {}
- def test_get_first(self):
- fields = {1: [10, 20], 2: [b"x"]}
- assert _get_first(fields, 1) == 10
- assert _get_first(fields, 2) == b"x"
- assert _get_first(fields, 99, "default") == "default"
- def test_get_all(self):
- fields = {1: [10, 20]}
- assert _get_all(fields, 1) == [10, 20]
- assert _get_all(fields, 99) == []
- # ── Protobuf codec: client hello / subscribe / ping ──────────────────────
- class TestClientFrames:
- """Test client-side protobuf message encoders."""
- def test_encode_client_hello_with_token(self):
- msg = _encode_client_hello("my-bearer-token")
- fields = _decode_fields(msg)
- # field 1 = protocol_version (varint)
- assert _get_first(fields, 1) == _REALTIME_PROTOCOL_VERSION
- # field 2 = bearer_token (bytes)
- token_val = _get_first(fields, 2)
- assert isinstance(token_val, bytes)
- assert token_val.decode("utf-8") == "my-bearer-token"
- def test_encode_client_hello_without_token(self):
- msg = _encode_client_hello("")
- fields = _decode_fields(msg)
- assert _get_first(fields, 1) == _REALTIME_PROTOCOL_VERSION
- assert 2 not in fields # no bearer_token field
- def test_encode_client_hello_protocol_version_is_1(self):
- msg = _encode_client_hello("x")
- fields = _decode_fields(msg)
- assert _get_first(fields, 1) == 1
- def test_encode_subscribe_events_with_rooms(self):
- msg = _encode_subscribe_events(retained_room_ids=["room1", "room2"])
- fields = _decode_fields(msg)
- # field 2 = repeated string
- room_vals = _get_all(fields, 2)
- assert len(room_vals) == 2
- assert room_vals[0].decode("utf-8") == "room1"
- assert room_vals[1].decode("utf-8") == "room2"
- def test_encode_subscribe_events_with_cursor(self):
- msg = _encode_subscribe_events(resume_cursor="cursor123")
- fields = _decode_fields(msg)
- cursor = _get_first(fields, 1)
- assert isinstance(cursor, bytes)
- assert cursor.decode("utf-8") == "cursor123"
- def test_encode_subscribe_events_empty(self):
- msg = _encode_subscribe_events()
- assert msg == b""
- def test_encode_ping_is_empty(self):
- assert _encode_ping() == b""
- def test_encode_client_frame_hello(self):
- hello = _encode_client_hello("token")
- frame = _encode_client_frame_hello(hello)
- fields = _decode_fields(frame)
- # field 1 = hello submessage
- assert 1 in fields
- inner = _get_first(fields, 1)
- assert isinstance(inner, bytes)
- # Decode inner to verify
- inner_fields = _decode_fields(inner)
- assert _get_first(inner_fields, 1) == _REALTIME_PROTOCOL_VERSION
- def test_encode_client_frame_subscribe(self):
- sub = _encode_subscribe_events(retained_room_ids=["r1"])
- frame = _encode_client_frame_subscribe(sub)
- fields = _decode_fields(frame)
- # field 2 = subscribe_events submessage
- assert 2 in fields
- def test_encode_client_frame_ping(self):
- ping = _encode_ping()
- frame = _encode_client_frame_ping(ping)
- fields = _decode_fields(frame)
- # field 3 = ping submessage (empty)
- assert 3 in fields
- def test_client_hello_roundtrip(self):
- """Encode a client hello, wrap in a frame, decode the frame, decode
- the inner hello, and verify values match."""
- original = _encode_client_hello("roundtrip-token")
- frame = _encode_client_frame_hello(original)
- decoded_frame = _decode_server_frame # not for client frames, but
- # We decode the frame manually
- frame_fields = _decode_fields(frame)
- inner_bytes = _get_first(frame_fields, 1)
- inner_fields = _decode_fields(inner_bytes)
- assert _get_first(inner_fields, 1) == _REALTIME_PROTOCOL_VERSION
- token_raw = _get_first(inner_fields, 2)
- assert token_raw.decode("utf-8") == "roundtrip-token"
- # ── Protobuf codec: server frame decoding ────────────────────────────────
- class TestServerFrameDecoding:
- """Test _decode_server_frame with synthetic frames."""
- def test_decode_server_hello_frame(self):
- # Build a RealtimeServerHello { protocol_version = 1 }
- hello_inner = _encode_field_varint(1, 1)
- # Wrap in RealtimeServerFrame { hello = 1 }
- frame = _encode_submessage(1, hello_inner)
- result = _decode_server_frame(frame)
- assert result["type"] == "hello"
- assert isinstance(result["data"], bytes)
- # Decode the hello data
- hello = _decode_server_hello(result["data"])
- assert hello["protocolVersion"] == 1
- def test_decode_subscribed_frame(self):
- # Build an empty submessage for field 2 (subscribed)
- frame = _encode_submessage(2, b"")
- result = _decode_server_frame(frame)
- assert result["type"] == "subscribed"
- def test_decode_error_frame(self):
- # Build RealtimeError { message = "bad", code = 500 }
- error_inner = _encode_field_string(1, "bad") + _encode_field_varint(2, 500)
- # Wrap in RealtimeServerFrame { error = 5 }
- frame = _encode_submessage(5, error_inner)
- result = _decode_server_frame(frame)
- assert result["type"] == "error"
- decoded = _decode_error(result["data"])
- assert decoded["message"] == "bad"
- assert decoded["code"] == 500
- def test_decode_pong_frame(self):
- frame = _encode_submessage(7, b"")
- result = _decode_server_frame(frame)
- assert result["type"] == "pong"
- def test_decode_heartbeat_frame(self):
- frame = _encode_submessage(4, b"")
- result = _decode_server_frame(frame)
- assert result["type"] == "heartbeat"
- def test_decode_close_frame(self):
- close_inner = _encode_field_string(1, "bye")
- frame = _encode_submessage(6, close_inner)
- result = _decode_server_frame(frame)
- assert result["type"] == "close"
- def test_decode_caught_up_frame(self):
- frame = _encode_submessage(8, b"")
- result = _decode_server_frame(frame)
- assert result["type"] == "caught_up"
- def test_decode_unknown_frame(self):
- # Empty frame → no fields → unknown
- result = _decode_server_frame(b"")
- assert result["type"] == "unknown"
- assert result["data"] is None
- # ── Protobuf codec: projection event decoding ────────────────────────────
- class TestProjectionEventDecoding:
- """Test _decode_projection_event and related decoders."""
- def test_decode_projection_event_basic(self):
- # Build RealtimeProjectionEvent {
- # id = "evt1",
- # actor_id = "user1",
- # resume_cursor = "cursor1"
- # }
- proj = (
- _encode_field_string(1, "evt1")
- + _encode_field_string(3, "user1")
- + _encode_field_string(4, "cursor1")
- )
- result = _decode_projection_event(proj)
- assert result["id"] == "evt1"
- assert result["actor_id"] == "user1"
- assert result["resume_cursor"] == "cursor1"
- assert result["operations"] == []
- def test_decode_projection_event_with_timestamp(self):
- ts_inner = _encode_field_varint(1, 1700000000) # seconds
- proj = (
- _encode_field_string(1, "evt2")
- + _encode_submessage(2, ts_inner)
- )
- result = _decode_projection_event(proj)
- assert result["id"] == "evt2"
- assert "1700000000" not in result["created_at"] # should be ISO format
- assert "T" in result["created_at"] # ISO format has T separator
- def test_decode_projection_event_with_operation(self):
- # Build a room_timeline_event_upsert operation
- room_id = _encode_field_string(1, "room123")
- # Build a minimal RoomTimelineEvent
- event_inner = _encode_field_string(1, "evt456")
- event_upsert = room_id + _encode_submessage(2, event_inner)
- # Wrap in RealtimeProjectionOperation { room_timeline_event_upsert = 10 }
- op = _encode_submessage(10, event_upsert)
- # Wrap in RealtimeProjectionEvent { operations = 5 }
- proj = _encode_field_string(1, "proj1") + _encode_submessage(5, op)
- result = _decode_projection_event(proj)
- assert len(result["operations"]) == 1
- op_result = result["operations"][0]
- assert op_result["type"] == "room_timeline_event_upsert"
- assert op_result["room_id"] == "room123"
- assert op_result["event"]["id"] == "evt456"
- def test_decode_projection_operation_unknown(self):
- # An operation with field 1 (room_upsert) — not handled specifically
- op = _encode_submessage(1, _encode_field_string(1, "room1"))
- result = _decode_projection_operation(op)
- assert result["type"] == "room_upsert"
- def test_decode_projection_operation_empty(self):
- result = _decode_projection_operation(b"")
- assert result["type"] == "empty"
- # ── Protobuf codec: room timeline event decoding ─────────────────────────
- class TestRoomTimelineEventDecoding:
- """Test _decode_room_timeline_event and _decode_message."""
- def test_decode_room_timeline_event_basic(self):
- # Build RoomTimelineEvent {
- # id = "evt1", room_id = "room1", kind = 1 (message_posted)
- # }
- ts_inner = _encode_field_varint(1, 1700000000)
- event = (
- _encode_field_string(1, "evt1")
- + _encode_submessage(2, ts_inner)
- + _encode_field_string(3, "room1")
- + _encode_field_varint(4, 1)
- )
- result = _decode_room_timeline_event(event)
- assert result["id"] == "evt1"
- assert result["roomId"] == "room1"
- assert result["kind"] == 1
- assert "T" in result["createdAt"]
- def test_decode_room_timeline_event_with_message(self):
- # Build a Message { id="m1", room_id="r1", actor_id="u1", body="hello" }
- ts_inner = _encode_field_varint(1, 1700000000)
- msg = (
- _encode_field_string(1, "m1")
- + _encode_field_string(2, "r1")
- + _encode_field_string(3, "u1")
- + _encode_field_string(4, "hello")
- + _encode_submessage(5, ts_inner)
- )
- # Wrap in MessagePosted { message = 1 }
- posted = _encode_submessage(1, msg)
- # Wrap in RoomTimelineEvent { message_posted = 5 }
- event = (
- _encode_field_string(1, "evt1")
- + _encode_field_string(3, "r1")
- + _encode_submessage(5, posted)
- )
- result = _decode_room_timeline_event(event)
- assert result["id"] == "evt1"
- assert result["messagePosted"] is not None
- assert result["messagePosted"]["message"]["id"] == "m1"
- assert result["messagePosted"]["message"]["body"] == "hello"
- assert result["messagePosted"]["message"]["actorId"] == "u1"
- def test_decode_message_with_thread(self):
- # Build Message with a thread
- thread_inner = _encode_field_string(1, "thread-root-123")
- msg = (
- _encode_field_string(1, "m1")
- + _encode_field_string(2, "r1")
- + _encode_field_string(3, "u1")
- + _encode_field_string(4, "threaded reply")
- + _encode_submessage(8, thread_inner)
- )
- result = _decode_message(msg)
- assert result["id"] == "m1"
- assert result["body"] == "threaded reply"
- assert result["thread"]["threadRootEventId"] == "thread-root-123"
- def test_decode_message_with_login_and_display(self):
- msg = (
- _encode_field_string(1, "m1")
- + _encode_field_string(2, "r1")
- + _encode_field_string(3, "u1")
- + _encode_field_string(4, "hi")
- + _encode_field_string(6, "alice")
- + _encode_field_string(7, "Alice Smith")
- )
- result = _decode_message(msg)
- assert result["actorLogin"] == "alice"
- assert result["actorDisplayName"] == "Alice Smith"
- def test_decode_thread(self):
- thread = _encode_field_string(1, "thread-abc")
- result = _decode_thread(thread)
- assert result["threadRootEventId"] == "thread-abc"
- def test_decode_thread_empty(self):
- result = _decode_thread(b"")
- assert result["threadRootEventId"] == ""
- def test_decode_message_posted_empty(self):
- result = _decode_message_posted(b"")
- assert result == {"message": {}}
- # ── Protobuf codec: timestamp decoding ────────────────────────────────────
- class TestTimestampDecoding:
- """Test _decode_timestamp."""
- def test_decode_timestamp_basic(self):
- ts = _encode_field_varint(1, 1700000000)
- result = _decode_timestamp(ts)
- assert "2023" in result # Nov 14, 2023
- assert result.endswith("Z")
- def test_decode_timestamp_with_nanos(self):
- ts = _encode_field_varint(1, 1700000000) + _encode_field_varint(2, 500000)
- result = _decode_timestamp(ts)
- assert "2023" in result
- # 500000 nanos = 0.5 seconds → microsecond=500000 → ".000500" in ISO
- assert ".000500" in result
- def test_decode_timestamp_zero(self):
- ts = b""
- result = _decode_timestamp(ts)
- assert result == ""
- def test_decode_timestamp_only_nanos(self):
- # No seconds, just nanos — should return "" since seconds == 0
- ts = _encode_field_varint(2, 1000000)
- result = _decode_timestamp(ts)
- assert result == ""
- # ── Protobuf codec: event envelope (transient events) ────────────────────
- class TestEventEnvelopeDecoding:
- """Test _decode_event_envelope for mention and DM notifications."""
- def test_decode_mention_notification(self):
- # Build MentionNotification { room_id = "r1", event_id = "e1" }
- mention_inner = _encode_field_string(1, "r1") + _encode_field_string(2, "e1")
- # Wrap in RealtimeEventEnvelope { mention_notification = 88 }
- envelope = (
- _encode_field_string(1, "envelope1")
- + _encode_submessage(88, mention_inner)
- )
- result = _decode_event_envelope(envelope)
- assert result["id"] == "envelope1"
- assert result["type"] == "mention_notification"
- assert result["data"]["roomId"] == "r1"
- assert result["data"]["eventId"] == "e1"
- def test_decode_dm_notification(self):
- # Build NewDirectMessageNotification { room_id = "r2", event_id = "e2" }
- dm_inner = _encode_field_string(1, "r2") + _encode_field_string(2, "e2")
- # Wrap in RealtimeEventEnvelope { new_direct_message_notification = 89 }
- envelope = (
- _encode_field_string(1, "envelope2")
- + _encode_submessage(89, dm_inner)
- )
- result = _decode_event_envelope(envelope)
- assert result["id"] == "envelope2"
- assert result["type"] == "new_direct_message_notification"
- assert result["data"]["roomId"] == "r2"
- assert result["data"]["eventId"] == "e2"
- def test_decode_event_envelope_unknown(self):
- envelope = _encode_field_string(1, "env3")
- result = _decode_event_envelope(envelope)
- assert result["id"] == "env3"
- assert result["type"] == "unknown"
- assert result["data"] == {}
- def test_decode_mention_notification_directly(self):
- mention_inner = _encode_field_string(1, "roomX") + _encode_field_string(2, "evtX")
- result = _decode_mention_notification(mention_inner)
- assert result["roomId"] == "roomX"
- assert result["eventId"] == "evtX"
- def test_decode_dm_notification_directly(self):
- dm_inner = _encode_field_string(1, "roomY") + _encode_field_string(2, "evtY")
- result = _decode_dm_notification(dm_inner)
- assert result["roomId"] == "roomY"
- assert result["eventId"] == "evtY"
- # ── Emoji shortcode conversion ───────────────────────────────────────────
- class TestEmojiShortcode:
- """Test _emoji_to_shortcode static method and the emoji mapping."""
- @pytest.mark.parametrize(
- "emoji,shortcode",
- [
- ("👍", "thumbsup"),
- ("👎", "thumbsdown"),
- ("❤️", "heart"),
- ("❤", "heart"),
- ("✅", "white_check_mark"),
- ("❌", "x"),
- ("👀", "eyes"),
- ("🎉", "tada"),
- ("😂", "joy"),
- ("🚀", "rocket"),
- ("🔥", "fire"),
- ("💯", "100"),
- ("🤔", "thinking"),
- ("👏", "clap"),
- ("🙏", "pray"),
- ("😅", "sweat_smile"),
- ("😴", "sleeping"),
- ("⏳", "hourglass"),
- ],
- )
- def test_known_emoji_to_shortcode(self, emoji, shortcode):
- assert ChattoAdapter._emoji_to_shortcode(emoji) == shortcode
- def test_unknown_emoji_passes_through(self):
- # 🦀 (crab) is not in the mapping
- assert ChattoAdapter._emoji_to_shortcode("🦀") == "🦀"
- def test_shortcode_passes_through(self):
- assert ChattoAdapter._emoji_to_shortcode("thumbsup") == "thumbsup"
- def test_empty_string_passes_through(self):
- assert ChattoAdapter._emoji_to_shortcode("") == ""
- def test_emoji_mapping_completeness(self):
- """Verify the mapping dict has the expected keys."""
- assert "👍" in _EMOJI_TO_SHORTCODE
- assert _EMOJI_TO_SHORTCODE["👍"] == "thumbsup"
- assert _EMOJI_TO_SHORTCODE["❤️"] == "heart"
- # ── Adapter instantiation and properties ──────────────────────────────────
- class TestAdapterInstantiation:
- """Test ChattoAdapter creation and property values."""
- def test_platform_name(self):
- adapter = _make_adapter()
- assert adapter.platform_name == "chatto"
- def test_supports_markdown(self):
- adapter = _make_adapter()
- assert adapter.supports_markdown is True
- def test_supports_reactions(self):
- adapter = _make_adapter()
- assert adapter.supports_reactions is True
- def test_supports_threads(self):
- adapter = _make_adapter()
- assert adapter.supports_threads is True
- def test_max_message_length(self):
- adapter = _make_adapter()
- assert adapter.MAX_MESSAGE_LENGTH == 10000
- def test_splits_long_messages(self):
- adapter = _make_adapter()
- assert adapter.splits_long_messages is True
- def test_typing_tasks_empty(self):
- adapter = _make_adapter()
- assert adapter._typing_tasks == {}
- def test_user_cache_empty(self):
- adapter = _make_adapter()
- assert adapter._user_cache == {}
- def test_base_url_from_extra(self):
- adapter = _make_adapter()
- assert adapter._base_url == "https://chat.example.com"
- def test_channels_from_extra(self):
- adapter = _make_adapter()
- assert adapter._channel_ids == ["room1"]
- def test_require_mention_default(self):
- adapter = _make_adapter(require_mention=False)
- assert adapter._require_mention is False
- def test_is_base_platform_adapter(self):
- from gateway.platforms.base import BasePlatformAdapter
- adapter = _make_adapter()
- assert isinstance(adapter, BasePlatformAdapter)
- # ── Registration and requirements ─────────────────────────────────────────
- class TestRegistration:
- """Test plugin registration and requirements checking."""
- def test_register_calls_register_platform(self):
- ctx = _MockPluginContext()
- register(ctx)
- assert "chatto" in ctx.registered_names
- kwargs = ctx.registered_kwargs
- assert kwargs["name"] == "chatto"
- assert kwargs["label"] == "Chatto"
- assert callable(kwargs["adapter_factory"])
- assert callable(kwargs["check_fn"])
- def test_register_adapter_factory_creates_adapter(self):
- ctx = _MockPluginContext()
- register(ctx)
- kwargs = ctx.registered_kwargs
- cfg = _make_config()
- adapter = kwargs["adapter_factory"](cfg)
- assert adapter is not None
- assert isinstance(adapter, ChattoAdapter)
- def test_check_requirements_true_when_env_set(self, monkeypatch):
- monkeypatch.setenv("CHATTO_URL", "https://chat.example.com")
- monkeypatch.setenv("CHATTO_LOGIN", "user")
- monkeypatch.setenv("CHATTO_PASSWORD", "pass")
- assert check_requirements() is True
- def test_check_requirements_false_when_url_missing(self, monkeypatch):
- monkeypatch.delenv("CHATTO_URL", raising=False)
- monkeypatch.setenv("CHATTO_LOGIN", "user")
- monkeypatch.setenv("CHATTO_PASSWORD", "pass")
- assert check_requirements() is False
- def test_check_requirements_false_when_login_missing(self, monkeypatch):
- monkeypatch.setenv("CHATTO_URL", "https://chat.example.com")
- monkeypatch.delenv("CHATTO_LOGIN", raising=False)
- monkeypatch.setenv("CHATTO_PASSWORD", "pass")
- assert check_requirements() is False
- def test_check_requirements_false_when_password_missing(self, monkeypatch):
- monkeypatch.setenv("CHATTO_URL", "https://chat.example.com")
- monkeypatch.setenv("CHATTO_LOGIN", "user")
- monkeypatch.delenv("CHATTO_PASSWORD", raising=False)
- assert check_requirements() is False
- def test_check_requirements_false_all_missing(self, monkeypatch):
- monkeypatch.delenv("CHATTO_URL", raising=False)
- monkeypatch.delenv("CHATTO_LOGIN", raising=False)
- monkeypatch.delenv("CHATTO_PASSWORD", raising=False)
- assert check_requirements() is False
- def test_validate_config_with_extra_url(self):
- cfg = MagicMock()
- cfg.extra = {"url": "https://chat.example.com"}
- # Also need login/password from env
- with patch.dict(os.environ, {"CHATTO_LOGIN": "u", "CHATTO_PASSWORD": "p"}):
- assert validate_config(cfg) is True
- def test_validate_config_missing_url(self):
- cfg = MagicMock()
- cfg.extra = {}
- with patch.dict(os.environ, {"CHATTO_LOGIN": "u", "CHATTO_PASSWORD": "p"}, clear=False):
- # Remove CHATTO_URL if set
- os.environ.pop("CHATTO_URL", None)
- assert validate_config(cfg) is False
- # ── Send method (mocked RPC) ──────────────────────────────────────────────
- class TestSend:
- """Test the send() method with mocked _rpc."""
- @pytest.mark.asyncio
- async def test_send_basic(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {"message": {"id": "evt123"}}))
- result = await adapter.send("room1", "Hello world")
- assert result.success is True
- assert result.message_id == "evt123"
- @pytest.mark.asyncio
- async def test_send_with_reply_to(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {"message": {"id": "evt456"}}))
- adapter._follow_thread = AsyncMock()
- await adapter.send("room1", "Reply", reply_to="thread-root-1")
- # Verify RPC body contains threadRootEventId
- call_args = adapter._rpc.call_args
- body = call_args.kwargs.get("body") or call_args.args[1]
- assert body["threadRootEventId"] == "thread-root-1"
- @pytest.mark.asyncio
- async def test_send_with_metadata_thread_id(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {"message": {"id": "evt789"}}))
- adapter._follow_thread = AsyncMock()
- await adapter.send("room1", "Reply", metadata={"thread_id": "thread456"})
- call_args = adapter._rpc.call_args
- body = call_args.kwargs.get("body") or call_args.args[1]
- assert body["threadRootEventId"] == "thread456"
- @pytest.mark.asyncio
- async def test_send_empty_content(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock()
- result = await adapter.send("room1", "")
- assert result.success is False
- assert "Empty" in (result.error or "")
- adapter._rpc.assert_not_called()
- @pytest.mark.asyncio
- async def test_send_failure(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(500, {"error": "server error"}))
- result = await adapter.send("room1", "Hello")
- assert result.success is False
- assert "server error" in (result.error or "")
- @pytest.mark.asyncio
- async def test_send_long_message_split(self):
- adapter = _make_adapter()
- # Return different message IDs for each call
- call_count = [0]
- async def mock_rpc(path, body, **kwargs):
- call_count[0] += 1
- return 200, {"message": {"id": f"evt-{call_count[0]}"}}
- adapter._rpc = AsyncMock(side_effect=mock_rpc)
- adapter._follow_thread = AsyncMock()
- # Create content longer than MAX_MESSAGE_LENGTH (10000)
- long_content = "A" * 12000
- result = await adapter.send("room1", long_content)
- assert result.success is True
- assert result.message_id == "evt-1"
- # Should have been called multiple times
- assert adapter._rpc.call_count > 1
- @pytest.mark.asyncio
- async def test_send_marks_seen(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {"message": {"id": "new_evt"}}))
- await adapter.send("room1", "Hello")
- assert "new_evt" in adapter._seen.get("room1", {})
- # ── Reactions (mocked RPC) ────────────────────────────────────────────────
- class TestReactions:
- """Test send_reaction and remove_reaction."""
- @pytest.mark.asyncio
- async def test_send_reaction_success(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {"added": True}))
- result = await adapter.send_reaction("room1", "evt1", "👍")
- assert result is True
- # Verify emoji was converted to shortcode
- call_args = adapter._rpc.call_args
- body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
- assert body["emoji"] == "thumbsup"
- @pytest.mark.asyncio
- async def test_send_reaction_heart(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- await adapter.send_reaction("room1", "evt1", "❤️")
- call_args = adapter._rpc.call_args
- body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
- assert body["emoji"] == "heart"
- @pytest.mark.asyncio
- async def test_remove_reaction_success(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- result = await adapter.remove_reaction("room1", "evt1", "❤️")
- assert result is True
- call_args = adapter._rpc.call_args
- body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
- assert body["emoji"] == "heart"
- @pytest.mark.asyncio
- async def test_send_reaction_failure(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(500, {}))
- result = await adapter.send_reaction("room1", "evt1", "👍")
- assert result is False
- @pytest.mark.asyncio
- async def test_remove_reaction_failure(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(404, {}))
- result = await adapter.remove_reaction("room1", "evt1", "❤️")
- assert result is False
- @pytest.mark.asyncio
- async def test_send_reaction_exception(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(side_effect=Exception("network error"))
- result = await adapter.send_reaction("room1", "evt1", "👍")
- assert result is False
- # ── Message edit/delete (mocked RPC) ──────────────────────────────────────
- class TestEditDelete:
- """Test edit_message and delete_message."""
- @pytest.mark.asyncio
- async def test_edit_message_success(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- result = await adapter.edit_message("room1", "evt1", "new text")
- assert result is True
- call_args = adapter._rpc.call_args
- body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
- assert body["body"] == "new text"
- assert body["eventId"] == "evt1"
- @pytest.mark.asyncio
- async def test_edit_message_failure(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(404, {}))
- result = await adapter.edit_message("room1", "evt1", "new text")
- assert result is False
- @pytest.mark.asyncio
- async def test_delete_message_success(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- result = await adapter.delete_message("room1", "evt1")
- assert result is True
- @pytest.mark.asyncio
- async def test_delete_message_failure(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(404, {}))
- result = await adapter.delete_message("room1", "evt1")
- assert result is False
- @pytest.mark.asyncio
- async def test_edit_message_exception(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(side_effect=Exception("boom"))
- result = await adapter.edit_message("room1", "evt1", "text")
- assert result is False
- @pytest.mark.asyncio
- async def test_delete_message_exception(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(side_effect=Exception("boom"))
- result = await adapter.delete_message("room1", "evt1")
- assert result is False
- # ── Typing indicator lifecycle ─────────────────────────────────────────────
- class TestTypingIndicator:
- """Test send_typing and stop_typing lifecycle."""
- @pytest.mark.asyncio
- async def test_send_typing_creates_task(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- await adapter.send_typing("room1")
- assert "room1" in adapter._typing_tasks
- assert isinstance(adapter._typing_tasks["room1"], asyncio.Task)
- # Clean up
- await adapter.stop_typing("room1")
- @pytest.mark.asyncio
- async def test_send_typing_no_duplicate(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- await adapter.send_typing("room1")
- first_task = adapter._typing_tasks["room1"]
- await adapter.send_typing("room1")
- assert adapter._typing_tasks["room1"] is first_task
- await adapter.stop_typing("room1")
- @pytest.mark.asyncio
- async def test_stop_typing_cancels_task(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- await adapter.send_typing("room1")
- assert "room1" in adapter._typing_tasks
- await adapter.stop_typing("room1")
- assert "room1" not in adapter._typing_tasks
- @pytest.mark.asyncio
- async def test_stop_typing_when_not_running(self):
- adapter = _make_adapter()
- # Should not raise even if no task exists
- await adapter.stop_typing("room1")
- @pytest.mark.asyncio
- async def test_typing_loop_calls_rpc(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- await adapter.send_typing("room1")
- # Allow the loop to run one iteration
- await asyncio.sleep(0.05)
- # The typing loop should have called _rpc at least once
- assert adapter._rpc.call_count >= 1
- # Verify it was called with the typing path
- first_call = adapter._rpc.call_args_list[0]
- path = first_call.args[0] if len(first_call.args) > 0 else first_call.kwargs.get("path")
- assert "UpdateTypingIndicator" in path
- body = first_call.args[1] if len(first_call.args) > 1 else first_call.kwargs.get("body")
- assert body["roomId"] == "room1"
- assert body["typing"] is True
- await adapter.stop_typing("room1")
- # ── Read state and notifications (mocked RPC) ─────────────────────────────
- class TestReadStateAndNotifications:
- """Test mark_room_as_read, mark_thread_as_read, dismiss notifications."""
- @pytest.mark.asyncio
- async def test_mark_room_as_read_success(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- result = await adapter.mark_room_as_read("room1")
- assert result is True
- @pytest.mark.asyncio
- async def test_mark_room_as_read_failure(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(500, {}))
- result = await adapter.mark_room_as_read("room1")
- assert result is False
- @pytest.mark.asyncio
- async def test_mark_thread_as_read_success(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- result = await adapter.mark_thread_as_read("room1", "thread-root-1")
- assert result is True
- call_args = adapter._rpc.call_args
- body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
- assert body["roomId"] == "room1"
- assert body["threadRootEventId"] == "thread-root-1"
- @pytest.mark.asyncio
- async def test_mark_thread_as_read_failure(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(404, {}))
- result = await adapter.mark_thread_as_read("room1", "thread-root-1")
- assert result is False
- @pytest.mark.asyncio
- async def test_dismiss_all_notifications_success(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- result = await adapter.dismiss_all_notifications()
- assert result is True
- @pytest.mark.asyncio
- async def test_dismiss_all_notifications_failure(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(500, {}))
- result = await adapter.dismiss_all_notifications()
- assert result is False
- @pytest.mark.asyncio
- async def test_dismiss_notification_success(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- result = await adapter.dismiss_notification("notif123")
- assert result is True
- call_args = adapter._rpc.call_args
- body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
- assert body["notificationId"] == "notif123"
- @pytest.mark.asyncio
- async def test_dismiss_notification_failure(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(404, {}))
- result = await adapter.dismiss_notification("notif123")
- assert result is False
- # ── DM initiation and room creation (mocked RPC) ──────────────────────────
- class TestDMAndRoomCreation:
- """Test start_dm and create_room."""
- @pytest.mark.asyncio
- async def test_start_dm_success(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {"room": {"id": "room789"}}))
- result = await adapter.start_dm("user123")
- assert result == "room789"
- call_args = adapter._rpc.call_args
- body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
- assert body["participantIds"] == ["user123"]
- @pytest.mark.asyncio
- async def test_start_dm_failure(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(500, {}))
- result = await adapter.start_dm("user123")
- assert result is None
- @pytest.mark.asyncio
- async def test_start_dm_no_room_id(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {"room": {}}))
- result = await adapter.start_dm("user123")
- assert result is None
- @pytest.mark.asyncio
- async def test_start_dm_sets_room_kind(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {"room": {"id": "dm-room-1"}}))
- await adapter.start_dm("user123")
- assert adapter._room_kinds["dm-room-1"] == "ROOM_KIND_DM"
- @pytest.mark.asyncio
- async def test_create_room_success(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {"room": {"id": "room789"}}))
- result = await adapter.create_room("test-room", "description")
- assert result == "room789"
- call_args = adapter._rpc.call_args
- body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
- assert body["name"] == "test-room"
- assert body["description"] == "description"
- @pytest.mark.asyncio
- async def test_create_room_failure(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(500, {}))
- result = await adapter.create_room("test-room", "description")
- assert result is None
- @pytest.mark.asyncio
- async def test_create_room_sets_room_kind(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {"room": {"id": "group-room-1"}}))
- await adapter.create_room("test-room")
- assert adapter._room_kinds["group-room-1"] == "ROOM_KIND_GROUP"
- assert adapter._room_names["group-room-1"] == "test-room"
- # ── User lookup (mocked RPC) ──────────────────────────────────────────────
- class TestUserLookup:
- """Test get_user, list_users, batch_get_users with caching."""
- @pytest.mark.asyncio
- async def test_get_user_success(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(
- return_value=(200, {"user": {"id": "u1", "login": "alice", "displayName": "Alice"}})
- )
- result = await adapter.get_user("u1")
- assert result is not None
- assert result["id"] == "u1"
- assert result["login"] == "alice"
- @pytest.mark.asyncio
- async def test_get_user_caches(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(
- return_value=(200, {"user": {"id": "u1", "login": "alice"}})
- )
- await adapter.get_user("u1")
- # Second call should use cache — no second RPC
- await adapter.get_user("u1")
- assert adapter._rpc.call_count == 1
- assert "u1" in adapter._user_cache
- @pytest.mark.asyncio
- async def test_get_user_empty_id(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock()
- result = await adapter.get_user("")
- assert result is None
- adapter._rpc.assert_not_called()
- @pytest.mark.asyncio
- async def test_get_user_failure(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(404, {}))
- result = await adapter.get_user("u1")
- assert result is None
- @pytest.mark.asyncio
- async def test_get_user_exception(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(side_effect=Exception("boom"))
- result = await adapter.get_user("u1")
- assert result is None
- @pytest.mark.asyncio
- async def test_list_users_success(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(
- return_value=(200, {"users": [{"id": "u1"}, {"id": "u2"}]})
- )
- result = await adapter.list_users()
- assert len(result) == 2
- assert result[0]["id"] == "u1"
- # Should cache all returned users
- assert "u1" in adapter._user_cache
- assert "u2" in adapter._user_cache
- @pytest.mark.asyncio
- async def test_list_users_failure(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(500, {}))
- result = await adapter.list_users()
- assert result == []
- @pytest.mark.asyncio
- async def test_batch_get_users_success(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(
- return_value=(200, {"users": [{"id": "u1"}, {"id": "u2"}]})
- )
- result = await adapter.batch_get_users(["u1", "u2"])
- assert len(result) == 2
- # Should cache results
- assert "u1" in adapter._user_cache
- assert "u2" in adapter._user_cache
- @pytest.mark.asyncio
- async def test_batch_get_users_empty_list(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock()
- result = await adapter.batch_get_users([])
- assert result == []
- adapter._rpc.assert_not_called()
- @pytest.mark.asyncio
- async def test_batch_get_users_uses_cache(self):
- adapter = _make_adapter()
- # Pre-populate cache
- adapter._user_cache["u1"] = {"id": "u1", "login": "alice"}
- adapter._rpc = AsyncMock(
- return_value=(200, {"users": [{"id": "u2"}]})
- )
- result = await adapter.batch_get_users(["u1", "u2"])
- assert len(result) == 2
- # Should have only fetched u2 from server
- call_args = adapter._rpc.call_args
- body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
- assert body["userIds"] == ["u2"]
- @pytest.mark.asyncio
- async def test_batch_get_users_all_cached(self):
- adapter = _make_adapter()
- adapter._user_cache["u1"] = {"id": "u1"}
- adapter._rpc = AsyncMock()
- result = await adapter.batch_get_users(["u1"])
- assert len(result) == 1
- adapter._rpc.assert_not_called()
- @pytest.mark.asyncio
- async def test_batch_get_users_failure(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(500, {}))
- result = await adapter.batch_get_users(["u1"])
- assert result == []
- # ── Presence and custom status (mocked RPC) ──────────────────────────────
- class TestPresenceAndStatus:
- """Test set_presence, set_custom_status, clear_custom_status."""
- @pytest.mark.asyncio
- async def test_set_presence_online(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- result = await adapter.set_presence("online")
- assert result is True
- call_args = adapter._rpc.call_args
- body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
- assert body["status"] == 1
- @pytest.mark.asyncio
- async def test_set_presence_dnd(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- result = await adapter.set_presence("dnd")
- assert result is True
- call_args = adapter._rpc.call_args
- body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
- assert body["status"] == 3
- @pytest.mark.asyncio
- async def test_set_presence_away(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- result = await adapter.set_presence("away")
- assert result is True
- call_args = adapter._rpc.call_args
- body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
- assert body["status"] == 2
- @pytest.mark.asyncio
- async def test_set_presence_do_not_disturb(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- result = await adapter.set_presence("do_not_disturb")
- assert result is True
- call_args = adapter._rpc.call_args
- body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
- assert body["status"] == 3
- @pytest.mark.asyncio
- async def test_set_presence_unknown(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- result = await adapter.set_presence("invisible")
- assert result is False
- adapter._rpc.assert_not_called()
- @pytest.mark.asyncio
- async def test_set_presence_failure(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(500, {}))
- result = await adapter.set_presence("online")
- assert result is False
- @pytest.mark.asyncio
- async def test_set_custom_status_success(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- result = await adapter.set_custom_status("Processing...")
- assert result is True
- call_args = adapter._rpc.call_args
- body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
- assert body["status"] == "Processing..."
- @pytest.mark.asyncio
- async def test_set_custom_status_empty(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock()
- result = await adapter.set_custom_status("")
- assert result is False
- adapter._rpc.assert_not_called()
- @pytest.mark.asyncio
- async def test_set_custom_status_truncates(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- long_status = "A" * 200
- await adapter.set_custom_status(long_status)
- call_args = adapter._rpc.call_args
- body = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs.get("body")
- assert len(body["status"]) == 100
- @pytest.mark.asyncio
- async def test_set_custom_status_failure(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(500, {}))
- result = await adapter.set_custom_status("test")
- assert result is False
- @pytest.mark.asyncio
- async def test_clear_custom_status_success(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- result = await adapter.clear_custom_status()
- assert result is True
- @pytest.mark.asyncio
- async def test_clear_custom_status_failure(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(500, {}))
- result = await adapter.clear_custom_status()
- assert result is False
- # ── Message dispatch (mocked) ─────────────────────────────────────────────
- class TestMessageDispatch:
- """Test _dispatch_message with mocked handler.
- The base class ``handle_message`` does complex session management and
- spawns background tasks, so we mock it to verify _dispatch_message calls
- it with the right MessageEvent.
- """
- @pytest.mark.asyncio
- async def test_dispatch_message_calls_handler(self):
- adapter = _make_adapter()
- adapter._message_handler = AsyncMock()
- adapter.handle_message = AsyncMock()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- adapter._room_names["room1"] = "General"
- adapter._room_kinds["room1"] = "ROOM_KIND_GROUP"
- adapter._require_mention = False # don't require mention for this test
- msg = {
- "id": "evt1",
- "roomId": "room1",
- "actorId": "user1",
- "body": "Hello bot",
- "createdAt": "",
- "actorLogin": "alice",
- "actorDisplayName": "Alice",
- "thread": {},
- }
- await adapter._dispatch_message(msg, "room1")
- assert adapter.handle_message.called
- event = adapter.handle_message.call_args.args[0]
- assert event.text == "Hello bot"
- assert event.message_id == "evt1"
- @pytest.mark.asyncio
- async def test_dispatch_message_self_echo_suppressed(self):
- adapter = _make_adapter()
- adapter._message_handler = AsyncMock()
- adapter.handle_message = AsyncMock()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- msg = {
- "id": "evt1",
- "roomId": "room1",
- "actorId": "bot-user-id", # matches adapter._user_id
- "body": "My own message",
- "createdAt": "",
- "actorLogin": "hermes_bot",
- "actorDisplayName": "Hermes Bot",
- "thread": {},
- }
- await adapter._dispatch_message(msg, "room1")
- adapter.handle_message.assert_not_called()
- @pytest.mark.asyncio
- async def test_dispatch_message_empty_body_skipped(self):
- adapter = _make_adapter()
- adapter._message_handler = AsyncMock()
- adapter.handle_message = AsyncMock()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- msg = {
- "id": "evt1",
- "roomId": "room1",
- "actorId": "user1",
- "body": "",
- "createdAt": "",
- "thread": {},
- }
- await adapter._dispatch_message(msg, "room1")
- adapter.handle_message.assert_not_called()
- @pytest.mark.asyncio
- async def test_dispatch_message_no_handler(self):
- adapter = _make_adapter()
- adapter._message_handler = None
- adapter._rpc = AsyncMock(return_value=(200, {}))
- # Should not raise
- msg = {
- "id": "evt1",
- "roomId": "room1",
- "actorId": "user1",
- "body": "Hello",
- "createdAt": "",
- "thread": {},
- }
- await adapter._dispatch_message(msg, "room1")
- @pytest.mark.asyncio
- async def test_dispatch_message_mark_room_read_called(self):
- adapter = _make_adapter()
- adapter._message_handler = AsyncMock()
- adapter.handle_message = AsyncMock()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- adapter._room_kinds["room1"] = "ROOM_KIND_GROUP"
- adapter._require_mention = False # don't require mention for this test
- msg = {
- "id": "evt1",
- "roomId": "room1",
- "actorId": "user1",
- "body": "Hello",
- "createdAt": "",
- "actorLogin": "alice",
- "thread": {},
- }
- await adapter._dispatch_message(msg, "room1")
- # mark_room_as_read and dismiss_all_notifications should have been called
- # via _rpc — check for MarkRoomAsRead and DismissAllNotifications paths
- rpc_paths = [c.args[0] for c in adapter._rpc.call_args_list if len(c.args) > 0]
- assert any("MarkRoomAsRead" in p for p in rpc_paths)
- assert any("DismissAllNotifications" in p for p in rpc_paths)
- @pytest.mark.asyncio
- async def test_dispatch_message_dm_always_responds(self):
- adapter = _make_adapter()
- adapter._message_handler = AsyncMock()
- adapter.handle_message = AsyncMock()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- adapter._room_kinds["room1"] = "ROOM_KIND_DM"
- adapter._require_mention = True # even with require_mention, DMs respond
- msg = {
- "id": "evt1",
- "roomId": "room1",
- "actorId": "user1",
- "body": "Hello without mention",
- "createdAt": "",
- "actorLogin": "alice",
- "thread": {},
- }
- await adapter._dispatch_message(msg, "room1")
- assert adapter.handle_message.called
- @pytest.mark.asyncio
- async def test_dispatch_message_require_mention_no_mention_skipped(self):
- adapter = _make_adapter()
- adapter._message_handler = AsyncMock()
- adapter.handle_message = AsyncMock()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- adapter._room_kinds["room1"] = "ROOM_KIND_GROUP"
- adapter._require_mention = True
- msg = {
- "id": "evt1",
- "roomId": "room1",
- "actorId": "user1",
- "body": "Hello without mention",
- "createdAt": "",
- "actorLogin": "alice",
- "thread": {},
- }
- await adapter._dispatch_message(msg, "room1")
- adapter.handle_message.assert_not_called()
- @pytest.mark.asyncio
- async def test_dispatch_message_with_mention(self):
- adapter = _make_adapter()
- adapter._message_handler = AsyncMock()
- adapter.handle_message = AsyncMock()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- adapter._room_kinds["room1"] = "ROOM_KIND_GROUP"
- adapter._require_mention = True
- msg = {
- "id": "evt1",
- "roomId": "room1",
- "actorId": "user1",
- "body": "@hermes_bot do something",
- "createdAt": "",
- "actorLogin": "alice",
- "thread": {},
- }
- await adapter._dispatch_message(msg, "room1")
- assert adapter.handle_message.called
- @pytest.mark.asyncio
- async def test_dispatch_message_with_thread(self):
- adapter = _make_adapter()
- adapter._message_handler = AsyncMock()
- adapter.handle_message = AsyncMock()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- adapter._room_kinds["room1"] = "ROOM_KIND_GROUP"
- adapter._require_mention = False # don't require mention for this test
- msg = {
- "id": "evt1",
- "roomId": "room1",
- "actorId": "user1",
- "body": "Hello",
- "createdAt": "",
- "actorLogin": "alice",
- "thread": {"threadRootEventId": "thread-root-123"},
- }
- await adapter._dispatch_message(msg, "room1")
- assert adapter.handle_message.called
- # Verify the event was created with the thread_id
- event = adapter.handle_message.call_args.args[0]
- assert event.source.thread_id == "thread-root-123"
- @pytest.mark.asyncio
- async def test_dispatch_message_strips_mention_prefix(self):
- adapter = _make_adapter()
- adapter._message_handler = AsyncMock()
- adapter.handle_message = AsyncMock()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- adapter._room_kinds["room1"] = "ROOM_KIND_GROUP"
- adapter._require_mention = True
- msg = {
- "id": "evt1",
- "roomId": "room1",
- "actorId": "user1",
- "body": "@hermes_bot please help",
- "createdAt": "",
- "actorLogin": "alice",
- "thread": {},
- }
- await adapter._dispatch_message(msg, "room1")
- event = adapter.handle_message.call_args.args[0]
- # The mention prefix should be stripped
- assert not event.text.startswith("@hermes_bot")
- assert "please help" in event.text
- # ── Processing lifecycle hooks ─────────────────────────────────────────────
- class TestProcessingLifecycle:
- """Test on_processing_start and on_processing_complete reaction hooks."""
- @pytest.mark.asyncio
- async def test_on_processing_start_adds_eyes(self):
- adapter = _make_adapter()
- adapter.send_reaction = AsyncMock(return_value=True)
- event = MagicMock()
- event.message_id = "evt1"
- event.source = MagicMock()
- event.source.chat_id = "room1"
- event.raw_message = {}
- await adapter.on_processing_start(event)
- adapter.send_reaction.assert_called_once_with("room1", "evt1", "👀")
- @pytest.mark.asyncio
- async def test_on_processing_complete_success(self):
- from gateway.platforms.base import ProcessingOutcome
- adapter = _make_adapter()
- adapter.send_reaction = AsyncMock(return_value=True)
- adapter.remove_reaction = AsyncMock(return_value=True)
- event = MagicMock()
- event.message_id = "evt1"
- event.source = MagicMock()
- event.source.chat_id = "room1"
- event.raw_message = {}
- await adapter.on_processing_complete(event, ProcessingOutcome.SUCCESS)
- adapter.remove_reaction.assert_called_once_with("room1", "evt1", "👀")
- adapter.send_reaction.assert_called_once_with("room1", "evt1", "✅")
- @pytest.mark.asyncio
- async def test_on_processing_complete_failure(self):
- from gateway.platforms.base import ProcessingOutcome
- adapter = _make_adapter()
- adapter.send_reaction = AsyncMock(return_value=True)
- adapter.remove_reaction = AsyncMock(return_value=True)
- event = MagicMock()
- event.message_id = "evt1"
- event.source = MagicMock()
- event.source.chat_id = "room1"
- event.raw_message = {}
- await adapter.on_processing_complete(event, ProcessingOutcome.FAILURE)
- adapter.send_reaction.assert_called_once_with("room1", "evt1", "❌")
- @pytest.mark.asyncio
- async def test_reactions_disabled(self, monkeypatch):
- adapter = _make_adapter()
- monkeypatch.setenv("CHATTO_REACTIONS", "false")
- adapter.send_reaction = AsyncMock()
- event = MagicMock()
- event.message_id = "evt1"
- event.source = MagicMock()
- event.source.chat_id = "room1"
- event.raw_message = {}
- await adapter.on_processing_start(event)
- adapter.send_reaction.assert_not_called()
- # ── Attachment upload (mocked RPC) ────────────────────────────────────────
- class TestAttachmentUpload:
- """Test the chunked asset upload flow."""
- @pytest.mark.asyncio
- async def test_upload_asset_success(self):
- adapter = _make_adapter()
- # Create a temporary file
- with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as f:
- f.write(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
- tmp_path = f.name
- try:
- rpc_responses = [
- (200, {"upload": {"id": "upload-1"}}), # CreateUpload
- (200, {}), # UploadChunk
- (200, {"asset": {"id": "asset-1"}}), # CompleteUpload
- ]
- adapter._rpc = AsyncMock(side_effect=rpc_responses)
- asset_id = await adapter._upload_asset("room1", tmp_path)
- assert asset_id == "asset-1"
- assert adapter._rpc.call_count == 3
- # Verify the first call was CreateUpload
- first_call = adapter._rpc.call_args_list[0]
- path = first_call.args[0]
- assert "CreateUpload" in path
- body = first_call.args[1]
- assert body["roomId"] == "room1"
- assert "sha256" in body
- # Verify last call was CompleteUpload
- last_call = adapter._rpc.call_args_list[-1]
- path = last_call.args[0]
- assert "CompleteUpload" in path
- finally:
- os.unlink(tmp_path)
- @pytest.mark.asyncio
- async def test_upload_asset_create_fails(self):
- adapter = _make_adapter()
- with tempfile.NamedTemporaryFile(delete=False, suffix=".txt") as f:
- f.write(b"test data")
- tmp_path = f.name
- try:
- adapter._rpc = AsyncMock(return_value=(500, {"error": "fail"}))
- result = await adapter._upload_asset("room1", tmp_path)
- assert result is None
- finally:
- os.unlink(tmp_path)
- @pytest.mark.asyncio
- async def test_upload_asset_empty_file(self):
- adapter = _make_adapter()
- with tempfile.NamedTemporaryFile(delete=False, suffix=".txt") as f:
- f.write(b"")
- tmp_path = f.name
- try:
- adapter._rpc = AsyncMock()
- result = await adapter._upload_asset("room1", tmp_path)
- assert result is None
- adapter._rpc.assert_not_called()
- finally:
- os.unlink(tmp_path)
- @pytest.mark.asyncio
- async def test_send_image_file_success(self):
- adapter = _make_adapter()
- with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as f:
- f.write(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
- tmp_path = f.name
- try:
- rpc_responses = [
- (200, {"upload": {"id": "upload-1"}}), # CreateUpload
- (200, {}), # UploadChunk
- (200, {"asset": {"id": "asset-1"}}), # CompleteUpload
- (200, {"message": {"id": "msg-1"}}), # CreateMessage
- ]
- adapter._rpc = AsyncMock(side_effect=rpc_responses)
- result = await adapter.send_image_file("room1", tmp_path, caption="Test image")
- assert result.success is True
- assert result.message_id == "msg-1"
- # Verify the CreateMessage call had attachmentAssetIds
- create_msg_call = adapter._rpc.call_args_list[-1]
- body = create_msg_call.args[1]
- assert body["attachmentAssetIds"] == ["asset-1"]
- assert body["body"] == "Test image"
- finally:
- os.unlink(tmp_path)
- @pytest.mark.asyncio
- async def test_send_image_file_upload_fails_fallback(self):
- adapter = _make_adapter()
- with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as f:
- f.write(b"\x89PNG" + b"\x00" * 100)
- tmp_path = f.name
- try:
- # Upload fails, then fallback send succeeds
- rpc_responses = [
- (500, {"error": "fail"}), # CreateUpload fails
- (200, {"message": {"id": "fallback-msg"}}), # Fallback send
- ]
- adapter._rpc = AsyncMock(side_effect=rpc_responses)
- result = await adapter.send_image_file("room1", tmp_path)
- assert result.success is True
- assert result.message_id == "fallback-msg"
- finally:
- os.unlink(tmp_path)
- # ── Seen tracking ─────────────────────────────────────────────────────────
- class TestSeenTracking:
- """Test _mark_seen, _is_seen, and the _SEEN_CAP."""
- def test_mark_and_check_seen(self):
- adapter = _make_adapter()
- adapter._mark_seen("room1", "evt1")
- assert adapter._is_seen("room1", "evt1") is True
- assert adapter._is_seen("room1", "evt2") is False
- assert adapter._is_seen("room2", "evt1") is False
- def test_mark_seen_evicts_old_beyond_cap(self):
- adapter = _make_adapter()
- # Add more than _SEEN_CAP events
- for i in range(_SEEN_CAP + 10):
- adapter._mark_seen("room1", f"evt{i}")
- # The first events should have been evicted
- assert adapter._is_seen("room1", "evt0") is False
- # The most recent should still be there
- assert adapter._is_seen("room1", f"evt{_SEEN_CAP + 9}") is True
- # Total should not exceed cap
- assert len(adapter._seen["room1"]) <= _SEEN_CAP
- # ── WebSocket URL building ────────────────────────────────────────────────
- class TestWebSocketURL:
- """Test _websocket_url conversion."""
- def test_https_to_wss(self):
- adapter = _make_adapter()
- adapter._base_url = "https://chat.example.com"
- url = adapter._websocket_url()
- assert url.startswith("wss://")
- assert "/api/realtime" in url
- def test_http_to_ws(self):
- adapter = _make_adapter()
- adapter._base_url = "http://localhost:8080"
- url = adapter._websocket_url()
- assert url.startswith("ws://")
- assert "/api/realtime" in url
- def test_websocket_url_with_path(self):
- adapter = _make_adapter()
- adapter._base_url = "https://chat.example.com/subpath"
- url = adapter._websocket_url()
- assert "/subpath/api/realtime" in url
- def test_websocket_url_invalid_scheme(self):
- adapter = _make_adapter()
- adapter._base_url = "ftp://chat.example.com"
- with pytest.raises(ValueError, match="must use http"):
- adapter._websocket_url()
- # ── Get chat info ─────────────────────────────────────────────────────────
- class TestGetChatInfo:
- """Test get_chat_info."""
- @pytest.mark.asyncio
- async def test_get_chat_info_group(self):
- adapter = _make_adapter()
- adapter._room_names["room1"] = "General"
- adapter._room_kinds["room1"] = "ROOM_KIND_GROUP"
- info = await adapter.get_chat_info("room1")
- assert info["name"] == "General"
- assert info["type"] == "group"
- @pytest.mark.asyncio
- async def test_get_chat_info_dm(self):
- adapter = _make_adapter()
- adapter._room_names["room2"] = "Alice"
- adapter._room_kinds["room2"] = "ROOM_KIND_DM"
- info = await adapter.get_chat_info("room2")
- assert info["name"] == "Alice"
- assert info["type"] == "dm"
- @pytest.mark.asyncio
- async def test_get_chat_info_unknown_room(self):
- adapter = _make_adapter()
- info = await adapter.get_chat_info("unknown-room")
- assert info["name"] == "unknown-room"
- assert info["type"] == "group" # default
- # ── Handle projection event ──────────────────────────────────────────────
- class TestHandleProjectionEvent:
- """Test _handle_projection_event and _handle_timeline_event_upsert."""
- @pytest.mark.asyncio
- async def test_handle_projection_event_updates_cursor(self):
- adapter = _make_adapter()
- # Build a projection event with a resume cursor
- proj = _encode_field_string(4, "new-cursor-123")
- await adapter._handle_projection_event(proj)
- assert adapter._resume_cursor == "new-cursor-123"
- @pytest.mark.asyncio
- async def test_handle_projection_event_empty_data(self):
- adapter = _make_adapter()
- await adapter._handle_projection_event(b"")
- # Should not raise, cursor unchanged
- @pytest.mark.asyncio
- async def test_handle_timeline_event_dedup(self):
- adapter = _make_adapter()
- adapter._message_handler = AsyncMock()
- adapter.handle_message = AsyncMock()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- adapter._room_kinds["room1"] = "ROOM_KIND_GROUP"
- adapter._require_mention = False # don't require mention for this test
- op = {
- "room_id": "room1",
- "event": {
- "id": "evt-dedup",
- "messagePosted": {
- "message": {
- "id": "m1",
- "roomId": "room1",
- "actorId": "user1",
- "body": "Hello",
- "createdAt": "",
- "thread": {},
- }
- },
- },
- }
- # First call should dispatch
- await adapter._handle_timeline_event_upsert(op)
- assert adapter.handle_message.called
- # Reset mock
- adapter.handle_message.reset_mock()
- # Second call should be deduped
- await adapter._handle_timeline_event_upsert(op)
- adapter.handle_message.assert_not_called()
- @pytest.mark.asyncio
- async def test_handle_timeline_event_no_message_posted(self):
- adapter = _make_adapter()
- adapter._message_handler = AsyncMock()
- op = {
- "room_id": "room1",
- "event": {
- "id": "evt1",
- "messagePosted": None,
- },
- }
- await adapter._handle_timeline_event_upsert(op)
- adapter._message_handler.assert_not_called()
- @pytest.mark.asyncio
- async def test_handle_timeline_event_empty_event(self):
- adapter = _make_adapter()
- adapter._message_handler = AsyncMock()
- op = {"room_id": "room1", "event": {}}
- await adapter._handle_timeline_event_upsert(op)
- adapter._message_handler.assert_not_called()
- # ── Handle transient event ────────────────────────────────────────────────
- class TestHandleTransientEvent:
- """Test _handle_transient_event."""
- @pytest.mark.asyncio
- async def test_handle_transient_mention(self):
- adapter = _make_adapter()
- # Build a transient event envelope with a mention notification
- mention_inner = _encode_field_string(1, "room1") + _encode_field_string(2, "evt1")
- envelope = _encode_field_string(1, "env1") + _encode_submessage(88, mention_inner)
- # Should not raise
- await adapter._handle_transient_event(envelope)
- @pytest.mark.asyncio
- async def test_handle_transient_dm(self):
- adapter = _make_adapter()
- dm_inner = _encode_field_string(1, "room2") + _encode_field_string(2, "evt2")
- envelope = _encode_field_string(1, "env2") + _encode_submessage(89, dm_inner)
- await adapter._handle_transient_event(envelope)
- @pytest.mark.asyncio
- async def test_handle_transient_empty(self):
- adapter = _make_adapter()
- await adapter._handle_transient_event(b"")
- # ── Disconnect ────────────────────────────────────────────────────────────
- class TestDisconnect:
- """Test disconnect cleanup."""
- @pytest.mark.asyncio
- async def test_disconnect_cancels_typing_tasks(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- adapter._ws_active = False # avoid websocket teardown
- adapter._liveness_task = None
- adapter._ws_task = None
- await adapter.send_typing("room1")
- assert "room1" in adapter._typing_tasks
- await adapter.disconnect()
- assert "room1" not in adapter._typing_tasks
- @pytest.mark.asyncio
- async def test_disconnect_clears_token(self):
- adapter = _make_adapter()
- adapter._rpc = AsyncMock(return_value=(200, {}))
- adapter._ws_active = False
- adapter._liveness_task = None
- adapter._ws_task = None
- assert adapter._token is not None
- await adapter.disconnect()
- assert adapter._token is None
|