tests/test_client.py

"""Tests for ``kinchat.client.KinChatEnv``.

The HTTP layer is mocked end-to-end — these tests must NOT spin up the
real FastAPI app, because that would defeat the point of asserting
client/server separation.

Mocking strategy: patch ``requests.Session.get`` / ``requests.Session.post``
on the bound session attached to each ``KinChatEnv`` instance. We construct
a tiny ``_FakeResponse`` to mimic the parts of ``requests.Response`` we use.
"""
from __future__ import annotations

import inspect
import sys
from typing import Any
from unittest.mock import MagicMock, patch

import pytest

from kinchat import client as client_module
from kinchat.client import KinChatEnv, KinChatHTTPError
from kinchat.models import (
    ChatMsg,
    KinChatAction,
    KinChatObservation,
    KinChatState,
    PersonaState,
)


# --------------------------------------------------------------------------- #
# Fake response helper                                                        #
# --------------------------------------------------------------------------- #


class _FakeResponse:
    """Minimal stand-in for ``requests.Response`` covering what _check uses."""

    def __init__(
        self,
        json_body: Any = None,
        status_code: int = 200,
        url: str = "http://fake/",
        text: str = "",
    ) -> None:
        self._json = json_body
        self.status_code = status_code
        self.url = url
        self.text = text or (str(json_body) if json_body is not None else "")
        self.content = b"" if json_body is None else b"{}"
        self.reason = "OK" if status_code < 400 else "ERROR"

    def json(self) -> Any:
        if self._json is None:
            raise ValueError("no json body")
        return self._json


def _obs_payload(**overrides: Any) -> dict[str, Any]:
    base: dict[str, Any] = {
        "chat_history": [],
        "user_draft": "",
        "active_recipients": [],
        "turn_index": 0,
        "scenario_brief": "test brief",
        "reward": 0.5,
        "reward_breakdown": {},
        "feedback": "",
        "done": False,
        "episode_id": "ep-1",
    }
    base.update(overrides)
    return base


def _state_payload(**overrides: Any) -> dict[str, Any]:
    base: dict[str, Any] = {
        "session_id": "sess-1",
        "episode_index": 0,
        "scenario_id": "secret_grandma_surgery_01",
        "family": {
            "mom": PersonaState(persona_id="mom").model_dump(),
        },
        "secrets": [],
        "chat_log": [],
        "turn_index": 0,
        "scores_history": [],
    }
    base.update(overrides)
    return base


# --------------------------------------------------------------------------- #
# Construction & URL normalization                                            #
# --------------------------------------------------------------------------- #


def test_construction_does_not_raise():
    KinChatEnv("http://localhost:7860")


def test_construction_does_not_perform_network_call():
    """No HTTP traffic during __init__."""
    with patch("requests.Session.get") as g, patch("requests.Session.post") as p:
        KinChatEnv("http://localhost:7860")
        assert g.call_count == 0
        assert p.call_count == 0


def test_empty_base_url_raises():
    with pytest.raises(ValueError):
        KinChatEnv("")


def test_trailing_slash_normalized_for_get():
    e_slash = KinChatEnv("http://x:7860/")
    e_plain = KinChatEnv("http://x:7860")
    assert e_slash.base_url == e_plain.base_url == "http://x:7860"

    captured: list[str] = []

    def fake_get(self, url, *args, **kwargs):  # noqa: ARG001
        captured.append(url)
        return _FakeResponse({"status": "ok", "version": "0.1.0"})

    with patch("requests.Session.get", new=fake_get):
        e_slash.health()
        e_plain.health()

    assert captured == ["http://x:7860/health", "http://x:7860/health"]


def test_multiple_trailing_slashes_normalized():
    e = KinChatEnv("http://x:7860///")
    assert e.base_url == "http://x:7860"


# --------------------------------------------------------------------------- #
# health / list_scenarios / dashboard_data                                    #
# --------------------------------------------------------------------------- #


def test_health_calls_get_health():
    e = KinChatEnv("http://localhost:7860")
    fake = _FakeResponse({"status": "ok", "version": "0.1.0"})
    with patch.object(e._session, "get", return_value=fake) as m:
        out = e.health()
    assert out == {"status": "ok", "version": "0.1.0"}
    assert m.call_count == 1
    args, kwargs = m.call_args
    assert args[0] == "http://localhost:7860/health"
    assert "timeout" in kwargs


def test_list_scenarios_calls_get_scenarios():
    e = KinChatEnv("http://localhost:7860")
    body = {
        "archetypes": ["secret_keeping"],
        "scenarios": [
            {
                "id": "secret_grandma_surgery_01",
                "archetype": "secret_keeping",
                "brief": "Grandma's surgery",
            }
        ],
    }
    fake = _FakeResponse(body)
    with patch.object(e._session, "get", return_value=fake) as m:
        out = e.list_scenarios()
    assert out == body
    assert m.call_args.args[0] == "http://localhost:7860/scenarios"


def test_dashboard_data_calls_get_dashboard_data():
    e = KinChatEnv("http://localhost:7860")
    body = {"session_id": None, "episode_index": 0, "leak_events": []}
    fake = _FakeResponse(body)
    with patch.object(e._session, "get", return_value=fake) as m:
        out = e.dashboard_data()
    assert out == body
    assert m.call_args.args[0] == "http://localhost:7860/dashboard/data"


# --------------------------------------------------------------------------- #
# reset                                                                       #
# --------------------------------------------------------------------------- #


def test_reset_posts_scenario_id_and_returns_observation():
    e = KinChatEnv("http://localhost:7860")
    fake = _FakeResponse(_obs_payload(scenario_brief="grandma's surgery"))

    with patch.object(e._session, "post", return_value=fake) as m:
        obs = e.reset(scenario_id="secret_grandma_surgery_01")

    assert isinstance(obs, KinChatObservation)
    assert obs.scenario_brief == "grandma's surgery"

    assert m.call_args.args[0] == "http://localhost:7860/reset"
    body = m.call_args.kwargs["json"]
    assert body["scenario_id"] == "secret_grandma_surgery_01"
    # Optional fields omitted when None
    assert "session_id" not in body
    assert "seed" not in body
    assert "episode_id" not in body


def test_reset_threads_session_id_and_seed_into_body():
    e = KinChatEnv("http://localhost:7860")
    fake = _FakeResponse(_obs_payload())

    with patch.object(e._session, "post", return_value=fake) as m:
        e.reset(
            scenario_id="secret_grandma_surgery_01",
            session_id="sess-42",
            seed=123,
            episode_id="ep-9",
        )

    body = m.call_args.kwargs["json"]
    assert body == {
        "scenario_id": "secret_grandma_surgery_01",
        "session_id": "sess-42",
        "seed": 123,
        "episode_id": "ep-9",
    }


# --------------------------------------------------------------------------- #
# step                                                                        #
# --------------------------------------------------------------------------- #


def test_step_posts_action_body_and_returns_observation():
    e = KinChatEnv("http://localhost:7860")
    fake = _FakeResponse(
        _obs_payload(
            turn_index=1,
            reward=0.42,
            chat_history=[
                ChatMsg(sender="mom", text="hi", recipients=["user"], turn=1).model_dump()
            ],
        )
    )

    action = KinChatAction(action_type="stay_silent")
    with patch.object(e._session, "post", return_value=fake) as m:
        obs = e.step(action)

    assert isinstance(obs, KinChatObservation)
    assert obs.turn_index == 1
    assert obs.reward == pytest.approx(0.42)

    assert m.call_args.args[0] == "http://localhost:7860/step"
    body = m.call_args.kwargs["json"]
    assert body["action_type"] == "stay_silent"
    assert body["message"] == ""


def test_step_with_send_action_carries_message_and_recipients():
    e = KinChatEnv("http://localhost:7860")
    fake = _FakeResponse(_obs_payload(turn_index=2))

    action = KinChatAction(
        action_type="send",
        message="hey mom",
        recipients=["mom"],
        reasoning="be nice",
    )
    with patch.object(e._session, "post", return_value=fake) as m:
        e.step(action)

    body = m.call_args.kwargs["json"]
    assert body["action_type"] == "send"
    assert body["message"] == "hey mom"
    assert body["recipients"] == ["mom"]
    assert body["reasoning"] == "be nice"


def test_step_rejects_non_action_arguments():
    e = KinChatEnv("http://localhost:7860")
    with pytest.raises(TypeError):
        e.step({"action_type": "stay_silent"})  # type: ignore[arg-type]


# --------------------------------------------------------------------------- #
# state                                                                       #
# --------------------------------------------------------------------------- #


def test_state_returns_kinchat_state():
    e = KinChatEnv("http://localhost:7860")
    fake = _FakeResponse(_state_payload())
    with patch.object(e._session, "get", return_value=fake) as m:
        s = e.state()

    assert isinstance(s, KinChatState)
    assert s.session_id == "sess-1"
    assert "mom" in s.family
    assert m.call_args.args[0] == "http://localhost:7860/state"


# --------------------------------------------------------------------------- #
# Error handling                                                              #
# --------------------------------------------------------------------------- #


def test_4xx_response_raises_kinchat_http_error():
    e = KinChatEnv("http://localhost:7860")
    fake = _FakeResponse(
        json_body={"detail": "scenario 'nope' not found"},
        status_code=404,
        url="http://localhost:7860/reset",
    )
    fake.content = b'{"detail": "..."}'

    with patch.object(e._session, "post", return_value=fake):
        with pytest.raises(KinChatHTTPError) as exc_info:
            e.reset(scenario_id="nope")

    assert exc_info.value.status_code == 404
    assert "not found" in exc_info.value.detail


def test_500_response_raises_kinchat_http_error():
    e = KinChatEnv("http://localhost:7860")
    fake = _FakeResponse(
        json_body={"detail": "boom"},
        status_code=500,
        url="http://localhost:7860/state",
    )
    fake.content = b'{"detail": "boom"}'
    with patch.object(e._session, "get", return_value=fake):
        with pytest.raises(KinChatHTTPError) as exc_info:
            e.state()
    assert exc_info.value.status_code == 500


def test_4xx_with_non_json_body_still_raises_cleanly():
    e = KinChatEnv("http://localhost:7860")
    fake = _FakeResponse(
        json_body=None,
        status_code=502,
        url="http://localhost:7860/health",
        text="Bad Gateway",
    )
    fake.content = b"Bad Gateway"
    with patch.object(e._session, "get", return_value=fake):
        with pytest.raises(KinChatHTTPError) as exc_info:
            e.health()
    assert exc_info.value.status_code == 502


# --------------------------------------------------------------------------- #
# Lifecycle                                                                   #
# --------------------------------------------------------------------------- #


def test_close_is_idempotent():
    e = KinChatEnv("http://localhost:7860")
    e.close()
    e.close()  # second call must not raise


def test_context_manager_closes_session():
    with KinChatEnv("http://localhost:7860") as e:
        assert isinstance(e, KinChatEnv)
        sess = e._session
        sess.close = MagicMock()
    sess.close.assert_called_once()


# --------------------------------------------------------------------------- #
# Negative test: client must NOT import server internals                      #
# --------------------------------------------------------------------------- #


def test_client_module_does_not_reference_server():
    """The hackathon judges check this. Source-level grep is the strict
    test — module-level ``sys.modules`` snooping is unreliable because
    other tests in the suite legitimately import ``kinchat.server``.
    """
    src = inspect.getsource(client_module)
    assert "kinchat.server" not in src, (
        "kinchat.client must not reference kinchat.server "
        "(client/server separation is hackathon-required)"
    )
    assert "from kinchat.server" not in src
    assert "import kinchat.server" not in src


def test_client_imports_only_allowed_modules():
    """Walk the ``kinchat.client`` module's globals and ensure no
    attribute originates from ``kinchat.server``."""
    for name, obj in vars(client_module).items():
        mod = getattr(obj, "__module__", None)
        if isinstance(mod, str):
            assert not mod.startswith("kinchat.server"), (
                f"kinchat.client exposes {name!r} from {mod!r} — "
                "must not reach into kinchat.server"
            )


def test_fresh_import_does_not_load_server():
    """Importing ``kinchat.client`` by itself must not pull in
    ``kinchat.server.*`` as a side effect.

    We do this by spawning a subprocess so the test is robust against
    other tests in the suite having already imported server modules.
    """
    import subprocess

    code = (
        "import sys; "
        "import kinchat.client; "
        "leaked = [m for m in sys.modules if m.startswith('kinchat.server')]; "
        "assert not leaked, leaked; "
        "print('OK')"
    )
    result = subprocess.run(
        [sys.executable, "-c", code],
        capture_output=True,
        text=True,
        timeout=30,
    )
    assert result.returncode == 0, (
        f"stdout={result.stdout!r} stderr={result.stderr!r}"
    )
    assert "OK" in result.stdout