268 lines
15 KiB
Python
268 lines
15 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
"""Live-browser acceptance of trainee KIO -> instructor review -> DDS status.
|
|||
|
|
|
|||
|
|
Only use with a disposable loopback stack/database. The script creates unique
|
|||
|
|
accounts and one group, then removes every row it created in a finally block.
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import argparse
|
|||
|
|
import asyncio
|
|||
|
|
import json
|
|||
|
|
import re
|
|||
|
|
import secrets
|
|||
|
|
import sys
|
|||
|
|
import time
|
|||
|
|
from pathlib import Path
|
|||
|
|
from urllib.parse import urlsplit
|
|||
|
|
from uuid import UUID
|
|||
|
|
|
|||
|
|
from playwright.async_api import async_playwright
|
|||
|
|
from sqlalchemy import delete, or_, select
|
|||
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|||
|
|
|
|||
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|||
|
|
sys.path.insert(0, str(ROOT / "backend"))
|
|||
|
|
|
|||
|
|
|
|||
|
|
def require_loopback(value: str, name: str) -> None:
|
|||
|
|
if urlsplit(value).hostname not in {"127.0.0.1", "localhost", "::1"}:
|
|||
|
|
raise ValueError(f"{name} must use loopback")
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def login(page, base: str, user: str, password: str, target: str) -> None:
|
|||
|
|
await page.goto(f"{base}{target}", wait_until="domcontentloaded")
|
|||
|
|
login_heading = page.get_by_role("heading", name="Учебный симулятор системы-112")
|
|||
|
|
if await login_heading.is_visible():
|
|||
|
|
await page.get_by_label("Логин").fill(user)
|
|||
|
|
await page.get_by_label("Пароль").fill(password)
|
|||
|
|
await page.get_by_role("button", name="Войти", exact=True).click()
|
|||
|
|
await page.get_by_role("heading", name={"/profile": "Профиль курсанта", "/instructor": "Рабочее место преподавателя"}[target]).wait_for()
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def main(args: argparse.Namespace) -> int:
|
|||
|
|
require_loopback(args.frontend_url, "frontend-url")
|
|||
|
|
require_loopback(args.database_url, "database-url")
|
|||
|
|
|
|||
|
|
from app.api.auth import hash_password
|
|||
|
|
from app.db.models import AuditLog, Group, Scenario as ScenarioRow, ScenarioSubmission, Session, Trainee, User
|
|||
|
|
from app.domain import ekp
|
|||
|
|
from app.scenarios import store
|
|||
|
|
|
|||
|
|
scenarios = store.load_from_disk(args.scenarios)
|
|||
|
|
source = next((item for item in scenarios if item.id == args.source_scenario), None)
|
|||
|
|
if source is None or source.ground_truth.incident_code is None or source.ground_truth.dds is None:
|
|||
|
|
raise ValueError(f"source scenario must be a classified DDS case: {args.source_scenario}")
|
|||
|
|
incident_group = ekp.incident(source.ground_truth.incident_code).group
|
|||
|
|
service = source.ground_truth.dds.value
|
|||
|
|
suffix = secrets.token_hex(5)
|
|||
|
|
teacher_login = f"browser-i-{suffix}"
|
|||
|
|
trainee_login = f"browser-t-{suffix}"
|
|||
|
|
teacher_password = secrets.token_urlsafe(24)
|
|||
|
|
trainee_password = secrets.token_urlsafe(24)
|
|||
|
|
teacher_name = f"Browser instructor {suffix}"
|
|||
|
|
trainee_name = f"Browser trainee {suffix}"
|
|||
|
|
group = Group(name=f"browser-smoke-{suffix}", owner_login=teacher_login)
|
|||
|
|
trainee = Trainee(name=trainee_name)
|
|||
|
|
session_ids: list[UUID] = []
|
|||
|
|
submission_id: UUID | None = None
|
|||
|
|
published_scenario_id: str | None = None
|
|||
|
|
session_id: UUID | None = None
|
|||
|
|
result: dict = {"checks": {}, "source_scenario": source.id}
|
|||
|
|
engine = create_async_engine(args.database_url, pool_pre_ping=True)
|
|||
|
|
factory = async_sessionmaker(engine, expire_on_commit=False)
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
async with factory() as db:
|
|||
|
|
db.add(group)
|
|||
|
|
await db.flush()
|
|||
|
|
trainee.group_id = group.id
|
|||
|
|
db.add(trainee)
|
|||
|
|
await db.flush()
|
|||
|
|
db.add_all([
|
|||
|
|
User(login=teacher_login, full_name=teacher_name,
|
|||
|
|
password_hash=hash_password(teacher_password), role="instructor", blocked=False),
|
|||
|
|
User(login=trainee_login, full_name=trainee_name,
|
|||
|
|
password_hash=hash_password(trainee_password), role="trainee",
|
|||
|
|
trainee_id=trainee.id, service=service, blocked=False),
|
|||
|
|
])
|
|||
|
|
await db.commit()
|
|||
|
|
|
|||
|
|
async with async_playwright() as playwright:
|
|||
|
|
browser = await playwright.chromium.launch(headless=True)
|
|||
|
|
student_page = await browser.new_page(ignore_https_errors=True)
|
|||
|
|
await login(student_page, args.frontend_url, trainee_login, trainee_password, "/profile")
|
|||
|
|
|
|||
|
|
title = f"КИО из браузера {suffix}"
|
|||
|
|
address = f"Москва, учебная улица, дом {suffix[:2]}"
|
|||
|
|
description = "В учебном помещении виден дым, требуется проверка и выезд службы."
|
|||
|
|
await student_page.get_by_label("Название").fill(title)
|
|||
|
|
await student_page.locator('input[aria-label="Адрес"]').fill(address)
|
|||
|
|
await student_page.locator('textarea[aria-label="Описание"]').fill(description)
|
|||
|
|
await student_page.get_by_label("Категория происшествия").select_option(value=str(incident_group))
|
|||
|
|
await student_page.get_by_role("group", name="Что случилось?").get_by_role(
|
|||
|
|
"button", name="Пожар", exact=True
|
|||
|
|
).click()
|
|||
|
|
for index, sign in enumerate(source.signs, start=1):
|
|||
|
|
picker = student_page.get_by_label(f"Признак происшествия {index}")
|
|||
|
|
await picker.wait_for(state="visible")
|
|||
|
|
await picker.select_option(label=sign)
|
|||
|
|
await student_page.get_by_role("button", name="Сохранить черновик преподавателю").click()
|
|||
|
|
await student_page.get_by_text("Черновик КИО отправлен и ожидает проверки.", exact=True).wait_for()
|
|||
|
|
result["checks"]["student_submitted_kio_from_profile"] = True
|
|||
|
|
|
|||
|
|
async with factory() as db:
|
|||
|
|
deadline = time.monotonic() + 10
|
|||
|
|
submission = None
|
|||
|
|
while time.monotonic() < deadline:
|
|||
|
|
submission = await db.scalar(select(ScenarioSubmission).where(ScenarioSubmission.title == title))
|
|||
|
|
if submission is not None:
|
|||
|
|
break
|
|||
|
|
await asyncio.sleep(0.1)
|
|||
|
|
if submission is None:
|
|||
|
|
raise RuntimeError("browser-submitted KIO was not persisted in PostgreSQL")
|
|||
|
|
submission_id = submission.id
|
|||
|
|
|
|||
|
|
instructor_page = await browser.new_page(ignore_https_errors=True)
|
|||
|
|
await login(instructor_page, args.frontend_url, teacher_login, teacher_password, "/instructor")
|
|||
|
|
proposal = instructor_page.locator(".student-scenario-proposal").filter(has_text=title)
|
|||
|
|
await proposal.wait_for(state="visible")
|
|||
|
|
await proposal.get_by_role("button", name="Проверил(а), утвердить в банк ДДС").click()
|
|||
|
|
await proposal.wait_for(state="hidden")
|
|||
|
|
result["checks"]["instructor_approved_in_browser"] = True
|
|||
|
|
|
|||
|
|
await instructor_page.get_by_label("Источник карточек").select_option(label="Сформированные курсантами")
|
|||
|
|
scenario_select = instructor_page.get_by_label("Сценарий занятия")
|
|||
|
|
await scenario_select.wait_for()
|
|||
|
|
deadline = time.monotonic() + 10
|
|||
|
|
selected_value = None
|
|||
|
|
while time.monotonic() < deadline:
|
|||
|
|
options = await scenario_select.locator("option").all()
|
|||
|
|
for option in options:
|
|||
|
|
if title in (await option.text_content() or ""):
|
|||
|
|
selected_value = await option.get_attribute("value")
|
|||
|
|
break
|
|||
|
|
if selected_value:
|
|||
|
|
break
|
|||
|
|
await asyncio.sleep(0.1)
|
|||
|
|
if not selected_value:
|
|||
|
|
raise RuntimeError("approved trainee KIO did not enter instructor's DDS scenario bank")
|
|||
|
|
published_scenario_id = selected_value
|
|||
|
|
await scenario_select.select_option(value=selected_value)
|
|||
|
|
trainee_select = instructor_page.get_by_label("Учётная запись курсанта")
|
|||
|
|
trainee_options = await trainee_select.locator("option").all()
|
|||
|
|
trainee_value = None
|
|||
|
|
for option in trainee_options:
|
|||
|
|
if trainee_name in (await option.text_content() or ""):
|
|||
|
|
trainee_value = await option.get_attribute("value")
|
|||
|
|
break
|
|||
|
|
if not trainee_value:
|
|||
|
|
raise RuntimeError("temporary trainee account is missing from instructor's trainee list")
|
|||
|
|
await trainee_select.select_option(value=trainee_value)
|
|||
|
|
await instructor_page.get_by_role("button", name="Начать занятие").click()
|
|||
|
|
await instructor_page.wait_for_function("new URL(location.href).searchParams.has('session')", timeout=15000)
|
|||
|
|
session_id = UUID(urlsplit(instructor_page.url).query.split("session=", 1)[1].split("&", 1)[0])
|
|||
|
|
session_ids.append(session_id)
|
|||
|
|
await instructor_page.get_by_role("link", name=re.compile("открыть ДДС")).wait_for()
|
|||
|
|
result["checks"]["instructor_started_dds_lesson"] = True
|
|||
|
|
|
|||
|
|
await student_page.goto(f"{args.frontend_url}/dds?session={session_id}&role=dds_{service}", wait_until="domcontentloaded")
|
|||
|
|
await student_page.locator(".dds-connection-live").filter(has_text="на связи").wait_for(timeout=15000)
|
|||
|
|
queue_row = student_page.locator("tr").filter(has_text=title)
|
|||
|
|
await queue_row.wait_for(state="visible", timeout=15000)
|
|||
|
|
await queue_row.get_by_role("button", name="Открыть карточку").click()
|
|||
|
|
await student_page.locator(".dds-case-address").get_by_text(address, exact=True).wait_for()
|
|||
|
|
result["checks"]["approved_kio_received_in_live_dds"] = True
|
|||
|
|
|
|||
|
|
service_buttons = student_page.locator(".dds-service-recipient-list .dds-service-recipient")
|
|||
|
|
await service_buttons.first.wait_for(state="visible")
|
|||
|
|
result["recipient_buttons"] = await service_buttons.all_text_contents()
|
|||
|
|
managed_button = None
|
|||
|
|
for index in range(await service_buttons.count()):
|
|||
|
|
candidate = service_buttons.nth(index)
|
|||
|
|
if await candidate.is_enabled():
|
|||
|
|
managed_button = candidate
|
|||
|
|
break
|
|||
|
|
if managed_button is None:
|
|||
|
|
raise RuntimeError(f"no manageable DDS service button: {result['recipient_buttons']}")
|
|||
|
|
managed_button_text = await managed_button.locator("b").inner_text()
|
|||
|
|
await managed_button.click()
|
|||
|
|
classification = await student_page.locator(".dds-case-classification").inner_text()
|
|||
|
|
result["actual_managed_service_button"] = " ".join(managed_button_text.split())
|
|||
|
|
result["service_assignment_matches_card"] = result["actual_managed_service_button"] in classification
|
|||
|
|
if not result["service_assignment_matches_card"]:
|
|||
|
|
raise RuntimeError("enabled managed-service button differs from the service shown on the card")
|
|||
|
|
await student_page.locator(".dds-case-drawer-status").wait_for(state="visible", timeout=5000)
|
|||
|
|
status_controls = student_page.locator('select[aria-label^="Следующий статус "]')
|
|||
|
|
if await status_controls.count() == 0:
|
|||
|
|
raise RuntimeError("DDS status pane opened without a status control: " +
|
|||
|
|
(await student_page.locator("body").inner_text())[-1600:])
|
|||
|
|
status_select = status_controls.first
|
|||
|
|
await status_select.select_option(label="Принята")
|
|||
|
|
await student_page.get_by_label("Основание статуса").fill("Карточка получена и принята ДДС.")
|
|||
|
|
await student_page.get_by_label("Полученные сведения").fill("Направление реагирования подтверждено.")
|
|||
|
|
await student_page.get_by_role("button", name="Сохранить статус").click()
|
|||
|
|
await student_page.locator(".dds-status-history").get_by_text("Принята", exact=False).first.wait_for()
|
|||
|
|
result["checks"]["dds_primary_status_saved"] = True
|
|||
|
|
|
|||
|
|
crew_select = student_page.get_by_label(f"Наряд {service}")
|
|||
|
|
crew_options = await crew_select.locator("option").all()
|
|||
|
|
crew_value = None
|
|||
|
|
for option in crew_options:
|
|||
|
|
candidate = await option.get_attribute("value")
|
|||
|
|
if candidate:
|
|||
|
|
crew_value = candidate
|
|||
|
|
break
|
|||
|
|
if crew_value:
|
|||
|
|
await crew_select.select_option(value=crew_value)
|
|||
|
|
await status_select.select_option(label="Начало реагирования")
|
|||
|
|
await student_page.get_by_label("Основание статуса").fill("Получен доклад о начале реагирования.")
|
|||
|
|
await student_page.get_by_label("Полученные сведения").fill("Бригада выехала к месту происшествия.")
|
|||
|
|
await student_page.get_by_role("button", name="Сохранить статус").click()
|
|||
|
|
history = student_page.locator(".dds-status-history")
|
|||
|
|
await history.get_by_text("Начало реагирования", exact=False).first.wait_for()
|
|||
|
|
result["checks"]["dds_followup_status_saved"] = True
|
|||
|
|
|
|||
|
|
if args.output:
|
|||
|
|
Path(args.output).parent.mkdir(parents=True, exist_ok=True)
|
|||
|
|
await student_page.screenshot(path=args.output, full_page=True)
|
|||
|
|
result["session_id"] = str(session_id)
|
|||
|
|
result["scenario_id"] = published_scenario_id
|
|||
|
|
result["managed_service"] = service
|
|||
|
|
result["all_checks_passed"] = all(result["checks"].values())
|
|||
|
|
await browser.close()
|
|||
|
|
|
|||
|
|
rendered = json.dumps(result, ensure_ascii=False, indent=2)
|
|||
|
|
print(rendered)
|
|||
|
|
if args.json_output:
|
|||
|
|
Path(args.json_output).parent.mkdir(parents=True, exist_ok=True)
|
|||
|
|
Path(args.json_output).write_text(rendered + "\n", encoding="utf-8")
|
|||
|
|
return 0 if result["all_checks_passed"] else 1
|
|||
|
|
finally:
|
|||
|
|
async with factory() as db:
|
|||
|
|
await db.execute(delete(Session).where(or_(
|
|||
|
|
Session.trainee_id == trainee.id,
|
|||
|
|
Session.id.in_(session_ids) if session_ids else False,
|
|||
|
|
)))
|
|||
|
|
if submission_id is not None:
|
|||
|
|
await db.execute(delete(ScenarioSubmission).where(ScenarioSubmission.id == submission_id))
|
|||
|
|
if published_scenario_id:
|
|||
|
|
await db.execute(delete(ScenarioRow).where(ScenarioRow.id == published_scenario_id))
|
|||
|
|
await db.execute(delete(AuditLog).where(AuditLog.actor.in_([teacher_login, trainee_login])))
|
|||
|
|
await db.execute(delete(User).where(User.login.in_([teacher_login, trainee_login])))
|
|||
|
|
await db.execute(delete(Trainee).where(Trainee.name == trainee_name))
|
|||
|
|
await db.execute(delete(Group).where(Group.name == group.name))
|
|||
|
|
await db.commit()
|
|||
|
|
await engine.dispose()
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|||
|
|
parser.add_argument("--frontend-url", required=True)
|
|||
|
|
parser.add_argument("--database-url", required=True)
|
|||
|
|
parser.add_argument("--scenarios", type=Path, default=ROOT / "scenarios")
|
|||
|
|
parser.add_argument("--source-scenario", default="t01-1-fire-container")
|
|||
|
|
parser.add_argument("--output", help="optional DDS screenshot path")
|
|||
|
|
parser.add_argument("--json-output", help="optional machine-readable evidence path")
|
|||
|
|
raise SystemExit(asyncio.run(main(parser.parse_args())))
|