befa5c897f
- ReviewQueueRepository: filtered/paginated queue query + item loader. - Admin panel: /admin/review list/detail with extracted-text preview. - Operator actions: send-to-analysis, complete, reject with Document Slot compensation and row-level status guard. - /code-review cleanups applied; actions lock via DocumentRepository. Closes issues 017, 018.
335 lines
12 KiB
Python
335 lines
12 KiB
Python
"""Admin manual-review queue HTTP seam tests (issue 017 + 018).
|
||
|
||
Drives the server-rendered admin panel through the ASGI transport with real
|
||
Postgres / RabbitMQ / MinIO. Covers list/detail/filters, RBAC, CSRF guard, and
|
||
all three review actions.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import datetime as dt
|
||
import uuid
|
||
from decimal import Decimal
|
||
from typing import Any
|
||
|
||
import aio_pika
|
||
import httpx
|
||
import pytest
|
||
from sqlalchemy import text
|
||
|
||
from contract_check.core.db.session import create_session_factory
|
||
from contract_check.core.mq.messages import AnalyzeRequested
|
||
from contract_check.core.mq.topology import QUEUE_ANALYZE
|
||
from contract_check.core.s3 import extracted_key
|
||
from contract_check.core.s3.minio_storage import MinioStorage
|
||
from contract_check.core.security.passwords import hash_password
|
||
|
||
pytestmark = pytest.mark.integration
|
||
|
||
|
||
_TEST_TEXT = "Договор поставки. " * 50
|
||
|
||
|
||
def _storage(infra: dict[str, str]) -> MinioStorage:
|
||
return MinioStorage.from_endpoint_url(
|
||
endpoint_url=infra["s3_endpoint_url"],
|
||
access_key=infra["s3_access_key"],
|
||
secret_key=infra["s3_secret_key"],
|
||
bucket="contract-check-docs",
|
||
)
|
||
|
||
|
||
def _session_factory() -> Any:
|
||
return create_session_factory()
|
||
|
||
|
||
async def _seed_admin(db_session: Any, *, email: str, password: str) -> str:
|
||
result = await db_session.execute(
|
||
text(
|
||
"INSERT INTO users (email, password_hash, role, is_active) "
|
||
"VALUES (:e, :p, 'admin', TRUE) RETURNING id"
|
||
),
|
||
{"e": email, "p": hash_password(password)},
|
||
)
|
||
await db_session.commit()
|
||
return str(result.scalar_one())
|
||
|
||
|
||
async def _login_as(client: httpx.AsyncClient, email: str, password: str) -> None:
|
||
r = await client.post(
|
||
"/admin/login", data={"email": email, "password": password}, follow_redirects=False
|
||
)
|
||
assert r.status_code == 303
|
||
assert r.headers["location"].startswith("/admin/users")
|
||
|
||
|
||
async def _seed_manual_review_doc(
|
||
infra: dict[str, str],
|
||
*,
|
||
telegram_id: int | None = None,
|
||
created_at: Any,
|
||
confidence: float = 0.5,
|
||
) -> tuple[uuid.UUID, uuid.UUID]:
|
||
if telegram_id is None:
|
||
telegram_id = int(uuid.uuid4().int % 1_000_000_000)
|
||
sess = _session_factory()
|
||
store = _storage(infra)
|
||
async with sess() as session:
|
||
result = await session.execute(
|
||
text("INSERT INTO users (telegram_id, credits_left) VALUES (:t, 5) RETURNING id"),
|
||
{"t": telegram_id},
|
||
)
|
||
user_id = result.scalar_one()
|
||
await session.commit()
|
||
|
||
document_id = uuid.uuid4()
|
||
ext_key = extracted_key(str(user_id), str(document_id))
|
||
await store.put(ext_key, _TEST_TEXT.encode("utf-8"), content_type="text/plain; charset=utf-8")
|
||
|
||
async with sess() as session:
|
||
await session.execute(
|
||
text(
|
||
"INSERT INTO documents "
|
||
"(id, user_id, s3_key, extracted_s3_key, filename, mime, bytes, status, stage) "
|
||
"VALUES (:id, :uid, :s3, :ext, 'contract.pdf', 'application/pdf', :bytes, "
|
||
" 'manual_review', 'manual_review')"
|
||
),
|
||
{
|
||
"id": document_id,
|
||
"uid": user_id,
|
||
"s3": f"users/{user_id}/docs/{document_id}.pdf",
|
||
"ext": ext_key,
|
||
"bytes": len(_TEST_TEXT),
|
||
},
|
||
)
|
||
await session.execute(
|
||
text(
|
||
"INSERT INTO prescreen_results "
|
||
"(id, document_id, correlation_id, contract_type, party_a, party_b, "
|
||
" total_amount, currency, start_date, end_date, has_penalty_clause, "
|
||
" has_termination_clause, has_arbitration, confidence_score, routing_decision, "
|
||
" prescreened_at, auto_summary, auto_findings, extractor_version) "
|
||
"VALUES (:id, :d, :cid, 'supply', 'ООО Продавец', 'ООО Покупатель', "
|
||
" :ta, 'RUB', '2025-01-01', '2025-12-31', TRUE, TRUE, FALSE, "
|
||
" :cs, 'manual_review', now(), 'summary', '[{\"note\":\"ok\"}]', 'heuristic-v2')"
|
||
),
|
||
{
|
||
"id": uuid.uuid4(),
|
||
"d": document_id,
|
||
"cid": uuid.uuid4(),
|
||
"ta": Decimal("100000.00"),
|
||
"cs": confidence,
|
||
},
|
||
)
|
||
await session.execute(
|
||
text("UPDATE documents SET created_at = :ca WHERE id = :id"),
|
||
{"ca": created_at, "id": document_id},
|
||
)
|
||
await session.commit()
|
||
return user_id, document_id
|
||
|
||
|
||
async def test_review_queue_redirects_without_session(client: httpx.AsyncClient) -> None:
|
||
r = await client.get("/admin/review", follow_redirects=False)
|
||
assert r.status_code == 303
|
||
assert r.headers["location"].startswith("/admin/login")
|
||
|
||
|
||
async def test_review_queue_list_and_filters(
|
||
client: httpx.AsyncClient, db_session: Any, infra: dict[str, str]
|
||
) -> None:
|
||
admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
|
||
await _seed_admin(db_session, email=admin_email, password="adminpass-123")
|
||
await _login_as(client, admin_email, "adminpass-123")
|
||
|
||
user_id, doc_id = await _seed_manual_review_doc(
|
||
infra,
|
||
created_at=dt.datetime(2030, 3, 15, tzinfo=dt.UTC),
|
||
confidence=0.8,
|
||
)
|
||
await _seed_manual_review_doc(
|
||
infra,
|
||
created_at=dt.datetime(2030, 6, 1, tzinfo=dt.UTC),
|
||
confidence=0.4,
|
||
)
|
||
|
||
# Unfiltered list in the unique date range.
|
||
r = await client.get(
|
||
"/admin/review?date_from=2030-01-01&date_to=2030-12-31", follow_redirects=False
|
||
)
|
||
assert r.status_code == 200
|
||
assert b"contract.pdf" in r.content
|
||
assert "ООО Продавец".encode() in r.content
|
||
|
||
# Date filter excludes the later document.
|
||
r = await client.get(
|
||
"/admin/review?date_from=2030-03-01&date_to=2030-03-31", follow_redirects=False
|
||
)
|
||
assert r.status_code == 200
|
||
assert str(doc_id).encode() in r.content
|
||
|
||
# Confidence filter.
|
||
r = await client.get(
|
||
"/admin/review?date_from=2030-01-01&date_to=2030-12-31&conf_min=0.7",
|
||
follow_redirects=False,
|
||
)
|
||
assert r.status_code == 200
|
||
assert b"80.0%" in r.content
|
||
|
||
|
||
async def test_review_detail_with_preview(
|
||
client: httpx.AsyncClient, db_session: Any, infra: dict[str, str]
|
||
) -> None:
|
||
admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
|
||
await _seed_admin(db_session, email=admin_email, password="adminpass-123")
|
||
await _login_as(client, admin_email, "adminpass-123")
|
||
|
||
user_id, doc_id = await _seed_manual_review_doc(
|
||
infra, created_at=dt.datetime(2030, 4, 1, tzinfo=dt.UTC)
|
||
)
|
||
|
||
r = await client.get(f"/admin/review/{doc_id}", follow_redirects=False)
|
||
assert r.status_code == 200
|
||
assert "ООО Продавец".encode() in r.content
|
||
assert "Договор поставки.".encode() in r.content
|
||
assert f"/admin/review/{doc_id}/send-to-analysis".encode() in r.content
|
||
|
||
# Non-existent document returns 404.
|
||
r = await client.get(f"/admin/review/{uuid.uuid4()}", follow_redirects=False)
|
||
assert r.status_code == 404
|
||
|
||
|
||
async def test_review_action_requires_htmx_header(
|
||
client: httpx.AsyncClient, db_session: Any, infra: dict[str, str]
|
||
) -> None:
|
||
admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
|
||
await _seed_admin(db_session, email=admin_email, password="adminpass-123")
|
||
await _login_as(client, admin_email, "adminpass-123")
|
||
|
||
_user_id, doc_id = await _seed_manual_review_doc(
|
||
infra, created_at=dt.datetime(2030, 5, 1, tzinfo=dt.UTC)
|
||
)
|
||
|
||
r = await client.post(
|
||
f"/admin/review/{doc_id}/complete",
|
||
data={"note": "note"},
|
||
follow_redirects=False,
|
||
)
|
||
assert r.status_code == 400
|
||
|
||
|
||
async def test_review_complete_action(
|
||
client: httpx.AsyncClient, db_session: Any, infra: dict[str, str]
|
||
) -> None:
|
||
admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
|
||
await _seed_admin(db_session, email=admin_email, password="adminpass-123")
|
||
await _login_as(client, admin_email, "adminpass-123")
|
||
|
||
_user_id, doc_id = await _seed_manual_review_doc(
|
||
infra, created_at=dt.datetime(2030, 6, 1, tzinfo=dt.UTC)
|
||
)
|
||
|
||
r = await client.post(
|
||
f"/admin/review/{doc_id}/complete",
|
||
data={"note": "Проверено оператором."},
|
||
headers={"HX-Request": "true"},
|
||
follow_redirects=False,
|
||
)
|
||
assert r.status_code == 303
|
||
assert "/admin/review" in r.headers["location"]
|
||
|
||
sess = _session_factory()
|
||
async with sess() as session:
|
||
status = (
|
||
await session.execute(text("SELECT status FROM documents WHERE id = :d"), {"d": doc_id})
|
||
).scalar_one()
|
||
assert status == "done"
|
||
report = (
|
||
await session.execute(
|
||
text("SELECT markdown FROM reports WHERE document_id = :d"), {"d": doc_id}
|
||
)
|
||
).scalar_one()
|
||
assert "Проверено оператором." in report
|
||
|
||
|
||
async def test_review_reject_action(
|
||
client: httpx.AsyncClient, db_session: Any, infra: dict[str, str]
|
||
) -> None:
|
||
admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
|
||
await _seed_admin(db_session, email=admin_email, password="adminpass-123")
|
||
await _login_as(client, admin_email, "adminpass-123")
|
||
|
||
user_id, doc_id = await _seed_manual_review_doc(
|
||
infra, created_at=dt.datetime(2030, 7, 1, tzinfo=dt.UTC)
|
||
)
|
||
|
||
sess = _session_factory()
|
||
async with sess() as session:
|
||
from contract_check.core.db.repositories import CreditsRepository
|
||
|
||
await CreditsRepository(session).reserve(user_id, document_id=doc_id)
|
||
await session.commit()
|
||
|
||
r = await client.post(
|
||
f"/admin/review/{doc_id}/reject",
|
||
data={"reason": "Некорректный формат"},
|
||
headers={"HX-Request": "true"},
|
||
follow_redirects=False,
|
||
)
|
||
assert r.status_code == 303
|
||
assert "/admin/review" in r.headers["location"]
|
||
|
||
async with sess() as session:
|
||
status = (
|
||
await session.execute(text("SELECT status FROM documents WHERE id = :d"), {"d": doc_id})
|
||
).scalar_one()
|
||
assert status == "failed"
|
||
balance = (
|
||
await session.execute(
|
||
text("SELECT credits_left FROM users WHERE id = :u"), {"u": user_id}
|
||
)
|
||
).scalar_one()
|
||
assert balance == 5
|
||
|
||
|
||
async def test_review_send_to_analysis_action(
|
||
client: httpx.AsyncClient, db_session: Any, infra: dict[str, str]
|
||
) -> None:
|
||
admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
|
||
await _seed_admin(db_session, email=admin_email, password="adminpass-123")
|
||
await _login_as(client, admin_email, "adminpass-123")
|
||
|
||
_user_id, doc_id = await _seed_manual_review_doc(
|
||
infra, created_at=dt.datetime(2030, 8, 1, tzinfo=dt.UTC)
|
||
)
|
||
|
||
# Clean the shared analyze queue so we can consume exactly one message.
|
||
connection = await aio_pika.connect(infra["rabbitmq_url"])
|
||
try:
|
||
channel = await connection.channel()
|
||
queue = await channel.declare_queue(QUEUE_ANALYZE, passive=True)
|
||
await queue.purge()
|
||
|
||
r = await client.post(
|
||
f"/admin/review/{doc_id}/send-to-analysis",
|
||
headers={"HX-Request": "true"},
|
||
follow_redirects=False,
|
||
)
|
||
assert r.status_code == 303
|
||
|
||
msg = await queue.get(timeout=5)
|
||
body = msg.body.decode("utf-8")
|
||
analyze_msg = AnalyzeRequested.model_validate_json(body)
|
||
assert analyze_msg.document_id == doc_id
|
||
assert analyze_msg.attempt == 0
|
||
assert analyze_msg.prescreen_meta is not None
|
||
await msg.ack()
|
||
finally:
|
||
await connection.close()
|
||
|
||
sess = _session_factory()
|
||
async with sess() as session:
|
||
status = (
|
||
await session.execute(text("SELECT status FROM documents WHERE id = :d"), {"d": doc_id})
|
||
).scalar_one()
|
||
assert status == "analyzing"
|