348 lines
15 KiB
Python
348 lines
15 KiB
Python
"""Экспорт разбора: содержимое, безопасность CSV, кириллица и доступ."""
|
||
|
||
import asyncio
|
||
import csv
|
||
import io
|
||
from datetime import UTC, datetime
|
||
from types import SimpleNamespace
|
||
from uuid import uuid4
|
||
|
||
import pytest
|
||
from app.session.pg_store import PostgresSessionStore
|
||
from app.session.store import MemorySessionStore
|
||
from fastapi import HTTPException
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app.api.auth import Principal
|
||
from app.api.http import sessions
|
||
from app.config import get_settings
|
||
from app.domain.events import DdsCardReport, SessionReport
|
||
from app.domain.roles import Role
|
||
from app.main import app
|
||
from app.scoring.export import _cell, certificate_pdf, to_csv, to_pdf
|
||
|
||
|
||
def sample_report(*, long: bool = False) -> SessionReport:
|
||
at = datetime(2026, 9, 21, 12, 0, tzinfo=UTC)
|
||
long_text = "Заявитель сообщает о дыме в учебном помещении. " * (240 if long else 1)
|
||
return SessionReport.model_validate({
|
||
"session_id": str(uuid4()),
|
||
"scenario_id": "=1+1",
|
||
"mode": "training",
|
||
"attempt": 2,
|
||
"criteria": {"decision_time_limit_seconds": 45, "allowed_errors": 1,
|
||
"require_correct_grammar": True},
|
||
"failed_metrics": 1,
|
||
"passed": True,
|
||
"transcript": [
|
||
{"ref": f"u{i}", "speaker": "caller", "text": long_text if i == 0 else "<вопрос> \t=cmd", "at": at}
|
||
for i in range(18 if long else 2)
|
||
],
|
||
"findings": [{
|
||
"code": "E1", "source": "slots", "summary": "+HYPERLINK(\"x\")",
|
||
"fact": "Адрес не уточнён", "norm": "Уточнить адрес происшествия",
|
||
"ref": "ГОСТ", "at": at,
|
||
}],
|
||
"metrics": [{
|
||
"key": "address", "title": "Адрес", "fact": "Не назван",
|
||
"norm": "Адрес должен быть уточнён", "passed": False,
|
||
}],
|
||
"competencies": [{"competency": "interview", "value": 65}],
|
||
"reference_questions": [{"checklist_id": "q_address", "question": "Назовите адрес?"}],
|
||
"missed_checklist": ["q_address"],
|
||
"hints_used": [{"checklist_id": "q_address", "question": "Уточните адрес", "at": at}],
|
||
"self_assessment": {"missed": ["q_address"], "comment": "@SUM(1,2)", "submitted_at": at},
|
||
"self_assessment_diff": {"noticed": ["q_address"], "unnoticed": [], "overcautious": []},
|
||
"notes": [{"transcript_ref": "u0", "text": "Внимательнее к адресу", "author": "Преподаватель"}],
|
||
"score_auto": 70,
|
||
"score_final": 75,
|
||
"overridden_by": "Преподаватель",
|
||
"override_comment": "Ручная корректировка",
|
||
})
|
||
|
||
|
||
def test_csv_contains_sections_and_blocks_formula_injection():
|
||
rows = list(csv.reader(io.StringIO(to_csv(sample_report()).decode("utf-8-sig"))))
|
||
assert rows[0] == ["Раздел", "№", "Поле", "Значение", "Дополнительно"]
|
||
assert any(row[0] == "Транскрипт" and row[3] == "<вопрос> \t=cmd" for row in rows)
|
||
assert any(row[0] == "Занятие" and row[3] == "'=1+1" for row in rows)
|
||
assert any(row[0] == "Ошибки" and row[3] == "'+HYPERLINK(\"x\")" for row in rows)
|
||
assert any(row[0] == "Самооценка" and row[3] == "'@SUM(1,2)" for row in rows)
|
||
assert ["Критерии", "", "Лимит реакции на доклад бригады, с", "не применяется", ""] in rows
|
||
assert _cell(" =cmd") == "' =cmd"
|
||
assert _cell("\tОбычный текст") == "'\tОбычный текст"
|
||
|
||
|
||
def test_csv_shows_enabled_report_reaction_limit():
|
||
report = sample_report()
|
||
report.criteria.dds_report_reaction_enabled = True
|
||
report.criteria.dds_report_reaction_limit_seconds = 20
|
||
|
||
rows = list(csv.reader(io.StringIO(to_csv(report).decode("utf-8-sig"))))
|
||
|
||
assert ["Критерии", "", "Лимит реакции на доклад бригады, с", "20", ""] in rows
|
||
|
||
|
||
def test_card_error_message_to_112_is_in_csv_and_pdf(tmp_path):
|
||
report = sample_report()
|
||
message = "В карточке дом 16; по докладу бригады фактически дом 14."
|
||
report.card_results = [DdsCardReport(
|
||
card_id=uuid4(), scenario_id="dds-card-address-error", score_auto=100,
|
||
reply_text="", metrics=[], findings=[], actions=[{
|
||
"type": "card.error_report_112", "text": message,
|
||
"at": "2026-09-21T12:05:00+00:00",
|
||
}],
|
||
)]
|
||
|
||
rows = list(csv.reader(io.StringIO(to_csv(report).decode("utf-8-sig"))))
|
||
assert any(row[0] == "Сообщение в 112" and row[3] == message for row in rows)
|
||
|
||
import shutil
|
||
import subprocess
|
||
|
||
if shutil.which("pdftotext"):
|
||
path = tmp_path / "dds-error.pdf"
|
||
path.write_bytes(to_pdf(report))
|
||
extracted = subprocess.check_output(["pdftotext", str(path), "-"], text=True)
|
||
assert "Сообщения в 112 об ошибках карточек" in extracted
|
||
assert message in extracted
|
||
|
||
|
||
def test_pdf_supports_cyrillic_and_spans_pages(tmp_path):
|
||
data = to_pdf(sample_report(long=True))
|
||
assert data.startswith(b"%PDF-")
|
||
path = tmp_path / "report.pdf"
|
||
path.write_bytes(data)
|
||
# Poppler даёт проверку извлекаемого текста, не только сигнатуры файла.
|
||
import shutil
|
||
import subprocess
|
||
|
||
if shutil.which("pdftotext") and shutil.which("pdfinfo"):
|
||
info = subprocess.check_output(["pdfinfo", str(path)], text=True)
|
||
pages = int(next(line.split(":", 1)[1].strip() for line in info.splitlines() if line.startswith("Pages:")))
|
||
assert pages >= 2
|
||
extracted = subprocess.check_output(["pdftotext", str(path), "-"], text=True)
|
||
assert "Отчёт по учебному занятию" in extracted
|
||
assert "Адрес должен быть уточнён" in extracted
|
||
assert "Заявитель сообщает о дыме" in extracted
|
||
assert "Лимит реакции на доклад бригады" in extracted
|
||
assert "не применяется" in extracted
|
||
|
||
|
||
def test_certificate_pdf_contains_saved_result(tmp_path):
|
||
data = certificate_pdf(
|
||
trainee_name="Петров Пётр Сергеевич",
|
||
trainee_id=uuid4(),
|
||
group_name="ДДС-17",
|
||
attempts=4,
|
||
average_score=87.25,
|
||
issued_at="2026-09-23",
|
||
)
|
||
assert data.startswith(b"%PDF-")
|
||
path = tmp_path / "certificate.pdf"
|
||
path.write_bytes(data)
|
||
import shutil
|
||
import subprocess
|
||
|
||
if shutil.which("pdftotext"):
|
||
extracted = subprocess.check_output(["pdftotext", str(path), "-"], text=True)
|
||
assert "СЕРТИФИКАТ" in extracted
|
||
assert "Петров Пётр Сергеевич" in extracted
|
||
assert "87.2 из 100" in extracted
|
||
assert "не заменяет квалификационный документ" in extracted
|
||
|
||
|
||
@pytest.fixture
|
||
def client(monkeypatch):
|
||
report = sample_report()
|
||
owner_login = "demo-instructor" if get_settings().demo_no_db else "dev"
|
||
state = SimpleNamespace(score={"score_auto": 70}, trainee_id=uuid4(), owner_login=owner_login)
|
||
audit_events = []
|
||
|
||
async def capture_audit(actor, role, action, object_id=None, detail=""):
|
||
audit_events.append((actor, role, action, object_id))
|
||
|
||
state.audit_events = audit_events
|
||
monkeypatch.setattr(sessions, "_live", lambda session_id: (state, object()))
|
||
monkeypatch.setattr(sessions, "build_report", lambda session_id, state, scenario: report)
|
||
monkeypatch.setattr(sessions, "audit_required", capture_audit)
|
||
with TestClient(app) as test_client:
|
||
# These endpoint tests exercise the in-memory live-report path. Durable
|
||
# report readiness is covered by the isolated PostgreSQL integration suite.
|
||
monkeypatch.setattr(sessions.hub, "store", MemorySessionStore())
|
||
test_client.post("/api/auth/dev-token")
|
||
yield test_client, state, report
|
||
|
||
|
||
def test_export_routes_return_downloads_with_json_report_rights(client):
|
||
browser, state, report = client
|
||
json_response = browser.get(f"/api/sessions/{report.session_id}/report")
|
||
assert json_response.status_code == 200, json_response.text
|
||
|
||
csv_response = browser.get(f"/api/sessions/{report.session_id}/report.csv")
|
||
assert csv_response.status_code == 200, csv_response.text
|
||
assert csv_response.headers["content-type"].startswith("text/csv")
|
||
assert csv_response.content.startswith(b"\xef\xbb\xbf")
|
||
assert "attachment" in csv_response.headers["content-disposition"]
|
||
|
||
pdf_response = browser.get(f"/api/sessions/{report.session_id}/report.pdf")
|
||
assert pdf_response.status_code == 200
|
||
assert pdf_response.headers["content-type"] == "application/pdf"
|
||
assert pdf_response.content.startswith(b"%PDF-")
|
||
assert state.audit_events == [
|
||
("dev", "instructor", "report.read", str(report.session_id)),
|
||
("dev", "instructor", "report.export.csv", str(report.session_id)),
|
||
("dev", "instructor", "report.export.pdf", str(report.session_id)),
|
||
]
|
||
|
||
state.score = None
|
||
assert browser.get(f"/api/sessions/{report.session_id}/report.csv").status_code == 409
|
||
assert browser.get(f"/api/sessions/{report.session_id}/report.pdf").status_code == 409
|
||
|
||
|
||
def test_trainee_cannot_export_another_persons_report(client, monkeypatch):
|
||
browser, _state, report = client
|
||
monkeypatch.setattr(
|
||
sessions, "require",
|
||
lambda request: Principal(login="trainee", full_name="Учебный", role=Role.TRAINEE, trainee_id=uuid4()),
|
||
)
|
||
for suffix in ("csv", "pdf"):
|
||
assert browser.get(f"/api/sessions/{report.session_id}/report.{suffix}").status_code == 403
|
||
assert not client[1].audit_events
|
||
|
||
|
||
def test_report_export_fails_closed_when_access_audit_is_unavailable(client, monkeypatch):
|
||
from fastapi import HTTPException
|
||
|
||
browser, _state, report = client
|
||
|
||
async def unavailable(*_args, **_kwargs):
|
||
raise HTTPException(status_code=503, detail="audit_unavailable")
|
||
|
||
monkeypatch.setattr(sessions, "audit_required", unavailable)
|
||
response = browser.get(f"/api/sessions/{report.session_id}/report.pdf")
|
||
assert response.status_code == 503
|
||
assert response.json() == {"detail": "audit_unavailable"}
|
||
|
||
|
||
def test_archived_report_survives_missing_live_session(monkeypatch):
|
||
archived = sample_report()
|
||
session_row = SimpleNamespace(id=archived.session_id, trainee_id=uuid4(), owner_login="teacher")
|
||
score_row = SimpleNamespace(
|
||
score_auto=70.0,
|
||
score_final=82.0,
|
||
overridden_by="Преподаватель",
|
||
override_comment="проверено после занятия",
|
||
report={"full_report": archived.model_dump(mode="json")},
|
||
)
|
||
|
||
class FakeDb:
|
||
async def get(self, model, key):
|
||
return session_row
|
||
|
||
async def scalar(self, statement):
|
||
return score_row
|
||
|
||
monkeypatch.setattr(sessions, "_live", lambda session_id: (_ for _ in ()).throw(
|
||
HTTPException(status_code=404, detail="session_not_found")
|
||
))
|
||
monkeypatch.setattr(
|
||
sessions, "require",
|
||
lambda request: Principal(login="teacher", full_name="Преподаватель", role=Role.INSTRUCTOR),
|
||
)
|
||
restored = asyncio.run(sessions._report_data(archived.session_id, object(), FakeDb()))
|
||
assert restored.session_id == archived.session_id
|
||
assert restored.score_auto == 70
|
||
assert restored.score_final == 82
|
||
assert restored.override_comment == "проверено после занятия"
|
||
|
||
|
||
def test_archived_score_override_updates_json_exports_and_audit(monkeypatch):
|
||
archived = sample_report()
|
||
session_row = SimpleNamespace(id=archived.session_id, trainee_id=uuid4(), owner_login="teacher")
|
||
score_row = SimpleNamespace(
|
||
score_auto=70.0,
|
||
score_final=70.0,
|
||
overridden_by=None,
|
||
override_comment=None,
|
||
report={"full_report": archived.model_dump(mode="json")},
|
||
)
|
||
|
||
class FakeDb:
|
||
def __init__(self):
|
||
self.added = []
|
||
self.commits = 0
|
||
|
||
async def get(self, model, key):
|
||
assert key == archived.session_id
|
||
return session_row
|
||
|
||
async def scalar(self, statement):
|
||
return score_row
|
||
|
||
def add(self, row):
|
||
self.added.append(row)
|
||
|
||
async def flush(self):
|
||
pass
|
||
|
||
async def rollback(self):
|
||
pass
|
||
|
||
async def commit(self):
|
||
self.commits += 1
|
||
|
||
async def __aenter__(self):
|
||
return self
|
||
|
||
async def __aexit__(self, *_args):
|
||
return None
|
||
|
||
db = FakeDb()
|
||
monkeypatch.setattr(sessions.hub, "store", PostgresSessionStore(lambda: db))
|
||
monkeypatch.setattr(sessions.hub, "get", lambda session_id: None)
|
||
monkeypatch.setattr(sessions, "_live", lambda session_id: (_ for _ in ()).throw(
|
||
HTTPException(status_code=404, detail="session_not_found")
|
||
))
|
||
monkeypatch.setattr(
|
||
sessions, "require",
|
||
lambda request, *roles: Principal(login="teacher", full_name="Преподаватель", role=Role.INSTRUCTOR),
|
||
)
|
||
audit_events = []
|
||
|
||
async def capture_audit(actor, role, action, object_id=None, detail=""):
|
||
audit_events.append((actor, role, action, object_id))
|
||
|
||
monkeypatch.setattr(sessions, "audit_required", capture_audit)
|
||
|
||
corrected = asyncio.run(sessions.override(
|
||
archived.session_id,
|
||
sessions.ScoreOverride(score_final=84.5, comment="проверена запись переговоров"),
|
||
object(),
|
||
db,
|
||
))
|
||
|
||
assert db.commits == 1, "оценка и аудит должны фиксироваться одной транзакцией"
|
||
assert score_row.score_auto == 70.0
|
||
assert score_row.score_final == 84.5
|
||
assert score_row.overridden_by == "teacher"
|
||
assert score_row.report["full_report"]["score_final"] == 84.5
|
||
assert corrected.score_auto == 70.0 and corrected.score_final == 84.5
|
||
assert corrected.override_comment == "проверена запись переговоров"
|
||
audit = db.added[0]
|
||
assert audit.action == "score.override" and audit.actor == "teacher"
|
||
assert "84.5" in audit.detail and "comment_chars=" in audit.detail
|
||
assert "проверена запись переговоров" not in audit.detail
|
||
|
||
report = asyncio.run(sessions.report(archived.session_id, object(), db))
|
||
assert report.score_final == 84.5 and report.score_auto == 70.0
|
||
csv_response = asyncio.run(sessions.report_csv(archived.session_id, object(), db))
|
||
assert "84.5" in csv_response.body.decode("utf-8-sig")
|
||
assert "проверена запись переговоров" in csv_response.body.decode("utf-8-sig")
|
||
pdf_response = asyncio.run(sessions.report_pdf(archived.session_id, object(), db))
|
||
assert pdf_response.body.startswith(b"%PDF-")
|
||
assert audit_events == [
|
||
("teacher", "instructor", "report.read", str(archived.session_id)),
|
||
("teacher", "instructor", "report.export.csv", str(archived.session_id)),
|
||
("teacher", "instructor", "report.export.pdf", str(archived.session_id)),
|
||
]
|