matt/central
1
0
Fork 0
mirror of https://github.com/zvx-echo6/central.git synced 2026-08-26 09:21:36 +00:00
Code Issues Projects Releases Packages Wiki Activity

gui: add consumers admin page (view + delete JetStream consumers) (#117)

Browse source 
* gui: add consumers admin page (view + delete JetStream consumers)

Adds GET /consumers (list all stream consumers grouped by stream) and
POST /consumers/{stream}/{consumer}/delete. Archive-* consumers are
protected in both the template (no delete button rendered) and the POST
handler (hard refuse before touching NATS). CSRF validated, audit
logged via CONSUMER_DELETE action, DB conn acquired same pattern as
api_keys_delete.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* gui: fix consumers_info coroutine usage + list-returning test mock + None-guard counts

- 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>

* gui: make consumers_delete guards DB-independent + add rendered-HTML tests

Builds on the consumers_info coroutine fix:

- consumers_delete: acquire the DB pool only when actually writing the
  audit (after the CSRF / archive-guard / NATS-unavailable early exits)
  and use a local `get_js` import. Previously `pool = get_pool()` ran at
  the top, so the CSRF-reject, archive-refuse and NATS-down paths all
  needed an initialized DB pool, and the module-level get_js bound at
  import time ignored test patches of central.gui.nats.get_js. Mirrors
  the local-import pattern the streams routes already use.
- tests: add TestConsumersListHtmlRender — renders consumers_list.html
  through the real Jinja2 environment and asserts the consumer NAME
  reaches the HTML body, the central-owned label gates on `protected`,
  and None counts render an em dash rather than the literal "None".
  Stronger than the context-dict checks; the coroutine/None regressions
  cannot return. All 4 delete-route tests now pass (were failing on an
  uninitialized-pool RuntimeError).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Ubuntu <zvx@cortex.echo6.co>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
malice 2026-06-28 19:41:39 -06:00 • committed by GitHub
commit a38eaaa4fd
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 568 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

1
src/central/gui/audit.py
View file

@ -14,6 +14,7 @@ STREAM_UPDATE = "stream.update"
API_KEY_CREATE = "api_key.create"
API_KEY_ROTATE = "api_key.rotate"
API_KEY_DELETE = "api_key.delete"
CONSUMER_DELETE = "consumer.delete"
SYSTEM_UPDATE = "system.update"
MONITORING_AREA_CREATE = "monitoring_area.create"
MONITORING_AREA_UPDATE = "monitoring_area.update"

117
src/central/gui/routes.py
View file

@ -44,6 +44,7 @@ from central.gui.audit import (
AUTH_LOGIN_FAILED,
AUTH_LOGOUT,
AUTH_PASSWORD_CHANGE,
CONSUMER_DELETE,
MONITORING_AREA_CREATE,
MONITORING_AREA_DELETE,
MONITORING_AREA_UPDATE,
@ -2194,6 +2195,122 @@ async def streams_update(
return RedirectResponse(url="/streams", status_code=302)
# =============================================================================
# Consumers routes
# =============================================================================
@router.get("/consumers", response_class=HTMLResponse)
async def consumers_list(request: Request) -> HTMLResponse:
"""List all JetStream consumers across all registered streams."""
from central.gui.nats import get_js
templates = _get_templates()
operator = request.state.operator
js = get_js()
streams_data = []
for stream_entry in STREAM_REGISTRY:
stream_name = stream_entry.name
consumers = []
stream_error = None
if js is not None:
try:
for ci in await js.consumers_info(stream_name):
consumers.append({
"name": ci.name,
"num_pending": ci.num_pending,
"num_ack_pending": ci.num_ack_pending,
"num_redelivered": ci.num_redelivered,
"num_waiting": ci.num_waiting,
"created": ci.created,
"protected": ci.name.startswith("archive-"),
})
except Exception as e:
logger.warning(
"consumers_info failed",
extra={"stream": stream_name, "err": type(e).__name__},
)
stream_error = f"unavailable: {type(e).__name__}"
else:
stream_error = "NATS unavailable"
streams_data.append({
"stream": stream_name,
"consumers": consumers,
"error": stream_error,
})
csrf_token = request.state.csrf_token
response = templates.TemplateResponse(
request=request,
name="consumers_list.html",
context={
"operator": operator,
"csrf_token": csrf_token,
"streams": streams_data,
},
)
return response
@router.post("/consumers/{stream}/{consumer}/delete", response_class=HTMLResponse)
async def consumers_delete(request: Request, stream: str, consumer: str) -> Response:
"""Delete a JetStream consumer."""
from central.gui.nats import get_js
operator = request.state.operator
form = await request.form()
form_csrf = form.get("csrf_token", "")
if not form_csrf or form_csrf != request.state.csrf_token:
raise CsrfValidationError("Invalid CSRF token")
# Hard guard: never delete archive-* consumers even if path is forged
if consumer.startswith("archive-"):
return RedirectResponse("/consumers", status_code=302)
js = get_js()
if js is None:
return RedirectResponse("/consumers", status_code=302)
# Capture before state from NATS for audit log
try:
before_info = await js.consumer_info(stream, consumer)
before = {
"name": before_info.name,
"stream": stream,
"num_pending": before_info.num_pending,
}
except Exception:
before = {"name": consumer, "stream": stream}
# Delete the consumer
try:
await js.delete_consumer(stream, consumer)
except Exception:
logger.exception(
"delete_consumer failed",
extra={"stream": stream, "consumer": consumer},
)
return RedirectResponse("/consumers", status_code=302)
# Write audit log
pool = get_pool()
async with pool.acquire() as conn:
await write_audit(
conn,
CONSUMER_DELETE,
operator_id=operator.id,
target=f"{stream}/{consumer}",
before=before,
after=None,
)
return RedirectResponse("/consumers", status_code=302)
# =============================================================================
# Enrichment config route
# =============================================================================

1
src/central/gui/templates/base.html
View file

@ -18,6 +18,7 @@
<a href="/events">Events</a>
<a href="/telemetry">Telemetry</a>
<a href="/streams">Streams</a>
<a href="/consumers">Consumers</a>
<a href="/enrichment">Enrichment</a>
<a href="/monitoring-area">Monitoring Area</a>
<a href="/api-keys">API Keys</a>

70
src/central/gui/templates/consumers_list.html Normal file
View file

@ -0,0 +1,70 @@
{% extends "base.html" %}
{% block title %}Central — Consumers{% endblock %}
{% block content %}
<h1>Consumers</h1>
<p class="muted">JetStream consumers across all registered streams. A consumer with high
<em>Pending</em> and zero <em>Waiting</em> has accumulated unacknowledged messages and
has no active subscriber — it is safe to delete if it is not a central-owned consumer.</p>
<div class="cols">
{% for stream in streams %}
<article>
<header><strong>{{ stream.stream }}</strong></header>
{% if stream.error %}
<p class="error" style="margin-top: 0.5rem;">({{ stream.error }})</p>
{% elif stream.consumers %}
<table style="width: 100%; border-collapse: collapse; margin-top: 0.5rem;">
<thead>
<tr>
<th style="text-align: left; padding: 0.25rem 0.5rem;">Name</th>
<th style="text-align: right; padding: 0.25rem 0.5rem;">Pending</th>
<th style="text-align: right; padding: 0.25rem 0.5rem;">Ack Pending</th>
<th style="text-align: right; padding: 0.25rem 0.5rem;">Redelivered</th>
<th style="text-align: right; padding: 0.25rem 0.5rem;">Waiting</th>
<th style="text-align: left; padding: 0.25rem 0.5rem;">Created</th>
<th style="text-align: center; padding: 0.25rem 0.5rem;">Action</th>
</tr>
</thead>
<tbody>
{% for c in stream.consumers %}
<tr>
<td style="padding: 0.25rem 0.5rem;">{{ c.name }}</td>
<td style="text-align: right; padding: 0.25rem 0.5rem;">{{ c.num_pending if c.num_pending is not none else '—' }}</td>
<td style="text-align: right; padding: 0.25rem 0.5rem;">{{ c.num_ack_pending if c.num_ack_pending is not none else '—' }}</td>
<td style="text-align: right; padding: 0.25rem 0.5rem;">{{ c.num_redelivered if c.num_redelivered is not none else '—' }}</td>
<td style="text-align: right; padding: 0.25rem 0.5rem;">{{ c.num_waiting if c.num_waiting is not none else '—' }}</td>
<td style="padding: 0.25rem 0.5rem;">{{ c.created.isoformat() if c.created else '—' }}</td>
<td style="text-align: center; padding: 0.25rem 0.5rem;">
{% if c.protected %}
<span class="muted" style="font-size: 0.85em;">central-owned</span>
{% else %}
<form method="post" action="/consumers/{{ stream.stream }}/{{ c.name }}/delete" style="margin: 0;">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<button type="submit" class="btn-danger" style="height: 28px; padding: 0 10px; font-size: 0.85em;"
onclick="return confirm('Delete consumer {{ c.name }} on {{ stream.stream }}? This cannot be undone.')">Delete</button>
</form>
{% endif %}
</td>
</tr>
{% endfor %}
</tbody>
</table>
{% else %}
<p class="muted" style="margin-top: 0.5rem;">(no consumers)</p>
{% endif %}
</article>
{% endfor %}
</div>
<p class="muted" style="margin-top: 1rem; font-size: 0.9em;">
<strong>Legend:</strong> <em>Pending</em> = messages not yet delivered to this consumer;
<em>Ack Pending</em> = delivered but not yet acknowledged;
<em>Waiting</em> = active pull requests from a live subscriber.
A consumer with high <em>Pending</em> and zero <em>Waiting</em> is abandoned — no subscriber
is pulling from it and messages are piling up.
Consumers marked <em>central-owned</em> (archive-*) are managed by central and cannot be deleted here.
</p>
{% endblock %}

379
tests/test_consumers.py Normal file
View file

@ -0,0 +1,379 @@
"""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 TestConsumersListHtmlRender:
"""Render consumers_list.html through the real Jinja2 environment.
Stronger than the context-dict checks above: these prove the values
actually reach the rendered HTML body. Guards two regressions:
- the consumer NAME must appear in the rendered HTML (proves the
``await js.consumers_info(...)`` list reaches the template, not the
coroutine/async-iterator bug)
- Optional[int] count fields that are None must not render the literal
string ``None`` (they are guarded to an em dash).
"""
PROTECTED_LABEL = '<span class="muted" style="font-size: 0.85em;">central-owned</span>'
def _render(self, streams):
from central.gui import templates as templates_mod
template = templates_mod.env.get_template("consumers_list.html")
return template.render(
operator=MagicMock(username="testop"),
csrf_token="test_csrf",
streams=streams,
)
def test_consumer_name_appears_in_html(self):
streams = [
{
"stream": "CENTRAL_WX",
"error": None,
"consumers": [
{
"name": "meshai-wx",
"num_pending": 1000,
"num_ack_pending": 0,
"num_redelivered": 0,
"num_waiting": 0,
"created": datetime(2026, 5, 17, 12, 0, 0, tzinfo=timezone.utc),
"protected": False,
},
],
},
]
html = self._render(streams)
assert "meshai-wx" in html
# Non-protected consumer renders a delete form
assert "/consumers/CENTRAL_WX/meshai-wx/delete" in html
# ...and not the central-owned label span (which only the legend prose
# mentions, so we match the exact span markup, not the bare phrase)
assert self.PROTECTED_LABEL not in html
def test_protected_consumer_renders_label_not_button(self):
streams = [
{
"stream": "CENTRAL_WX",
"error": None,
"consumers": [
{
"name": "archive-CENTRAL_WX",
"num_pending": 5,
"num_ack_pending": 0,
"num_redelivered": 0,
"num_waiting": 1,
"created": datetime(2026, 5, 17, 12, 0, 0, tzinfo=timezone.utc),
"protected": True,
},
],
},
]
html = self._render(streams)
assert "archive-CENTRAL_WX" in html
assert self.PROTECTED_LABEL in html
# No delete form for the protected consumer
assert "/consumers/CENTRAL_WX/archive-CENTRAL_WX/delete" not in html
def test_none_counts_render_dash_not_literal_none(self):
streams = [
{
"stream": "CENTRAL_WX",
"error": None,
"consumers": [
{
"name": "meshai-wx",
"num_pending": None,
"num_ack_pending": None,
"num_redelivered": None,
"num_waiting": None,
"created": None,
"protected": False,
},
],
},
]
html = self._render(streams)
assert "meshai-wx" in html
# The literal "None" must never leak into a rendered table cell
assert ">None<" not in html
# The guarded fallback em dash is rendered instead
assert "—" in html
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"