Merge pull request 'test(gameplay): add canonical loop smoke evidence (#302)' (#304) from dev/issue-302-canonical-loop-evidence into main

lobby/management/commands/smoke_staging.py

@@ -10,7 +10,7 @@ from fupogfakta.models import Category, GameSession, Player, Question, RoundQues
 
 
 class Command(BaseCommand):
-    help = "Run minimal staging smoke flow for lobby gameplay"
+    help = "Run canonical gameplay smoke/regression flow for bluff -> guess -> reveal -> scoreboard"
 
     def add_arguments(self, parser):
         parser.add_argument(
@@ -18,6 +18,26 @@ class Command(BaseCommand):
             help="Optional path to write smoke result artifact as JSON",
         )
 
+    def _fail(self, step: str, detail: str, payload=None):
+        message = f"{step} failed: {detail}"
+        if payload is not None:
+            message += f" | payload={json.dumps(payload, sort_keys=True)}"
+        raise CommandError(message)
+
+    def _expect_status(self, response, expected_status: int, step: str):
+        if response.status_code != expected_status:
+            try:
+                payload = response.json()
+            except ValueError:
+                payload = {"raw": response.content.decode("utf-8", errors="replace")}
+            self._fail(step, f"expected HTTP {expected_status}, got {response.status_code}", payload)
+        return response.json()
+
+    def _expect_session_status(self, payload: dict, expected_status: str, step: str):
+        actual_status = payload.get("session", {}).get("status")
+        if actual_status != expected_status:
+            self._fail(step, f"expected session.status={expected_status}, got {actual_status}", payload)
+
     def handle(self, *args, **options):
         GameSession.objects.all().delete()
         Player.objects.all().delete()
@@ -30,11 +50,14 @@ class Command(BaseCommand):
         category.is_active = True
         category.save(update_fields=["is_active"])
 
-        Question.objects.get_or_create(
+        question, _ = Question.objects.get_or_create(
             category=category,
             prompt="Smoke prompt?",
             defaults={"correct_answer": "Correct", "is_active": True},
         )
+        if not question.is_active:
+            question.is_active = True
+            question.save(update_fields=["is_active"])
 
         User = get_user_model()
         host, _ = User.objects.get_or_create(username="smoke-host")
@@ -42,111 +65,254 @@ class Command(BaseCommand):
         host.is_staff = True
         host.save()
 
+        artifact = {
+            "ok": True,
+            "command": "python manage.py smoke_staging --artifact <path>",
+            "generated_at": datetime.now(timezone.utc).isoformat(),
+            "question": {
+                "prompt": question.prompt,
+                "correct_answer": question.correct_answer,
+            },
+            "steps": [],
+        }
+
         host_client = Client()
         host_client.force_login(host)
 
-        create_res = host_client.post("/lobby/sessions/create", content_type="application/json")
-        if create_res.status_code != 201:
-            raise CommandError(f"create_session failed: {create_res.status_code} {create_res.content!r}")
-
-        code = create_res.json()["session"]["code"]
+        create_payload = self._expect_status(
+            host_client.post("/lobby/sessions/create", content_type="application/json"),
+            201,
+            "create_session",
+        )
+        code = create_payload["session"]["code"]
+        artifact["session_code"] = code
+        artifact["steps"].append(
+            {
+                "step": "create_session",
+                "session_status": create_payload["session"]["status"],
+            }
+        )
 
         players = []
         for nickname in ["P1", "P2", "P3"]:
-            join_res = Client().post(
-                "/lobby/sessions/join",
-                data=json.dumps({"code": code, "nickname": nickname}),
-                content_type="application/json",
+            join_payload = self._expect_status(
+                Client().post(
+                    "/lobby/sessions/join",
+                    data=json.dumps({"code": code, "nickname": nickname}),
+                    content_type="application/json",
+                ),
+                201,
+                f"join_session[{nickname}]",
             )
-            if join_res.status_code != 201:
-                raise CommandError(f"join_session failed for {nickname}: {join_res.status_code}")
-            players.append(join_res.json()["player"])
-
-        start_res = host_client.post(
-            f"/lobby/sessions/{code}/rounds/start",
-            data=json.dumps({"category_slug": category.slug}),
-            content_type="application/json",
+            players.append(join_payload["player"])
+        artifact["players"] = [player["nickname"] for player in players]
+        artifact["steps"].append(
+            {
+                "step": "join_players",
+                "players_count": len(players),
+            }
         )
-        if start_res.status_code != 201:
-            raise CommandError(f"start_round failed: {start_res.status_code}")
 
-        round_question_id = start_res.json()["round_question"]["id"]
+        start_payload = self._expect_status(
+            host_client.post(
+                f"/lobby/sessions/{code}/rounds/start",
+                data=json.dumps({"category_slug": category.slug}),
+                content_type="application/json",
+            ),
+            201,
+            "start_round",
+        )
+        self._expect_session_status(start_payload, GameSession.Status.LIE, "start_round")
+
+        round_question_id = start_payload["round_question"]["id"]
+        artifact["round_question_id"] = round_question_id
+        artifact["steps"].append(
+            {
+                "step": "start_round",
+                "session_status": start_payload["session"]["status"],
+                "round_question_id": round_question_id,
+            }
+        )
 
         answers = []
+        lie_transition_payload = None
         for player in players:
-            nick = player["nickname"]
-            lie_res = Client().post(
-                f"/lobby/sessions/{code}/questions/{round_question_id}/lies/submit",
-                data=json.dumps(
-                    {
-                        "player_id": player["id"],
-                        "session_token": player["session_token"],
-                        "text": f"Lie from {nick}",
-                    }
+            nickname = player["nickname"]
+            lie_payload = self._expect_status(
+                Client().post(
+                    f"/lobby/sessions/{code}/questions/{round_question_id}/lies/submit",
+                    data=json.dumps(
+                        {
+                            "player_id": player["id"],
+                            "session_token": player["session_token"],
+                            "text": f"Lie from {nickname}",
+                        }
+                    ),
+                    content_type="application/json",
                 ),
-                content_type="application/json",
+                201,
+                f"submit_lie[{nickname}]",
             )
-            if lie_res.status_code != 201:
-                raise CommandError(f"submit_lie failed for {nick}: {lie_res.status_code}")
-            if lie_res.json().get("answers"):
-                answers = lie_res.json()["answers"]
+            if lie_payload.get("answers"):
+                answers = lie_payload["answers"]
+                lie_transition_payload = lie_payload
 
         if not answers:
-            detail_res = host_client.get(f"/lobby/sessions/{code}")
-            if detail_res.status_code != 200:
-                raise CommandError(f"session_detail after lies failed: {detail_res.status_code}")
-            answers = detail_res.json().get("round_question", {}).get("answers", [])
+            detail_payload = self._expect_status(host_client.get(f"/lobby/sessions/{code}"), 200, "session_detail_after_lies")
+            answers = detail_payload.get("round_question", {}).get("answers", [])
+            self._expect_session_status(detail_payload, GameSession.Status.GUESS, "session_detail_after_lies")
+            lie_transition_payload = detail_payload
 
         if not answers:
-            raise CommandError("canonical lie->guess transition returned empty answers")
+            self._fail("auto_guess_transition", "canonical lie->guess transition returned empty answers")
 
-        for player in players:
-            nick = player["nickname"]
-            selected = next((a for a in answers if a.get("player_id") != player["id"]), answers[0])
-            guess_res = Client().post(
-                f"/lobby/sessions/{code}/questions/{round_question_id}/guesses/submit",
-                data=json.dumps(
-                    {
-                        "player_id": player["id"],
-                        "session_token": player["session_token"],
-                        "selected_text": selected["text"],
-                    }
-                ),
-                content_type="application/json",
+        if not any(answer.get("text") == question.correct_answer for answer in answers):
+            self._fail("auto_guess_transition", "mixed answers missing correct answer", {"answers": answers})
+        if len(answers) < len(players) + 1:
+            self._fail(
+                "auto_guess_transition",
+                "mixed answers shorter than expected bluff set",
+                {"answers": answers, "players_count": len(players)},
             )
-            if guess_res.status_code != 201:
-                raise CommandError(f"submit_guess failed for {nick}: {guess_res.status_code}")
 
-        detail_res = host_client.get(f"/lobby/sessions/{code}")
-        if detail_res.status_code != 200:
-            raise CommandError(f"session_detail after guesses failed: {detail_res.status_code}")
-        if detail_res.json()["session"]["status"] != GameSession.Status.SCOREBOARD:
-            raise CommandError("canonical guess->reveal->scoreboard transition did not reach scoreboard")
+        self._expect_session_status(lie_transition_payload, GameSession.Status.GUESS, "auto_guess_transition")
+        artifact["steps"].append(
+            {
+                "step": "auto_guess_transition",
+                "session_status": lie_transition_payload["session"]["status"],
+                "answers": [answer["text"] for answer in answers],
+            }
+        )
 
-        finish_res = host_client.post(f"/lobby/sessions/{code}/finish", content_type="application/json")
-        if finish_res.status_code != 200:
-            raise CommandError(f"finish_game failed: {finish_res.status_code}")
+        answer_texts = {answer["text"] for answer in answers}
+        correct_answer = next((answer["text"] for answer in answers if answer.get("text") == question.correct_answer), None)
+        if correct_answer is None:
+            self._fail("submit_guesses", "could not resolve correct answer from mixed answers", {"answers": answers})
+
+        guess_plan = {
+            players[0]["nickname"]: "Lie from P2",
+            players[1]["nickname"]: correct_answer,
+            players[2]["nickname"]: "Lie from P1",
+        }
+        missing_guess_targets = {text for text in guess_plan.values() if text not in answer_texts}
+        if missing_guess_targets:
+            self._fail(
+                "submit_guesses",
+                "expected bluff targets missing from mixed answers",
+                {"answers": answers, "missing_guess_targets": sorted(missing_guess_targets)},
+            )
+        artifact["guess_plan"] = guess_plan
+
+        guess_payloads = []
+        for player in players:
+            nickname = player["nickname"]
+            guess_payload = self._expect_status(
+                Client().post(
+                    f"/lobby/sessions/{code}/questions/{round_question_id}/guesses/submit",
+                    data=json.dumps(
+                        {
+                            "player_id": player["id"],
+                            "session_token": player["session_token"],
+                            "selected_text": guess_plan[nickname],
+                        }
+                    ),
+                    content_type="application/json",
+                ),
+                201,
+                f"submit_guess[{nickname}]",
+            )
+            guess_payloads.append(guess_payload)
+
+        reveal_payload = guess_payloads[-1]
+        self._expect_session_status(reveal_payload, GameSession.Status.REVEAL, "auto_reveal_transition")
+        if not reveal_payload.get("phase_transition", {}).get("auto_advanced"):
+            self._fail("auto_reveal_transition", "expected auto_advanced=true on final guess", reveal_payload)
+        reveal = reveal_payload.get("reveal")
+        if not reveal:
+            self._fail("auto_reveal_transition", "missing canonical reveal payload", reveal_payload)
+        if reveal.get("correct_answer") != question.correct_answer:
+            self._fail(
+                "auto_reveal_transition",
+                "reveal payload returned wrong correct answer",
+                {"expected": question.correct_answer, "reveal": reveal},
+            )
+        if len(reveal.get("lies", [])) != len(players):
+            self._fail("auto_reveal_transition", "unexpected lie count in reveal payload", reveal)
+        if len(reveal.get("guesses", [])) != len(players):
+            self._fail("auto_reveal_transition", "unexpected guess count in reveal payload", reveal)
+
+        fooled_guesses = [guess for guess in reveal["guesses"] if not guess.get("is_correct")]
+        correct_guesses = [guess for guess in reveal["guesses"] if guess.get("is_correct")]
+        if len(fooled_guesses) != 2:
+            self._fail("auto_reveal_transition", "expected exactly two bluff guesses", reveal)
+        if len(correct_guesses) != 1:
+            self._fail("auto_reveal_transition", "expected exactly one correct guess", reveal)
+        if any(guess.get("fooled_player_id") is None for guess in fooled_guesses):
+            self._fail("auto_reveal_transition", "bluff guesses missing fooled_player_id", reveal)
+
+        artifact["steps"].append(
+            {
+                "step": "submit_guesses",
+                "guess_results": [
+                    {
+                        "player_id": payload["guess"]["player_id"],
+                        "selected_text": payload["guess"]["selected_text"],
+                        "is_correct": payload["guess"]["is_correct"],
+                        "fooled_player_id": payload["guess"].get("fooled_player_id"),
+                    }
+                    for payload in guess_payloads
+                ],
+            }
+        )
+        artifact["steps"].append(
+            {
+                "step": "auto_reveal_transition",
+                "session_status": reveal_payload["session"]["status"],
+                "reveal": {
+                    "correct_answer": reveal["correct_answer"],
+                    "lies_count": len(reveal["lies"]),
+                    "guesses_count": len(reveal["guesses"]),
+                    "fooled_player_ids": sorted(guess["fooled_player_id"] for guess in fooled_guesses),
+                    "correct_guess_player_ids": sorted(guess["player_id"] for guess in correct_guesses),
+                },
+            }
+        )
+
+        detail_payload = self._expect_status(host_client.get(f"/lobby/sessions/{code}"), 200, "session_detail_after_guesses")
+        self._expect_session_status(detail_payload, GameSession.Status.SCOREBOARD, "auto_scoreboard_transition")
+        if detail_payload.get("reveal") != reveal:
+            self._fail("auto_scoreboard_transition", "scoreboard promotion changed canonical reveal payload", detail_payload)
+        scoreboard = detail_payload.get("scoreboard")
+        if not scoreboard:
+            self._fail("auto_scoreboard_transition", "missing scoreboard payload after promotion", detail_payload)
+        if len(scoreboard) != len(players):
+            self._fail("auto_scoreboard_transition", "unexpected scoreboard length", detail_payload)
+        if not detail_payload.get("phase_view_model", {}).get("readiness", {}).get("scoreboard_ready"):
+            self._fail("auto_scoreboard_transition", "scoreboard_ready=false after promotion", detail_payload)
+
+        artifact["steps"].append(
+            {
+                "step": "auto_scoreboard_transition",
+                "session_status": detail_payload["session"]["status"],
+                "leaderboard": scoreboard,
+            }
+        )
+
+        finish_payload = self._expect_status(
+            host_client.post(f"/lobby/sessions/{code}/finish", content_type="application/json"),
+            200,
+            "finish_game",
+        )
+        self._expect_session_status(finish_payload, GameSession.Status.FINISHED, "finish_game")
+        artifact["steps"].append(
+            {
+                "step": "finish_game",
+                "session_status": finish_payload["session"]["status"],
+            }
+        )
 
         artifact_path = options.get("artifact")
         if artifact_path:
-            artifact = {
-                "ok": True,
-                "command": "smoke_staging",
-                "generated_at": datetime.now(timezone.utc).isoformat(),
-                "session_code": code,
-                "players": [player["nickname"] for player in players],
-                "round_question_id": round_question_id,
-                "steps": [
-                    "create_session",
-                    "join_players",
-                    "start_round",
-                    "submit_lies",
-                    "auto_guess_transition",
-                    "submit_guesses",
-                    "auto_reveal_to_scoreboard",
-                    "finish_game",
-                ],
-            }
             output_path = Path(artifact_path)
             output_path.parent.mkdir(parents=True, exist_ok=True)
             output_path.write_text(json.dumps(artifact, indent=2) + "\n", encoding="utf-8")

lobby/tests.py

@@ -2017,7 +2017,7 @@ class SmokeStagingCommandTests(TestCase):
         self.assertEqual(session.status, GameSession.Status.FINISHED)
         self.assertEqual(Player.objects.filter(session=session).count(), 3)
 
-    def test_smoke_staging_writes_artifact_when_requested(self):
+    def test_smoke_staging_writes_phase_evidence_artifact_when_requested(self):
         with tempfile.TemporaryDirectory() as tmp_dir:
             artifact_path = Path(tmp_dir) / "smoke.json"
             call_command("smoke_staging", artifact=str(artifact_path))
@@ -2025,24 +2025,40 @@ class SmokeStagingCommandTests(TestCase):
             self.assertTrue(artifact_path.exists())
             payload = json.loads(artifact_path.read_text(encoding="utf-8"))
             self.assertTrue(payload["ok"])
-            self.assertEqual(payload["command"], "smoke_staging")
+            self.assertEqual(payload["command"], "python manage.py smoke_staging --artifact <path>")
             self.assertEqual(payload["players"], ["P1", "P2", "P3"])
             self.assertIn("generated_at", payload)
             self.assertIn("session_code", payload)
+            self.assertEqual(payload["question"]["correct_answer"], "Correct")
+            self.assertEqual(payload["guess_plan"]["P2"], "Correct")
+
+            step_names = [step["step"] for step in payload["steps"]]
             self.assertEqual(
-                payload["steps"],
+                step_names,
                 [
                     "create_session",
                     "join_players",
                     "start_round",
-                    "submit_lies",
                     "auto_guess_transition",
                     "submit_guesses",
-                    "auto_reveal_to_scoreboard",
+                    "auto_reveal_transition",
+                    "auto_scoreboard_transition",
                     "finish_game",
                 ],
             )
 
+            reveal_step = payload["steps"][5]
+            self.assertEqual(reveal_step["session_status"], GameSession.Status.REVEAL)
+            self.assertEqual(reveal_step["reveal"]["correct_answer"], "Correct")
+            self.assertEqual(reveal_step["reveal"]["lies_count"], 3)
+            self.assertEqual(reveal_step["reveal"]["guesses_count"], 3)
+            self.assertEqual(len(reveal_step["reveal"]["fooled_player_ids"]), 2)
+            self.assertEqual(len(reveal_step["reveal"]["correct_guess_player_ids"]), 1)
+
+            scoreboard_step = payload["steps"][6]
+            self.assertEqual(scoreboard_step["session_status"], GameSession.Status.SCOREBOARD)
+            self.assertEqual(len(scoreboard_step["leaderboard"]), 3)
+
 
 class I18nResolverTests(TestCase):
     def test_resolve_locale_accepts_language_tags_and_normalizes_to_supported_base_locale(self):