From 7b9987e78f1b10d46570ac26cb0ae07b2209d2ba Mon Sep 17 00:00:00 2001 From: GGlamer <52128225+Gamer201760@users.noreply.github.com> Date: Sun, 27 Sep 2026 22:40:20 +0300 Subject: [PATCH] =?UTF-8?q?test:=20=D1=81=D0=BA=D0=B2=D0=BE=D0=B7=D0=BD?= =?UTF-8?q?=D0=B0=D1=8F=20=D1=80=D0=B5=D0=B0=D0=BA=D1=86=D0=B8=D1=8F=20?= =?UTF-8?q?=D0=BD=D0=B0=20=D0=B4=D0=BE=D0=BA=D0=BB=D0=B0=D0=B4=20=D1=87?= =?UTF-8?q?=D0=B5=D1=80=D0=B5=D0=B7=20=D1=81=D1=82=D0=B0=D0=BD=D1=86=D0=B8?= =?UTF-8?q?=D1=8E,=20=D0=BE=D1=82=D0=BA=D0=B0=D0=B7=20=D0=BE=D1=82=20?= =?UTF-8?q?=D1=80=D0=B0=D0=B1=D0=BE=D1=82=20=D0=B8=20=D0=B2=D0=BE=D1=81?= =?UTF-8?q?=D1=81=D1=82=D0=B0=D0=BD=D0=BE=D0=B2=D0=BB=D0=B5=D0=BD=D0=B8?= =?UTF-8?q?=D0=B5=20=D0=B8=D0=B7=20checkpoint?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/tests/test_dds_exercise.py | 114 ++++++++++++++++++++++- backend/tests/test_session_checkpoint.py | 35 ++++++- 2 files changed, 146 insertions(+), 3 deletions(-) diff --git a/backend/tests/test_dds_exercise.py b/backend/tests/test_dds_exercise.py index b027fad..70a4e49 100644 --- a/backend/tests/test_dds_exercise.py +++ b/backend/tests/test_dds_exercise.py @@ -2,7 +2,7 @@ import asyncio import time -from datetime import datetime +from datetime import datetime, timedelta from uuid import uuid4 import pytest @@ -800,7 +800,117 @@ def test_complete_dds_workflow_with_training_calls_and_status_updates(client): assert all(report.crew == crew for report in state.desk.active.phone_reports) assert not score["findings"] reaction = [m for m in score["metrics"] if m["key"] == "dds_report_reaction"] - assert reaction and all(m["passed"] for m in reaction), "каждый доклад отмечен вовремя" + assert len(reaction) == 1 and reaction[0]["passed"], "каждый доклад отмечен вовремя" + assert "с нарушением 0 из 4 докладов" in reaction[0]["fact"] + finally: + control.__exit__(None, None, None) + + +STATUS_COMMENT = "Основание: доклад бригады.\nСведения: передано старшему бригады." + + +def accept_with_crew(station): + read_until(station, "card.received") + snapshot = read_until(station, "station.state")["snapshot"] + service = snapshot["services"][0] + crew = next(option for option in snapshot["crew_options"] + if option.startswith(service + " — ")) + station.send_json({"type": "card.status", "service": service, "status": "accepted", + "comment": STATUS_COMMENT}) + read_until(station, "station.state") + station.send_json({"type": "crew.select", "crew": crew}) + read_until(station, "station.state") + return service + + +def send_status(station, service, status): + station.send_json({"type": "card.status", "service": service, "status": status, + "comment": STATUS_COMMENT}) + read_until(station, "station.state") + + +@pytest.mark.parametrize(("reaction", "passed"), [ + ("fast", True), ("slow", False), ("missing", False), ("early", True), +]) +def test_dds_report_reaction_goes_from_criteria_to_report(client, reaction, passed): + """Лимит из `scenario.start` доходит до метрики, отметки D6 и отчёта.""" + session_id, control = start(client, criteria={ + "dds_report_reaction_limit_seconds": 20, + "score_weights": {"dds_report_reaction": 3.5}, + }) + try: + state = hub.get(session_id) + with client.websocket_connect(f"/ws/station/{session_id}") as station: + service = accept_with_crew(station) + if reaction == "early": + send_status(station, service, "responding") + complete_phone_call(station, state, "dispatched") + read_until(station, "station.state") + if reaction in {"fast", "slow"}: + send_status(station, service, "responding") + if reaction == "slow": + # Доклад на 70 с раньше статуса: время доклада — серверное. + state.desk.active.phone_reports[0].at -= timedelta(seconds=70) + station.send_json({"type": "station.finish"}) + read_until(station, "score.ready") + + score = wait_for(lambda: hub.get(session_id).score) + metrics = [item for item in score["metrics"] if item["key"] == "dds_report_reaction"] + assert len(metrics) == 1 + metric = metrics[0] + assert metric["passed"] is passed + assert metric["weight"] == 3.5 + assert metric["norm"] == "отметить статус по докладу за 20 с" + assert "«Начало реагирования» по докладу бригады" in metric["fact"] + findings = [item for item in score["findings"] + if item["metric_key"] == "dds_report_reaction"] + assert [item["code"] for item in findings] == ([] if passed else ["D6"]) + assert all(item["competency"] == "norms" for item in findings) + if reaction == "slow": + assert "70 с (+50 с сверх норматива)" in metric["fact"] + assert findings[0]["fact"] == metric["fact"] + if reaction == "missing": + assert metric["fact"].endswith(": не отмечен") and metric["credit"] == 0.0 + if reaction == "early": + assert metric["fact"].endswith("отмечен не позже доклада") + + report = client.get(f"/api/sessions/{session_id}/report").json() + assert report["criteria"]["dds_report_reaction_limit_seconds"] == 20 + assert [item for item in report["metrics"] + if item["key"] == "dds_report_reaction"] == metrics + if reaction == "slow": + csv_text = client.get(f"/api/sessions/{session_id}/report.csv").content.decode("utf-8-sig") + assert "Лимит реакции на доклад бригады" in csv_text + assert "70 с (+50 с сверх норматива)" in csv_text + finally: + control.__exit__(None, None, None) + + +def test_refusal_after_crew_report_does_not_lower_score_without_finding(client): + """Бригада доложила о прибытии, но работы не проводились: отказ с причиной + — терминальный исход, неотмеченное «Прибытие» не штрафуется.""" + session_id, control = start(client) + try: + state = hub.get(session_id) + with client.websocket_connect(f"/ws/station/{session_id}") as station: + service = accept_with_crew(station) + complete_phone_call(station, state, "dispatched") + read_until(station, "station.state") + send_status(station, service, "responding") + complete_phone_call(station, state, "arrived") + read_until(station, "station.state") + send_status(station, service, "refused") + station.send_json({"type": "station.finish"}) + read_until(station, "score.ready") + + score = wait_for(lambda: hub.get(session_id).score) + assert [report.phase for report in state.desk.active.phone_reports] == [ + "dispatched", "arrived", + ] + metric = next(item for item in score["metrics"] if item["key"] == "dds_report_reaction") + assert metric["passed"], "прибытие при отказе не требуется" + assert "о выезде" in metric["fact"] and "докладов" not in metric["fact"] + assert not [item for item in score["findings"] if item["code"] == "D6"] finally: control.__exit__(None, None, None) diff --git a/backend/tests/test_session_checkpoint.py b/backend/tests/test_session_checkpoint.py index d5842e8..e572a22 100644 --- a/backend/tests/test_session_checkpoint.py +++ b/backend/tests/test_session_checkpoint.py @@ -12,12 +12,13 @@ from app.domain.events import ( SessionMode, ) from app.domain.kio import KIO -from app.domain.statuses import PhoneCallPending, ServiceStatus +from app.domain.statuses import PhoneCallPending, PhoneReportRecord, ServiceStatus from app.domain.taxonomy import ErrorCode, Finding, FindingSource from app.domain.timers import TimerCode from app.scenarios.loader import load_file from app.session.checkpoint import dump_state, load_state from app.session.dds import prepare_handoff_queue, prepare_queue +from app.session.finish import score_dds_card from app.session.state import DdsCardRecord, SessionState, now_utc LIBRARY = Path(__file__).resolve().parents[2] / "scenarios" @@ -85,6 +86,38 @@ def test_active_dds_session_round_trips_without_losing_work(): assert timer.limit_ms == 45_000 +def test_report_reaction_keeps_times_and_limit_after_restore(): + before = dds_state() + before.criteria.dds_report_reaction_limit_seconds = 20 + card = before.desk.active + service = before.card_services(card)[0] + card.phone_reports.append(PhoneReportRecord( + service=service, crew=card.crew_selected, phase="dispatched", + text="Бригада выехала", at=now_utc(), + )) + card.set_service_status(service, ServiceStatus.RESPONDING, + "Основание: доклад.\nСведения: бригада выехала.") + card.phone_reports[0].at -= timedelta(seconds=70) + + restored = load_state(dump_state(before), now_utc()) + + assert restored.criteria.dds_report_reaction_limit_seconds == 20 + assert restored.desk.active.phone_reports == card.phone_reports + + def reaction(state): + result = score_dds_card(state, state.desk.active) + metric = next(item for item in result.metrics if item.key == "dds_report_reaction") + findings = [item for item in result.findings if item.metric_key == "dds_report_reaction"] + return metric, findings + + metric, findings = reaction(restored) + assert not metric.passed + assert metric.norm == "отметить статус по докладу за 20 с" + assert "70 с (+50 с сверх норматива)" in metric.fact + assert [item.code for item in findings] == [ErrorCode.D6] + assert reaction(restored) == reaction(before), "оценка после восстановления та же" + + def completed_card() -> DdsCardRecord: return DdsCardRecord( card_id=uuid4(),