"""Tests for the pure conversation-identity resolver. All tests are offline; no provider, local-model, or network calls are made. The duplicate-header tests use a real ``starlette.Headers`` to prove the module behaves identically across both header container shapes. """ from starlette.datastructures import Headers from conversation_identity import ( ConversationIdentity, parse_conversation_id, resolve_identity, ) class TestParseConversationId: def test_valid_id_returns_value(self) -> None: assert parse_conversation_id("abc-123._:XYZ") == "abc-123._:XYZ" def test_valid_single_char(self) -> None: assert parse_conversation_id("a") == "a" def test_valid_max_length_128(self) -> None: value = "a" * 128 assert parse_conversation_id(value) == value def test_129_char_id_ignored(self) -> None: assert parse_conversation_id("a" * 129) is None def test_space_ignored(self) -> None: assert parse_conversation_id("abc def") is None def test_slash_ignored(self) -> None: assert parse_conversation_id("abc/def") is None def test_newline_ignored(self) -> None: assert parse_conversation_id("abc\ndef") is None def test_empty_string_ignored(self) -> None: assert parse_conversation_id("") is None def test_non_string_ignored(self) -> None: assert parse_conversation_id(123) is None assert parse_conversation_id(None) is None assert parse_conversation_id(b"abc") is None class TestResolveIdentitySessionKey: def test_valid_conversation_gives_c_prefixed_key(self) -> None: ident = resolve_identity({"X-Router-Conversation": "conv1"}, "fp1234567890abcdef") assert ident.session_key == "c:conv1" def test_uppercase_header_name_accepted(self) -> None: ident = resolve_identity({"X-ROUTER-CONVERSATION": "conv1"}, None) assert ident.session_key == "c:conv1" def test_header_with_any_case_accepted(self) -> None: for name in ( "x-router-conversation", "X-Router-Conversation", "X-ROUTER-CONVERSATION", ): ident = resolve_identity({name: "conv1"}, None) assert ident.session_key == "c:conv1" def test_fingerprint_fallback_returned_untouched(self) -> None: fp = "0123456789abcdef" ident = resolve_identity({}, fp) assert ident.session_key is fp def test_invalid_conversation_falls_back_to_fingerprint(self) -> None: fp = "0123456789abcdef" ident = resolve_identity({"X-Router-Conversation": "bad value"}, fp) assert ident.session_key is fp def test_fingerprint_shaped_value_still_gets_c_prefix(self) -> None: # A 16-hex value passed as a conversation id is a namespace check: it # must get the c: prefix so it cannot collide with an unfingerprinted # session key. ident = resolve_identity({"X-Router-Conversation": "0123456789abcdef"}, None) assert ident.session_key == "c:0123456789abcdef" def test_fingerprint_none_when_no_conversation_and_no_fingerprint(self) -> None: ident = resolve_identity({}, None) assert ident.session_key is None class TestResolveIdentityAgent: def test_valid_agent(self) -> None: ident = resolve_identity({"X-Router-Agent": "my-agent_2"}, None) assert ident.agent == "my-agent_2" def test_agent_absent(self) -> None: ident = resolve_identity({}, None) assert ident.agent is None def test_agent_65_chars_ignored(self) -> None: ident = resolve_identity({"X-Router-Agent": "a" * 65}, None) assert ident.agent is None def test_agent_64_chars_accepted(self) -> None: ident = resolve_identity({"X-Router-Agent": "a" * 64}, None) assert ident.agent == "a" * 64 def test_agent_with_disallowed_char_ignored(self) -> None: ident = resolve_identity({"X-Router-Agent": "my agent"}, None) assert ident.agent is None class TestResolveIdentityParent: def test_valid_parent_gives_c_prefixed_key(self) -> None: ident = resolve_identity({"X-Router-Parent": "parent1"}, None) assert ident.parent_key == "c:parent1" def test_parent_absent(self) -> None: ident = resolve_identity({}, None) assert ident.parent_key is None def test_invalid_parent_ignored(self) -> None: ident = resolve_identity({"X-Router-Parent": "has space"}, None) assert ident.parent_key is None class TestDuplicateHeaders: def test_starlette_headers_first_wins(self) -> None: # starlette.Headers yields lowercased names and .get() returns the # first value; the module must read the first occurrence. headers = Headers( { "X-Router-Conversation": "first", "x-router-conversation": "second", } ) ident = resolve_identity(headers, None) assert ident.session_key == "c:first" def test_plain_dict_lowercase_collision_first_wins(self) -> None: headers = { "X-Router-Conversation": "A", "x-router-conversation": "B", } ident = resolve_identity(headers, None) assert ident.session_key == "c:A" def test_starlette_headers_invalid_first_not_replaced_by_second(self) -> None: # The FIRST occurrence wins even when its value is invalid: a later, # valid occurrence must not be adopted. headers = Headers( { "X-Router-Conversation": "bad value", "x-router-conversation": "good", } ) fp = "0123456789abcdef" ident = resolve_identity(headers, fp) assert ident.session_key is fp def test_plain_dict_invalid_first_wins_over_valid_second(self) -> None: headers = { "X-Router-Conversation": "bad value", "x-router-conversation": "good", } fp = "0123456789abcdef" ident = resolve_identity(headers, fp) assert ident.session_key is fp def test_agent_invalid_first_not_replaced_by_second(self) -> None: headers = Headers( { "X-Router-Agent": "has space", "x-router-agent": "good-agent", } ) ident = resolve_identity(headers, None) assert ident.agent is None class TestDataclassShape: def test_exposes_exactly_expected_fields(self) -> None: assert list(ConversationIdentity.__dataclass_fields__.keys()) == [ "session_key", "agent", "parent_key", ] def test_is_frozen(self) -> None: assert ConversationIdentity.__dataclass_params__.frozen is True