weirsoe-party-protocol/party/tests/test_brain.py

import asyncio
from unittest import mock

from asgiref.sync import sync_to_async
from channels.testing.websocket import WebsocketCommunicator
from django.test import SimpleTestCase, TransactionTestCase, override_settings

from party import rooms, stt, tts
from party.brain import GameBrain
from party.brain.providers import OllamaProvider, OpenAICompatibleProvider, build_provider
from partyhub.asgi import application

from .test_narration import fake_synth
from .test_ws import HOST_HEADERS, make_bank, recv_until


class FakeProvider:
    """Svarer som en sprogmodel ville, ud fra hvilket skema der bedes om."""
    name = "fake"

    def __init__(self):
        self.calls = []

    async def complete_json(self, system, user, schema, **kw):
        self.calls.append(user)
        props = schema["properties"]
        if "is_truth" in props:
            bad = '"løgn": "grim løgn"' in user
            return {"is_truth": False, "inappropriate": bad, "message": "Prøv en pænere løgn." if bad else ""}
        if "comments" in props:
            import json, re
            data = json.loads(user.split("Data (JSON):\n", 1)[1])
            return {"comments": [{"id": a["id"], "line": f"KOMMENTAR {a['text']}"} for a in data["svar"]]}
        if "teases" in props:
            return {"teases": []}
        if "lines" in props:
            return {"lines": []}
        return None


class ProviderFactoryTests(SimpleTestCase):
    def test_build(self):
        self.assertIsNone(build_provider("", "x"))
        self.assertIsNone(build_provider("ollama", ""))
        self.assertIsInstance(build_provider("ollama", "gemma"), OllamaProvider)
        self.assertIsInstance(build_provider("openai", "gpt"), OpenAICompatibleProvider)

    def test_timeout_and_errors_return_none(self):
        class Slow(OllamaProvider):
            async def _complete(self, *a):
                await asyncio.sleep(5)

        class Broken(OllamaProvider):
            async def _complete(self, *a):
                raise RuntimeError("nede")
        self.assertIsNone(asyncio.run(Slow("m").complete_json("s", "u", {}, timeout=0.05)))
        self.assertIsNone(asyncio.run(Broken("m").complete_json("s", "u", {})))


@override_settings(TRUSTED_PROXIES=["127.0.0.1"], DEV_HOST_USER=None)
class BrainInGameTests(TransactionTestCase):
    def setUp(self):
        rooms.all_rooms().clear()
        make_bank()
        from fupogfakta import engine as _engine
        self._intro, _engine.INTRO_SECONDS = _engine.INTRO_SECONDS, 0.2
        self.addCleanup(setattr, _engine, "INTRO_SECONDS", self._intro)
        self.fake = FakeProvider()

    async def _room(self):
        from fupogfakta.models import load_game_questions
        with mock.patch.object(rooms, "get_brain", return_value=GameBrain(self.fake)):
            return rooms.create_room(await sync_to_async(load_game_questions)(), "agw", "http://testserver")

    async def test_judge_rejects_lie_and_comments_reach_reveal(self):
        room = await self._room()
        with mock.patch.object(tts, "enabled", return_value=True), mock.patch.object(tts, "synth", fake_synth):
            host = WebsocketCommunicator(application, f"/ws/host/{room.code}/", headers=HOST_HEADERS)
            host.scope["client"] = ("127.0.0.1", 5000)
            await host.connect()
            phones = []
            for name in ["Asger", "Sus"]:
                c = WebsocketCommunicator(application, f"/ws/play/{room.code}/")
                await c.connect()
                await c.send_json_to({"action": "join", "name": name})
                await recv_until(c, "joined")
                phones.append(c)
            await phones[0].send_json_to({"action": "start"})
            st = (await recv_until(host, "state", "category"))["state"]
            chooser = phones[0] if st["category"]["chooser_id"] == "p1" else phones[1]
            await chooser.send_json_to({"action": "choose_category", "category_id": st["category"]["options"][0]["id"]})
            for c in phones:
                await recv_until(c, "state", "lie")
            await phones[0].send_json_to({"action": "submit_lie", "text": "grim løgn"})
            err = await recv_until(phones[0], "error")
            self.assertEqual(err["message"], "Prøv en pænere løgn.")
            self.assertNotIn("p1", room.game.lies)                       # afvist og fjernet
            await phones[0].send_json_to({"action": "submit_lie", "text": "pæn løgn"})
            await phones[1].send_json_to({"action": "submit_lie", "text": "anden løgn"})
            for c in phones:
                mine = (await recv_until(c, "state", "guess"))["state"]
                await c.send_json_to({"action": "guess", "answer_id": mine["answers"][0]["id"]})
            for _ in range(40):
                st = (await recv_until(host, "state", "reveal"))["state"]
                if st["reveal"]["started_at"]:
                    break
            texts = [s["audio"]["text"] for s in st["reveal"]["steps"] if s.get("audio")]
            self.assertTrue(any("KOMMENTAR" in t for t in texts), texts)  # hjernens kommentar blev læst op
            self.assertTrue(room.memory)                                  # afsløringen blev husket
            for c in phones:
                await c.disconnect()
            await host.disconnect()

    async def test_mic_feeds_memory(self):
        room = await self._room()

        async def fake_transcribe(audio, ctype):
            return {"text": "det er helt sikkert ugler", "laughter": True}
        with mock.patch.object(stt, "transcribe", fake_transcribe):
            res = await room.hear(b"x" * 5000, "audio/webm")
        self.assertTrue(res["laughter"])
        self.assertEqual(room.mic_laughs, 1)
        self.assertTrue(any("grinede" in m for m in room.memory))
        self.assertTrue(any("helt sikkert ugler" in m for m in room.memory))
        self.assertTrue(room.mic_on())