agw
2da781e361
CI / test (push) Successful in 16s
Claude-Session: https://claude.ai/code/session_01JmhJG6JnhXfzNiKv8f2vHu
134 lines
6.7 KiB
Python
134 lines
6.7 KiB
Python
import asyncio
|
|
import random
|
|
from unittest import mock
|
|
|
|
from channels.testing.websocket import WebsocketCommunicator
|
|
from django.test import SimpleTestCase, TransactionTestCase, override_settings
|
|
|
|
from fupogfakta import engine
|
|
from fupogfakta.engine import Game
|
|
from party import rooms, tts
|
|
from party.narrator import Narrator, join_names, number_words, safe, speakable_prompt
|
|
from partyhub.asgi import application
|
|
|
|
from .test_ws import HOST_HEADERS, make_bank, recv_until
|
|
|
|
|
|
class NarratorTests(SimpleTestCase):
|
|
def test_number_words(self):
|
|
cases = {0: "nul", 1: "én", 3: "tre", 21: "enogtyve", 47: "syvogfyrre", 100: "et hundrede",
|
|
1500: "et tusind fem hundrede", 10500: "ti tusind fem hundrede", 2000: "to tusind"}
|
|
for n, words in cases.items():
|
|
self.assertEqual(number_words(n), words)
|
|
|
|
def test_names_and_safety(self):
|
|
self.assertEqual(join_names(["A"]), "A")
|
|
self.assertEqual(join_names(["A", "B", "C"]), "A, B og C")
|
|
self.assertEqual(safe("Bo[laughs]"), "Bolaughs")
|
|
self.assertEqual(speakable_prompt("En flok _____ kaldes."), "En flok blank kaldes.")
|
|
self.assertEqual(speakable_prompt("_____ kan ikke mærke chili."), "Blank kan ikke mærke chili.")
|
|
n = Narrator(random.Random(0))
|
|
for _ in range(10):
|
|
line = n.line("reveal_truth_found", sandhed="kongevand", navne="Sus")
|
|
self.assertRegex(line, r"^(\[[a-z ]+\] )*[A-ZÆØÅ]")
|
|
|
|
def test_lines_are_filled_and_vary(self):
|
|
n = Narrator(random.Random(1))
|
|
seen = {n.line("lobby_join", navn="Sus") for _ in range(6)}
|
|
self.assertGreater(len(seen), 3)
|
|
self.assertTrue(all("Sus" in s and "{" not in s for s in seen))
|
|
self.assertNotIn("{", n.line("winner", vinder="Bo", point=10500))
|
|
|
|
def test_reveal_step_choice(self):
|
|
n = Narrator(random.Random(2))
|
|
names = {"p1": "Asger", "p2": "Sus", "p3": "Jørn"}
|
|
lie = {"text": "katte", "is_truth": False, "house": False, "authors": ["p1"], "guessed_by": ["p2"], "audience_votes": 0}
|
|
self.assertIn("Asger", n.reveal_step(lie, names, 3))
|
|
house = dict(lie, authors=[], house=True)
|
|
self.assertIn("katte", n.reveal_step(house, names, 3))
|
|
nobody = {"text": "ugler", "is_truth": True, "house": False, "authors": [], "guessed_by": [], "audience_votes": 0}
|
|
self.assertIn("ugler", n.reveal_step(nobody, names, 3))
|
|
unpicked = dict(lie, guessed_by=[])
|
|
self.assertIsNone(n.reveal_step(unpicked, names, 3))
|
|
|
|
|
|
class EngineEventTests(SimpleTestCase):
|
|
def test_events_and_reveal_hold(self):
|
|
from fupogfakta.tests.test_engine import bank
|
|
g = Game("TEST", bank(), rng=random.Random(3))
|
|
a, b = g.join("A"), g.join("B")
|
|
g.start(None, 0)
|
|
types = [e["type"] for e in g.drain_events()]
|
|
self.assertEqual(types[:4], ["join", "join", "game_start", "round_intro"])
|
|
self.assertIn("category", types)
|
|
g.tick(100) # kategorien vælges af tiden
|
|
self.assertEqual([e["type"] for e in g.drain_events()], ["category_timeout", "lie_start"])
|
|
g.submit_lie(a.id, "katte", 100)
|
|
with self.assertRaises(engine.GameError):
|
|
g.submit_lie(b.id, g.question["answer"], 100)
|
|
self.assertEqual([e["type"] for e in g.drain_events()], ["lie_was_truth"])
|
|
g.submit_lie(b.id, "hunde", 100)
|
|
truth = next(x.id for x in g.answers if x.is_truth)
|
|
g.guess(a.id, truth, 100)
|
|
g.guess(b.id, truth, 100)
|
|
self.assertEqual(g.phase, "reveal")
|
|
g.hold_reveal(100, max_wait=5)
|
|
self.assertIsNone(g.snapshot(100, host=True)["reveal"]["started_at"])
|
|
g.tick(106) # lyden kom aldrig: afsløringen starter alligevel
|
|
self.assertEqual(g.phase, "reveal")
|
|
self.assertIsNotNone(g.reveal_started_at)
|
|
g.start_reveal_timeline(106, [5000] * len(g.reveal_steps), 4000)
|
|
self.assertAlmostEqual(g.deadline, 106 + (5 * len(g.reveal_steps) + 4))
|
|
|
|
|
|
async def fake_synth(text):
|
|
return {"url": f"/tts/{abs(hash(text)) % 10**32:032d}.mp3", "ms": 2000 + len(text) * 10}
|
|
|
|
|
|
@override_settings(TRUSTED_PROXIES=["127.0.0.1"], DEV_HOST_USER=None)
|
|
class NarrationOverWebSocketTests(TransactionTestCase):
|
|
def setUp(self):
|
|
rooms.all_rooms().clear()
|
|
make_bank()
|
|
|
|
async def test_host_gets_narration_and_reveal_audio(self):
|
|
from asgiref.sync import sync_to_async
|
|
from fupogfakta.models import load_game_questions
|
|
room = rooms.create_room(await sync_to_async(load_game_questions)(), "agw", "http://testserver")
|
|
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 i, c in enumerate(phones):
|
|
await recv_until(c, "state", "lie")
|
|
await c.send_json_to({"action": "submit_lie", "text": f"løgn {i}"})
|
|
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"]})
|
|
# Først venter afsløringen på lyden, så starter den med en længde pr. trin.
|
|
for _ in range(30):
|
|
st = (await recv_until(host, "state", "reveal"))["state"]
|
|
if st["reveal"]["started_at"]:
|
|
break
|
|
r = st["reveal"]
|
|
self.assertTrue(all(s.get("audio") for s in r["steps"]))
|
|
self.assertEqual(len(r["steps_ms"]), len(r["steps"]))
|
|
self.assertGreaterEqual(min(r["steps_ms"]), 3500)
|
|
texts = [n["text"] for n in st["narration"]]
|
|
self.assertTrue(any("Asger" in t or "Sus" in t for t in texts), texts) # join/kategori-replik
|
|
self.assertTrue(any("blank" in t for t in texts), texts) # spørgsmålet blev læst op
|
|
for c in phones:
|
|
await c.disconnect()
|
|
await host.disconnect()
|