319 lines
11 KiB
Python
319 lines
11 KiB
Python
"""端到端模拟两人实时竞赛，覆盖同分不同时长的平局判定场景。"""
|
|
|
|
import pytest
|
|
from django.test import Client
|
|
from django.utils import timezone
|
|
from datetime import timedelta
|
|
|
|
from accounts.models import User
|
|
from contest.models import (
|
|
Contest,
|
|
ContestAttempt,
|
|
ContestQuestion,
|
|
Question,
|
|
QuestionVersion,
|
|
RealtimeMatch,
|
|
)
|
|
|
|
|
|
def _setup_realtime_contest(db, slug="tiebreak-contest", duration=60):
|
|
question = Question.objects.create(
|
|
slug=f"{slug}-q",
|
|
track=Question.Track.STANDARD,
|
|
)
|
|
version = QuestionVersion.objects.create(
|
|
question=question,
|
|
version=1,
|
|
prompt="18 + 24",
|
|
answer="42",
|
|
explanation="18 + 24 = 42",
|
|
)
|
|
contest = Contest.objects.create(
|
|
slug=slug,
|
|
title="同分决胜测试赛",
|
|
kind=Contest.Kind.REALTIME,
|
|
track=Question.Track.STANDARD,
|
|
status=Contest.Status.PUBLISHED,
|
|
duration_seconds=duration,
|
|
)
|
|
ContestQuestion.objects.create(
|
|
contest=contest,
|
|
question_version=version,
|
|
order=1,
|
|
points=100,
|
|
)
|
|
return contest
|
|
|
|
|
|
def _create_user(username, nickname):
|
|
return User.objects.create_user(
|
|
username=username,
|
|
password="StrongPass_2026",
|
|
nickname=nickname,
|
|
)
|
|
|
|
|
|
def _client(user):
|
|
c = Client()
|
|
c.force_login(user)
|
|
return c
|
|
|
|
|
|
def _create_and_join(first_client, second_client, contest):
|
|
created = first_client.post(
|
|
f"/api/v1/contests/{contest.slug}/challenges/",
|
|
{},
|
|
content_type="application/json",
|
|
)
|
|
assert created.status_code == 201
|
|
code = created.json()["challenge_code"]
|
|
|
|
joined = second_client.post(
|
|
"/api/v1/contests/challenges/join/",
|
|
{"challenge_code": code},
|
|
content_type="application/json",
|
|
)
|
|
assert joined.status_code == 200
|
|
assert joined.json()["status"] == "active"
|
|
match_id = joined.json()["match_id"]
|
|
|
|
first_state = first_client.get(f"/api/v1/contests/matches/{match_id}/").json()
|
|
second_state = second_client.get(f"/api/v1/contests/matches/{match_id}/").json()
|
|
return match_id, first_state, second_state
|
|
|
|
|
|
def _set_start(match, user, seconds_ago):
|
|
"""统一设置 match 和 attempt 的 started_at。"""
|
|
t = timezone.now() - timedelta(seconds=seconds_ago)
|
|
match.started_at = t
|
|
match.save(update_fields=["started_at"])
|
|
match.attempts.filter(user=user).update(started_at=t)
|
|
|
|
|
|
def _submit(client, attempt_id, answer, key):
|
|
return client.post(
|
|
f"/api/v1/contests/attempts/{attempt_id}/submit/",
|
|
{"answers": [{"order": 1, "answer": answer}]},
|
|
content_type="application/json",
|
|
HTTP_IDEMPOTENCY_KEY=key,
|
|
)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_同分但一方更快则快者胜(db):
|
|
"""两人答对同一题，但提交时间不同，用时短者获胜。"""
|
|
contest = _setup_realtime_contest(db, "tiebreak-fast")
|
|
first = _create_user("tie_p1", "快方")
|
|
second = _create_user("tie_p2", "慢方")
|
|
fc, sc = _client(first), _client(second)
|
|
|
|
match_id, fs, ss = _create_and_join(fc, sc, contest)
|
|
fa, sa = fs["attempt"]["attempt_id"], ss["attempt"]["attempt_id"]
|
|
|
|
match = RealtimeMatch.objects.get(id=match_id)
|
|
_set_start(match, first, 5)
|
|
r1 = _submit(fc, fa, "42", "tie-fast-one")
|
|
assert r1.status_code == 200
|
|
|
|
_set_start(match, second, 15)
|
|
r2 = _submit(sc, sa, "42", "tie-fast-two")
|
|
assert r2.status_code == 200
|
|
|
|
final = fc.get(f"/api/v1/contests/matches/{match_id}/").json()
|
|
assert final["status"] == "completed"
|
|
assert final["result"]["winner"] == "self"
|
|
assert final["attempt"]["score"] == 100
|
|
assert final["opponent"]["score"] == 100
|
|
assert final["attempt"]["duration_ms"] < final["opponent"]["duration_ms"]
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_同分且同时长则平局(db):
|
|
"""两人答对同一题且用时完全相同，判平局。"""
|
|
contest = _setup_realtime_contest(db, "tiebreak-draw")
|
|
first = _create_user("draw_p1", "甲")
|
|
second = _create_user("draw_p2", "乙")
|
|
fc, sc = _client(first), _client(second)
|
|
|
|
match_id, fs, ss = _create_and_join(fc, sc, contest)
|
|
fa, sa = fs["attempt"]["attempt_id"], ss["attempt"]["attempt_id"]
|
|
|
|
match = RealtimeMatch.objects.get(id=match_id)
|
|
_set_start(match, first, 5)
|
|
_submit(fc, fa, "42", "draw-one")
|
|
|
|
_set_start(match, second, 5)
|
|
_submit(sc, sa, "42", "draw-two")
|
|
|
|
final = fc.get(f"/api/v1/contests/matches/{match_id}/").json()
|
|
assert final["status"] == "completed"
|
|
assert final["result"]["winner"] == "draw"
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_一方答对一方答错则答对者胜(db):
|
|
"""一人答对一人答错，答对者胜（不受时间影响）。"""
|
|
contest = _setup_realtime_contest(db, "tiebreak-correct")
|
|
first = _create_user("corr_p1", "答对方")
|
|
second = _create_user("corr_p2", "答错方")
|
|
fc, sc = _client(first), _client(second)
|
|
|
|
match_id, fs, ss = _create_and_join(fc, sc, contest)
|
|
fa, sa = fs["attempt"]["attempt_id"], ss["attempt"]["attempt_id"]
|
|
|
|
match = RealtimeMatch.objects.get(id=match_id)
|
|
_set_start(match, first, 3)
|
|
_submit(fc, fa, "42", "corr-one")
|
|
|
|
_set_start(match, second, 10)
|
|
_submit(sc, sa, "0", "corr-two")
|
|
|
|
final = fc.get(f"/api/v1/contests/matches/{match_id}/").json()
|
|
assert final["status"] == "completed"
|
|
assert final["result"]["winner"] == "self"
|
|
assert final["attempt"]["score"] == 100
|
|
assert final["opponent"]["score"] == 0
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_双方都答错则快者胜(db):
|
|
"""两人都答错，分数相同（0 分），比较用时，快者胜。"""
|
|
contest = _setup_realtime_contest(db, "tiebreak-both-wrong")
|
|
first = _create_user("wrong_p1", "快错")
|
|
second = _create_user("wrong_p2", "慢错")
|
|
fc, sc = _client(first), _client(second)
|
|
|
|
match_id, fs, ss = _create_and_join(fc, sc, contest)
|
|
fa, sa = fs["attempt"]["attempt_id"], ss["attempt"]["attempt_id"]
|
|
|
|
match = RealtimeMatch.objects.get(id=match_id)
|
|
_set_start(match, first, 5)
|
|
_submit(fc, fa, "0", "wrong-one")
|
|
|
|
_set_start(match, second, 12)
|
|
_submit(sc, sa, "0", "wrong-two")
|
|
|
|
final = fc.get(f"/api/v1/contests/matches/{match_id}/").json()
|
|
assert final["status"] == "completed"
|
|
assert final["result"]["winner"] == "self"
|
|
assert final["attempt"]["score"] == 0
|
|
assert final["opponent"]["score"] == 0
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_超时提交导致分数为零_同分时比较用时(db):
|
|
"""一方超时提交（分数为 0），另一方正常提交也得 0 分，同分快者胜。"""
|
|
contest = _setup_realtime_contest(db, "tiebreak-timeout", duration=10)
|
|
first = _create_user("to_p1", "超时方")
|
|
second = _create_user("to_p2", "正常零分方")
|
|
fc, sc = _client(first), _client(second)
|
|
|
|
match_id, fs, ss = _create_and_join(fc, sc, contest)
|
|
fa, sa = fs["attempt"]["attempt_id"], ss["attempt"]["attempt_id"]
|
|
|
|
match = RealtimeMatch.objects.get(id=match_id)
|
|
_set_start(match, first, 20)
|
|
r1 = _submit(fc, fa, "42", "to-one")
|
|
assert r1.status_code == 200
|
|
assert r1.json()["attempt"]["status"] == "expired"
|
|
|
|
_set_start(match, second, 3)
|
|
_submit(sc, sa, "0", "to-two")
|
|
|
|
final = fc.get(f"/api/v1/contests/matches/{match_id}/").json()
|
|
assert final["status"] == "completed"
|
|
assert final["attempt"]["score"] == 0
|
|
assert final["opponent"]["score"] == 0
|
|
assert final["result"]["winner"] == "opponent"
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_随机匹配同分快者胜(db):
|
|
"""通过随机匹配（非联机码）也能正确触发同分快者胜。"""
|
|
contest = _setup_realtime_contest(db, "tiebreak-random")
|
|
first = _create_user("rnd_p1", "随机快")
|
|
second = _create_user("rnd_p2", "随机慢")
|
|
fc, sc = _client(first), _client(second)
|
|
|
|
r1 = fc.post(
|
|
f"/api/v1/contests/{contest.slug}/matchmaking/",
|
|
{},
|
|
content_type="application/json",
|
|
)
|
|
r2 = sc.post(
|
|
f"/api/v1/contests/{contest.slug}/matchmaking/",
|
|
{},
|
|
content_type="application/json",
|
|
)
|
|
assert r1.json()["status"] == "active" or r2.json()["status"] == "active"
|
|
|
|
match_id = r1.json().get("match_id") or r2.json()["match_id"]
|
|
fs = fc.get(f"/api/v1/contests/matches/{match_id}/").json()
|
|
ss = sc.get(f"/api/v1/contests/matches/{match_id}/").json()
|
|
fa, sa = fs["attempt"]["attempt_id"], ss["attempt"]["attempt_id"]
|
|
|
|
match = RealtimeMatch.objects.get(id=match_id)
|
|
_set_start(match, first, 4)
|
|
_submit(fc, fa, "42", "rnd-one")
|
|
|
|
_set_start(match, second, 8)
|
|
_submit(sc, sa, "42", "rnd-two")
|
|
|
|
final = fc.get(f"/api/v1/contests/matches/{match_id}/").json()
|
|
assert final["status"] == "completed"
|
|
assert final["result"]["winner"] == "self"
|
|
assert final["attempt"]["score"] == 100
|
|
assert final["opponent"]["score"] == 100
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_rating_变化在同分快者胜时正确(db):
|
|
"""同分快者胜时，Elo rating 按正常胜负变化（不平局）。"""
|
|
contest = _setup_realtime_contest(db, "tiebreak-rating")
|
|
first = _create_user("rate_p1", "Rating快")
|
|
second = _create_user("rate_p2", "Rating慢")
|
|
fc, sc = _client(first), _client(second)
|
|
|
|
match_id, fs, ss = _create_and_join(fc, sc, contest)
|
|
fa, sa = fs["attempt"]["attempt_id"], ss["attempt"]["attempt_id"]
|
|
|
|
match = RealtimeMatch.objects.get(id=match_id)
|
|
_set_start(match, first, 5)
|
|
_submit(fc, fa, "42", "rate-one")
|
|
|
|
_set_start(match, second, 12)
|
|
_submit(sc, sa, "42", "rate-two")
|
|
|
|
first.refresh_from_db()
|
|
second.refresh_from_db()
|
|
assert first.rating > 1000
|
|
assert second.rating < 1000
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_双方都超时则由_refresh_自动结算判平(db):
|
|
"""双方都超时（分数都为 0），refresh_match_state 自动结算，duration 统一为时限，判平。"""
|
|
contest = _setup_realtime_contest(db, "tiebreak-both-timeout", duration=10)
|
|
first = _create_user("both_to_p1", "快超时")
|
|
second = _create_user("both_to_p2", "慢超时")
|
|
fc, sc = _client(first), _client(second)
|
|
|
|
match_id, fs, ss = _create_and_join(fc, sc, contest)
|
|
fa, sa = fs["attempt"]["attempt_id"], ss["attempt"]["attempt_id"]
|
|
|
|
# 两个人都不提交，直接让 match 超时
|
|
match = RealtimeMatch.objects.get(id=match_id)
|
|
match.started_at = timezone.now() - timedelta(seconds=15)
|
|
match.save(update_fields=["started_at"])
|
|
match.attempts.update(started_at=match.started_at)
|
|
|
|
from contest.services import refresh_match_state
|
|
refresh_match_state(match.id)
|
|
|
|
final = fc.get(f"/api/v1/contests/matches/{match_id}/").json()
|
|
assert final["status"] == "completed"
|
|
assert final["attempt"]["score"] == 0
|
|
assert final["opponent"]["score"] == 0
|
|
# 双方都超时，duration_ms 相同（都约 15000ms），判平
|
|
assert final["result"]["winner"] == "draw"
|