agw
9f6d134288
CI / test (push) Successful in 58s
Spørgsmål og forklaring i en kategori med voice_id læses af den stemme; resten af spillet bruger speakeren. Forvarmningen og cachen følger stemmen. Claude-Session: https://claude.ai/code/session_01T4tydM7urcwizGC8jeJU1K
253 lines
12 KiB
Python
253 lines
12 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
|
|
|
|
|
|
|
|
def setUpModule():
|
|
from fupogfakta.engine import Game
|
|
assert Game.show_rules is True # standarden i drift
|
|
Game.show_rules = False # reglerne testes for sig (RulesTests)
|
|
|
|
|
|
def tearDownModule():
|
|
from fupogfakta.engine import Game
|
|
Game.show_rules = True
|
|
|
|
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", "category_intro"])
|
|
g.tick(100 + engine.INTRO_SECONDS)
|
|
self.assertEqual([e["type"] for e in g.drain_events()], ["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_submitted", "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, voice=None):
|
|
return {"url": f"/tts/{abs(hash(text)) % 10**32:032d}.mp3", "ms": 40}
|
|
|
|
|
|
@override_settings(TRUSTED_PROXIES=["127.0.0.1"], DEV_HOST_USER=None)
|
|
class NarrationOverWebSocketTests(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)
|
|
|
|
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()
|
|
|
|
|
|
class SpeechHoldTests(SimpleTestCase):
|
|
def test_transitions_wait_for_speech(self):
|
|
from fupogfakta.tests.test_engine import bank
|
|
g = Game("TEST", bank(), rng=random.Random(5))
|
|
a, b = g.join("A"), g.join("B")
|
|
g.start(None, 0)
|
|
g.hold_until = 10 # speakeren taler til t=10
|
|
g.choose_category(g.chooser_id, g.category_options[0]["id"], 1)
|
|
self.assertEqual(g.phase, "category") # valgt, men skærmen skifter ikke endnu
|
|
self.assertFalse(g.tick(5))
|
|
g.hold_until = None
|
|
self.assertTrue(g.tick(10))
|
|
self.assertEqual(g.phase, "intro") # kategorien præsenteres
|
|
self.assertTrue(g.tick(10 + engine.INTRO_SECONDS))
|
|
self.assertEqual(g.phase, "lie")
|
|
g.hold_until = 20
|
|
g.submit_lie(a.id, "katte", 11)
|
|
g.submit_lie(b.id, "hunde", 11)
|
|
self.assertEqual(g.phase, "lie") # alle har løjet, men speakeren læser stadig op
|
|
self.assertFalse(g.tick(15))
|
|
self.assertTrue(g.tick(20))
|
|
self.assertEqual(g.phase, "guess")
|
|
self.assertIn("lie_all_done", [e["type"] for e in g.drain_events()])
|
|
g.hold_until = 99
|
|
g.skip(21) # værtens "spring over" venter ikke
|
|
self.assertEqual(g.phase, "reveal")
|
|
|
|
|
|
@override_settings(TRUSTED_PROXIES=["127.0.0.1"], DEV_HOST_USER=None)
|
|
class TeaseTests(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)
|
|
|
|
async def test_fooled_player_gets_teased_when_stamp_lands(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(rooms, "tease_images", return_value=["/static/party/tease/x.png"]), \
|
|
mock.patch.object(rooms, "TEASE_CHANCE", 1.0):
|
|
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})
|
|
phones.append((c, (await recv_until(c, "joined"))["player_id"]))
|
|
await phones[0][0].send_json_to({"action": "start"})
|
|
st = (await recv_until(phones[0][0], "state", "category"))["state"]
|
|
chooser = next(c for c, pid in phones if pid == st["category"]["chooser_id"])
|
|
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}"})
|
|
# Sus vælger Asgers løgn og bliver narret
|
|
(asger, a_id), (sus, s_id) = phones
|
|
st = (await recv_until(sus, "state", "guess"))["state"]
|
|
await recv_until(asger, "state", "guess")
|
|
asger_lie = next(a["id"] for a in st["answers"] if a["text"] == "løgn 0")
|
|
await sus.send_json_to({"action": "guess", "answer_id": asger_lie})
|
|
truth_or_other = next(a["id"] for a in st["answers"] if a["text"] not in ("løgn 0", "løgn 1"))
|
|
await asger.send_json_to({"action": "guess", "answer_id": truth_or_other})
|
|
# Drilleriet kommer, når stemplet lander på Asgers løgn. Ligger husets løgn før i
|
|
# afsløringen, går der et helt trin (fire sekunder) uden beskeder, så vent rigeligt.
|
|
for _ in range(40):
|
|
msg = await sus.receive_json_from(timeout=15)
|
|
if msg["type"] == "tease":
|
|
break
|
|
self.assertEqual(msg["type"], "tease")
|
|
self.assertTrue(msg["text"])
|
|
self.assertEqual(msg["image"], "/static/party/tease/x.png")
|
|
for c, _ in phones:
|
|
await c.disconnect()
|
|
|
|
|
|
class TruthSpamTests(SimpleTestCase):
|
|
"""Fejl fra første rigtige spil: alle kendte svaret og spammede det, og spillet kom aldrig videre."""
|
|
|
|
def test_truth_warning_only_once_and_hold_is_capped(self):
|
|
import time as _time
|
|
from fupogfakta.tests.test_engine import bank
|
|
from party.rooms import DUE_HOLD_MAX, Room
|
|
g = Game("TEST", bank(), rng=random.Random(7))
|
|
a, b = g.join("A"), g.join("B")
|
|
g.start(None, 0)
|
|
g.tick(100)
|
|
g.tick(100 + engine.INTRO_SECONDS)
|
|
g.drain_events()
|
|
for _ in range(6):
|
|
with self.assertRaises(engine.GameError):
|
|
g.submit_lie(a.id, g.question["answer"], 110)
|
|
self.assertEqual([e["type"] for e in g.drain_events()], ["lie_was_truth"]) # kun én gang
|
|
|
|
room = Room("TEST", g, "agw")
|
|
now = _time.time()
|
|
g.deadline = now - 1 # tiden er gået
|
|
room.speech_end = now + 300 # speakeren har (fejlagtigt) meget i kø
|
|
room._update_hold()
|
|
self.assertLessEqual(g.hold_until, g.deadline + DUE_HOLD_MAX)
|