mirror of
https://github.com/zvx-echo6/central.git
synced 2026-08-26 17:31:39 +00:00
40ea23e904
- routes.py: change `async for ci in js.consumers_info(stream_name)` to `for ci in await js.consumers_info(stream_name)` — nats-py 2.14.0 consumers_info() is a plain coroutine returning list[ConsumerInfo], not an async iterable; the old form threw TypeError silently (swallowed by except), causing every stream to show "unavailable" and zero consumers. - test_consumers.py: replace async-generator mock with AsyncMock returning a list, matching the real API; also fix inline consumers_info_raising in the error test (remove dead yield); add explicit regression guard asserting consumer names appear in the template context. - consumers_list.html: guard num_pending/num_ack_pending/num_redelivered/ num_waiting with `… if … is not none else '—'` to prevent "None" in cells. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
280 lines
11 KiB
Python
280 lines
11 KiB
Python
"""Tests for consumers admin routes (GET /consumers, POST /consumers/{s}/{c}/delete)."""
|
|
|
|
import os
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
# Set required env vars before importing central modules
|
|
os.environ.setdefault("CENTRAL_DB_DSN", "postgresql://test:test@localhost/test")
|
|
os.environ.setdefault("CENTRAL_CSRF_SECRET", "testsecret12345678901234567890ab")
|
|
os.environ.setdefault("CENTRAL_NATS_URL", "nats://localhost:4222")
|
|
|
|
|
|
def _make_consumer_info(name: str, num_pending: int = 0, num_ack_pending: int = 0,
|
|
num_redelivered: int = 0, num_waiting: int = 0):
|
|
ci = MagicMock()
|
|
ci.name = name
|
|
ci.num_pending = num_pending
|
|
ci.num_ack_pending = num_ack_pending
|
|
ci.num_redelivered = num_redelivered
|
|
ci.num_waiting = num_waiting
|
|
ci.created = datetime(2026, 5, 17, 12, 0, 0, tzinfo=timezone.utc)
|
|
return ci
|
|
|
|
|
|
def _make_js_with_consumers(consumers_by_stream: dict):
|
|
"""Build a mock JetStreamContext whose consumers_info is a coroutine returning a list."""
|
|
mock_js = MagicMock()
|
|
mock_js.consumers_info = AsyncMock(
|
|
side_effect=lambda stream, **kw: consumers_by_stream.get(stream, [])
|
|
)
|
|
mock_js.consumer_info = AsyncMock()
|
|
mock_js.delete_consumer = AsyncMock()
|
|
return mock_js
|
|
|
|
|
|
class TestConsumersListNatsUnavailable:
|
|
"""GET /consumers when NATS is down shows per-stream error."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nats_unavailable_shows_error_per_stream(self):
|
|
from central.gui.routes import consumers_list
|
|
|
|
mock_request = MagicMock()
|
|
mock_request.state.operator = MagicMock(id=1, username="testop")
|
|
mock_request.state.csrf_token = "test_csrf"
|
|
|
|
mock_templates = MagicMock()
|
|
mock_templates.TemplateResponse.return_value = MagicMock()
|
|
|
|
with patch("central.gui.routes._get_templates", return_value=mock_templates):
|
|
with patch("central.gui.nats.get_js", return_value=None):
|
|
await consumers_list(mock_request)
|
|
|
|
call_args = mock_templates.TemplateResponse.call_args
|
|
context = call_args.kwargs.get("context", call_args[1].get("context"))
|
|
streams = context["streams"]
|
|
# All streams should show the NATS unavailable error
|
|
assert all(s["error"] == "NATS unavailable" for s in streams)
|
|
# And no consumers listed
|
|
assert all(s["consumers"] == [] for s in streams)
|
|
|
|
|
|
class TestConsumersListWithConsumers:
|
|
"""GET /consumers with live NATS returns consumers per stream."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_consumers_listed_with_protected_flag(self):
|
|
from central.gui.routes import consumers_list
|
|
|
|
mock_request = MagicMock()
|
|
mock_request.state.operator = MagicMock(id=1, username="testop")
|
|
mock_request.state.csrf_token = "test_csrf"
|
|
|
|
mock_templates = MagicMock()
|
|
mock_templates.TemplateResponse.return_value = MagicMock()
|
|
|
|
consumers_by_stream = {
|
|
"CENTRAL_WX": [
|
|
_make_consumer_info("archive-CENTRAL_WX", num_pending=5, num_waiting=1),
|
|
_make_consumer_info("meshai-wx", num_pending=1000, num_waiting=0),
|
|
],
|
|
}
|
|
mock_js = _make_js_with_consumers(consumers_by_stream)
|
|
|
|
with patch("central.gui.routes._get_templates", return_value=mock_templates):
|
|
with patch("central.gui.nats.get_js", return_value=mock_js):
|
|
await consumers_list(mock_request)
|
|
|
|
call_args = mock_templates.TemplateResponse.call_args
|
|
context = call_args.kwargs.get("context", call_args[1].get("context"))
|
|
streams = context["streams"]
|
|
|
|
wx = next(s for s in streams if s["stream"] == "CENTRAL_WX")
|
|
assert wx["error"] is None
|
|
assert len(wx["consumers"]) == 2
|
|
|
|
archive_c = next(c for c in wx["consumers"] if c["name"] == "archive-CENTRAL_WX")
|
|
assert archive_c["protected"] is True
|
|
assert archive_c["num_pending"] == 5
|
|
|
|
meshai_c = next(c for c in wx["consumers"] if c["name"] == "meshai-wx")
|
|
assert meshai_c["protected"] is False
|
|
assert meshai_c["num_pending"] == 1000
|
|
assert meshai_c["num_waiting"] == 0
|
|
|
|
# Regression guard: consumer names must appear in the template context so
|
|
# they are rendered into the HTML body (guards against the coroutine/iterator
|
|
# bug where consumers_info was consumed as an async-iterable instead of awaited).
|
|
consumer_names_in_context = {c["name"] for c in wx["consumers"]}
|
|
assert "archive-CENTRAL_WX" in consumer_names_in_context
|
|
assert "meshai-wx" in consumer_names_in_context
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stream_with_no_consumers_shows_empty(self):
|
|
from central.gui.routes import consumers_list
|
|
|
|
mock_request = MagicMock()
|
|
mock_request.state.operator = MagicMock(id=1, username="testop")
|
|
mock_request.state.csrf_token = "test_csrf"
|
|
|
|
mock_templates = MagicMock()
|
|
mock_templates.TemplateResponse.return_value = MagicMock()
|
|
|
|
mock_js = _make_js_with_consumers({}) # No consumers on any stream
|
|
|
|
with patch("central.gui.routes._get_templates", return_value=mock_templates):
|
|
with patch("central.gui.nats.get_js", return_value=mock_js):
|
|
await consumers_list(mock_request)
|
|
|
|
call_args = mock_templates.TemplateResponse.call_args
|
|
context = call_args.kwargs.get("context", call_args[1].get("context"))
|
|
streams = context["streams"]
|
|
assert all(s["consumers"] == [] for s in streams)
|
|
assert all(s["error"] is None for s in streams)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_one_stream_error_does_not_break_page(self):
|
|
from central.gui.routes import consumers_list
|
|
|
|
mock_request = MagicMock()
|
|
mock_request.state.operator = MagicMock(id=1, username="testop")
|
|
mock_request.state.csrf_token = "test_csrf"
|
|
|
|
mock_templates = MagicMock()
|
|
mock_templates.TemplateResponse.return_value = MagicMock()
|
|
|
|
mock_js = MagicMock()
|
|
|
|
async def consumers_info_raising(stream_name):
|
|
if stream_name == "CENTRAL_FIRE":
|
|
raise RuntimeError("stream not found")
|
|
# other streams: empty list (coroutine returning a list, not an async generator)
|
|
return []
|
|
|
|
mock_js.consumers_info = consumers_info_raising
|
|
|
|
with patch("central.gui.routes._get_templates", return_value=mock_templates):
|
|
with patch("central.gui.nats.get_js", return_value=mock_js):
|
|
await consumers_list(mock_request)
|
|
|
|
call_args = mock_templates.TemplateResponse.call_args
|
|
context = call_args.kwargs.get("context", call_args[1].get("context"))
|
|
streams = context["streams"]
|
|
|
|
fire = next(s for s in streams if s["stream"] == "CENTRAL_FIRE")
|
|
assert "unavailable" in fire["error"]
|
|
assert fire["consumers"] == []
|
|
|
|
|
|
class TestConsumersDeleteArchiveGuard:
|
|
"""POST /consumers/{stream}/archive-*/delete must be refused."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archive_consumer_refused_redirects(self):
|
|
from central.gui.routes import consumers_delete
|
|
|
|
mock_request = MagicMock()
|
|
mock_request.state.operator = MagicMock(id=1)
|
|
mock_request.state.csrf_token = "tok"
|
|
form_data = MagicMock()
|
|
form_data.get.side_effect = lambda k, d="": {"csrf_token": "tok"}.get(k, d)
|
|
mock_request.form = AsyncMock(return_value=form_data)
|
|
|
|
with patch("central.gui.nats.get_js", return_value=MagicMock()):
|
|
result = await consumers_delete(mock_request, "CENTRAL_WX", "archive-CENTRAL_WX")
|
|
|
|
assert result.status_code == 302
|
|
assert result.headers["location"] == "/consumers"
|
|
|
|
|
|
class TestConsumersDeleteSuccess:
|
|
"""POST /consumers/{stream}/{consumer}/delete happy path."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_non_protected_consumer_audits_and_redirects(self):
|
|
from central.gui.routes import consumers_delete
|
|
|
|
mock_request = MagicMock()
|
|
mock_request.state.operator = MagicMock(id=1)
|
|
mock_request.state.csrf_token = "tok"
|
|
form_data = MagicMock()
|
|
form_data.get.side_effect = lambda k, d="": {"csrf_token": "tok"}.get(k, d)
|
|
mock_request.form = AsyncMock(return_value=form_data)
|
|
|
|
before_ci = _make_consumer_info("meshai-wx", num_pending=500)
|
|
mock_js = MagicMock()
|
|
mock_js.consumer_info = AsyncMock(return_value=before_ci)
|
|
mock_js.delete_consumer = AsyncMock()
|
|
|
|
mock_conn = AsyncMock()
|
|
mock_pool = MagicMock()
|
|
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
|
|
mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
captured_audit = {}
|
|
|
|
async def capture_audit(conn, action, operator_id=None, target=None, before=None, after=None):
|
|
captured_audit["action"] = action
|
|
captured_audit["operator_id"] = operator_id
|
|
captured_audit["target"] = target
|
|
captured_audit["before"] = before
|
|
captured_audit["after"] = after
|
|
|
|
with patch("central.gui.nats.get_js", return_value=mock_js):
|
|
with patch("central.gui.routes.get_pool", return_value=mock_pool):
|
|
with patch("central.gui.routes.write_audit", side_effect=capture_audit):
|
|
result = await consumers_delete(mock_request, "CENTRAL_WX", "meshai-wx")
|
|
|
|
assert result.status_code == 302
|
|
assert result.headers["location"] == "/consumers"
|
|
|
|
mock_js.delete_consumer.assert_awaited_once_with("CENTRAL_WX", "meshai-wx")
|
|
|
|
assert captured_audit["action"] == "consumer.delete"
|
|
assert captured_audit["operator_id"] == 1
|
|
assert captured_audit["target"] == "CENTRAL_WX/meshai-wx"
|
|
assert captured_audit["before"]["name"] == "meshai-wx"
|
|
assert captured_audit["after"] is None
|
|
|
|
|
|
class TestConsumersDeleteCsrfGuard:
|
|
"""POST /consumers/{stream}/{consumer}/delete CSRF mismatch raises."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_csrf_mismatch_raises(self):
|
|
from central.gui.routes import consumers_delete
|
|
from central.gui.auth import CsrfValidationError
|
|
|
|
mock_request = MagicMock()
|
|
mock_request.state.operator = MagicMock(id=1)
|
|
mock_request.state.csrf_token = "real_token"
|
|
form_data = MagicMock()
|
|
form_data.get.side_effect = lambda k, d="": {"csrf_token": "wrong_token"}.get(k, d)
|
|
mock_request.form = AsyncMock(return_value=form_data)
|
|
|
|
with pytest.raises(CsrfValidationError):
|
|
await consumers_delete(mock_request, "CENTRAL_WX", "meshai-wx")
|
|
|
|
|
|
class TestConsumersDeleteNatsUnavailable:
|
|
"""POST /consumers/{stream}/{consumer}/delete when NATS is down redirects."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nats_unavailable_redirects(self):
|
|
from central.gui.routes import consumers_delete
|
|
|
|
mock_request = MagicMock()
|
|
mock_request.state.operator = MagicMock(id=1)
|
|
mock_request.state.csrf_token = "tok"
|
|
form_data = MagicMock()
|
|
form_data.get.side_effect = lambda k, d="": {"csrf_token": "tok"}.get(k, d)
|
|
mock_request.form = AsyncMock(return_value=form_data)
|
|
|
|
with patch("central.gui.nats.get_js", return_value=None):
|
|
result = await consumers_delete(mock_request, "CENTRAL_WX", "meshai-wx")
|
|
|
|
assert result.status_code == 302
|
|
assert result.headers["location"] == "/consumers"
|