257 lines
8.6 KiB
Python
257 lines
8.6 KiB
Python
from datetime import timedelta
|
|
|
|
import pytest
|
|
from django.test import Client
|
|
from django.utils import timezone
|
|
|
|
from accounts.models import User
|
|
from contest.models import (
|
|
Contest,
|
|
ContestQuestion,
|
|
Question,
|
|
QuestionVersion,
|
|
RealtimeMatch,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def realtime_api_setup(db):
|
|
question = Question.objects.create(
|
|
slug="realtime-api-question",
|
|
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="realtime-api",
|
|
title="API 联机赛",
|
|
kind=Contest.Kind.REALTIME,
|
|
track=Question.Track.STANDARD,
|
|
status=Contest.Status.PUBLISHED,
|
|
duration_seconds=60,
|
|
)
|
|
ContestQuestion.objects.create(
|
|
contest=contest,
|
|
question_version=version,
|
|
order=1,
|
|
points=100,
|
|
)
|
|
first = User.objects.create_user(
|
|
username="api_player_one",
|
|
password="StrongPass_2026",
|
|
nickname="API 玩家一",
|
|
)
|
|
second = User.objects.create_user(
|
|
username="api_player_two",
|
|
password="StrongPass_2026",
|
|
nickname="API 玩家二",
|
|
)
|
|
first_client = Client()
|
|
second_client = Client()
|
|
first_client.force_login(first)
|
|
second_client.force_login(second)
|
|
return contest, first_client, second_client
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_challenge_api_创建加入状态提交形成完整闭环(realtime_api_setup):
|
|
contest, first_client, second_client = realtime_api_setup
|
|
created = first_client.post(
|
|
f"/api/v1/contests/{contest.slug}/challenges/",
|
|
{},
|
|
content_type="application/json",
|
|
)
|
|
code = created.json()["challenge_code"]
|
|
|
|
joined = second_client.post(
|
|
"/api/v1/contests/challenges/join/",
|
|
{"challenge_code": code.lower()},
|
|
content_type="application/json",
|
|
)
|
|
match_id = joined.json()["match_id"]
|
|
first_state = first_client.get(f"/api/v1/contests/matches/{match_id}/")
|
|
|
|
assert created.status_code == 201
|
|
assert joined.status_code == 200
|
|
assert joined.json()["status"] == "active"
|
|
assert first_state.json()["opponent"]["nickname"] == "API 玩家二"
|
|
|
|
first_attempt = first_state.json()["attempt"]["attempt_id"]
|
|
second_attempt = joined.json()["attempt"]["attempt_id"]
|
|
first_submit = first_client.post(
|
|
f"/api/v1/contests/attempts/{first_attempt}/submit/",
|
|
{"answers": [{"order": 1, "answer": "42"}]},
|
|
content_type="application/json",
|
|
HTTP_IDEMPOTENCY_KEY="api-submit-one",
|
|
)
|
|
second_submit = second_client.post(
|
|
f"/api/v1/contests/attempts/{second_attempt}/submit/",
|
|
{"answers": [{"order": 1, "answer": "0"}]},
|
|
content_type="application/json",
|
|
HTTP_IDEMPOTENCY_KEY="api-submit-two",
|
|
)
|
|
|
|
assert first_submit.json()["status"] == "active"
|
|
assert "correct_answer" not in first_submit.json()["attempt"]["questions"][0]
|
|
assert second_submit.json()["status"] == "completed"
|
|
assert second_submit.json()["result"]["winner"] == "opponent"
|
|
final_state = first_client.get(f"/api/v1/contests/matches/{match_id}/").json()
|
|
assert final_state["result"]["winner"] == "self"
|
|
assert final_state["attempt"]["questions"][0]["correct_answer"] == "42"
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_challenge_api_无效联机码返回具体原因_request_id_和诊断日志(
|
|
realtime_api_setup,
|
|
caplog,
|
|
):
|
|
_, _, second_client = realtime_api_setup
|
|
caplog.set_level("INFO", logger="common.api")
|
|
|
|
response = second_client.post(
|
|
"/api/v1/contests/challenges/join/",
|
|
{"challenge_code": "ABC234"},
|
|
content_type="application/json",
|
|
HTTP_X_REQUEST_ID="challenge-invalid-test",
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response["X-Request-ID"] == "challenge-invalid-test"
|
|
assert response.json()["error"] == {
|
|
"code": "invalid",
|
|
"message": "联机码不存在",
|
|
"details": {"challenge_code": "联机码不存在"},
|
|
"request_id": "challenge-invalid-test",
|
|
}
|
|
assert "api_request_error method=POST" in caplog.text
|
|
assert "path=/api/v1/contests/challenges/join/" in caplog.text
|
|
assert "fields=challenge_code message=联机码不存在" in caplog.text
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_challenge_api_创建者不能加入自己的联机码(realtime_api_setup):
|
|
contest, first_client, _ = realtime_api_setup
|
|
created = first_client.post(
|
|
f"/api/v1/contests/{contest.slug}/challenges/",
|
|
{},
|
|
content_type="application/json",
|
|
)
|
|
|
|
response = first_client.post(
|
|
"/api/v1/contests/challenges/join/",
|
|
{"challenge_code": created.json()["challenge_code"]},
|
|
content_type="application/json",
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.json()["error"]["message"] == "不能加入自己创建的约战"
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_challenge_api_过期联机码返回具体原因(realtime_api_setup):
|
|
contest, first_client, second_client = realtime_api_setup
|
|
created = first_client.post(
|
|
f"/api/v1/contests/{contest.slug}/challenges/",
|
|
{},
|
|
content_type="application/json",
|
|
)
|
|
RealtimeMatch.objects.filter(id=created.json()["match_id"]).update(
|
|
expires_at=timezone.now() - timedelta(seconds=1)
|
|
)
|
|
|
|
response = second_client.post(
|
|
"/api/v1/contests/challenges/join/",
|
|
{"challenge_code": created.json()["challenge_code"]},
|
|
content_type="application/json",
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.json()["error"]["message"] == "联机码已失效或已被使用"
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_challenge_api_已使用联机码返回具体原因(realtime_api_setup):
|
|
contest, first_client, second_client = realtime_api_setup
|
|
created = first_client.post(
|
|
f"/api/v1/contests/{contest.slug}/challenges/",
|
|
{},
|
|
content_type="application/json",
|
|
)
|
|
second_client.post(
|
|
"/api/v1/contests/challenges/join/",
|
|
{"challenge_code": created.json()["challenge_code"]},
|
|
content_type="application/json",
|
|
)
|
|
third = User.objects.create_user(
|
|
username="api_player_three",
|
|
password="StrongPass_2026",
|
|
nickname="API 玩家三",
|
|
)
|
|
third_client = Client()
|
|
third_client.force_login(third)
|
|
|
|
response = third_client.post(
|
|
"/api/v1/contests/challenges/join/",
|
|
{"challenge_code": created.json()["challenge_code"]},
|
|
content_type="application/json",
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.json()["error"]["message"] == "联机码已失效或已被使用"
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_challenge_api_24点竞速完整闭环(realtime_api_setup):
|
|
contest, first_client, second_client = realtime_api_setup
|
|
created = first_client.post(
|
|
f"/api/v1/contests/{contest.slug}/challenges/",
|
|
{"game_kind": "twenty_four"},
|
|
content_type="application/json",
|
|
)
|
|
joined = second_client.post(
|
|
"/api/v1/contests/challenges/join/",
|
|
{"challenge_code": created.json()["challenge_code"]},
|
|
content_type="application/json",
|
|
)
|
|
|
|
assert joined.status_code == 200
|
|
assert joined.json()["game_kind"] == "twenty_four"
|
|
assert created.json()["attempt"] is None
|
|
first_state = first_client.get(
|
|
f"/api/v1/contests/matches/{joined.json()['match_id']}/"
|
|
).json()
|
|
assert first_state["attempt"]["puzzle"] == joined.json()["attempt"]["puzzle"]
|
|
|
|
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(joined.json()["attempt"]["puzzle"]["numbers"]))
|
|
]
|
|
first_submit = first_client.post(
|
|
f"/api/v1/contests/games/attempts/{first_state['attempt']['attempt_id']}/submit/",
|
|
{"expression": expression},
|
|
content_type="application/json",
|
|
HTTP_IDEMPOTENCY_KEY="api-game-first",
|
|
)
|
|
second_submit = second_client.post(
|
|
f"/api/v1/contests/games/attempts/{joined.json()['attempt']['attempt_id']}/submit/",
|
|
{"expression": expression},
|
|
content_type="application/json",
|
|
HTTP_IDEMPOTENCY_KEY="api-game-second",
|
|
)
|
|
|
|
assert first_submit.status_code == 200
|
|
assert first_submit.json()["status"] == "active"
|
|
assert first_submit.json()["attempt"]["status"] == "completed"
|
|
assert second_submit.status_code == 200
|
|
assert second_submit.json()["status"] == "completed"
|
|
assert second_submit.json()["result"]["winner"] in {"self", "opponent", "draw"}
|