363 lines
11 KiB
Python
363 lines
11 KiB
Python
from datetime import timedelta
|
|
|
|
import pytest
|
|
from django.utils import timezone
|
|
from rest_framework.exceptions import ValidationError
|
|
|
|
from accounts.models import User
|
|
from contest.models import (
|
|
Contest,
|
|
ContestAttempt,
|
|
ContestQuestion,
|
|
Question,
|
|
QuestionVersion,
|
|
RatingHistory,
|
|
RealtimeMatch,
|
|
)
|
|
from contest.services import (
|
|
cancel_waiting_match,
|
|
create_challenge,
|
|
finalize_match,
|
|
find_match,
|
|
join_challenge,
|
|
match_payload,
|
|
normalize_answer,
|
|
refresh_match_state,
|
|
start_attempt,
|
|
submit_attempt,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def user(db):
|
|
return User.objects.create_user(
|
|
username="contest_user",
|
|
password="StrongPass_2026",
|
|
nickname="比赛用户",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def daily_contest(db):
|
|
question = Question.objects.create(
|
|
slug="sum-question",
|
|
track=Question.Track.STANDARD,
|
|
)
|
|
version = QuestionVersion.objects.create(
|
|
question=question,
|
|
version=1,
|
|
prompt="17 + 25",
|
|
answer="42",
|
|
explanation="相加得 42",
|
|
)
|
|
contest = Contest.objects.create(
|
|
slug="daily-test",
|
|
title="测试今日赛",
|
|
kind=Contest.Kind.DAILY,
|
|
track=Question.Track.STANDARD,
|
|
status=Contest.Status.PUBLISHED,
|
|
duration_seconds=60,
|
|
)
|
|
ContestQuestion.objects.create(
|
|
contest=contest,
|
|
question_version=version,
|
|
order=1,
|
|
points=100,
|
|
)
|
|
return contest
|
|
|
|
|
|
@pytest.fixture
|
|
def realtime_contest(daily_contest):
|
|
daily_contest.kind = Contest.Kind.REALTIME
|
|
daily_contest.slug = "realtime-with-question"
|
|
daily_contest.title = "联机测试赛"
|
|
daily_contest.duration_seconds = 60
|
|
daily_contest.save(update_fields=["kind", "slug", "title", "duration_seconds"])
|
|
return daily_contest
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"raw, expected",
|
|
[
|
|
pytest.param(" 1.0 ", "1", id="小数标准化"),
|
|
pytest.param("ABC ", "abc", id="文本去空格并转小写"),
|
|
pytest.param("-0", "-0", id="保留十进制负零表示"),
|
|
],
|
|
)
|
|
def test_normalize_answer_标准化输入(raw, expected):
|
|
assert normalize_answer(raw) == expected
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_start_attempt_每日赛重复进入复用同一记录(user, daily_contest):
|
|
first = start_attempt(user, daily_contest)
|
|
second = start_attempt(user, daily_contest)
|
|
|
|
assert first["attempt_id"] == second["attempt_id"]
|
|
assert ContestAttempt.objects.count() == 1
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_submit_attempt_服务端判分且幂等重放(user, daily_contest):
|
|
started = start_attempt(user, daily_contest)
|
|
|
|
result = submit_attempt(
|
|
user,
|
|
started["attempt_id"],
|
|
[{"order": 1, "answer": "42.0"}],
|
|
"submission-1",
|
|
)
|
|
replay = submit_attempt(
|
|
user,
|
|
started["attempt_id"],
|
|
[{"order": 1, "answer": "0"}],
|
|
"submission-1",
|
|
)
|
|
|
|
assert result["status"] == ContestAttempt.Status.SUBMITTED
|
|
assert result["score"] == 100
|
|
assert result["correct_count"] == 1
|
|
assert result["questions"][0]["correct_answer"] == "42"
|
|
assert replay["score"] == 100
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_submit_attempt_缺少幂等键时拒绝(user, daily_contest):
|
|
started = start_attempt(user, daily_contest)
|
|
|
|
with pytest.raises(ValidationError, match="幂等键"):
|
|
submit_attempt(
|
|
user,
|
|
started["attempt_id"],
|
|
[{"order": 1, "answer": "42"}],
|
|
None,
|
|
)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_submit_attempt_畸形题号返回校验错误且记录保持进行中(user, daily_contest):
|
|
started = start_attempt(user, daily_contest)
|
|
|
|
with pytest.raises(ValidationError):
|
|
submit_attempt(
|
|
user,
|
|
started["attempt_id"],
|
|
[{"order": "first", "answer": "42"}],
|
|
"malformed-order",
|
|
)
|
|
|
|
attempt = ContestAttempt.objects.get(id=started["attempt_id"])
|
|
assert attempt.status == ContestAttempt.Status.ACTIVE
|
|
assert attempt.answers.count() == 0
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_submit_attempt_超过服务端时限不计分(user, daily_contest):
|
|
started = start_attempt(user, daily_contest)
|
|
ContestAttempt.objects.filter(id=started["attempt_id"]).update(
|
|
started_at=timezone.now() - timedelta(seconds=61)
|
|
)
|
|
|
|
result = submit_attempt(
|
|
user,
|
|
started["attempt_id"],
|
|
[{"order": 1, "answer": "42"}],
|
|
"late-submit",
|
|
)
|
|
|
|
assert result["status"] == ContestAttempt.Status.EXPIRED
|
|
assert result["score"] == 0
|
|
assert result["correct_count"] == 0
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_find_match_同分玩家配对并由_finalize_match_唯一结算_rating():
|
|
first = User.objects.create_user(
|
|
username="player_one",
|
|
password="StrongPass_2026",
|
|
nickname="玩家一",
|
|
)
|
|
second = User.objects.create_user(
|
|
username="player_two",
|
|
password="StrongPass_2026",
|
|
nickname="玩家二",
|
|
)
|
|
contest = Contest.objects.create(
|
|
slug="realtime-test",
|
|
title="测试实时赛",
|
|
kind=Contest.Kind.REALTIME,
|
|
track=Question.Track.STANDARD,
|
|
status=Contest.Status.PUBLISHED,
|
|
)
|
|
|
|
waiting = find_match(first, contest)
|
|
active = find_match(second, contest)
|
|
active.refresh_from_db()
|
|
|
|
assert waiting.id == active.id
|
|
assert active.status == RealtimeMatch.Status.ACTIVE
|
|
assert active.attempts.count() == 2
|
|
|
|
active.attempts.filter(user=first).update(
|
|
status=ContestAttempt.Status.SUBMITTED,
|
|
score=200,
|
|
)
|
|
active.attempts.filter(user=second).update(
|
|
status=ContestAttempt.Status.SUBMITTED,
|
|
score=100,
|
|
)
|
|
finalized = finalize_match(active.id)
|
|
first.refresh_from_db()
|
|
second.refresh_from_db()
|
|
|
|
assert finalized.status == RealtimeMatch.Status.COMPLETED
|
|
assert finalized.winner == first
|
|
assert first.rating == 1016
|
|
assert second.rating == 984
|
|
assert RatingHistory.objects.filter(match=active).count() == 2
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_challenge_code_创建者和加入者通过联机码进入同一场(realtime_contest):
|
|
first = User.objects.create_user(
|
|
username="challenge_owner",
|
|
password="StrongPass_2026",
|
|
nickname="房主",
|
|
)
|
|
second = User.objects.create_user(
|
|
username="challenge_guest",
|
|
password="StrongPass_2026",
|
|
nickname="访客",
|
|
)
|
|
|
|
waiting = create_challenge(first, realtime_contest)
|
|
active = join_challenge(second, waiting.challenge_code.lower())
|
|
owner_payload = match_payload(active, first)
|
|
guest_payload = match_payload(active, second)
|
|
|
|
assert len(waiting.challenge_code) == 6
|
|
assert active.id == waiting.id
|
|
assert active.match_type == RealtimeMatch.MatchType.CHALLENGE
|
|
assert active.status == RealtimeMatch.Status.ACTIVE
|
|
assert active.attempts.count() == 2
|
|
assert owner_payload["opponent"]["nickname"] == "访客"
|
|
assert guest_payload["opponent"]["nickname"] == "房主"
|
|
assert owner_payload["attempt"]["questions"] == guest_payload["attempt"]["questions"]
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_random_match_不会加入联机码约战(realtime_contest):
|
|
owner = User.objects.create_user(
|
|
username="private_owner",
|
|
password="StrongPass_2026",
|
|
nickname="约战房主",
|
|
)
|
|
random_player = User.objects.create_user(
|
|
username="random_player",
|
|
password="StrongPass_2026",
|
|
nickname="随机玩家",
|
|
)
|
|
|
|
challenge = create_challenge(owner, realtime_contest)
|
|
random_match = find_match(random_player, realtime_contest)
|
|
|
|
assert challenge.status == RealtimeMatch.Status.WAITING
|
|
assert random_match.id != challenge.id
|
|
assert random_match.match_type == RealtimeMatch.MatchType.RANDOM
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_realtime_submit_双方结束前不泄露答案且结束后结算(realtime_contest):
|
|
first = User.objects.create_user(
|
|
username="fair_player_one",
|
|
password="StrongPass_2026",
|
|
nickname="公平玩家一",
|
|
)
|
|
second = User.objects.create_user(
|
|
username="fair_player_two",
|
|
password="StrongPass_2026",
|
|
nickname="公平玩家二",
|
|
)
|
|
match = join_challenge(
|
|
second,
|
|
create_challenge(first, realtime_contest).challenge_code,
|
|
)
|
|
first_attempt = match.attempts.get(user=first)
|
|
second_attempt = match.attempts.get(user=second)
|
|
|
|
first_result = submit_attempt(
|
|
first,
|
|
first_attempt.id,
|
|
[{"order": 1, "answer": "42"}],
|
|
"fair-submit-one",
|
|
)
|
|
active_payload = match_payload(match, first)
|
|
|
|
assert "correct_answer" not in first_result["questions"][0]
|
|
assert active_payload["status"] == RealtimeMatch.Status.ACTIVE
|
|
assert active_payload["attempt"]["status"] == ContestAttempt.Status.SUBMITTED
|
|
|
|
submit_attempt(
|
|
second,
|
|
second_attempt.id,
|
|
[{"order": 1, "answer": "0"}],
|
|
"fair-submit-two",
|
|
)
|
|
match.refresh_from_db()
|
|
completed_payload = match_payload(match, first)
|
|
|
|
assert match.status == RealtimeMatch.Status.COMPLETED
|
|
assert completed_payload["result"]["winner"] == "self"
|
|
assert completed_payload["attempt"]["questions"][0]["correct_answer"] == "42"
|
|
assert completed_payload["opponent"]["score"] == 0
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_realtime_timeout_未提交玩家自动过期并完成比赛(realtime_contest):
|
|
first = User.objects.create_user(
|
|
username="timeout_one",
|
|
password="StrongPass_2026",
|
|
nickname="超时玩家一",
|
|
)
|
|
second = User.objects.create_user(
|
|
username="timeout_two",
|
|
password="StrongPass_2026",
|
|
nickname="超时玩家二",
|
|
)
|
|
match = join_challenge(
|
|
second,
|
|
create_challenge(first, realtime_contest).challenge_code,
|
|
)
|
|
RealtimeMatch.objects.filter(id=match.id).update(
|
|
started_at=timezone.now() - timedelta(seconds=61)
|
|
)
|
|
|
|
refreshed = refresh_match_state(match.id)
|
|
|
|
assert refreshed.status == RealtimeMatch.Status.COMPLETED
|
|
assert not refreshed.attempts.filter(status=ContestAttempt.Status.ACTIVE).exists()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_challenge_owner_可取消等待中的联机码(realtime_contest):
|
|
owner = User.objects.create_user(
|
|
username="cancel_owner",
|
|
password="StrongPass_2026",
|
|
nickname="取消房主",
|
|
)
|
|
waiting = create_challenge(owner, realtime_contest)
|
|
|
|
cancelled = cancel_waiting_match(owner, waiting.id)
|
|
|
|
assert cancelled.status == RealtimeMatch.Status.CANCELLED
|
|
with pytest.raises(ValidationError, match="失效"):
|
|
join_challenge(
|
|
User.objects.create_user(
|
|
username="late_guest",
|
|
password="StrongPass_2026",
|
|
nickname="迟到访客",
|
|
),
|
|
waiting.challenge_code,
|
|
)
|