mirror of
https://github.com/zvx-echo6/central.git
synced 2026-05-21 18:14:44 +02:00
1b404d02b5
Replace trivial status_code==200 assertions with checks that verify the filter values were actually parsed and passed to the template. These tests now fail if the handler ignores the filter parameters.
633 lines
25 KiB
Python
633 lines
25 KiB
Python
"""Tests for events feed frontend routes."""
|
|
|
|
import json
|
|
from datetime import datetime, timedelta, timezone
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from central.gui.routes import events_list, events_rows, events_json
|
|
|
|
|
|
class TestEventsFeedFrontendAuthenticated:
|
|
"""Test events feed frontend with authentication."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_events_no_filters_returns_html(self):
|
|
"""GET /events authenticated, no filters returns HTML with events."""
|
|
mock_request = MagicMock()
|
|
mock_request.state.operator = MagicMock(id=1, username="admin")
|
|
mock_request.state.csrf_token = "test_csrf_token"
|
|
mock_request.query_params = {}
|
|
|
|
mock_events = [
|
|
{
|
|
"id": f"event_{i}",
|
|
"time": datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc) - timedelta(hours=i),
|
|
"received": datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc) - timedelta(hours=i),
|
|
"adapter": "nws",
|
|
"category": "Weather Alert",
|
|
"subject": f"Test Alert {i}",
|
|
"geometry": '{"type": "Point", "coordinates": [-122.4, 37.8]}' if i % 2 == 0 else None,
|
|
"data": {},
|
|
"regions": [],
|
|
}
|
|
for i in range(5)
|
|
]
|
|
|
|
mock_conn = AsyncMock()
|
|
mock_conn.fetch.return_value = mock_events
|
|
mock_conn.fetchrow.return_value = {
|
|
"map_tile_url": "https://tile.openstreetmap.org/{z}/{x}/{y}.png",
|
|
"map_attribution": "OpenStreetMap",
|
|
}
|
|
|
|
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_templates.TemplateResponse.return_value = MagicMock(status_code=200)
|
|
|
|
with patch("central.gui.routes._get_templates", return_value=mock_templates):
|
|
with patch("central.gui.routes.get_pool", return_value=mock_pool):
|
|
result = await events_list(mock_request)
|
|
|
|
assert result.status_code == 200
|
|
call_args = mock_templates.TemplateResponse.call_args
|
|
context = call_args.kwargs.get("context", call_args[1].get("context"))
|
|
assert "events" in context
|
|
assert context["filter_error"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_events_adapter_filter(self):
|
|
"""GET /events?adapter=nws returns only nws events."""
|
|
mock_request = MagicMock()
|
|
mock_request.state.operator = MagicMock(id=1, username="admin")
|
|
mock_request.state.csrf_token = "test_csrf_token"
|
|
mock_request.query_params = {"adapter": "nws"}
|
|
|
|
mock_events = [
|
|
{
|
|
"id": "nws_event_1",
|
|
"time": datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc),
|
|
"received": datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc),
|
|
"adapter": "nws",
|
|
"category": "Alert",
|
|
"subject": "NWS Alert",
|
|
"geometry": None,
|
|
"data": {},
|
|
"regions": [],
|
|
},
|
|
]
|
|
|
|
mock_conn = AsyncMock()
|
|
mock_conn.fetch.return_value = mock_events
|
|
mock_conn.fetchrow.return_value = {
|
|
"map_tile_url": "https://tile.openstreetmap.org/{z}/{x}/{y}.png",
|
|
"map_attribution": "OpenStreetMap",
|
|
}
|
|
|
|
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_templates.TemplateResponse.return_value = MagicMock(status_code=200)
|
|
|
|
with patch("central.gui.routes._get_templates", return_value=mock_templates):
|
|
with patch("central.gui.routes.get_pool", return_value=mock_pool):
|
|
result = await events_list(mock_request)
|
|
|
|
assert result.status_code == 200
|
|
context = mock_templates.TemplateResponse.call_args.kwargs.get("context")
|
|
assert context["filter_values"]["adapter"] == "nws"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_events_since_until_filter(self):
|
|
"""GET /events?since=...&until=... filters by time window."""
|
|
mock_request = MagicMock()
|
|
mock_request.state.operator = MagicMock(id=1, username="admin")
|
|
mock_request.state.csrf_token = "test_csrf_token"
|
|
mock_request.query_params = {
|
|
"since": "2026-05-17T00:00:00",
|
|
"until": "2026-05-17T12:00:00",
|
|
}
|
|
|
|
mock_events = [
|
|
{
|
|
"id": "in_range",
|
|
"time": datetime(2026, 5, 17, 6, 0, tzinfo=timezone.utc),
|
|
"received": datetime(2026, 5, 17, 6, 0, tzinfo=timezone.utc),
|
|
"adapter": "nws",
|
|
"category": "Alert",
|
|
"subject": "In Range",
|
|
"geometry": None,
|
|
"data": {},
|
|
"regions": [],
|
|
},
|
|
]
|
|
|
|
mock_conn = AsyncMock()
|
|
mock_conn.fetch.return_value = mock_events
|
|
mock_conn.fetchrow.return_value = {
|
|
"map_tile_url": "https://tile.openstreetmap.org/{z}/{x}/{y}.png",
|
|
"map_attribution": "OpenStreetMap",
|
|
}
|
|
|
|
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_templates.TemplateResponse.return_value = MagicMock(status_code=200)
|
|
|
|
with patch("central.gui.routes._get_templates", return_value=mock_templates):
|
|
with patch("central.gui.routes.get_pool", return_value=mock_pool):
|
|
result = await events_list(mock_request)
|
|
|
|
assert result.status_code == 200
|
|
# Verify filter was actually parsed and passed to template
|
|
mock_templates.TemplateResponse.assert_called_once()
|
|
call_kwargs = mock_templates.TemplateResponse.call_args.kwargs
|
|
context = call_kwargs.get("context", call_kwargs)
|
|
assert context["filter_values"]["since"] == "2026-05-17T00:00:00"
|
|
assert context["filter_values"]["until"] == "2026-05-17T12:00:00"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_events_region_filter(self):
|
|
"""GET /events with full region bbox filters by location."""
|
|
mock_request = MagicMock()
|
|
mock_request.state.operator = MagicMock(id=1, username="admin")
|
|
mock_request.state.csrf_token = "test_csrf_token"
|
|
mock_request.query_params = {
|
|
"region_north": "49.5",
|
|
"region_south": "31",
|
|
"region_east": "-102",
|
|
"region_west": "-124.5",
|
|
}
|
|
|
|
mock_events = [
|
|
{
|
|
"id": "in_bbox",
|
|
"time": datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc),
|
|
"received": datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc),
|
|
"adapter": "nws",
|
|
"category": "Alert",
|
|
"subject": "In BBox",
|
|
"geometry": '{"type": "Point", "coordinates": [-120, 40]}',
|
|
"data": {},
|
|
"regions": [],
|
|
},
|
|
]
|
|
|
|
mock_conn = AsyncMock()
|
|
mock_conn.fetch.return_value = mock_events
|
|
mock_conn.fetchrow.return_value = {
|
|
"map_tile_url": "https://tile.openstreetmap.org/{z}/{x}/{y}.png",
|
|
"map_attribution": "OpenStreetMap",
|
|
}
|
|
|
|
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_templates.TemplateResponse.return_value = MagicMock(status_code=200)
|
|
|
|
with patch("central.gui.routes._get_templates", return_value=mock_templates):
|
|
with patch("central.gui.routes.get_pool", return_value=mock_pool):
|
|
result = await events_list(mock_request)
|
|
|
|
assert result.status_code == 200
|
|
# Verify region filter was actually parsed and passed to template
|
|
mock_templates.TemplateResponse.assert_called_once()
|
|
call_kwargs = mock_templates.TemplateResponse.call_args.kwargs
|
|
context = call_kwargs.get("context", call_kwargs)
|
|
assert context["filter_values"]["region_north"] == "49.5"
|
|
assert context["filter_values"]["region_south"] == "31"
|
|
assert context["filter_values"]["region_east"] == "-102"
|
|
assert context["filter_values"]["region_west"] == "-124.5"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_events_partial_region_shows_error_banner(self):
|
|
"""GET /events with partial region shows error banner, not 400."""
|
|
mock_request = MagicMock()
|
|
mock_request.state.operator = MagicMock(id=1, username="admin")
|
|
mock_request.state.csrf_token = "test_csrf_token"
|
|
mock_request.query_params = {"region_north": "49"}
|
|
|
|
mock_conn = AsyncMock()
|
|
mock_conn.fetchrow.return_value = {
|
|
"map_tile_url": "https://tile.openstreetmap.org/{z}/{x}/{y}.png",
|
|
"map_attribution": "OpenStreetMap",
|
|
}
|
|
|
|
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_templates.TemplateResponse.return_value = MagicMock(status_code=200)
|
|
|
|
with patch("central.gui.routes._get_templates", return_value=mock_templates):
|
|
with patch("central.gui.routes.get_pool", return_value=mock_pool):
|
|
result = await events_list(mock_request)
|
|
|
|
# Should be 200, not 400
|
|
assert result.status_code == 200
|
|
context = mock_templates.TemplateResponse.call_args.kwargs.get("context")
|
|
assert context["filter_error"] is not None
|
|
assert "region" in context["filter_error"].lower()
|
|
# Events should be empty due to validation error
|
|
assert context["events"] == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_events_with_limit_shows_next_button(self):
|
|
"""GET /events?limit=5 shows Next button when more events exist."""
|
|
mock_request = MagicMock()
|
|
mock_request.state.operator = MagicMock(id=1, username="admin")
|
|
mock_request.state.csrf_token = "test_csrf_token"
|
|
mock_request.query_params = {"limit": "5"}
|
|
|
|
# Return 6 events (limit+1) to trigger pagination
|
|
mock_events = [
|
|
{
|
|
"id": f"event_{i}",
|
|
"time": datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc) - timedelta(hours=i),
|
|
"received": datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc) - timedelta(hours=i),
|
|
"adapter": "nws",
|
|
"category": "Alert",
|
|
"subject": f"Event {i}",
|
|
"geometry": None,
|
|
"data": {},
|
|
"regions": [],
|
|
}
|
|
for i in range(6)
|
|
]
|
|
|
|
mock_conn = AsyncMock()
|
|
mock_conn.fetch.return_value = mock_events
|
|
mock_conn.fetchrow.return_value = {
|
|
"map_tile_url": "https://tile.openstreetmap.org/{z}/{x}/{y}.png",
|
|
"map_attribution": "OpenStreetMap",
|
|
}
|
|
|
|
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_templates.TemplateResponse.return_value = MagicMock(status_code=200)
|
|
|
|
with patch("central.gui.routes._get_templates", return_value=mock_templates):
|
|
with patch("central.gui.routes.get_pool", return_value=mock_pool):
|
|
result = await events_list(mock_request)
|
|
|
|
assert result.status_code == 200
|
|
context = mock_templates.TemplateResponse.call_args.kwargs.get("context")
|
|
assert context["next_cursor"] is not None
|
|
assert len(context["events"]) == 5 # Should be trimmed to limit
|
|
|
|
|
|
class TestEventsRowsFragment:
|
|
"""Test /events/rows HTMX fragment."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_events_rows_returns_fragment(self):
|
|
"""GET /events/rows returns table fragment, not full page."""
|
|
mock_request = MagicMock()
|
|
mock_request.state.operator = MagicMock(id=1, username="admin")
|
|
mock_request.query_params = {"limit": "5"}
|
|
|
|
mock_events = [
|
|
{
|
|
"id": "event_1",
|
|
"time": datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc),
|
|
"received": datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc),
|
|
"adapter": "nws",
|
|
"category": "Alert",
|
|
"subject": "Event 1",
|
|
"geometry": None,
|
|
"data": {},
|
|
"regions": [],
|
|
},
|
|
]
|
|
|
|
mock_conn = AsyncMock()
|
|
mock_conn.fetch.return_value = mock_events
|
|
|
|
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_templates.TemplateResponse.return_value = MagicMock(status_code=200)
|
|
|
|
with patch("central.gui.routes._get_templates", return_value=mock_templates):
|
|
with patch("central.gui.routes.get_pool", return_value=mock_pool):
|
|
result = await events_rows(mock_request)
|
|
|
|
assert result.status_code == 200
|
|
# Verify it uses the fragment template
|
|
call_args = mock_templates.TemplateResponse.call_args
|
|
assert call_args.kwargs.get("name") == "_events_rows.html"
|
|
|
|
|
|
class TestGeometrySummary:
|
|
"""Test geometry summary function."""
|
|
|
|
def test_geometry_summary_polygon(self):
|
|
"""Polygon geometry shows point count."""
|
|
from central.gui.routes import _geometry_summary
|
|
|
|
geom = {
|
|
"type": "Polygon",
|
|
"coordinates": [[[-122, 37], [-122, 38], [-121, 38], [-121, 37], [-122, 37]]]
|
|
}
|
|
summary = _geometry_summary(geom)
|
|
assert "Polygon" in summary
|
|
assert "5 pts" in summary
|
|
|
|
def test_geometry_summary_point(self):
|
|
"""Point geometry shows 'Point'."""
|
|
from central.gui.routes import _geometry_summary
|
|
|
|
geom = {"type": "Point", "coordinates": [-122.4, 37.8]}
|
|
summary = _geometry_summary(geom)
|
|
assert summary == "Point"
|
|
|
|
def test_geometry_summary_linestring(self):
|
|
"""LineString geometry shows point count."""
|
|
from central.gui.routes import _geometry_summary
|
|
|
|
geom = {
|
|
"type": "LineString",
|
|
"coordinates": [[-122, 37], [-121, 38], [-120, 39]]
|
|
}
|
|
summary = _geometry_summary(geom)
|
|
assert "Line" in summary
|
|
assert "3 pts" in summary
|
|
|
|
def test_geometry_summary_none(self):
|
|
"""None geometry shows 'None'."""
|
|
from central.gui.routes import _geometry_summary
|
|
|
|
summary = _geometry_summary(None)
|
|
assert summary == "None"
|
|
|
|
|
|
class TestDataGeometryAttribute:
|
|
"""Test that rows have valid geometry data attributes."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_event_with_geometry_has_valid_json(self):
|
|
"""Events with geometry have parseable JSON in data-geometry."""
|
|
mock_request = MagicMock()
|
|
mock_request.state.operator = MagicMock(id=1, username="admin")
|
|
mock_request.query_params = {}
|
|
|
|
mock_events = [
|
|
{
|
|
"id": "geom_event",
|
|
"time": datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc),
|
|
"received": datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc),
|
|
"adapter": "nws",
|
|
"category": "Alert",
|
|
"subject": "With Geometry",
|
|
"geometry": '{"type": "Polygon", "coordinates": [[[-122, 37], [-122, 38], [-121, 38], [-121, 37], [-122, 37]]]}',
|
|
"data": {},
|
|
"regions": [],
|
|
},
|
|
]
|
|
|
|
mock_conn = AsyncMock()
|
|
mock_conn.fetch.return_value = mock_events
|
|
|
|
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_templates.TemplateResponse.return_value = MagicMock(status_code=200)
|
|
|
|
with patch("central.gui.routes._get_templates", return_value=mock_templates):
|
|
with patch("central.gui.routes.get_pool", return_value=mock_pool):
|
|
result = await events_rows(mock_request)
|
|
|
|
context = mock_templates.TemplateResponse.call_args.kwargs.get("context")
|
|
event = context["events"][0]
|
|
# Geometry should be parsed dict, not string
|
|
assert isinstance(event["geometry"], dict)
|
|
assert event["geometry"]["type"] == "Polygon"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_event_without_geometry_has_none(self):
|
|
"""Events without geometry have None for geometry field."""
|
|
mock_request = MagicMock()
|
|
mock_request.state.operator = MagicMock(id=1, username="admin")
|
|
mock_request.query_params = {}
|
|
|
|
mock_events = [
|
|
{
|
|
"id": "no_geom_event",
|
|
"time": datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc),
|
|
"received": datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc),
|
|
"adapter": "nws",
|
|
"category": "Alert",
|
|
"subject": "No Geometry",
|
|
"geometry": None,
|
|
"data": {},
|
|
"regions": [],
|
|
},
|
|
]
|
|
|
|
mock_conn = AsyncMock()
|
|
mock_conn.fetch.return_value = mock_events
|
|
|
|
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_templates.TemplateResponse.return_value = MagicMock(status_code=200)
|
|
|
|
with patch("central.gui.routes._get_templates", return_value=mock_templates):
|
|
with patch("central.gui.routes.get_pool", return_value=mock_pool):
|
|
result = await events_rows(mock_request)
|
|
|
|
context = mock_templates.TemplateResponse.call_args.kwargs.get("context")
|
|
event = context["events"][0]
|
|
assert event["geometry"] is None
|
|
|
|
|
|
class TestCrossEndpointParity:
|
|
"""Test that /events.json and /events return the same filtered results."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_category_filter_both_endpoints(self):
|
|
"""Category filter works on both /events.json and /events."""
|
|
mock_events = [
|
|
{
|
|
"id": "weather_event",
|
|
"time": datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc),
|
|
"received": datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc),
|
|
"adapter": "nws",
|
|
"category": "Weather Alert",
|
|
"subject": "Weather Event",
|
|
"geometry": None,
|
|
"data": {},
|
|
"regions": [],
|
|
},
|
|
]
|
|
|
|
mock_conn = AsyncMock()
|
|
mock_conn.fetch.return_value = mock_events
|
|
mock_conn.fetchrow.return_value = {
|
|
"map_tile_url": "https://tile.openstreetmap.org/{z}/{x}/{y}.png",
|
|
"map_attribution": "OpenStreetMap",
|
|
}
|
|
|
|
mock_pool = MagicMock()
|
|
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
|
|
mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
query_params = {"category": "Weather Alert"}
|
|
|
|
# Test /events.json
|
|
json_request = MagicMock()
|
|
json_request.state.operator = MagicMock(id=1, username="admin")
|
|
json_request.query_params = query_params
|
|
|
|
with patch("central.gui.routes.get_pool", return_value=mock_pool):
|
|
json_response = await events_json(json_request)
|
|
|
|
json_data = json.loads(json_response.body)
|
|
assert len(json_data["events"]) == 1
|
|
assert json_data["events"][0]["category"] == "Weather Alert"
|
|
|
|
# Test /events
|
|
html_request = MagicMock()
|
|
html_request.state.operator = MagicMock(id=1, username="admin")
|
|
html_request.state.csrf_token = "test_csrf"
|
|
html_request.query_params = query_params
|
|
|
|
mock_templates = MagicMock()
|
|
mock_templates.TemplateResponse.return_value = MagicMock(status_code=200)
|
|
|
|
mock_conn.fetch.return_value = mock_events
|
|
|
|
with patch("central.gui.routes._get_templates", return_value=mock_templates):
|
|
with patch("central.gui.routes.get_pool", return_value=mock_pool):
|
|
await events_list(html_request)
|
|
|
|
html_context = mock_templates.TemplateResponse.call_args.kwargs.get("context")
|
|
assert len(html_context["events"]) == 1
|
|
assert html_context["events"][0]["category"] == "Weather Alert"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cursor_pagination_both_endpoints(self):
|
|
"""Cursor pagination works identically on both endpoints."""
|
|
first_page = [
|
|
{
|
|
"id": f"event_{i}",
|
|
"time": datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc) - timedelta(hours=i),
|
|
"received": datetime(2026, 5, 17, 12, 0, tzinfo=timezone.utc) - timedelta(hours=i),
|
|
"adapter": "nws",
|
|
"category": "Alert",
|
|
"subject": f"Event {i}",
|
|
"geometry": None,
|
|
"data": {},
|
|
"regions": [],
|
|
}
|
|
for i in range(3)
|
|
]
|
|
|
|
mock_conn = AsyncMock()
|
|
mock_conn.fetch.return_value = first_page
|
|
mock_conn.fetchrow.return_value = {
|
|
"map_tile_url": "https://tile.openstreetmap.org/{z}/{x}/{y}.png",
|
|
"map_attribution": "OpenStreetMap",
|
|
}
|
|
|
|
mock_pool = MagicMock()
|
|
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
|
|
mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
json_request = MagicMock()
|
|
json_request.state.operator = MagicMock(id=1, username="admin")
|
|
json_request.query_params = {"limit": "2"}
|
|
|
|
with patch("central.gui.routes.get_pool", return_value=mock_pool):
|
|
json_response = await events_json(json_request)
|
|
|
|
json_data = json.loads(json_response.body)
|
|
json_cursor = json_data["next_cursor"]
|
|
assert json_cursor is not None
|
|
|
|
html_request = MagicMock()
|
|
html_request.state.operator = MagicMock(id=1, username="admin")
|
|
html_request.state.csrf_token = "test_csrf"
|
|
html_request.query_params = {"limit": "2"}
|
|
|
|
mock_templates = MagicMock()
|
|
mock_templates.TemplateResponse.return_value = MagicMock(status_code=200)
|
|
|
|
mock_conn.fetch.return_value = first_page
|
|
|
|
with patch("central.gui.routes._get_templates", return_value=mock_templates):
|
|
with patch("central.gui.routes.get_pool", return_value=mock_pool):
|
|
await events_list(html_request)
|
|
|
|
html_context = mock_templates.TemplateResponse.call_args.kwargs.get("context")
|
|
html_cursor = html_context["next_cursor"]
|
|
|
|
assert json_cursor == html_cursor
|
|
|
|
|
|
class TestErrorSemantics:
|
|
"""Test error handling differences between JSON and HTML endpoints."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_json_endpoint_returns_400_on_invalid_limit(self):
|
|
"""/events.json?limit=0 returns 400 JSON error."""
|
|
mock_request = MagicMock()
|
|
mock_request.state.operator = MagicMock(id=1, username="admin")
|
|
mock_request.query_params = {"limit": "0"}
|
|
|
|
response = await events_json(mock_request)
|
|
|
|
assert response.status_code == 400
|
|
data = json.loads(response.body)
|
|
assert "error" in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_html_endpoint_returns_200_with_error_banner(self):
|
|
"""/events?limit=0 returns 200 HTML with error banner."""
|
|
mock_request = MagicMock()
|
|
mock_request.state.operator = MagicMock(id=1, username="admin")
|
|
mock_request.state.csrf_token = "test_csrf"
|
|
mock_request.query_params = {"limit": "0"}
|
|
|
|
mock_conn = AsyncMock()
|
|
mock_conn.fetchrow.return_value = {
|
|
"map_tile_url": "https://tile.openstreetmap.org/{z}/{x}/{y}.png",
|
|
"map_attribution": "OpenStreetMap",
|
|
}
|
|
|
|
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_templates.TemplateResponse.return_value = MagicMock(status_code=200)
|
|
|
|
with patch("central.gui.routes._get_templates", return_value=mock_templates):
|
|
with patch("central.gui.routes.get_pool", return_value=mock_pool):
|
|
result = await events_list(mock_request)
|
|
|
|
assert result.status_code == 200
|
|
context = mock_templates.TemplateResponse.call_args.kwargs.get("context")
|
|
assert context["filter_error"] is not None
|
|
assert "limit" in context["filter_error"].lower()
|
|
assert context["events"] == []
|