diff --git a/src/contract_check/api/admin/mq.py b/src/contract_check/api/admin/mq.py new file mode 100644 index 0000000..e0205fc --- /dev/null +++ b/src/contract_check/api/admin/mq.py @@ -0,0 +1,256 @@ +"""DLQ admin panel: list, peek, requeue, and purge dead-letter queues. + +All management API calls go through :class:`RabbitMQManagementClient` so the +admin surface is broker-agnostic and mockable at the HTTP seam. +""" + +from __future__ import annotations + +from typing import Annotated + +from fastapi import APIRouter, Depends, Form, HTTPException, Request, status +from fastapi.responses import HTMLResponse, RedirectResponse + +from src.contract_check.api.admin.auth import AdminUser, HtmxGuard, require_admin +from src.contract_check.api.admin.templating import templates +from src.contract_check.core.logging import get_logger +from src.contract_check.core.mq.management import ( + MqManagementDisabledError, + MqManagementError, + RabbitMQManagementClient, +) +from src.contract_check.core.mq.topology import DLQ_FOR + +log = get_logger(__name__) + +router = APIRouter( + prefix="/admin/mq", + tags=["admin-mq"], + dependencies=[Depends(require_admin)], +) + +_PREVIEW_LIMIT = 20 + + +def _client() -> RabbitMQManagementClient: + return RabbitMQManagementClient() + + +def _attach_toast(response: RedirectResponse, toast: str) -> None: + import json + + response.headers["HX-Trigger"] = json.dumps({"showToast": toast}) + + +def _safe_toast(toast: str) -> str: + # Keep toast text short and URL-safe for redirects. + return toast[:200] + + +@router.get("/queues", response_class=HTMLResponse, include_in_schema=False) +async def list_queues( + request: Request, + admin: Annotated[AdminUser, Depends(require_admin)], +) -> HTMLResponse: + """List all DLQs with live depth and status.""" + client = _client() + dlq_names = sorted(DLQ_FOR.values()) + rows: list[dict[str, object]] = [] + error: str | None = None + try: + await client.connect() + for name in dlq_names: + try: + info = await client.get_queue_info(name) + rows.append({"name": info.name, "messages": info.messages, "state": info.state}) + except MqManagementError as exc: + log.warning("dlq_info_failed", queue=name, error=str(exc)) + rows.append({"name": name, "messages": None, "state": "unknown"}) + except MqManagementDisabledError: + error = "Управление RabbitMQ отключено в конфигурации." + except MqManagementError as exc: + error = f"Управление RabbitMQ недоступно: {exc}" + finally: + await client.aclose() + + return templates.TemplateResponse( + request, + "mq_queues.html", + { + "request": request, + "rows": rows, + "error": error, + "total_messages": sum( + int(r["messages"]) if isinstance(r["messages"], int) else 0 for r in rows + ), + }, + ) + + +@router.get("/dlq/{queue_name}", response_class=HTMLResponse, include_in_schema=False) +async def dlq_detail( + request: Request, + queue_name: str, + admin: Annotated[AdminUser, Depends(require_admin)], +) -> HTMLResponse: + """Peek recent messages from one DLQ.""" + if queue_name not in DLQ_FOR.values(): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="unknown DLQ") + + client = _client() + messages: list[dict[str, object]] = [] + error: str | None = None + info: dict[str, object] = {"name": queue_name, "messages": 0, "state": "unknown"} + try: + await client.connect() + try: + qi = await client.get_queue_info(queue_name) + info = {"name": qi.name, "messages": qi.messages, "state": qi.state} + except MqManagementError as exc: + log.warning("dlq_info_failed", queue=queue_name, error=str(exc)) + error = f"Не удалось получить состояние очереди: {exc}" + try: + raw = await client.peek_messages(queue_name, _PREVIEW_LIMIT) + for m in raw: + payload_text = m.payload.decode("utf-8", errors="replace") + messages.append( + { + "correlation_id": m.correlation_id or "—", + "failure_class": m.failure_class, + "failure_error": (m.failure_error or "")[:240], + "payload_summary": ( + payload_text[:240] + "…" if len(payload_text) > 240 else payload_text + ), + "headers": m.headers, + } + ) + except MqManagementError as exc: + log.warning("dlq_peek_failed", queue=queue_name, error=str(exc)) + if error is None: + error = f"Не удалось прочитать сообщения: {exc}" + except MqManagementDisabledError: + error = "Управление RabbitMQ отключено в конфигурации." + finally: + await client.aclose() + + return templates.TemplateResponse( + request, + "mq_dlq_detail.html", + { + "request": request, + "queue": info, + "messages": messages, + "error": error, + }, + ) + + +@router.post("/dlq/{queue_name}/requeue", response_class=HTMLResponse, include_in_schema=False) +async def requeue_dlq( + request: Request, + queue_name: str, + admin: Annotated[AdminUser, Depends(require_admin)], + _: Annotated[None, HtmxGuard], +) -> RedirectResponse: + """Move a bounded batch from the DLQ back to its main exchange.""" + if queue_name not in DLQ_FOR.values(): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="unknown DLQ") + + client = _client() + try: + await client.connect() + result = await client.requeue_batch(queue_name) + except MqManagementError as exc: + log.warning( + "dlq_requeue_failed", queue=queue_name, error=str(exc), admin_id=str(admin.user_id) + ) + return RedirectResponse( + url=f"/admin/mq/dlq/{queue_name}?toast={_safe_toast(str(exc))}", + status_code=status.HTTP_303_SEE_OTHER, + ) + finally: + await client.aclose() + + toast = f"Возвращено в очередь: {result.messages_requeued}" + if result.errors: + toast += f", ошибок: {result.errors}" + log.info( + "admin_dlq_requeue", + queue=queue_name, + admin_user_id=str(admin.user_id), + requeued=result.messages_requeued, + errors=result.errors, + ) + response = RedirectResponse( + url=f"/admin/mq/dlq/{queue_name}", status_code=status.HTTP_303_SEE_OTHER + ) + _attach_toast(response, toast) + return response + + +@router.get( + "/dlq/{queue_name}/purge/confirm", + response_class=HTMLResponse, + include_in_schema=False, +) +async def purge_confirm( + request: Request, + queue_name: str, + admin: Annotated[AdminUser, Depends(require_admin)], +) -> HTMLResponse: + """Show typed-confirmation page before purge.""" + if queue_name not in DLQ_FOR.values(): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="unknown DLQ") + return templates.TemplateResponse( + request, + "mq_purge_confirm.html", + {"request": request, "queue_name": queue_name}, + ) + + +@router.post("/dlq/{queue_name}/purge", response_class=HTMLResponse, include_in_schema=False) +async def purge_dlq( + request: Request, + queue_name: str, + admin: Annotated[AdminUser, Depends(require_admin)], + _: Annotated[None, HtmxGuard], + confirmation: Annotated[str, Form()] = "", +) -> RedirectResponse: + """Purge a DLQ only after explicit typed confirmation.""" + if queue_name not in DLQ_FOR.values(): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="unknown DLQ") + + expected = f"purge {queue_name}" + if confirmation.strip().lower() != expected.lower(): + return RedirectResponse( + url=f"/admin/mq/dlq/{queue_name}/purge/confirm?toast=Неверное+подтверждение.+Введите+%27{expected}%27", + status_code=status.HTTP_303_SEE_OTHER, + ) + + client = _client() + try: + await client.connect() + result = await client.purge_queue(queue_name) + except MqManagementError as exc: + log.warning( + "dlq_purge_failed", queue=queue_name, error=str(exc), admin_id=str(admin.user_id) + ) + return RedirectResponse( + url=f"/admin/mq/dlq/{queue_name}?toast={_safe_toast(str(exc))}", + status_code=status.HTTP_303_SEE_OTHER, + ) + finally: + await client.aclose() + + toast = f"DLQ {queue_name} очищена: удалено {result.messages_removed} сообщений" + log.info( + "admin_dlq_purge", + queue=queue_name, + admin_user_id=str(admin.user_id), + removed=result.messages_removed, + ) + response = RedirectResponse( + url=f"/admin/mq/dlq/{queue_name}", status_code=status.HTTP_303_SEE_OTHER + ) + _attach_toast(response, toast) + return response diff --git a/src/contract_check/api/admin/router.py b/src/contract_check/api/admin/router.py index 3f8ac84..ee31df2 100644 --- a/src/contract_check/api/admin/router.py +++ b/src/contract_check/api/admin/router.py @@ -14,6 +14,7 @@ from fastapi.responses import HTMLResponse, RedirectResponse from src.contract_check.api.admin.auth import ADMIN_COOKIE, ADMIN_COOKIE_MAX_AGE, resolve_admin from src.contract_check.api.admin.billing import router as billing_router +from src.contract_check.api.admin.mq import router as mq_router from src.contract_check.api.admin.review import router as review_router from src.contract_check.api.admin.templating import templates from src.contract_check.api.admin.users import router as users_router @@ -30,6 +31,7 @@ router = APIRouter(tags=["admin"]) router.include_router(users_router) router.include_router(billing_router) router.include_router(review_router) +router.include_router(mq_router) @router.get("/admin", include_in_schema=False) diff --git a/src/contract_check/api/admin/templates/base.html b/src/contract_check/api/admin/templates/base.html index 3d8bc26..bbf84ad 100644 --- a/src/contract_check/api/admin/templates/base.html +++ b/src/contract_check/api/admin/templates/base.html @@ -71,6 +71,7 @@ Пользователи Счета Ручная проверка + DLQ
diff --git a/src/contract_check/api/admin/templates/mq_dlq_detail.html b/src/contract_check/api/admin/templates/mq_dlq_detail.html new file mode 100644 index 0000000..3fdf59a --- /dev/null +++ b/src/contract_check/api/admin/templates/mq_dlq_detail.html @@ -0,0 +1,47 @@ +{% extends "base.html" %} +{% block title %}DLQ {{ queue.name }} — Админка{% endblock %} +{% block content %} +
+ ← Назад к списку + + {{ queue.name }} · {{ queue.messages }} сообщений · {{ queue.state }} +
+ +{% if error %} +
+

{{ error }}

+
+{% endif %} + +
+
+ +
+ Очистить DLQ +
+ +
+ + + + + + + + + + + {% for m in messages %} + + + + + + + {% else %} + + {% endfor %} + +
Correlation IDКласс ошибкиСокращённая ошибкаСводка payload
{{ m.correlation_id }}{{ m.failure_class }}{{ m.failure_error or '—' }}{{ m.payload_summary }}
Очередь пуста
+
+{% endblock %} diff --git a/src/contract_check/api/admin/templates/mq_purge_confirm.html b/src/contract_check/api/admin/templates/mq_purge_confirm.html new file mode 100644 index 0000000..4a54a3f --- /dev/null +++ b/src/contract_check/api/admin/templates/mq_purge_confirm.html @@ -0,0 +1,17 @@ +{% extends "base.html" %} +{% block title %}Очистка {{ queue_name }} — Админка{% endblock %} +{% block content %} +
+ ← Назад +
+ +
+

Очистить {{ queue_name }}?

+

Это безвозвратно удалит все сообщения из DLQ. Для подтверждения введите purge {{ queue_name }}.

+ +
+ + +
+
+{% endblock %} diff --git a/src/contract_check/api/admin/templates/mq_queues.html b/src/contract_check/api/admin/templates/mq_queues.html new file mode 100644 index 0000000..e4c9265 --- /dev/null +++ b/src/contract_check/api/admin/templates/mq_queues.html @@ -0,0 +1,46 @@ +{% extends "base.html" %} +{% block title %}DLQ — Админка{% endblock %} +{% block content %} +
+
+

Очереди DLQ ({{ total_messages }})

+
+ + {% if error %} +
+

{{ error }}

+
+ {% endif %} + + + + + + + + + + + + {% for row in rows %} + + + + + + + {% else %} + + {% endfor %} + +
ОчередьСообщенийСостояние
{{ row.name }} + {% if row.messages is none %} + — + {% elif row.messages == 0 %} + 0 + {% else %} + {{ row.messages }} + {% endif %} + {{ row.state }}Просмотр
Нет DLQ
+
+{% endblock %} diff --git a/src/contract_check/core/config.py b/src/contract_check/core/config.py index d56d2eb..1906ff1 100644 --- a/src/contract_check/core/config.py +++ b/src/contract_check/core/config.py @@ -44,8 +44,12 @@ class Settings(BaseSettings): mq_prefetch_extract: int = 1 mq_prefetch_analyze: int = 3 mq_prefetch_notify: int = 5 + mq_prefetch_prescreen: int = 1 mq_max_attempts: int = 5 mq_retry_base_ms: int = 2000 + mq_management_enabled: bool = True + mq_management_timeout: float = 10.0 + mq_requeue_batch_size: int = 50 # --- object storage (MinIO) --- s3_endpoint_url: str diff --git a/src/contract_check/core/mq/management.py b/src/contract_check/core/mq/management.py new file mode 100644 index 0000000..b656dd9 --- /dev/null +++ b/src/contract_check/core/mq/management.py @@ -0,0 +1,365 @@ +"""Async RabbitMQ management API client. + +Derives URL and credentials from the broker AMQP URL, exactly like the +lazy-queue policy setup in :mod:`src.contract_check.core.mq.topology`. + +All operations return small typed dataclasses. Network/management-plugin +failures are mapped to :class:`MqManagementError` subclasses so callers +can render a clean error state instead of a traceback. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any +from urllib.parse import quote + +import httpx + +from src.contract_check.core.config import get_settings +from src.contract_check.core.http_logging import http_log_event_hooks +from src.contract_check.core.logging import get_logger +from src.contract_check.core.mq.topology import ( + DLQ_FOR, + EXCHANGE_MAIN, + EXCHANGE_NOTIFY, + H_ATTEMPT, + H_CORRELATION_ID, + _amqp_credentials, + _management_url_from_amqp, +) + +log = get_logger(__name__) + + +class MqManagementError(Exception): + """Base for RabbitMQ management API errors.""" + + +class MqManagementDisabledError(MqManagementError): + """Raised when the management client is disabled by configuration.""" + + +class MqManagementUnreachableError(MqManagementError): + """Raised when the management plugin is unreachable or disabled.""" + + +class MqManagementNotFoundError(MqManagementError): + """Raised when the management API returns 404 (queue not found).""" + + +class MqManagementResponseError(MqManagementError): + """Raised for unexpected management API status codes.""" + + def __init__(self, message: str, *, status_code: int, body: str = "") -> None: + super().__init__(message) + self.status_code = status_code + self.body = body + + +@dataclass(frozen=True, slots=True) +class QueueInfo: + """Minimal queue metadata returned by the management API.""" + + name: str + messages: int + state: str + + +@dataclass(frozen=True, slots=True) +class QueueMessage: + """One peeked message from a queue, with its headers and payload summary.""" + + payload: bytes + headers: dict[str, Any] + correlation_id: str | None + routing_key: str | None + + @property + def failure_class(self) -> str: + return str(self.headers.get("x-failure-class", "unknown")) + + @property + def failure_error(self) -> str: + return str(self.headers.get("x-failure-error", "")) + + +@dataclass(frozen=True, slots=True) +class PurgeResult: + messages_removed: int + + +@dataclass(frozen=True, slots=True) +class RequeueResult: + messages_requeued: int + errors: int + + +def _main_exchange_and_routing_key(queue_name: str) -> tuple[str, str]: + """Return the exchange and routing key to publish a requeued message to.""" + if queue_name == "notify.q": + return EXCHANGE_NOTIFY, "notify" + return EXCHANGE_MAIN, queue_name + + +class RabbitMQManagementClient: + """Small async client for the RabbitMQ management HTTP API.""" + + def __init__( + self, + amqp_url: str | None = None, + *, + timeout: float | None = None, + ) -> None: + settings = get_settings() + self._enabled = settings.mq_management_enabled + self._amqp_url = amqp_url or settings.rabbitmq_url + self._timeout = timeout or settings.mq_management_timeout + self._vhost = quote("/", safe="") + self._client: httpx.AsyncClient | None = None + + async def __aenter__(self) -> RabbitMQManagementClient: + await self.connect() + return self + + async def __aexit__(self, *exc: object) -> None: + await self.aclose() + + def _ensure_enabled(self) -> None: + if not self._enabled: + raise MqManagementDisabledError("RabbitMQ management client is disabled") + + def _management_url(self) -> str: + url = _management_url_from_amqp(self._amqp_url) + if not url: + raise MqManagementUnreachableError("cannot derive management URL from broker URL") + return url + + def _auth(self) -> tuple[str, str]: + creds = _amqp_credentials(self._amqp_url) + if not creds: + raise MqManagementUnreachableError( + "cannot derive management credentials from broker URL" + ) + return creds + + async def connect(self) -> None: + self._ensure_enabled() + if self._client is not None: + return + mgmt_url = self._management_url() + username, password = self._auth() + self._client = httpx.AsyncClient( + base_url=mgmt_url, + auth=(username, password), + timeout=httpx.Timeout(self._timeout, connect=5.0), + event_hooks=http_log_event_hooks(service="rabbitmq_mgmt"), + ) + + async def aclose(self) -> None: + if self._client is not None: + await self._client.aclose() + self._client = None + + @property + def client(self) -> httpx.AsyncClient: + if self._client is None: + raise RuntimeError("RabbitMQManagementClient not connected") + return self._client + + async def get_queue_info(self, queue_name: str) -> QueueInfo: + """Return name, depth, and state for a queue.""" + self._ensure_enabled() + url = f"/api/queues/{self._vhost}/{quote(queue_name, safe='')}" + try: + resp = await self.client.get(url) + except httpx.NetworkError as exc: + raise MqManagementUnreachableError(f"management API unreachable: {exc}") from exc + except httpx.TimeoutException as exc: + raise MqManagementUnreachableError(f"management API timeout: {exc}") from exc + if resp.status_code == 404: + raise MqManagementNotFoundError(f"queue {queue_name!r} not found") + if resp.status_code >= 400: + raise MqManagementResponseError( + f"management API error {resp.status_code}", + status_code=resp.status_code, + body=resp.text[:500], + ) + data = resp.json() + return QueueInfo( + name=str(data.get("name", queue_name)), + messages=int(data.get("messages", 0)), + state=str(data.get("state", "unknown")), + ) + + async def peek_messages(self, queue_name: str, count: int) -> list[QueueMessage]: + """Non-destructively peek at up to ``count`` messages from a queue.""" + self._ensure_enabled() + url = f"/api/queues/{self._vhost}/{quote(queue_name, safe='')}/get" + body: dict[str, Any] = { + "count": max(1, count), + "ackmode": "ack_requeue_true", + "encoding": "auto", + "truncate": 2000, + } + try: + resp = await self.client.post(url, json=body) + except httpx.NetworkError as exc: + raise MqManagementUnreachableError(f"management API unreachable: {exc}") from exc + except httpx.TimeoutException as exc: + raise MqManagementUnreachableError(f"management API timeout: {exc}") from exc + if resp.status_code == 404: + raise MqManagementNotFoundError(f"queue {queue_name!r} not found") + if resp.status_code >= 400: + raise MqManagementResponseError( + f"management API error {resp.status_code}", + status_code=resp.status_code, + body=resp.text[:500], + ) + rows = resp.json() + if not isinstance(rows, list): + return [] + out: list[QueueMessage] = [] + for row in rows: + if not isinstance(row, dict): + continue + payload = self._decode_payload(row) + headers = row.get("properties", {}).get("headers") or {} + out.append( + QueueMessage( + payload=payload, + headers=dict(headers), + correlation_id=row.get("properties", {}).get("correlation_id") + or headers.get(H_CORRELATION_ID), + routing_key=row.get("routing_key"), + ) + ) + return out + + async def purge_queue(self, queue_name: str) -> PurgeResult: + """Empty a queue and return the number of messages removed.""" + self._ensure_enabled() + url = f"/api/queues/{self._vhost}/{quote(queue_name, safe='')}/contents" + try: + resp = await self.client.delete(url) + except httpx.NetworkError as exc: + raise MqManagementUnreachableError(f"management API unreachable: {exc}") from exc + except httpx.TimeoutException as exc: + raise MqManagementUnreachableError(f"management API timeout: {exc}") from exc + if resp.status_code == 404: + raise MqManagementNotFoundError(f"queue {queue_name!r} not found") + if resp.status_code >= 400: + raise MqManagementResponseError( + f"management API error {resp.status_code}", + status_code=resp.status_code, + body=resp.text[:500], + ) + # Successful purge returns no body; the queue is now empty. + info = await self.get_queue_info(queue_name) + return PurgeResult(messages_removed=max(0, info.messages)) + + async def requeue_batch( + self, + dlq_name: str, + *, + batch_size: int | None = None, + ) -> RequeueResult: + """Move up to ``batch_size`` messages from a DLQ back to their main exchange. + + The ``x-attempt`` header is reset to 0 so retried messages get a fresh + retry budget. Unknown DLQ names are rejected explicitly. + """ + self._ensure_enabled() + main_queue = _main_queue_for_dlq(dlq_name) + if main_queue is None: + raise MqManagementNotFoundError(f"no main queue known for DLQ {dlq_name!r}") + + settings = get_settings() + batch_size = max(1, batch_size or settings.mq_requeue_batch_size) + + url = f"/api/queues/{self._vhost}/{quote(dlq_name, safe='')}/get" + body: dict[str, Any] = { + "count": batch_size, + "ackmode": "ack_requeue_false", + "encoding": "auto", + "truncate": 2000, + } + try: + resp = await self.client.post(url, json=body) + except httpx.NetworkError as exc: + raise MqManagementUnreachableError(f"management API unreachable: {exc}") from exc + except httpx.TimeoutException as exc: + raise MqManagementUnreachableError(f"management API timeout: {exc}") from exc + if resp.status_code == 404: + raise MqManagementNotFoundError(f"queue {dlq_name!r} not found") + if resp.status_code >= 400: + raise MqManagementResponseError( + f"management API error {resp.status_code}", + status_code=resp.status_code, + body=resp.text[:500], + ) + rows = resp.json() + if not isinstance(rows, list) or not rows: + return RequeueResult(messages_requeued=0, errors=0) + + exchange, routing_key = _main_exchange_and_routing_key(main_queue) + requeued = 0 + errors = 0 + for row in rows: + if not isinstance(row, dict): + continue + try: + await self._publish_to_main(row, exchange, routing_key) + requeued += 1 + except Exception as exc: # noqa: BLE001 — count failures, keep going + log.warning("requeue_publish_failed", dlq=dlq_name, error=str(exc)) + errors += 1 + return RequeueResult(messages_requeued=requeued, errors=errors) + + async def _publish_to_main( + self, + row: dict[str, Any], + exchange: str, + routing_key: str, + ) -> None: + """Publish one message back to the main exchange with attempt reset to 0.""" + url = f"/api/exchanges/{self._vhost}/{quote(exchange, safe='')}/publish" + props = dict(row.get("properties", {})) + headers = dict(props.get("headers") or {}) + headers[H_ATTEMPT] = 0 + props["headers"] = headers + if H_CORRELATION_ID not in headers and props.get("correlation_id"): + headers[H_CORRELATION_ID] = props["correlation_id"] + payload = self._decode_payload(row) + body: dict[str, Any] = { + "routing_key": routing_key, + "payload": payload.decode("utf-8", errors="replace"), + "payload_encoding": "string", + "properties": props, + } + resp = await self.client.post(url, json=body) + if resp.status_code >= 400: + raise MqManagementResponseError( + f"publish to {exchange!r} failed: {resp.status_code}", + status_code=resp.status_code, + body=resp.text[:500], + ) + + @staticmethod + def _decode_payload(row: dict[str, Any]) -> bytes: + payload = row.get("payload") + if payload is None: + return b"" + if isinstance(payload, bytes): + return payload + if isinstance(payload, str): + return payload.encode("utf-8", errors="replace") + return str(payload).encode("utf-8", errors="replace") + + +def _main_queue_for_dlq(dlq_name: str) -> str | None: + """Return the main queue that feeds the given DLQ.""" + for main, dlq in DLQ_FOR.items(): + if dlq == dlq_name: + return main + return None diff --git a/tests/unit/test_mq_management.py b/tests/unit/test_mq_management.py new file mode 100644 index 0000000..9e1a393 --- /dev/null +++ b/tests/unit/test_mq_management.py @@ -0,0 +1,242 @@ +"""Unit tests for the RabbitMQ management client. + +All real HTTP traffic is intercepted with ``respx``. We do not need a running +broker here — the integration suite covers live RabbitMQ behaviour. +""" + +from __future__ import annotations + +import json + +import httpx +import pytest +import respx + +from src.contract_check.core.mq.management import ( + MqManagementDisabledError, + MqManagementNotFoundError, + MqManagementResponseError, + MqManagementUnreachableError, + PurgeResult, + QueueInfo, + QueueMessage, + RabbitMQManagementClient, + RequeueResult, +) +from src.contract_check.core.mq.topology import DLQ_FOR + + +@pytest.fixture +def mgmt_client(monkeypatch: pytest.MonkeyPatch) -> RabbitMQManagementClient: + # Pin AMQP URL before Settings is imported/looked up anywhere else. + monkeypatch.setenv("RABBITMQ_URL", "amqp://admin:secret@rabbitmq:5672//") + from src.contract_check.core.config import get_settings + + get_settings.cache_clear() + return RabbitMQManagementClient() + + +def _mgmt_root(base: str) -> str: + return base.rstrip("/") + + +@pytest.mark.anyio +async def test_get_queue_info_ok(mgmt_client: RabbitMQManagementClient) -> None: + with respx.mock: + route = respx.get("http://rabbitmq:15672/api/queues/%2F/extract.q").mock( + return_value=httpx.Response( + 200, json={"name": "extract.q", "messages": 7, "state": "running"} + ) + ) + await mgmt_client.connect() + info = await mgmt_client.get_queue_info("extract.q") + assert info == QueueInfo(name="extract.q", messages=7, state="running") + assert route.called + await mgmt_client.aclose() + + +@pytest.mark.anyio +async def test_get_queue_info_not_found(mgmt_client: RabbitMQManagementClient) -> None: + with respx.mock: + respx.get("http://rabbitmq:15672/api/queues/%2F/missing").mock( + return_value=httpx.Response(404) + ) + await mgmt_client.connect() + with pytest.raises(MqManagementNotFoundError): + await mgmt_client.get_queue_info("missing") + await mgmt_client.aclose() + + +@pytest.mark.anyio +async def test_get_queue_info_unreachable(mgmt_client: RabbitMQManagementClient) -> None: + with respx.mock: + respx.get("http://rabbitmq:15672/api/queues/%2F/extract.q").mock( + side_effect=httpx.ConnectError("nope") + ) + await mgmt_client.connect() + with pytest.raises(MqManagementUnreachableError): + await mgmt_client.get_queue_info("extract.q") + await mgmt_client.aclose() + + +@pytest.mark.anyio +async def test_disabled_client_raises( + mgmt_client: RabbitMQManagementClient, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(mgmt_client, "_enabled", False) + with pytest.raises(MqManagementDisabledError): + await mgmt_client.connect() + + +@pytest.mark.anyio +async def test_peek_messages(mgmt_client: RabbitMQManagementClient) -> None: + payload = json.dumps({"document_id": "d1"}).encode() + with respx.mock: + respx.post("http://rabbitmq:15672/api/queues/%2F/extract.dlq/get").mock( + return_value=httpx.Response( + 200, + json=[ + { + "routing_key": "extract.q", + "payload": payload.decode(), + "properties": { + "correlation_id": "c1", + "headers": {"x-failure-class": "infra", "x-failure-error": "boom"}, + }, + } + ], + ) + ) + await mgmt_client.connect() + msgs = await mgmt_client.peek_messages("extract.dlq", 5) + assert len(msgs) == 1 + msg = msgs[0] + assert isinstance(msg, QueueMessage) + assert msg.correlation_id == "c1" + assert msg.routing_key == "extract.q" + assert msg.failure_class == "infra" + assert msg.failure_error == "boom" + assert msg.payload == payload + await mgmt_client.aclose() + + +@pytest.mark.anyio +async def test_peek_messages_empty_non_list(mgmt_client: RabbitMQManagementClient) -> None: + with respx.mock: + respx.post("http://rabbitmq:15672/api/queues/%2F/extract.dlq/get").mock( + return_value=httpx.Response(200, json={"garbage": True}) + ) + await mgmt_client.connect() + assert await mgmt_client.peek_messages("extract.dlq", 5) == [] + await mgmt_client.aclose() + + +@pytest.mark.anyio +async def test_purge_queue(mgmt_client: RabbitMQManagementClient) -> None: + with respx.mock: + respx.delete("http://rabbitmq:15672/api/queues/%2F/extract.dlq/contents").mock( + return_value=httpx.Response(204) + ) + respx.get("http://rabbitmq:15672/api/queues/%2F/extract.dlq").mock( + return_value=httpx.Response( + 200, json={"name": "extract.dlq", "messages": 0, "state": "running"} + ) + ) + await mgmt_client.connect() + assert await mgmt_client.purge_queue("extract.dlq") == PurgeResult(messages_removed=0) + await mgmt_client.aclose() + + +@pytest.mark.anyio +async def test_requeue_batch_unknown_dlq(mgmt_client: RabbitMQManagementClient) -> None: + await mgmt_client.connect() + with pytest.raises(MqManagementNotFoundError): + await mgmt_client.requeue_batch("not-a-known-dlq") + await mgmt_client.aclose() + + +@pytest.mark.anyio +async def test_requeue_batch_empty(mgmt_client: RabbitMQManagementClient) -> None: + dlq = list(DLQ_FOR.values())[0] + with respx.mock: + respx.post(f"http://rabbitmq:15672/api/queues/%2F/{dlq}/get").mock( + return_value=httpx.Response(200, json=[]) + ) + await mgmt_client.connect() + assert await mgmt_client.requeue_batch(dlq) == RequeueResult(messages_requeued=0, errors=0) + await mgmt_client.aclose() + + +@pytest.mark.anyio +async def test_requeue_batch_success(mgmt_client: RabbitMQManagementClient) -> None: + dlq = "extract.dlq" + main_queue = "extract.q" + payload = json.dumps({"document_id": "d1"}) + with respx.mock: + respx.post("http://rabbitmq:15672/api/queues/%2F/extract.dlq/get").mock( + return_value=httpx.Response( + 200, + json=[ + { + "routing_key": main_queue, + "payload": payload, + "properties": { + "correlation_id": "c1", + "headers": {"x-attempt": 5, "x-failure-class": "infra"}, + }, + } + ], + ) + ) + respx.post("http://rabbitmq:15672/api/exchanges/%2F/contracts.x/publish").mock( + return_value=httpx.Response(200, json={"routed": True}) + ) + await mgmt_client.connect() + result = await mgmt_client.requeue_batch(dlq, batch_size=10) + assert result == RequeueResult(messages_requeued=1, errors=0) + await mgmt_client.aclose() + + +@pytest.mark.anyio +async def test_requeue_batch_publish_failure_counts_as_error( + mgmt_client: RabbitMQManagementClient, +) -> None: + dlq = "extract.dlq" + main_queue = "extract.q" + with respx.mock: + respx.post("http://rabbitmq:15672/api/queues/%2F/extract.dlq/get").mock( + return_value=httpx.Response( + 200, + json=[ + { + "routing_key": main_queue, + "payload": "{}", + "properties": {"headers": {"x-attempt": 5}}, + } + ], + ) + ) + respx.post("http://rabbitmq:15672/api/exchanges/%2F/contracts.x/publish").mock( + return_value=httpx.Response(500, text="broker error") + ) + await mgmt_client.connect() + result = await mgmt_client.requeue_batch(dlq, batch_size=10) + assert result == RequeueResult(messages_requeued=0, errors=1) + await mgmt_client.aclose() + + +@pytest.mark.anyio +async def test_response_error_includes_status_and_body( + mgmt_client: RabbitMQManagementClient, +) -> None: + with respx.mock: + respx.get("http://rabbitmq:15672/api/queues/%2F/extract.q").mock( + return_value=httpx.Response(418, text="teapot") + ) + await mgmt_client.connect() + with pytest.raises(MqManagementResponseError) as exc_info: + await mgmt_client.get_queue_info("extract.q") + err = exc_info.value + assert err.status_code == 418 + assert err.body == "teapot" + await mgmt_client.aclose()