NotBigGhostandClaude Opus 5
976b8fc622
Объявление администрации показывается игроку окном в свой период, пока игрок
не закроет его («Понятно»). Отметка о закрытии хранится на сервере с номером
версии: правка с «показать заново» поднимает версию, и закрывшие прежнюю
увидят объявление снова — ответ помечен updated («обновлено»). Флаг
show_to_new_players=false прячет объявление от зарегистрировавшихся после
начала показа. Пересекающиеся объявления идут от старого к новому.
Текст приходит HTML-ом из редактора админки и сохраняется только после
очистки по белому списку (b, em, mark и mark.red, p, br): атрибуты
отбрасываются, script/style/svg — вместе с содержимым, текст экранируется
заново. Фронт вставляет только этот HTML.
API: GET /api/announcements/pending, POST /api/announcements/{id}/ack;
админка — список со статусом и счётчиком «закрыли N из M», создание, правка,
«снять с показа», удаление, всё в аудит. SSE-событие announcements активным
игрокам. Миграция 0015 идемпотентная.
Тесты: очистка (XSS-попытки, вложенные div), права, период и порядок,
«новые игроки», повторный показ, снятие, удаление, валидация. #84
Перенесено в main без рейтинга из a0a0e52; миграция 0015 — сразу от 0013
(как в ec0445f).
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
287 lines
12 KiB
Python
287 lines
12 KiB
Python
"""Объявления администрации (#84): очистка HTML, права, период показа, «новые игроки»,
|
||
повторный показ с пометкой «обновлено», снятие с показа и удаление."""
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlmodel import Session, select
|
||
|
||
from app.models import AnnouncementView, User
|
||
from app.services import notify
|
||
from app.services.announcement_service import sanitize_body
|
||
from tests.conftest import csrf_headers, login
|
||
|
||
|
||
def _iso(dt: datetime) -> str:
|
||
return dt.astimezone(timezone.utc).isoformat()
|
||
|
||
|
||
def _now() -> datetime:
|
||
return datetime.now(timezone.utc)
|
||
|
||
|
||
def _admin_login(client: TestClient, make_admin) -> None:
|
||
make_admin("admin", "secret123")
|
||
r = client.post(
|
||
"/api/admin/auth/login",
|
||
json={"username": "admin", "password": "secret123"},
|
||
headers=csrf_headers(client),
|
||
)
|
||
assert r.status_code == 200, r.text
|
||
|
||
|
||
def _create(client: TestClient, **over) -> dict:
|
||
body = {
|
||
"title": "Турнир",
|
||
"body_html": "<p>Суббота, <b>11:00</b></p>",
|
||
"starts_at": _iso(_now() - timedelta(hours=1)),
|
||
"ends_at": _iso(_now() + timedelta(days=1)),
|
||
"show_to_new_players": True,
|
||
}
|
||
body.update(over)
|
||
r = client.post("/api/admin/announcements", json=body, headers=csrf_headers(client))
|
||
assert r.status_code == 200, r.text
|
||
return r.json()
|
||
|
||
|
||
def _update(client: TestClient, ann: dict, **over) -> dict:
|
||
body = {
|
||
"title": ann["title"],
|
||
"body_html": ann["body_html"],
|
||
"starts_at": ann["starts_at"],
|
||
"ends_at": ann["ends_at"],
|
||
"show_to_new_players": ann["show_to_new_players"],
|
||
"reshow": False,
|
||
}
|
||
body.update(over)
|
||
r = client.put(
|
||
f"/api/admin/announcements/{ann['id']}", json=body, headers=csrf_headers(client)
|
||
)
|
||
assert r.status_code == 200, r.text
|
||
return r.json()
|
||
|
||
|
||
def _pending(client: TestClient) -> list[dict]:
|
||
r = client.get("/api/announcements/pending")
|
||
assert r.status_code == 200, r.text
|
||
return r.json()
|
||
|
||
|
||
def _ack(client: TestClient, ann_id: int, revision: int):
|
||
return client.post(
|
||
f"/api/announcements/{ann_id}/ack",
|
||
json={"revision": revision},
|
||
headers=csrf_headers(client),
|
||
)
|
||
|
||
|
||
def _admin_item(client: TestClient, ann_id: int) -> dict:
|
||
return next(a for a in client.get("/api/admin/announcements").json() if a["id"] == ann_id)
|
||
|
||
|
||
# ─── Очистка HTML ────────────────────────────────────────────────────────────
|
||
|
||
def test_sanitize_keeps_allowed_and_strips_everything_else():
|
||
raw = (
|
||
'<div>Привет <strong onclick="steal()">мир</strong></div>'
|
||
'<p style="color:red"><i>курсив</i> <mark class="red big">красный</mark> '
|
||
'<mark class="gold" onmouseover="x()">золотой</mark></p>'
|
||
"<script>alert(1)</script><style>p{display:none}</style>"
|
||
'<img src=x onerror="alert(1)"><a href="javascript:alert(1)">ссылка</a>'
|
||
"<svg><text>svg-текст</text></svg><b>"
|
||
)
|
||
html, text = sanitize_body(raw)
|
||
assert html == (
|
||
"<p>Привет <b>мир</b></p>"
|
||
'<p><em>курсив</em> <mark class="red">красный</mark> <mark>золотой</mark></p>'
|
||
"ссылка<b>"
|
||
)
|
||
# Содержимое script/style/svg в видимый текст (и в счётчик длины) не попадает.
|
||
assert text == "Привет миркурсив красный золотойссылка<b>"
|
||
|
||
|
||
def test_sanitize_fixes_nesting_and_keeps_line_breaks():
|
||
html, _ = sanitize_body("<div><div>один</div><div>два<br></div></div><b>жирный<br/>хвост")
|
||
# Вложенные <div> из contenteditable → плоские абзацы без пустых <p></p>;
|
||
# <br> внутри <b> не путается с самим <b>; незакрытое закрывается.
|
||
assert html == "<p>один</p><p>два<br></p><b>жирный<br>хвост</b>"
|
||
|
||
|
||
# ─── Права ───────────────────────────────────────────────────────────────────
|
||
|
||
def test_admin_endpoints_are_closed_to_players(client: TestClient):
|
||
assert client.get("/api/admin/announcements").status_code == 401
|
||
login(client, "Игрок")
|
||
assert client.get("/api/admin/announcements").status_code == 401
|
||
r = client.post(
|
||
"/api/admin/announcements",
|
||
json={
|
||
"title": "x",
|
||
"body_html": "y",
|
||
"starts_at": _iso(_now()),
|
||
"ends_at": _iso(_now() + timedelta(days=1)),
|
||
},
|
||
headers=csrf_headers(client),
|
||
)
|
||
assert r.status_code == 401
|
||
|
||
|
||
def test_pending_requires_player_session(client: TestClient):
|
||
assert client.get("/api/announcements/pending").status_code == 401
|
||
|
||
|
||
# ─── Показ ───────────────────────────────────────────────────────────────────
|
||
|
||
def test_created_announcement_is_shown_once(client: TestClient, make_admin, monkeypatch):
|
||
events: list[dict] = []
|
||
monkeypatch.setattr(notify.hub, "publish", lambda ids, ev: events.append(ev))
|
||
login(client, "Игрок")
|
||
_admin_login(client, make_admin)
|
||
|
||
ann = _create(client, body_html="<p>Сбор <b>в 11:00</b></p><script>alert(1)</script>")
|
||
assert ann["body_html"] == "<p>Сбор <b>в 11:00</b></p>"
|
||
assert ann["status"] == "live"
|
||
assert (ann["seen_count"], ann["audience_count"]) == (0, 1)
|
||
assert {"type": "announcements"} in events
|
||
|
||
items = _pending(client)
|
||
assert [(a["id"], a["revision"], a["updated"]) for a in items] == [(ann["id"], 1, False)]
|
||
assert _ack(client, ann["id"], 1).status_code == 200
|
||
assert _pending(client) == []
|
||
assert _admin_item(client, ann["id"])["seen_count"] == 1
|
||
|
||
logs = client.get("/api/admin/audit-logs?entity_type=announcement").json()["items"]
|
||
assert [(l["action"], l["entity_id"]) for l in logs] == [("create", ann["id"])]
|
||
|
||
|
||
def test_pending_respects_period_and_order(client: TestClient, make_admin):
|
||
login(client, "Игрок")
|
||
_admin_login(client, make_admin)
|
||
later = _create(client, title="Позже", starts_at=_iso(_now() - timedelta(minutes=30)))
|
||
earlier = _create(client, title="Раньше", starts_at=_iso(_now() - timedelta(hours=3)))
|
||
planned = _create(
|
||
client,
|
||
title="Завтра",
|
||
starts_at=_iso(_now() + timedelta(days=1)),
|
||
ends_at=_iso(_now() + timedelta(days=2)),
|
||
)
|
||
assert planned["status"] == "planned"
|
||
# Пересекающиеся периоды — от старого к новому; запланированного пока нет.
|
||
assert [a["title"] for a in _pending(client)] == ["Раньше", "Позже"]
|
||
assert earlier["id"] != later["id"]
|
||
|
||
# Период, который уже закончился, создать нельзя.
|
||
r = client.post(
|
||
"/api/admin/announcements",
|
||
json={
|
||
"title": "Прошлое",
|
||
"body_html": "текст",
|
||
"starts_at": _iso(_now() - timedelta(days=2)),
|
||
"ends_at": _iso(_now() - timedelta(days=1)),
|
||
},
|
||
headers=csrf_headers(client),
|
||
)
|
||
assert r.status_code == 422
|
||
|
||
|
||
def test_hidden_from_players_registered_after_start(client: TestClient, make_admin, engine):
|
||
login(client, "Старожил")
|
||
with Session(engine) as s:
|
||
old = s.exec(select(User).where(User.nickname == "Старожил")).one()
|
||
old.created_at = datetime.now(timezone.utc) - timedelta(days=2)
|
||
s.add(old)
|
||
s.commit()
|
||
_admin_login(client, make_admin)
|
||
ann = _create(client, show_to_new_players=False)
|
||
assert ann["audience_count"] == 1
|
||
|
||
assert [a["id"] for a in _pending(client)] == [ann["id"]]
|
||
login(client, "Новичок") # зарегистрирован уже после начала показа
|
||
assert _pending(client) == []
|
||
# Адресаты — по-прежнему только старожил.
|
||
assert _admin_item(client, ann["id"])["audience_count"] == 1
|
||
|
||
|
||
def test_reshow_brings_announcement_back_marked_updated(client: TestClient, make_admin):
|
||
login(client, "Игрок")
|
||
_admin_login(client, make_admin)
|
||
ann = _create(client)
|
||
assert _ack(client, ann["id"], 1).status_code == 200
|
||
|
||
# Правка без «показать заново» — закрывшие её не видят.
|
||
ann = _update(client, ann, body_html="<p>Сбор в 12:00</p>")
|
||
assert ann["revision"] == 1
|
||
assert _pending(client) == []
|
||
|
||
ann = _update(client, ann, body_html="<p>Сбор в 13:00</p>", reshow=True)
|
||
assert ann["revision"] == 2
|
||
assert ann["seen_count"] == 0 # закрывших новую версию ещё нет
|
||
items = _pending(client)
|
||
assert [(a["revision"], a["updated"], a["body_html"]) for a in items] == [
|
||
(2, True, "<p>Сбор в 13:00</p>")
|
||
]
|
||
|
||
# Пока окно висело, админ выпустил третью версию: закрыв вторую, игрок увидит третью.
|
||
ann = _update(client, ann, reshow=True)
|
||
assert _ack(client, ann["id"], 2).status_code == 200
|
||
assert [(a["revision"], a["updated"]) for a in _pending(client)] == [(3, True)]
|
||
assert _ack(client, ann["id"], 3).status_code == 200
|
||
assert _pending(client) == []
|
||
|
||
|
||
def test_stop_and_delete(client: TestClient, make_admin, engine):
|
||
login(client, "Игрок")
|
||
_admin_login(client, make_admin)
|
||
live = _create(client)
|
||
planned = _create(
|
||
client,
|
||
starts_at=_iso(_now() + timedelta(days=1)),
|
||
ends_at=_iso(_now() + timedelta(days=2)),
|
||
)
|
||
|
||
r = client.post(f"/api/admin/announcements/{planned['id']}/stop", headers=csrf_headers(client))
|
||
assert r.status_code == 422 # снять можно только идущее
|
||
|
||
r = client.post(f"/api/admin/announcements/{live['id']}/stop", headers=csrf_headers(client))
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["status"] == "finished"
|
||
assert _pending(client) == []
|
||
|
||
assert _ack(client, live["id"], 1).status_code == 200 # закрыть можно и снятое
|
||
r = client.delete(f"/api/admin/announcements/{live['id']}", headers=csrf_headers(client))
|
||
assert r.status_code == 200
|
||
with Session(engine) as s:
|
||
assert s.exec(select(AnnouncementView)).all() == [] # отметки ушли каскадом
|
||
assert client.delete(
|
||
f"/api/admin/announcements/{live['id']}", headers=csrf_headers(client)
|
||
).status_code == 404
|
||
assert _ack(client, live["id"], 1).status_code == 404
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("over", "message_part"),
|
||
[
|
||
({"title": " "}, "Заголовок"),
|
||
({"title": "Я" * 61}, "Заголовок"),
|
||
({"body_html": "<p> </p><script>текст</script>"}, "пустым"),
|
||
({"body_html": "<p>" + "а" * 601 + "</p>"}, "600"),
|
||
({"ends_at": _iso(_now() - timedelta(hours=2))}, "позже начала"),
|
||
],
|
||
)
|
||
def test_validation(client: TestClient, make_admin, over: dict, message_part: str):
|
||
_admin_login(client, make_admin)
|
||
body = {
|
||
"title": "Заголовок",
|
||
"body_html": "<p>текст</p>",
|
||
"starts_at": _iso(_now() - timedelta(hours=1)),
|
||
"ends_at": _iso(_now() + timedelta(days=1)),
|
||
}
|
||
body.update(over)
|
||
r = client.post("/api/admin/announcements", json=body, headers=csrf_headers(client))
|
||
assert r.status_code == 422, r.text
|
||
err = r.json()["error"]
|
||
assert err["code"] == "VALIDATION_ERROR"
|
||
assert message_part in err["message"]
|