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

from channels.testing.websocket import WebsocketCommunicator
from django.test import TransactionTestCase, override_settings

from fupogfakta.models import Category, Question
from partyhub.asgi import application
from party import rooms


def make_bank():
    for c in range(1, 6):
        cat = Category.objects.create(name=f"Kat {c}")
        for i in range(3):
            Question.objects.create(category=cat, prompt=f"Spørgsmål {c}-{i} _____.", answer=f"sandt{c}{i}",
                                    decoys=["x", "y", "z"])


async def recv_until(comm, type_, phase=None, limit=20):
    for _ in range(limit):
        msg = await comm.receive_json_from(timeout=3)
        if msg["type"] == type_ and (phase is None or msg["state"]["phase"] == phase):
            return msg
    raise AssertionError(f"fik aldrig {type_} {phase}")


HOST_HEADERS = [(b"x-authentik-username", b"agw"), (b"x-authentik-groups", b"Familie")]


@override_settings(TRUSTED_PROXIES=["127.0.0.1"], DEV_HOST_USER=None)

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 WebSocketGameTests(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)

    def _room(self):
        from fupogfakta.models import load_game_questions
        return rooms.create_room(load_game_questions(), "agw", "http://testserver")

    async def _host(self, code, headers=HOST_HEADERS):
        comm = WebsocketCommunicator(application, f"/ws/host/{code}/", headers=headers)
        comm.scope["client"] = ("127.0.0.1", 5000)
        return comm

    async def test_host_requires_authentik_header(self):
        room = await self._sync(self._room)
        comm = await self._host(room.code, headers=[])
        connected, code = await comm.connect()
        self.assertFalse(connected)

    async def test_header_from_untrusted_ip_is_ignored(self):
        room = await self._sync(self._room)
        comm = WebsocketCommunicator(application, f"/ws/host/{room.code}/", headers=HOST_HEADERS)
        comm.scope["client"] = ("203.0.113.9", 5000)
        connected, _ = await comm.connect()
        self.assertFalse(connected)

    async def test_unknown_room(self):
        comm = WebsocketCommunicator(application, "/ws/play/ZZZZ/")
        connected, _ = await comm.connect()
        self.assertTrue(connected)
        msg = await comm.receive_json_from()
        self.assertEqual(msg["code"], "unknown_room")
        await comm.disconnect()

    async def test_join_play_a_question_and_resume(self):
        room = await self._sync(self._room)
        host = await self._host(room.code)
        self.assertTrue((await host.connect())[0])
        await recv_until(host, "state", "lobby")

        phones = []
        for name in ["Asger", "Sus", "Jørn"]:
            c = WebsocketCommunicator(application, f"/ws/play/{room.code}/")
            await c.connect()
            await c.send_json_to({"action": "join", "name": name})
            joined = await recv_until(c, "joined")
            phones.append((c, joined))

        await phones[0][0].send_json_to({"action": "start"})
        st = (await recv_until(host, "state", "category"))["state"]
        chooser = next(c for c, j in phones if j["player_id"] == st["category"]["chooser_id"])
        await chooser.send_json_to({"action": "choose_category", "category_id": st["category"]["options"][0]["id"]})
        await recv_until(host, "state", "lie")

        for i, (c, _) in enumerate(phones):
            await c.send_json_to({"action": "submit_lie", "text": f"løgn {i}"})
        st = (await recv_until(host, "state", "guess"))["state"]
        self.assertGreaterEqual(len(st["answers"]), 4)

        for c, j in phones:
            mine = (await recv_until(c, "state", "guess"))["state"]
            await c.send_json_to({"action": "guess", "answer_id": mine["answers"][0]["id"]})
        st = (await recv_until(host, "state", "reveal"))["state"]
        self.assertTrue(st["reveal"]["steps"][-1]["is_truth"])

        # Telefonen mister forbindelsen og kommer tilbage med sit token.
        c0, j0 = phones[0]
        await c0.disconnect()
        again = WebsocketCommunicator(application, f"/ws/play/{room.code}/")
        await again.connect()
        await again.send_json_to({"action": "resume", "token": j0["token"]})
        self.assertEqual((await recv_until(again, "joined"))["player_id"], j0["player_id"])
        st = (await recv_until(again, "state"))["state"]
        self.assertEqual(st["me"]["id"], j0["player_id"])

        for c, _ in phones[1:]:
            await c.disconnect()
        await again.disconnect()
        await host.disconnect()

    async def test_errors_are_reported_not_raised(self):
        room = await self._sync(self._room)
        c = WebsocketCommunicator(application, f"/ws/play/{room.code}/")
        await c.connect()
        await c.send_json_to({"action": "start"})
        self.assertEqual((await c.receive_json_from())["code"], "unknown_token")
        await c.send_json_to({"action": "join", "name": "A"})
        await recv_until(c, "joined")
        await c.send_json_to({"action": "guess"})
        msg = await recv_until(c, "error")
        self.assertIn(msg["code"], ("wrong_phase", "bad_request"))
        await c.disconnect()

    @staticmethod
    async def _sync(fn):
        from asgiref.sync import sync_to_async
        return await sync_to_async(fn)()