feat: add unified contest pool and realtime game modes

This commit is contained in:
2026-08-10 00:31:21 +08:00
parent 97ca20413e
commit 40e9462842
18 changed files with 822 additions and 152 deletions
+119
View File
@@ -5,10 +5,12 @@ from django.utils import timezone
from rest_framework.exceptions import ValidationError
from accounts.models import User
from contest.game_services import submit_game
from contest.models import (
Contest,
ContestAttempt,
ContestQuestion,
MathGameAttempt,
Question,
QuestionVersion,
RatingHistory,
@@ -17,6 +19,7 @@ from contest.models import (
from contest.services import (
cancel_waiting_match,
create_challenge,
finalize_game_match,
finalize_match,
find_match,
join_challenge,
@@ -190,6 +193,21 @@ def test_find_match_同分玩家配对并由_finalize_match_唯一结算_rating(
track=Question.Track.STANDARD,
status=Contest.Status.PUBLISHED,
)
question = Question.objects.create(
slug="realtime-rating-question",
track=Question.Track.STANDARD,
)
version = QuestionVersion.objects.create(
question=question,
version=1,
prompt="1 + 1",
answer="2",
)
ContestQuestion.objects.create(
contest=contest,
question_version=version,
order=1,
)
waiting = find_match(first, contest)
active = find_match(second, contest)
@@ -360,3 +378,104 @@ def test_challenge_owner_可取消等待中的联机码(realtime_contest):
),
waiting.challenge_code,
)
@pytest.mark.django_db
def test_practice_attempt_从完整题池保存随机快照(user):
contest = Contest.objects.create(
slug="random-practice",
title="随机练习",
kind=Contest.Kind.PRACTICE,
track=Question.Track.STANDARD,
status=Contest.Status.PUBLISHED,
)
for index in range(12):
question = Question.objects.create(
slug=f"random-question-{index}",
track=Question.Track.STANDARD,
)
version = QuestionVersion.objects.create(
question=question,
version=1,
prompt=f"{index} + 1",
answer=str(index + 1),
)
ContestQuestion.objects.create(
contest=contest,
question_version=version,
order=index + 1,
)
payloads = [start_attempt(user, contest) for _ in range(8)]
snapshots = {
tuple(
ContestAttempt.objects.get(id=payload["attempt_id"]).question_order
)
for payload in payloads
}
assert all(len(payload["questions"]) == 8 for payload in payloads)
assert len(snapshots) > 1
@pytest.mark.django_db
def test_twenty_four_realtime_共享题面按完成时间结算_rating(realtime_contest):
first = User.objects.create_user(
username="game_match_one",
password="StrongPass_2026",
nickname="竞速玩家一",
)
second = User.objects.create_user(
username="game_match_two",
password="StrongPass_2026",
nickname="竞速玩家二",
)
waiting = create_challenge(
first,
realtime_contest,
RealtimeMatch.GameKind.TWENTY_FOUR,
)
match = join_challenge(second, waiting.challenge_code)
attempts = {
attempt.user_id: attempt for attempt in match.game_attempts.all()
}
first_attempt = attempts[first.id]
second_attempt = attempts[second.id]
assert first_attempt.puzzle == second_attempt.puzzle
assert first_attempt.solution == second_attempt.solution
solutions = {
(2, 3, 4, 9): "(2/3)*4*9",
(3, 5, 7, 13): "(7+5*13)/3",
(4, 7, 8, 8): "8*(4+7-8)",
}
expression = solutions[tuple(sorted(first_attempt.puzzle["numbers"]))]
MathGameAttempt.objects.filter(id=first_attempt.id).update(
started_at=timezone.now() - timedelta(seconds=2)
)
MathGameAttempt.objects.filter(id=second_attempt.id).update(
started_at=timezone.now() - timedelta(seconds=5)
)
submit_game(
first,
first_attempt.id,
{"expression": expression},
"game-race-first",
)
assert finalize_game_match(match.id).status == RealtimeMatch.Status.ACTIVE
submit_game(
second,
second_attempt.id,
{"expression": expression},
"game-race-second",
)
completed = finalize_game_match(match.id)
first.refresh_from_db()
second.refresh_from_db()
assert completed.status == RealtimeMatch.Status.COMPLETED
assert completed.winner_id == first.id
assert first.rating == 1016
assert second.rating == 984
assert RatingHistory.objects.filter(match=match).count() == 2