fix(gui): use get_msg().time for stream timestamps, fix badge layout

- nats-py StreamState doesn't expose first_ts/last_ts
- Fetch timestamps via js.get_msg(stream, seq=N).time instead
- Handle edge cases: empty streams, single-message streams, get_msg failures
- Fix badge overlap using flex layout instead of float:right
- Change label from "Max bytes (config)" to "Max bytes (current)"

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Matt Johnson 2026-05-17 23:59:31 +00:00
commit 16e31efd73
3 changed files with 287 additions and 23 deletions

View file

@ -1,9 +1,12 @@
"""Route handlers for Central GUI."""
import json
import logging
import re
from typing import Any
logger = logging.getLogger("central.gui.routes")
from fastapi import APIRouter, Depends, Form, Request
from fastapi.responses import HTMLResponse, RedirectResponse, Response
from fastapi_csrf_protect import CsrfProtect
@ -911,8 +914,12 @@ async def streams_list(
"managed_max_bytes": row["managed_max_bytes"],
"live_bytes": None,
"live_messages": None,
"live_first_seq": None,
"live_last_seq": None,
"live_first_ts": None,
"live_last_ts": None,
"first_ts_error": None,
"last_ts_error": None,
"error": None,
}
@ -922,10 +929,41 @@ async def streams_list(
info = await js.stream_info(row["name"])
stream_data["live_bytes"] = info.state.bytes
stream_data["live_messages"] = info.state.messages
stream_data["live_first_ts"] = info.state.first_ts
stream_data["live_last_ts"] = info.state.last_ts
except Exception:
stream_data["error"] = "unavailable"
stream_data["live_first_seq"] = info.state.first_seq
stream_data["live_last_seq"] = info.state.last_seq
# Fetch first / last message timestamps via get_msg
# RawStreamMsg has .time attribute (not .metadata.timestamp)
if info.state.first_seq > 0:
try:
first_msg = await js.get_msg(row["name"], seq=info.state.first_seq)
stream_data["live_first_ts"] = first_msg.time
except Exception as e:
logger.warning(
"get_msg first failed",
extra={"stream": row["name"], "err": type(e).__name__},
)
stream_data["live_first_ts"] = None
stream_data["first_ts_error"] = type(e).__name__
if info.state.last_seq > 0 and info.state.last_seq != info.state.first_seq:
try:
last_msg = await js.get_msg(row["name"], seq=info.state.last_seq)
stream_data["live_last_ts"] = last_msg.time
except Exception as e:
logger.warning(
"get_msg last failed",
extra={"stream": row["name"], "err": type(e).__name__},
)
stream_data["live_last_ts"] = None
stream_data["last_ts_error"] = type(e).__name__
elif info.state.last_seq == info.state.first_seq and info.state.first_seq > 0:
# Single message in stream
stream_data["live_last_ts"] = stream_data.get("live_first_ts")
except Exception as e:
logger.exception("Stream info failed", extra={"stream": row["name"]})
stream_data["error"] = f"unavailable: {type(e).__name__}"
else:
stream_data["error"] = "NATS unavailable"

View file

@ -8,10 +8,10 @@
<div class="grid">
{% for stream in streams %}
<article>
<header>
<header style="display: flex; justify-content: space-between; align-items: center;">
<strong>{{ stream.name }}</strong>
{% if stream.managed_max_bytes %}
<span style="float: right; font-size: 0.8em; background: var(--pico-secondary-background); padding: 0.2em 0.5em; border-radius: 4px;">Managed by supervisor</span>
<span style="font-size: 0.8em; background: var(--pico-secondary-background); padding: 0.2em 0.5em; border-radius: 4px;">Managed by supervisor</span>
{% endif %}
</header>
@ -21,10 +21,31 @@
<span style="color: var(--pico-color-red-500);">({{ stream.error }})</span>
{% else %}
<ul style="margin: 0.5rem 0;">
<li>Size: {{ stream.live_bytes|default(0)|filesizeformat if stream.live_bytes is not none else '(unavailable)' }}</li>
<li>Messages: {{ stream.live_messages if stream.live_messages is not none else '(unavailable)' }}</li>
<li>First message: {{ stream.live_first_ts.isoformat() if stream.live_first_ts else '(none)' }}</li>
<li>Last message: {{ stream.live_last_ts.isoformat() if stream.live_last_ts else '(none)' }}</li>
<li>Messages: {{ stream.live_messages }}</li>
<li>Size: {{ stream.live_bytes|filesizeformat }}</li>
{% if stream.live_messages == 0 %}
<li>First message: (empty)</li>
<li>Last message: (empty)</li>
{% else %}
<li>First message:
{% if stream.live_first_ts %}
{{ stream.live_first_ts.isoformat() }}
{% elif stream.first_ts_error %}
(lookup failed: {{ stream.first_ts_error }})
{% else %}
(unknown)
{% endif %}
</li>
<li>Last message:
{% if stream.live_last_ts %}
{{ stream.live_last_ts.isoformat() }}
{% elif stream.last_ts_error %}
(lookup failed: {{ stream.last_ts_error }})
{% else %}
(unknown)
{% endif %}
</li>
{% endif %}
</ul>
{% endif %}
</div>
@ -33,7 +54,7 @@
<strong>Configuration:</strong>
<ul style="margin: 0.5rem 0;">
<li>Max age: {{ (stream.max_age_s / 86400)|round(1) }} days ({{ stream.max_age_s }}s)</li>
<li>Max bytes (config): {{ stream.max_bytes_cfg|filesizeformat }}</li>
<li>Max bytes (current): {{ stream.max_bytes_cfg|filesizeformat }}</li>
</ul>
</div>

View file

@ -1,6 +1,7 @@
"""Tests for streams list and edit routes."""
import os
from datetime import datetime, timezone
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@ -28,8 +29,8 @@ class TestStreamsListAuthenticated:
"""Test streams list with authentication."""
@pytest.mark.asyncio
async def test_streams_list_returns_all_streams(self):
"""GET /streams authenticated returns 200 with all four streams."""
async def test_streams_list_returns_all_streams_with_timestamps(self):
"""GET /streams authenticated returns 200 with all four streams and timestamps."""
from central.gui.routes import streams_list
mock_request = MagicMock()
@ -55,15 +56,21 @@ class TestStreamsListAuthenticated:
mock_csrf.generate_csrf_tokens.return_value = ("token", "signed")
mock_csrf.set_csrf_cookie = MagicMock()
# Mock JetStream
# Mock JetStream with proper state fields
mock_js = AsyncMock()
mock_stream_info = MagicMock()
mock_stream_info.state.bytes = 1024000
mock_stream_info.state.messages = 100
mock_stream_info.state.first_ts = None
mock_stream_info.state.last_ts = None
mock_stream_info.state.first_seq = 1
mock_stream_info.state.last_seq = 100
mock_js.stream_info.return_value = mock_stream_info
# Mock get_msg for timestamps (RawStreamMsg has .time attribute)
test_ts = datetime(2026, 5, 17, 12, 0, 0, tzinfo=timezone.utc)
mock_msg = MagicMock()
mock_msg.time = test_ts
mock_js.get_msg.return_value = mock_msg
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
with patch("central.gui.nats.get_js", return_value=mock_js):
@ -78,6 +85,13 @@ class TestStreamsListAuthenticated:
assert "CENTRAL_QUAKE" in stream_names
assert "CENTRAL_WX" in stream_names
# Check that timestamps are populated
fire_stream = next(s for s in context["streams"] if s["name"] == "CENTRAL_FIRE")
assert fire_stream["live_bytes"] == 1024000
assert fire_stream["live_messages"] == 100
assert fire_stream["live_first_ts"] == test_ts
assert fire_stream["live_last_ts"] == test_ts
class TestStreamsListNatsUnavailable:
"""Test streams list when NATS is unavailable."""
@ -122,8 +136,8 @@ class TestStreamsListPartialFailure:
"""Test streams list with one stream's info raising."""
@pytest.mark.asyncio
async def test_one_stream_unavailable_others_render(self):
"""GET /streams with one stream error shows unavailable for that stream only."""
async def test_stream_info_raises_shows_exception_type(self):
"""GET /streams with stream_info error shows unavailable with exception type."""
from central.gui.routes import streams_list
mock_request = MagicMock()
@ -147,21 +161,26 @@ class TestStreamsListPartialFailure:
mock_csrf.generate_csrf_tokens.return_value = ("token", "signed")
mock_csrf.set_csrf_cookie = MagicMock()
# Mock JetStream - CENTRAL_FIRE raises, CENTRAL_WX works
# Mock JetStream - CENTRAL_FIRE raises ValueError, CENTRAL_WX works
mock_js = AsyncMock()
test_ts = datetime(2026, 5, 17, 12, 0, 0, tzinfo=timezone.utc)
def stream_info_side_effect(name):
if name == "CENTRAL_FIRE":
raise Exception("Stream not found")
raise ValueError("Stream not found")
mock_info = MagicMock()
mock_info.state.bytes = 2048
mock_info.state.messages = 50
mock_info.state.first_ts = None
mock_info.state.last_ts = None
mock_info.state.first_seq = 1
mock_info.state.last_seq = 50
return mock_info
mock_js.stream_info.side_effect = stream_info_side_effect
mock_msg = MagicMock()
mock_msg.time = test_ts
mock_js.get_msg.return_value = mock_msg
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
with patch("central.gui.nats.get_js", return_value=mock_js):
@ -173,11 +192,197 @@ class TestStreamsListPartialFailure:
fire_stream = next(s for s in context["streams"] if s["name"] == "CENTRAL_FIRE")
wx_stream = next(s for s in context["streams"] if s["name"] == "CENTRAL_WX")
assert fire_stream["error"] == "unavailable"
# Exception type should be in error message
assert "unavailable: ValueError" in fire_stream["error"]
assert wx_stream["error"] is None
assert wx_stream["live_bytes"] == 2048
class TestStreamsListEmptyStream:
"""Test streams list with empty stream (zero messages)."""
@pytest.mark.asyncio
async def test_empty_stream_no_get_msg_calls(self):
"""Empty stream (first_seq == 0) renders correctly, no get_msg calls made."""
from central.gui.routes import streams_list
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_conn = AsyncMock()
mock_conn.fetch.return_value = [
{"name": "CENTRAL_QUAKE", "max_age_s": 604800, "max_bytes": 1073741824, "managed_max_bytes": True, "updated_at": None},
]
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
mock_templates = MagicMock()
mock_response = MagicMock()
mock_templates.TemplateResponse.return_value = mock_response
mock_csrf = MagicMock()
mock_csrf.generate_csrf_tokens.return_value = ("token", "signed")
mock_csrf.set_csrf_cookie = MagicMock()
# Mock JetStream with empty stream (first_seq = 0)
mock_js = AsyncMock()
mock_stream_info = MagicMock()
mock_stream_info.state.bytes = 0
mock_stream_info.state.messages = 0
mock_stream_info.state.first_seq = 0
mock_stream_info.state.last_seq = 0
mock_js.stream_info.return_value = mock_stream_info
mock_js.get_msg = AsyncMock()
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
with patch("central.gui.nats.get_js", return_value=mock_js):
result = await streams_list(mock_request, mock_csrf)
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
quake_stream = context["streams"][0]
assert quake_stream["live_messages"] == 0
assert quake_stream["live_first_ts"] is None
assert quake_stream["live_last_ts"] is None
assert quake_stream["error"] is None
# get_msg should NOT have been called for empty stream
mock_js.get_msg.assert_not_called()
class TestStreamsListSingleMessage:
"""Test streams list with single-message stream."""
@pytest.mark.asyncio
async def test_single_message_stream_reuses_timestamp(self):
"""Single-message stream (first_seq == last_seq) calls get_msg once."""
from central.gui.routes import streams_list
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_conn = AsyncMock()
mock_conn.fetch.return_value = [
{"name": "CENTRAL_WX", "max_age_s": 604800, "max_bytes": 1073741824, "managed_max_bytes": True, "updated_at": None},
]
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
mock_templates = MagicMock()
mock_response = MagicMock()
mock_templates.TemplateResponse.return_value = mock_response
mock_csrf = MagicMock()
mock_csrf.generate_csrf_tokens.return_value = ("token", "signed")
mock_csrf.set_csrf_cookie = MagicMock()
# Mock JetStream with single message (first_seq == last_seq)
mock_js = AsyncMock()
mock_stream_info = MagicMock()
mock_stream_info.state.bytes = 512
mock_stream_info.state.messages = 1
mock_stream_info.state.first_seq = 1
mock_stream_info.state.last_seq = 1
mock_js.stream_info.return_value = mock_stream_info
test_ts = datetime(2026, 5, 17, 12, 0, 0, tzinfo=timezone.utc)
mock_msg = MagicMock()
mock_msg.time = test_ts
mock_js.get_msg.return_value = mock_msg
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
with patch("central.gui.nats.get_js", return_value=mock_js):
result = await streams_list(mock_request, mock_csrf)
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
wx_stream = context["streams"][0]
assert wx_stream["live_messages"] == 1
assert wx_stream["live_first_ts"] == test_ts
assert wx_stream["live_last_ts"] == test_ts
# get_msg should have been called once (for first message only)
assert mock_js.get_msg.call_count == 1
class TestStreamsListGetMsgFailure:
"""Test streams list when get_msg fails for a message."""
@pytest.mark.asyncio
async def test_get_msg_first_fails_shows_lookup_error(self):
"""get_msg raises for first message shows lookup failed, rest renders normally."""
from central.gui.routes import streams_list
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_conn = AsyncMock()
mock_conn.fetch.return_value = [
{"name": "CENTRAL_WX", "max_age_s": 604800, "max_bytes": 1073741824, "managed_max_bytes": True, "updated_at": None},
]
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
mock_templates = MagicMock()
mock_response = MagicMock()
mock_templates.TemplateResponse.return_value = mock_response
mock_csrf = MagicMock()
mock_csrf.generate_csrf_tokens.return_value = ("token", "signed")
mock_csrf.set_csrf_cookie = MagicMock()
# Mock JetStream
mock_js = AsyncMock()
mock_stream_info = MagicMock()
mock_stream_info.state.bytes = 2048
mock_stream_info.state.messages = 50
mock_stream_info.state.first_seq = 1
mock_stream_info.state.last_seq = 50
mock_js.stream_info.return_value = mock_stream_info
test_ts = datetime(2026, 5, 17, 12, 0, 0, tzinfo=timezone.utc)
# First call fails, second succeeds
def get_msg_side_effect(name, seq):
if seq == 1:
raise KeyError("Message not found")
mock_msg = MagicMock()
mock_msg.time = test_ts
return mock_msg
mock_js.get_msg.side_effect = get_msg_side_effect
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
with patch("central.gui.nats.get_js", return_value=mock_js):
result = await streams_list(mock_request, mock_csrf)
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
wx_stream = context["streams"][0]
# Main stream info still works
assert wx_stream["live_bytes"] == 2048
assert wx_stream["live_messages"] == 50
assert wx_stream["error"] is None
# First timestamp failed
assert wx_stream["live_first_ts"] is None
assert wx_stream["first_ts_error"] == "KeyError"
# Last timestamp succeeded
assert wx_stream["live_last_ts"] == test_ts
assert wx_stream["last_ts_error"] is None
class TestStreamsUpdate:
"""Test stream update POST handler."""