d55c8bc8cd
Самое крупное расхождение с ТЗ: входа не было вовсе, экраны открывались ссылкой с номером занятия, и пускало знание адреса. - Таблицы users и audit_log, миграция. Пароль argon2, сессия — подписанная cookie; роль на сокетах читается из той же cookie в момент рукопожатия, отдельного протокола авторизации в канале нет. - Разграничение: control — преподавателю, observe — преподавателю и админу, call и station — обучающемуся и преподавателю. Отказ приходит событием error с кодом forbidden. - Обучающийся не видит чужого: история подменяет фильтр на его собственный идентификатор, разбор и профиль сверяют trainee_id. ТЗ запрещает доступ к чужим результатам, а не только к чужим экранам. - Администратору закрыта правка оценок — ТЗ запрещает это прямо. - make users заводит по записи на роль и печатает случайные пароли один раз: зашитый в репозиторий admin/admin пережил бы сдачу. - Экран входа и проверка роли на каждом маршруте фронта. Наши инструменты не сломались: make lesson и тесты входят через dev-token за флагом dev_auth_bypass, на стенде точка отвечает 404 — выключенной функции не должно быть видно вовсе. У тестов появился conftest.py. Role уехала в домен и в generated.ts через EventCatalog.principal: иначе фронт переписывал бы список ролей руками. 181 тест зелёный (14 новых), make typecheck чистый.
217 lines
9.6 KiB
Python
217 lines
9.6 KiB
Python
"""Цепочка 112 → ДДС: карточка уходит снимком, диспетчер подтверждает или отбивает."""
|
||
|
||
import time
|
||
from uuid import uuid4
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app.main import app
|
||
from app.session.hub import hub
|
||
|
||
|
||
@pytest.fixture
|
||
def client():
|
||
with TestClient(app) as test_client:
|
||
# Сокеты закрыты ролями (lct-23): тесты входят так же,
|
||
# как `make lesson`, — через dev-token за флагом.
|
||
test_client.post("/api/auth/dev-token")
|
||
hub.journal = None
|
||
yield test_client
|
||
|
||
|
||
def wait_for(predicate, timeout: float = 3.0):
|
||
deadline = time.monotonic() + timeout
|
||
while time.monotonic() < deadline:
|
||
value = predicate()
|
||
if value:
|
||
return value
|
||
time.sleep(0.02)
|
||
raise AssertionError("не дождались")
|
||
|
||
|
||
def read_until(ws, event_type: str, limit: int = 20) -> dict:
|
||
for _ in range(limit):
|
||
message = ws.receive_json()
|
||
if message["type"] == event_type:
|
||
return message
|
||
raise AssertionError(f"событие {event_type} не пришло")
|
||
|
||
|
||
def lesson(client):
|
||
session_id = uuid4()
|
||
control = client.websocket_connect(f"/ws/control/{session_id}")
|
||
socket = control.__enter__()
|
||
socket.send_json({"type": "scenario.start", "scenario_id": "fire-apartment-l2",
|
||
"trainee": "Иванов", "mode": "training"})
|
||
wait_for(lambda: hub.get(session_id))
|
||
return session_id, control, socket
|
||
|
||
|
||
def test_card_goes_to_the_dispatcher_frozen(client):
|
||
"""Снимок не меняется после передачи: оператор не дописывает задним числом."""
|
||
session_id, control, _ = lesson(client)
|
||
try:
|
||
with client.websocket_connect(f"/ws/station/{session_id}?role=dds_01") as station:
|
||
with client.websocket_connect(f"/ws/call/{session_id}") as trainee:
|
||
trainee.send_json({"type": "call.answer"})
|
||
trainee.send_json({"type": "kio.patch", "fields": {"address": "улица Ленина, 14", "dds": "01"}})
|
||
wait_for(lambda: hub.get(session_id).kio.address)
|
||
trainee.send_json({"type": "dds.dispatch", "service": "01"})
|
||
received = read_until(station, "card.received")
|
||
|
||
# Правка после передачи в снимок не попадает.
|
||
trainee.send_json({"type": "kio.patch", "fields": {"floor": "5"}})
|
||
wait_for(lambda: hub.get(session_id).kio.floor == "5")
|
||
|
||
assert received["card"]["address"] == "улица Ленина, 14"
|
||
assert received["card"]["floor"] is None, "снимок изменился после передачи"
|
||
assert received["from_operator"] == "Иванов"
|
||
finally:
|
||
control.__exit__(None, None, None)
|
||
|
||
|
||
def test_station_joining_late_still_gets_the_card(client):
|
||
"""Диспетчер садится за АРМ, когда вызов уже идёт."""
|
||
session_id, control, _ = lesson(client)
|
||
try:
|
||
with client.websocket_connect(f"/ws/call/{session_id}") as trainee:
|
||
trainee.send_json({"type": "call.answer"})
|
||
trainee.send_json({"type": "dds.dispatch", "service": "01"})
|
||
wait_for(lambda: hub.get(session_id).dispatched_card)
|
||
|
||
with client.websocket_connect(f"/ws/station/{session_id}") as station:
|
||
assert read_until(station, "card.received")["card"] is not None
|
||
finally:
|
||
control.__exit__(None, None, None)
|
||
|
||
|
||
def test_acknowledgement_stops_the_four_second_norm(client):
|
||
from app.domain.timers import TimerCode
|
||
|
||
session_id, control, _ = lesson(client)
|
||
try:
|
||
state = hub.get(session_id)
|
||
with client.websocket_connect(f"/ws/station/{session_id}") as station:
|
||
with client.websocket_connect(f"/ws/call/{session_id}") as trainee:
|
||
trainee.send_json({"type": "call.answer"})
|
||
trainee.send_json({"type": "dds.dispatch", "service": "01"})
|
||
read_until(station, "card.received")
|
||
station.send_json({"type": "card.ack"})
|
||
measured = wait_for(lambda: state.timers.measured_ms(TimerCode.DDS_ACK) is not None)
|
||
assert measured
|
||
finally:
|
||
control.__exit__(None, None, None)
|
||
|
||
|
||
def test_bounced_card_becomes_e6_with_the_reason(client):
|
||
"""Неполнота КИО перестаёт быть процентом в отчёте и становится
|
||
сорванным выездом с конкретной причиной."""
|
||
session_id, control, _ = lesson(client)
|
||
try:
|
||
state = hub.get(session_id)
|
||
with client.websocket_connect(f"/ws/station/{session_id}") as station:
|
||
with client.websocket_connect(f"/ws/call/{session_id}") as trainee:
|
||
trainee.send_json({"type": "call.answer"})
|
||
trainee.send_json({"type": "dds.dispatch", "service": "01"})
|
||
read_until(station, "card.received")
|
||
station.send_json({"type": "card.bounce", "missing_fields": ["floor", "victims_count"],
|
||
"comment": "куда ехать без этажа"})
|
||
wait_for(lambda: state.bounced_fields)
|
||
trainee.send_json({"type": "call.hangup"})
|
||
wait_for(lambda: state.score is not None)
|
||
|
||
response = client.get(f"/api/sessions/{session_id}/report").json()
|
||
chain = [finding for finding in response["findings"] if finding["code"] == "E6"]
|
||
assert chain, "возврат карточки должен попасть в разбор"
|
||
assert "floor" in chain[0]["summary"] and "выезд сорван" in chain[0]["summary"]
|
||
finally:
|
||
control.__exit__(None, None, None)
|
||
|
||
|
||
# ── статусы реагирования (lct-33) ──
|
||
|
||
|
||
def dispatched(client):
|
||
"""Занятие, доведённое до передачи карточки в ДДС, и подключённая станция."""
|
||
session_id, control, _ = lesson(client)
|
||
station_ctx = client.websocket_connect(f"/ws/station/{session_id}?role=dds_01")
|
||
station = station_ctx.__enter__()
|
||
call_ctx = client.websocket_connect(f"/ws/call/{session_id}")
|
||
trainee = call_ctx.__enter__()
|
||
trainee.send_json({"type": "call.answer"})
|
||
trainee.send_json(
|
||
{"type": "kio.patch", "fields": {"signs": ["жилой дом", "балкон", "открытое пламя"]}}
|
||
)
|
||
wait_for(lambda: hub.get(session_id).kio.incident_code)
|
||
trainee.send_json({"type": "dds.dispatch", "service": "01"})
|
||
read_until(station, "card.received")
|
||
return session_id, station, (control, station_ctx, call_ctx)
|
||
|
||
|
||
def close_all(contexts):
|
||
for context in contexts:
|
||
context.__exit__(None, None, None)
|
||
|
||
|
||
def test_station_sees_the_notify_list_from_the_classifier(client):
|
||
"""Диспетчеру есть что отмечать только потому, что список оповещения
|
||
посчитан по ЕКП, а не выбран оператором (lct-32)."""
|
||
session_id, station, contexts = dispatched(client)
|
||
try:
|
||
state = read_until(station, "station.state")
|
||
assert "Служба 101" in state["snapshot"]["services"]
|
||
assert len(state["snapshot"]["services"]) > 1
|
||
assert state["snapshot"]["statuses"]["Служба 101"] == "added"
|
||
finally:
|
||
close_all(contexts)
|
||
|
||
|
||
def test_dispatcher_sets_a_status_and_the_card_follows(client):
|
||
session_id, station, contexts = dispatched(client)
|
||
try:
|
||
read_until(station, "station.state")
|
||
station.send_json({"type": "card.status", "service": "Служба 101", "status": "accepted"})
|
||
state = read_until(station, "station.state")
|
||
assert state["snapshot"]["statuses"]["Служба 101"] == "accepted"
|
||
assert "responding" in state["snapshot"]["available"]["Служба 101"]
|
||
assert "accepted" not in state["snapshot"]["available"]["Служба 101"]
|
||
finally:
|
||
close_all(contexts)
|
||
|
||
|
||
def test_out_of_order_status_is_rejected_with_a_reason(client):
|
||
"""Автомат живёт на сервере: фронт может ошибиться, состояние — нет."""
|
||
session_id, station, contexts = dispatched(client)
|
||
try:
|
||
read_until(station, "station.state")
|
||
station.send_json({"type": "card.status", "service": "Служба 101", "status": "arrived"})
|
||
error = read_until(station, "error")
|
||
assert "Принята" in error["message"]
|
||
assert hub.get(session_id).status_log == []
|
||
finally:
|
||
close_all(contexts)
|
||
|
||
|
||
def test_refusal_without_a_comment_is_rejected(client):
|
||
session_id, station, contexts = dispatched(client)
|
||
try:
|
||
read_until(station, "station.state")
|
||
station.send_json({"type": "card.status", "service": "Служба 101", "status": "declined"})
|
||
error = read_until(station, "error")
|
||
assert "комментар" in error["message"]
|
||
finally:
|
||
close_all(contexts)
|
||
|
||
|
||
def test_ack_button_still_works_and_counts_as_accepted(client):
|
||
"""Кнопка подтверждения из живой цепочки (lct-20) осталась и означает
|
||
«Принята» у главной службы — иначе два экрана учили бы разному."""
|
||
session_id, station, contexts = dispatched(client)
|
||
try:
|
||
read_until(station, "station.state")
|
||
station.send_json({"type": "card.ack"})
|
||
state = read_until(station, "station.state")
|
||
assert state["snapshot"]["statuses"]["Служба 101"] == "accepted"
|
||
finally:
|
||
close_all(contexts)
|