6krrt/tests/test_conversation_identity.py

"""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