matt/central
1
0
Fork 0
mirror of https://github.com/zvx-echo6/central.git synced 2026-05-21 18:14:44 +02:00
Code Issues Projects Releases Packages Wiki Activity

feat(gui): implement first-run setup wizard (1b-8) (#24)

Browse source 
* feat(gui): implement first-run setup wizard (1b-8)

Add a 5-step setup wizard that replaces the single-step /setup:
1. Create Operator - create initial operator account
2. System Settings - configure map tile URL and attribution
3. API Keys - optionally add API keys for adapters
4. Configure Adapters - enable/disable adapters with region picker
5. Finish Setup - review and complete setup

Key changes:
- Update middleware to handle wizard URL structure and step routing
- Add wizard routes for each step with proper auth checks
- Create new templates using base_wizard.html for consistent styling
- Add audit events for system.update and setup.complete
- Update tests for new middleware behavior

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(gui): handle CSRF errors on wizard paths

Update csrf_exception_handler to re-render wizard forms with error
message instead of redirecting to /login when CSRF validation fails.

- /setup/operator: re-render with error
- /setup/system: re-render with current system values + error
- /setup/keys: re-render with current keys list + error
- /setup/adapters: re-render with current adapter config + error
- /setup/finish: re-render with summary data + error
- /setup: redirect to /setup (middleware routes to appropriate step)

Add error display to setup_keys.html and setup_finish.html templates.
Add 7 new CSRF handler tests for wizard paths.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(gui): region picker render + click-to-draw

Bug A: Maps render blank on /setup/adapters for FIRMS and USGS
because Leaflet computed zero dimensions before container layout
settled. Fix: add setTimeout invalidateSize() after map creation.

Bug B: No click-to-draw functionality - only drag corners. Fix:
add L.Control.Draw for rectangle drawing with CREATED event handler
to replace existing rectangle.

Both fixes applied to:
- setup_adapters.html (wizard inline JS)
- _region_picker.html (standalone edit page)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(gui): handle revisiting /setup/operator after operator created

When an operator already exists, /setup/operator now shows a
confirmation page instead of the create form. This prevents:
- Unique constraint violations on duplicate username
- Silent creation of duplicate operators

GET /setup/operator: queries config.operators; if any exist,
renders confirmation state with existing_operator context.

POST /setup/operator: checks operator count before INSERT; if
non-zero, renders confirmation state without inserting.

Template updated with conditional to show "Operator Already
Configured" message when existing_operator is set.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(csrf): replace fastapi-csrf-protect with session-bound CSRF

Fixes CSRF race condition where every GET rotated the CSRF token,
causing POST failures when users had multiple tabs or slow connections.

Changes:
- Remove fastapi-csrf-protect dependency
- Add session-bound CSRF tokens stored in config.sessions table
- Add pre-auth CSRF for unauthenticated routes (/login, /setup/operator)
- Add csrf.py module for pre-auth token generation/validation
- Update routes to use new CSRF token handling
- Add migration 013 to add csrf_token column to sessions

The session-bound approach ensures CSRF tokens remain stable for the
duration of a session, eliminating the race condition.

Note: Route tests (test_wizard.py, test_adapters.py, etc.) need
refactoring to mock get_settings() instead of CsrfProtect dependency.
Core auth/CSRF handler tests pass (74 tests).

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* test(csrf): update test suite for session-bound CSRF tokens

- Add CSRF fixtures to conftest.py for pre-auth and session CSRF
- Update test_wizard.py: use bypass_pre_auth_csrf and patch_route_settings
- Update test_adapters.py: set request.state.csrf_token and form mock data
- Update test_api_keys.py: add CSRF token to form data for POST routes
- Update test_streams.py: change return_value to side_effect for CSRF support
- Update test_region_picker.py: add CSRF token handling
- Update test_config_store.py: set CENTRAL_CSRF_SECRET env var in fixture

All 285 tests now pass with session-bound CSRF validation.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>
Co-authored-by: Matt Johnson <mj@k7zvx.com>
This commit is contained in:
malice 2026-05-17 22:06:22 -06:00 • committed by GitHub
commit 494ad1c799
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
28 changed files with 2897 additions and 377 deletions
Download patch file Download diff file Expand all files Collapse all files

46
tests/conftest.py
View file

@ -48,3 +48,49 @@ def mock_conn():
conn.fetchval = AsyncMock()
conn.execute = AsyncMock()
return conn
# CSRF fixtures for route tests
@pytest.fixture
def bypass_pre_auth_csrf():
"""Patch pre-auth CSRF validation to always pass.
Use for tests of pre-auth routes: /login, /setup/operator
"""
with patch("central.gui.routes.validate_pre_auth_csrf", return_value=True):
with patch("central.gui.routes.generate_pre_auth_csrf", return_value=("test_csrf_token", "test_signed_token")):
yield
@pytest.fixture
def bypass_session_csrf():
"""Create a mock request with session CSRF properly configured.
Use for tests of authenticated routes that check request.state.csrf_token.
Returns a configured mock_request.
"""
request = MagicMock()
request.state.csrf_token = "test_csrf_token_12345"
request.state.operator = MagicMock()
request.state.operator.id = 1
request.state.operator.username = "testuser"
# Mock form() to return dict with matching CSRF token
form_data = {"csrf_token": "test_csrf_token_12345"}
async def mock_form():
return form_data
request.form = mock_form
request._form_data = form_data # Allow tests to modify form data
return request
@pytest.fixture
def patch_route_settings():
"""Patch get_settings in routes module."""
with patch("central.gui.routes.get_settings") as mock:
mock.return_value.csrf_secret = "test-csrf-secret-for-testing-only-32chars"
yield mock

72
tests/test_adapters.py
View file

@ -55,13 +55,9 @@ class TestAdaptersListAuthenticated:
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()
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
result = await adapters_list(mock_request, mock_csrf)
result = await adapters_list(mock_request)
# Verify template was called with adapters
call_args = mock_templates.TemplateResponse.call_args
@ -105,13 +101,9 @@ class TestAdaptersEditForm:
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()
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
result = await adapters_edit_form(mock_request, "nws", mock_csrf)
result = await adapters_edit_form(mock_request, "nws")
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
@ -133,11 +125,8 @@ class TestAdaptersEditForm:
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
mock_csrf = MagicMock()
mock_csrf.generate_csrf_tokens.return_value = ("token", "signed")
with patch("central.gui.routes.get_pool", return_value=mock_pool):
result = await adapters_edit_form(mock_request, "nonexistent", mock_csrf)
result = await adapters_edit_form(mock_request, "nonexistent")
assert result.status_code == 404
@ -156,7 +145,9 @@ class TestAdaptersEditSubmit:
# Mock form data
mock_form = MagicMock()
mock_request.state.csrf_token = "test_csrf_token"
mock_form.get.side_effect = lambda k, d="": {
"csrf_token": "test_csrf_token",
"cadence_s": "120",
"contact_email": "new@example.com",
"region_north": "49.0",
@ -183,12 +174,9 @@ class TestAdaptersEditSubmit:
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
with patch("central.gui.routes.get_pool", return_value=mock_pool):
with patch("central.gui.routes.write_audit", new_callable=AsyncMock) as mock_audit:
result = await adapters_edit_submit(mock_request, "nws", mock_csrf)
result = await adapters_edit_submit(mock_request, "nws")
assert result.status_code == 302
assert result.headers["location"] == "/adapters"
@ -204,7 +192,9 @@ class TestAdaptersEditSubmit:
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_form = MagicMock()
mock_request.state.csrf_token = "test_csrf_token"
mock_form.get.side_effect = lambda k, d="": {
"csrf_token": "test_csrf_token",
"cadence_s": "30",
"contact_email": "test@example.com",
"region_north": "49.0",
@ -239,14 +229,9 @@ class TestAdaptersEditSubmit:
mock_response.status_code = 200
mock_templates.TemplateResponse.return_value = mock_response
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
mock_csrf.generate_csrf_tokens.return_value = ("token", "signed")
mock_csrf.set_csrf_cookie = MagicMock()
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
result = await adapters_edit_submit(mock_request, "nws", mock_csrf)
result = await adapters_edit_submit(mock_request, "nws")
# Should re-render form with error
call_args = mock_templates.TemplateResponse.call_args
@ -263,7 +248,9 @@ class TestAdaptersEditSubmit:
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_form = MagicMock()
mock_request.state.csrf_token = "test_csrf_token"
mock_form.get.side_effect = lambda k, d="": {
"csrf_token": "test_csrf_token",
"cadence_s": "300",
"api_key_alias": "nonexistent_key",
"region_north": "49.5",
@ -299,14 +286,9 @@ class TestAdaptersEditSubmit:
mock_response.status_code = 200
mock_templates.TemplateResponse.return_value = mock_response
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
mock_csrf.generate_csrf_tokens.return_value = ("token", "signed")
mock_csrf.set_csrf_cookie = MagicMock()
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
result = await adapters_edit_submit(mock_request, "firms", mock_csrf)
result = await adapters_edit_submit(mock_request, "firms")
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
@ -322,7 +304,9 @@ class TestAdaptersEditSubmit:
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_form = MagicMock()
mock_request.state.csrf_token = "test_csrf_token"
mock_form.get.side_effect = lambda k, d="": {
"csrf_token": "test_csrf_token",
"cadence_s": "120",
"feed": "invalid_feed",
"region_north": "49.0",
@ -357,14 +341,9 @@ class TestAdaptersEditSubmit:
mock_response.status_code = 200
mock_templates.TemplateResponse.return_value = mock_response
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
mock_csrf.generate_csrf_tokens.return_value = ("token", "signed")
mock_csrf.set_csrf_cookie = MagicMock()
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
result = await adapters_edit_submit(mock_request, "usgs_quake", mock_csrf)
result = await adapters_edit_submit(mock_request, "usgs_quake")
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
@ -383,7 +362,9 @@ class TestAdaptersAudit:
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_form = MagicMock()
mock_request.state.csrf_token = "test_csrf_token"
mock_form.get.side_effect = lambda k, d="": {
"csrf_token": "test_csrf_token",
"cadence_s": "120",
"contact_email": "new@example.com",
"region_north": "49.0",
@ -410,9 +391,6 @@ class TestAdaptersAudit:
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
captured_audit = {}
async def capture_audit(conn, action, operator_id=None, target=None, before=None, after=None):
@ -423,7 +401,7 @@ class TestAdaptersAudit:
with patch("central.gui.routes.get_pool", return_value=mock_pool):
with patch("central.gui.routes.write_audit", side_effect=capture_audit):
result = await adapters_edit_submit(mock_request, "nws", mock_csrf)
result = await adapters_edit_submit(mock_request, "nws")
assert captured_audit["action"] == "adapter.update"
assert captured_audit["target"] == "nws"
@ -449,7 +427,9 @@ class TestAdaptersJsonbRegression:
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_form = MagicMock()
mock_request.state.csrf_token = "test_csrf_token"
mock_form.get.side_effect = lambda k, d="": {
"csrf_token": "test_csrf_token",
"cadence_s": "120",
"contact_email": "test@example.com",
"region_north": "49.0",
@ -476,12 +456,9 @@ class TestAdaptersJsonbRegression:
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
with patch("central.gui.routes.get_pool", return_value=mock_pool):
with patch("central.gui.routes.write_audit", new_callable=AsyncMock):
await adapters_edit_submit(mock_request, "nws", mock_csrf)
await adapters_edit_submit(mock_request, "nws")
# Get the settings argument passed to execute (3rd positional arg after query)
call_args = mock_conn.execute.call_args
@ -502,7 +479,9 @@ class TestAdaptersJsonbRegression:
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_form = MagicMock()
mock_request.state.csrf_token = "test_csrf_token"
mock_form.get.side_effect = lambda k, d="": {
"csrf_token": "test_csrf_token",
"cadence_s": "120",
"contact_email": "new@example.com",
"region_north": "49.0",
@ -529,9 +508,6 @@ class TestAdaptersJsonbRegression:
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
captured_audit = {}
async def capture_audit(conn, action, operator_id=None, target=None, before=None, after=None):
@ -540,7 +516,7 @@ class TestAdaptersJsonbRegression:
with patch("central.gui.routes.get_pool", return_value=mock_pool):
with patch("central.gui.routes.write_audit", side_effect=capture_audit):
await adapters_edit_submit(mock_request, "nws", mock_csrf)
await adapters_edit_submit(mock_request, "nws")
# CRITICAL: before and after must be dicts, NOT strings
assert isinstance(captured_audit["before"], dict), f"before should be dict, got {type(captured_audit['before'])}"

86
tests/test_api_keys.py
View file

@ -75,13 +75,9 @@ class TestApiKeysListAuthenticated:
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
mock_csrf = MagicMock()
mock_csrf.generate_csrf_tokens.return_value = ("token", "signed")
mock_csrf.set_csrf_cookie = MagicMock()
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
result = await api_keys_list(mock_request, mock_csrf)
result = await api_keys_list(mock_request)
# Check template was called with correct context
call_args = mock_templates.TemplateResponse.call_args
@ -104,7 +100,8 @@ class TestApiKeysCreate:
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="admin")
form_data = {"alias": "test1", "plaintext_key": "secret-api-key-123"}
mock_request.state.csrf_token = "test_csrf_token"
form_data = {"csrf_token": "test_csrf_token", "alias": "test1", "plaintext_key": "secret-api-key-123"}
mock_request.form = AsyncMock(return_value=form_data)
mock_conn = AsyncMock()
@ -119,13 +116,10 @@ class TestApiKeysCreate:
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
with patch("central.gui.routes.get_pool", return_value=mock_pool):
with patch("central.crypto.encrypt", return_value=b"encrypted_data"):
with patch("central.gui.routes.write_audit", new_callable=AsyncMock):
result = await api_keys_create(mock_request, mock_csrf)
result = await api_keys_create(mock_request)
assert result.status_code == 302
assert result.headers["location"] == "/api-keys"
@ -136,7 +130,8 @@ class TestApiKeysCreate:
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="admin")
form_data = {"alias": "firms", "plaintext_key": "secret-key"}
mock_request.state.csrf_token = "test_csrf_token"
form_data = {"csrf_token": "test_csrf_token", "alias": "firms", "plaintext_key": "secret-key"}
mock_request.form = AsyncMock(return_value=form_data)
mock_templates = MagicMock()
@ -150,15 +145,10 @@ class TestApiKeysCreate:
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
mock_csrf.generate_csrf_tokens.return_value = ("token", "signed")
mock_csrf.set_csrf_cookie = MagicMock()
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.crypto.encrypt", return_value=b"encrypted"):
result = await api_keys_create(mock_request, mock_csrf)
result = await api_keys_create(mock_request)
# Should re-render form with error
call_args = mock_templates.TemplateResponse.call_args
@ -172,7 +162,8 @@ class TestApiKeysCreate:
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="admin")
form_data = {"alias": "", "plaintext_key": "secret-key"}
mock_request.state.csrf_token = "test_csrf_token"
form_data = {"csrf_token": "test_csrf_token", "alias": "", "plaintext_key": "secret-key"}
mock_request.form = AsyncMock(return_value=form_data)
mock_templates = MagicMock()
@ -183,14 +174,9 @@ class TestApiKeysCreate:
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
mock_csrf.generate_csrf_tokens.return_value = ("token", "signed")
mock_csrf.set_csrf_cookie = MagicMock()
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
result = await api_keys_create(mock_request, mock_csrf)
result = await api_keys_create(mock_request)
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
@ -203,7 +189,8 @@ class TestApiKeysCreate:
mock_request.state.operator = MagicMock(id=1, username="admin")
# Test with space
form_data = {"alias": "test key", "plaintext_key": "secret-key"}
mock_request.state.csrf_token = "test_csrf_token"
form_data = {"csrf_token": "test_csrf_token", "alias": "test key", "plaintext_key": "secret-key"}
mock_request.form = AsyncMock(return_value=form_data)
mock_templates = MagicMock()
@ -214,14 +201,9 @@ class TestApiKeysCreate:
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
mock_csrf.generate_csrf_tokens.return_value = ("token", "signed")
mock_csrf.set_csrf_cookie = MagicMock()
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
result = await api_keys_create(mock_request, mock_csrf)
result = await api_keys_create(mock_request)
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
@ -233,7 +215,8 @@ class TestApiKeysCreate:
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="admin")
form_data = {"alias": "test-key", "plaintext_key": "secret-key"}
mock_request.state.csrf_token = "test_csrf_token"
form_data = {"csrf_token": "test_csrf_token", "alias": "test-key", "plaintext_key": "secret-key"}
mock_request.form = AsyncMock(return_value=form_data)
mock_templates = MagicMock()
@ -244,14 +227,9 @@ class TestApiKeysCreate:
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
mock_csrf.generate_csrf_tokens.return_value = ("token", "signed")
mock_csrf.set_csrf_cookie = MagicMock()
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
result = await api_keys_create(mock_request, mock_csrf)
result = await api_keys_create(mock_request)
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
@ -267,7 +245,8 @@ class TestApiKeysRotate:
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="admin")
form_data = {"new_plaintext_key": "new-secret-key-456"}
mock_request.state.csrf_token = "test_csrf_token"
form_data = {"csrf_token": "test_csrf_token", "new_plaintext_key": "new-secret-key-456"}
mock_request.form = AsyncMock(return_value=form_data)
mock_conn = AsyncMock()
@ -290,13 +269,10 @@ class TestApiKeysRotate:
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
with patch("central.gui.routes.get_pool", return_value=mock_pool):
with patch("central.crypto.encrypt", return_value=b"new_encrypted"):
with patch("central.gui.routes.write_audit", new_callable=AsyncMock) as mock_audit:
result = await api_keys_rotate(mock_request, "test1", mock_csrf)
result = await api_keys_rotate(mock_request, "test1")
assert result.status_code == 302
# Check audit was called with no plaintext
@ -313,6 +289,8 @@ class TestApiKeysDelete:
"""POST /api-keys/{alias}/delete with references shows error."""
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="admin")
mock_request.state.csrf_token = "test_csrf_token"
mock_request.form = AsyncMock(return_value={"csrf_token": "test_csrf_token"})
mock_templates = MagicMock()
mock_templates.TemplateResponse.return_value = MagicMock()
@ -331,14 +309,9 @@ class TestApiKeysDelete:
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
mock_csrf.generate_csrf_tokens.return_value = ("token", "signed")
mock_csrf.set_csrf_cookie = MagicMock()
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
result = await api_keys_delete(mock_request, "firms", mock_csrf)
result = await api_keys_delete(mock_request, "firms")
# Should re-render with error
call_args = mock_templates.TemplateResponse.call_args
@ -351,6 +324,8 @@ class TestApiKeysDelete:
"""POST /api-keys/{alias}/delete without references deletes and redirects."""
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="admin")
mock_request.state.csrf_token = "test_csrf_token"
mock_request.form = AsyncMock(return_value={"csrf_token": "test_csrf_token"})
mock_conn = AsyncMock()
mock_conn.fetchrow.return_value = {
@ -367,12 +342,9 @@ class TestApiKeysDelete:
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
with patch("central.gui.routes.get_pool", return_value=mock_pool):
with patch("central.gui.routes.write_audit", new_callable=AsyncMock) as mock_audit:
result = await api_keys_delete(mock_request, "test1", mock_csrf)
result = await api_keys_delete(mock_request, "test1")
assert result.status_code == 302
assert result.headers["location"] == "/api-keys"
@ -388,7 +360,8 @@ class TestApiKeysAuditNoPlaintext:
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="admin")
form_data = {"alias": "newkey", "plaintext_key": "super-secret-value"}
mock_request.state.csrf_token = "test_csrf_token"
form_data = {"csrf_token": "test_csrf_token", "alias": "newkey", "plaintext_key": "super-secret-value"}
mock_request.form = AsyncMock(return_value=form_data)
mock_conn = AsyncMock()
@ -401,13 +374,10 @@ class TestApiKeysAuditNoPlaintext:
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
with patch("central.gui.routes.get_pool", return_value=mock_pool):
with patch("central.crypto.encrypt", return_value=b"encrypted"):
with patch("central.gui.routes.write_audit", new_callable=AsyncMock) as mock_audit:
await api_keys_create(mock_request, mock_csrf)
await api_keys_create(mock_request)
# Check audit call arguments
call_kwargs = mock_audit.call_args.kwargs

12
tests/test_auth.py
View file

@ -92,29 +92,33 @@ class TestSessionManagement:
mock_conn = MagicMock()
mock_conn.execute = AsyncMock()
token, expires_at = await create_session(mock_conn, operator_id=1, lifetime_days=90)
token, expires_at, csrf_token = await create_session(mock_conn, operator_id=1, lifetime_days=90)
assert len(token) == 43
assert len(csrf_token) == 64 # 32 bytes hex = 64 chars
mock_conn.execute.assert_called_once()
call_args = mock_conn.execute.call_args
assert "INSERT INTO config.sessions" in call_args[0][0]
@pytest.mark.asyncio
async def test_get_session_found(self):
"""get_session returns Operator when session exists."""
"""get_session returns (Operator, csrf_token) when session exists."""
mock_conn = MagicMock()
mock_conn.fetchrow = AsyncMock(return_value={
"id": 1,
"username": "testuser",
"created_at": datetime.now(timezone.utc),
"password_changed_at": datetime.now(timezone.utc),
"csrf_token": "test_csrf_token_12345",
})
operator = await get_session(mock_conn, "valid-token")
result = await get_session(mock_conn, "valid-token")
assert operator is not None
assert result is not None
operator, csrf_token = result
assert operator.id == 1
assert operator.username == "testuser"
assert csrf_token == "test_csrf_token_12345"
@pytest.mark.asyncio
async def test_get_session_not_found(self):

1
tests/test_config_store.py
View file

@ -39,6 +39,7 @@ def setup_master_key(master_key_path: Path, monkeypatch: pytest.MonkeyPatch) ->
clear_key_cache()
monkeypatch.setenv("CENTRAL_DB_DSN", TEST_DB_DSN)
monkeypatch.setenv("CENTRAL_MASTER_KEY_PATH", str(master_key_path))
monkeypatch.setenv("CENTRAL_CSRF_SECRET", "test-csrf-secret-for-testing-only-32chars")
@pytest_asyncio.fixture

200
tests/test_csrf_handler.py
View file

@ -14,23 +14,18 @@ class TestCsrfExceptionHandlerRegistered:
"""Verify CSRF exception handler is properly registered."""
def test_csrf_exception_handler_is_registered(self):
"""The app has a CsrfProtectError exception handler registered."""
"""The app has a CsrfValidationError exception handler registered."""
from central.gui import app
from fastapi_csrf_protect.exceptions import CsrfProtectError
from central.gui.auth import CsrfValidationError
assert CsrfProtectError in app.exception_handlers, \
"CsrfProtectError handler should be registered"
assert CsrfValidationError in app.exception_handlers, \
"CsrfValidationError handler should be registered"
def test_csrf_subclasses_are_caught(self):
"""MissingTokenError and TokenValidationError inherit from CsrfProtectError."""
from fastapi_csrf_protect.exceptions import (
CsrfProtectError,
MissingTokenError,
TokenValidationError,
)
def test_csrf_validation_error_is_exception(self):
"""CsrfValidationError is a proper Exception subclass."""
from central.gui.auth import CsrfValidationError
assert issubclass(MissingTokenError, CsrfProtectError)
assert issubclass(TokenValidationError, CsrfProtectError)
assert issubclass(CsrfValidationError, Exception)
class TestCsrfExceptionHandlerBehavior:
@ -40,10 +35,10 @@ class TestCsrfExceptionHandlerBehavior:
"""CSRF handler checks request path for /login."""
import inspect
from central.gui import _create_app
from fastapi_csrf_protect.exceptions import CsrfProtectError
from central.gui.auth import CsrfValidationError
app = _create_app()
handler = app.exception_handlers.get(CsrfProtectError)
handler = app.exception_handlers.get(CsrfValidationError)
# Verify handler source contains /login path check
source = inspect.getsource(handler)
@ -54,17 +49,16 @@ class TestCsrfExceptionHandlerBehavior:
async def test_logout_csrf_error_redirects_to_login(self):
"""CSRF error on /logout should redirect to /login."""
from central.gui import _create_app
from fastapi_csrf_protect.exceptions import TokenValidationError
from central.gui.auth import CsrfValidationError
from fastapi.responses import RedirectResponse
app = _create_app()
from fastapi_csrf_protect.exceptions import CsrfProtectError
handler = app.exception_handlers.get(CsrfProtectError)
handler = app.exception_handlers.get(CsrfValidationError)
mock_request = MagicMock()
mock_request.url.path = "/logout"
exc = TokenValidationError("Invalid token")
exc = CsrfValidationError("Invalid token")
result = await handler(mock_request, exc)
@ -75,17 +69,16 @@ class TestCsrfExceptionHandlerBehavior:
async def test_adapters_csrf_error_redirects_to_adapters(self):
"""CSRF error on /adapters/{name} should redirect to /adapters."""
from central.gui import _create_app
from fastapi_csrf_protect.exceptions import TokenValidationError
from central.gui.auth import CsrfValidationError
from fastapi.responses import RedirectResponse
app = _create_app()
from fastapi_csrf_protect.exceptions import CsrfProtectError
handler = app.exception_handlers.get(CsrfProtectError)
handler = app.exception_handlers.get(CsrfValidationError)
mock_request = MagicMock()
mock_request.url.path = "/adapters/nws"
exc = TokenValidationError("Invalid token")
exc = CsrfValidationError("Invalid token")
result = await handler(mock_request, exc)
@ -94,16 +87,171 @@ class TestCsrfExceptionHandlerBehavior:
class TestCsrfHandlerNoTraceback:
"""Verify exception handler doesn't expose Python internals."""
"""Verify exception handler does not expose Python internals."""
def test_handler_exists_and_is_async(self):
"""The CSRF handler should be an async function."""
import inspect
from central.gui import _create_app
from fastapi_csrf_protect.exceptions import CsrfProtectError
from central.gui.auth import CsrfValidationError
app = _create_app()
handler = app.exception_handlers.get(CsrfProtectError)
handler = app.exception_handlers.get(CsrfValidationError)
assert handler is not None
assert inspect.iscoroutinefunction(handler)
class TestCsrfHandlerWizardPaths:
"""Test CSRF exception handler for wizard paths."""
@pytest.mark.asyncio
async def test_setup_operator_csrf_error_renders_form_with_error(self):
"""CSRF error on /setup/operator re-renders form with error message."""
from central.gui import _create_app
from central.gui.auth import CsrfValidationError
app = _create_app()
handler = app.exception_handlers.get(CsrfValidationError)
mock_request = MagicMock()
mock_request.url.path = "/setup/operator"
exc = CsrfValidationError("Invalid token")
result = await handler(mock_request, exc)
# Should be HTML response, not redirect
assert hasattr(result, "body")
assert result.status_code == 200
body = result.body.decode() if hasattr(result.body, "decode") else str(result.body)
assert "session expired" in body.lower()
@pytest.mark.asyncio
async def test_setup_system_csrf_error_renders_form_with_error(self):
"""CSRF error on /setup/system re-renders form with error message."""
from central.gui import _create_app
from central.gui.auth import CsrfValidationError
app = _create_app()
handler = app.exception_handlers.get(CsrfValidationError)
mock_request = MagicMock()
mock_request.url.path = "/setup/system"
exc = CsrfValidationError("Invalid token")
with patch("central.gui.db.get_pool", return_value=None):
result = await handler(mock_request, exc)
assert hasattr(result, "body")
assert result.status_code == 200
body = result.body.decode() if hasattr(result.body, "decode") else str(result.body)
assert "session expired" in body.lower()
@pytest.mark.asyncio
async def test_setup_keys_csrf_error_renders_form_with_error(self):
"""CSRF error on /setup/keys re-renders form with error message."""
from central.gui import _create_app
from central.gui.auth import CsrfValidationError
app = _create_app()
handler = app.exception_handlers.get(CsrfValidationError)
mock_request = MagicMock()
mock_request.url.path = "/setup/keys"
exc = CsrfValidationError("Invalid token")
with patch("central.gui.db.get_pool", return_value=None):
result = await handler(mock_request, exc)
assert hasattr(result, "body")
assert result.status_code == 200
body = result.body.decode() if hasattr(result.body, "decode") else str(result.body)
assert "session expired" in body.lower()
@pytest.mark.asyncio
async def test_setup_adapters_csrf_error_renders_form_with_error(self):
"""CSRF error on /setup/adapters re-renders form with error message."""
from central.gui import _create_app
from central.gui.auth import CsrfValidationError
app = _create_app()
handler = app.exception_handlers.get(CsrfValidationError)
mock_request = MagicMock()
mock_request.url.path = "/setup/adapters"
exc = CsrfValidationError("Invalid token")
with patch("central.gui.db.get_pool", return_value=None):
result = await handler(mock_request, exc)
assert hasattr(result, "body")
assert result.status_code == 200
body = result.body.decode() if hasattr(result.body, "decode") else str(result.body)
assert "session expired" in body.lower()
@pytest.mark.asyncio
async def test_setup_finish_csrf_error_renders_form_with_error(self):
"""CSRF error on /setup/finish re-renders form with error message."""
from central.gui import _create_app
from central.gui.auth import CsrfValidationError
app = _create_app()
handler = app.exception_handlers.get(CsrfValidationError)
mock_request = MagicMock()
mock_request.url.path = "/setup/finish"
exc = CsrfValidationError("Invalid token")
with patch("central.gui.db.get_pool", return_value=None):
result = await handler(mock_request, exc)
assert hasattr(result, "body")
assert result.status_code == 200
body = result.body.decode() if hasattr(result.body, "decode") else str(result.body)
assert "session expired" in body.lower()
@pytest.mark.asyncio
async def test_setup_base_csrf_error_redirects_to_setup(self):
"""CSRF error on /setup redirects to /setup (middleware routes to step)."""
from central.gui import _create_app
from central.gui.auth import CsrfValidationError
from fastapi.responses import RedirectResponse
app = _create_app()
handler = app.exception_handlers.get(CsrfValidationError)
mock_request = MagicMock()
mock_request.url.path = "/setup"
exc = CsrfValidationError("Invalid token")
result = await handler(mock_request, exc)
assert isinstance(result, RedirectResponse)
assert result.status_code == 302
@pytest.mark.asyncio
async def test_login_csrf_error_still_works(self):
"""CSRF error on /login still renders login form with error (regression test)."""
from central.gui import _create_app
from central.gui.auth import CsrfValidationError
app = _create_app()
handler = app.exception_handlers.get(CsrfValidationError)
mock_request = MagicMock()
mock_request.url.path = "/login"
exc = CsrfValidationError("Invalid token")
result = await handler(mock_request, exc)
assert hasattr(result, "body")
assert result.status_code == 200
body = result.body.decode() if hasattr(result.body, "decode") else str(result.body)
assert "session expired" in body.lower()

108
tests/test_csrf_race_condition.py Normal file
View file

@ -0,0 +1,108 @@
"""
Integration test for CSRF race condition fix.
This test verifies that the session-bound CSRF implementation fixes the race
condition where interleaved GET requests would invalidate CSRF tokens.
See: PR #24 - Central 1b-8 fix-up phase 2
"""
import pytest
class TestCsrfRaceConditionFix:
"""Verify that interleaved GETs don't break CSRF validation."""
def test_session_bound_csrf_consistent_across_gets(self):
"""Session-bound CSRF tokens remain consistent across multiple GETs.
This was the core bug: fastapi-csrf-protect rotated tokens on every GET,
causing race conditions when users had multiple tabs or slow connections.
With session-bound CSRF, the token is stored in the session row and
remains constant until the session is destroyed.
"""
from unittest.mock import MagicMock, AsyncMock
from central.gui.auth import get_session
# Mock a session with a csrf_token
mock_conn = MagicMock()
mock_conn.fetchrow = AsyncMock(return_value={
"id": 1,
"username": "testuser",
"created_at": "2024-01-01T00:00:00Z",
"password_changed_at": "2024-01-01T00:00:00Z",
"csrf_token": "fixed_csrf_token_12345",
})
import asyncio
async def test():
# First GET
result1 = await get_session(mock_conn, "test-token")
assert result1 is not None
op1, csrf1 = result1
# Second GET (simulating interleaved request)
result2 = await get_session(mock_conn, "test-token")
assert result2 is not None
op2, csrf2 = result2
# CSRF tokens should be identical (the fix!)
assert csrf1 == csrf2 == "fixed_csrf_token_12345"
asyncio.run(test())
def test_pre_auth_csrf_tokens_independently_valid(self):
"""Pre-auth CSRF tokens are independently valid.
For unauthenticated routes, each GET generates a new token+cookie pair.
Each pair should validate independently, allowing the original token
to work even if another GET happened in between.
"""
from central.gui.csrf import generate_pre_auth_csrf, validate_pre_auth_csrf
from unittest.mock import MagicMock
secret = "testsecret12345678901234567890ab"
# First GET generates token1 + cookie1
token1, signed1 = generate_pre_auth_csrf(secret)
# Second GET generates token2 + cookie2
token2, signed2 = generate_pre_auth_csrf(secret)
# Tokens should be different (fresh random tokens)
assert token1 != token2
assert signed1 != signed2
# But each pair should validate independently
mock_request1 = MagicMock()
mock_request1.cookies = {"central_preauth_csrf": signed1}
mock_request2 = MagicMock()
mock_request2.cookies = {"central_preauth_csrf": signed2}
# Original token still validates with original cookie
assert validate_pre_auth_csrf(mock_request1, token1, secret) is True
# Second token validates with second cookie
assert validate_pre_auth_csrf(mock_request2, token2, secret) is True
# Cross-validation should fail
assert validate_pre_auth_csrf(mock_request1, token2, secret) is False
assert validate_pre_auth_csrf(mock_request2, token1, secret) is False
def test_csrf_token_generation_is_secure(self):
"""CSRF tokens are cryptographically secure."""
from central.gui.auth import generate_csrf_token
# Generate multiple tokens
tokens = [generate_csrf_token() for _ in range(100)]
# All tokens should be unique
assert len(set(tokens)) == 100
# Tokens should be 64 hex chars (32 bytes)
for token in tokens:
assert len(token) == 64
assert all(c in "0123456789abcdef" for c in token)

47
tests/test_region_picker.py
View file

@ -51,13 +51,9 @@ class TestRegionPickerInTemplate:
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()
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
result = await adapters_edit_form(mock_request, "firms", mock_csrf)
result = await adapters_edit_form(mock_request, "firms")
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
@ -79,7 +75,9 @@ class TestRegionValidation:
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_form = MagicMock()
mock_request.state.csrf_token = "test_csrf_token"
mock_form.get.side_effect = lambda k, d="": {
"csrf_token": "test_csrf_token",
"cadence_s": "300",
"api_key_alias": "firms",
"region_north": "45.0",
@ -109,9 +107,6 @@ class TestRegionValidation:
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
captured_settings = {}
async def capture_execute(query, *args):
@ -122,7 +117,7 @@ class TestRegionValidation:
with patch("central.gui.routes.get_pool", return_value=mock_pool):
with patch("central.gui.routes.write_audit", new_callable=AsyncMock):
result = await adapters_edit_submit(mock_request, "firms", mock_csrf)
result = await adapters_edit_submit(mock_request, "firms")
assert result.status_code == 302
assert captured_settings["settings"]["region"]["north"] == 45.0
@ -139,7 +134,9 @@ class TestRegionValidation:
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_form = MagicMock()
mock_request.state.csrf_token = "test_csrf_token"
mock_form.get.side_effect = lambda k, d="": {
"csrf_token": "test_csrf_token",
"cadence_s": "300",
"api_key_alias": "firms",
"region_north": "30.0", # Less than south!
@ -175,14 +172,9 @@ class TestRegionValidation:
mock_response.status_code = 200
mock_templates.TemplateResponse.return_value = mock_response
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
mock_csrf.generate_csrf_tokens.return_value = ("token", "signed")
mock_csrf.set_csrf_cookie = MagicMock()
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
result = await adapters_edit_submit(mock_request, "firms", mock_csrf)
result = await adapters_edit_submit(mock_request, "firms")
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
@ -198,7 +190,9 @@ class TestRegionValidation:
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_form = MagicMock()
mock_request.state.csrf_token = "test_csrf_token"
mock_form.get.side_effect = lambda k, d="": {
"csrf_token": "test_csrf_token",
"cadence_s": "300",
"api_key_alias": "firms",
"region_north": "45.0",
@ -234,14 +228,9 @@ class TestRegionValidation:
mock_response.status_code = 200
mock_templates.TemplateResponse.return_value = mock_response
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
mock_csrf.generate_csrf_tokens.return_value = ("token", "signed")
mock_csrf.set_csrf_cookie = MagicMock()
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
result = await adapters_edit_submit(mock_request, "firms", mock_csrf)
result = await adapters_edit_submit(mock_request, "firms")
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
@ -257,7 +246,9 @@ class TestRegionValidation:
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_form = MagicMock()
mock_request.state.csrf_token = "test_csrf_token"
mock_form.get.side_effect = lambda k, d="": {
"csrf_token": "test_csrf_token",
"cadence_s": "300",
"api_key_alias": "firms",
"region_north": "95.0", # > 90!
@ -293,14 +284,9 @@ class TestRegionValidation:
mock_response.status_code = 200
mock_templates.TemplateResponse.return_value = mock_response
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
mock_csrf.generate_csrf_tokens.return_value = ("token", "signed")
mock_csrf.set_csrf_cookie = MagicMock()
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
result = await adapters_edit_submit(mock_request, "firms", mock_csrf)
result = await adapters_edit_submit(mock_request, "firms")
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
@ -319,7 +305,9 @@ class TestRegionAuditLog:
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_form = MagicMock()
mock_request.state.csrf_token = "test_csrf_token"
mock_form.get.side_effect = lambda k, d="": {
"csrf_token": "test_csrf_token",
"cadence_s": "300",
"api_key_alias": "firms",
"region_north": "45.0",
@ -352,9 +340,6 @@ class TestRegionAuditLog:
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
captured_audit = {}
async def capture_audit(conn, action, operator_id=None, target=None, before=None, after=None):
@ -363,7 +348,7 @@ class TestRegionAuditLog:
with patch("central.gui.routes.get_pool", return_value=mock_pool):
with patch("central.gui.routes.write_audit", side_effect=capture_audit):
result = await adapters_edit_submit(mock_request, "firms", mock_csrf)
result = await adapters_edit_submit(mock_request, "firms")
# Before should have old region
assert captured_audit["before"]["settings"]["region"]["north"] == 49.5

2
tests/test_session_auth.py
View file

@ -43,6 +43,7 @@ class TestSessionMiddleware:
"username": "admin",
"created_at": datetime.now(timezone.utc),
"password_changed_at": datetime.now(timezone.utc),
"csrf_token": "mock_csrf_token_12345",
})
mock_conn.__aenter__ = AsyncMock(return_value=mock_conn)
mock_conn.__aexit__ = AsyncMock()
@ -99,6 +100,7 @@ class TestSessionMiddleware:
"username": "admin",
"created_at": datetime.now(timezone.utc),
"password_changed_at": datetime.now(timezone.utc),
"csrf_token": "mock_csrf_token_12345",
})
mock_conn.__aenter__ = AsyncMock(return_value=mock_conn)
mock_conn.__aexit__ = AsyncMock()

50
tests/test_setup_gate.py
View file

@ -12,8 +12,8 @@ class TestSetupGateMiddleware:
"""Tests for SetupGateMiddleware."""
@pytest.mark.asyncio
async def test_allows_setup_route_when_incomplete(self):
"""SetupGateMiddleware allows /setup when setup_complete=False."""
async def test_allows_setup_subpath_when_incomplete(self):
"""SetupGateMiddleware allows /setup/operator when setup_complete=False."""
mock_pool = MagicMock()
mock_conn = MagicMock()
mock_conn.fetchrow = AsyncMock(return_value={"setup_complete": False})
@ -21,6 +21,31 @@ class TestSetupGateMiddleware:
mock_conn.__aexit__ = AsyncMock()
mock_pool.acquire = MagicMock(return_value=mock_conn)
with patch("central.gui.middleware.get_pool", return_value=mock_pool):
app = FastAPI()
@app.get("/setup/operator")
async def setup_operator():
return {"message": "operator"}
app.add_middleware(SetupGateMiddleware)
client = TestClient(app)
response = client.get("/setup/operator")
assert response.status_code == 200
assert response.json() == {"message": "operator"}
@pytest.mark.asyncio
async def test_redirects_setup_base_to_wizard_step(self):
"""SetupGateMiddleware redirects /setup to wizard step when incomplete."""
mock_pool = MagicMock()
mock_conn = MagicMock()
mock_conn.fetchrow = AsyncMock(return_value={"setup_complete": False})
mock_conn.fetchval = AsyncMock(return_value=0) # No operators
mock_conn.__aenter__ = AsyncMock(return_value=mock_conn)
mock_conn.__aexit__ = AsyncMock()
mock_pool.acquire = MagicMock(return_value=mock_conn)
with patch("central.gui.middleware.get_pool", return_value=mock_pool):
app = FastAPI()
@ -28,12 +53,16 @@ class TestSetupGateMiddleware:
async def setup():
return {"message": "setup"}
@app.get("/setup/operator")
async def setup_operator():
return {"message": "operator"}
app.add_middleware(SetupGateMiddleware)
client = TestClient(app)
client = TestClient(app, follow_redirects=False)
response = client.get("/setup")
assert response.status_code == 200
assert response.json() == {"message": "setup"}
assert response.status_code == 302
assert response.headers["location"] == "/setup/operator"
@pytest.mark.asyncio
async def test_allows_health_when_incomplete(self):
@ -135,7 +164,7 @@ class TestSetupGateMiddleware:
@pytest.mark.asyncio
async def test_redirects_setup_when_complete(self):
"""SetupGateMiddleware redirects /setup to / when setup_complete=True."""
"""SetupGateMiddleware redirects /setup/* to / when setup_complete=True."""
mock_pool = MagicMock()
mock_conn = MagicMock()
mock_conn.fetchrow = AsyncMock(return_value={"setup_complete": True})
@ -154,9 +183,18 @@ class TestSetupGateMiddleware:
async def setup():
return {"message": "setup"}
@app.get("/setup/operator")
async def setup_operator():
return {"message": "operator"}
app.add_middleware(SetupGateMiddleware)
client = TestClient(app, follow_redirects=False)
# Both /setup and /setup/operator should redirect to /
response = client.get("/setup")
assert response.status_code == 302
assert response.headers["location"] == "/"
response = client.get("/setup/operator")
assert response.status_code == 302
assert response.headers["location"] == "/"

92
tests/test_streams.py
View file

@ -52,10 +52,6 @@ class TestStreamsListAuthenticated:
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 proper state fields
mock_js = AsyncMock()
mock_stream_info = MagicMock()
@ -74,7 +70,7 @@ class TestStreamsListAuthenticated:
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)
result = await streams_list(mock_request)
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
@ -117,14 +113,10 @@ class TestStreamsListNatsUnavailable:
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()
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=None):
result = await streams_list(mock_request, mock_csrf)
result = await streams_list(mock_request)
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
@ -157,10 +149,6 @@ class TestStreamsListPartialFailure:
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 - CENTRAL_FIRE raises ValueError, CENTRAL_WX works
mock_js = AsyncMock()
test_ts = datetime(2026, 5, 17, 12, 0, 0, tzinfo=timezone.utc)
@ -184,7 +172,7 @@ class TestStreamsListPartialFailure:
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)
result = await streams_list(mock_request)
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
@ -222,10 +210,6 @@ class TestStreamsListEmptyStream:
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()
@ -239,7 +223,7 @@ class TestStreamsListEmptyStream:
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)
result = await streams_list(mock_request)
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
@ -278,10 +262,6 @@ class TestStreamsListSingleMessage:
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()
@ -299,7 +279,7 @@ class TestStreamsListSingleMessage:
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)
result = await streams_list(mock_request)
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
@ -337,10 +317,6 @@ class TestStreamsListGetMsgFailure:
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()
@ -365,7 +341,7 @@ class TestStreamsListGetMsgFailure:
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)
result = await streams_list(mock_request)
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
@ -394,8 +370,12 @@ class TestStreamsUpdate:
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_request.state.csrf_token = "test_csrf_token"
mock_form = MagicMock()
mock_form.get.return_value = "1209600" # 14 days
mock_form.get.side_effect = lambda k, d="": {
"csrf_token": "test_csrf_token",
"max_age_s": "1209600",
}.get(k, d)
mock_request.form = AsyncMock(return_value=mock_form)
mock_conn = AsyncMock()
@ -406,9 +386,6 @@ class TestStreamsUpdate:
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
captured_audit = {}
async def capture_audit(conn, action, operator_id=None, target=None, before=None, after=None):
@ -419,7 +396,7 @@ class TestStreamsUpdate:
with patch("central.gui.routes.get_pool", return_value=mock_pool):
with patch("central.gui.routes.write_audit", side_effect=capture_audit):
result = await streams_update(mock_request, "CENTRAL_WX", mock_csrf)
result = await streams_update(mock_request, "CENTRAL_WX")
assert result.status_code == 302
assert result.headers["location"] == "/streams"
@ -438,8 +415,12 @@ class TestStreamsUpdate:
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_request.state.csrf_token = "test_csrf_token"
mock_form = MagicMock()
mock_form.get.return_value = "60" # 1 minute - too small
mock_form.get.side_effect = lambda k, d="": {
"csrf_token": "test_csrf_token",
"max_age_s": "60",
}.get(k, d)
mock_request.form = AsyncMock(return_value=mock_form)
mock_conn = AsyncMock()
@ -458,15 +439,10 @@ class TestStreamsUpdate:
mock_response.status_code = 200
mock_templates.TemplateResponse.return_value = mock_response
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
mock_csrf.generate_csrf_tokens.return_value = ("token", "signed")
mock_csrf.set_csrf_cookie = MagicMock()
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=None):
result = await streams_update(mock_request, "CENTRAL_WX", mock_csrf)
result = await streams_update(mock_request, "CENTRAL_WX")
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
@ -480,9 +456,10 @@ class TestStreamsUpdate:
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_request.state.csrf_token = "test_csrf_token"
mock_form = MagicMock()
mock_form.get.return_value = "999999999" # Way too large
mock_form.get.side_effect = lambda k, d="": {"csrf_token": "test_csrf_token", "max_age_s": "999999999"}.get(k, d) # Way too large
mock_request.form = AsyncMock(return_value=mock_form)
mock_conn = AsyncMock()
@ -501,15 +478,10 @@ class TestStreamsUpdate:
mock_response.status_code = 200
mock_templates.TemplateResponse.return_value = mock_response
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
mock_csrf.generate_csrf_tokens.return_value = ("token", "signed")
mock_csrf.set_csrf_cookie = MagicMock()
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=None):
result = await streams_update(mock_request, "CENTRAL_WX", mock_csrf)
result = await streams_update(mock_request, "CENTRAL_WX")
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
@ -523,8 +495,12 @@ class TestStreamsUpdate:
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_request.state.csrf_token = "test_csrf_token"
mock_form = MagicMock()
mock_form.get.return_value = "604800"
mock_form.get.side_effect = lambda k, d="": {
"csrf_token": "test_csrf_token",
"max_age_s": "604800",
}.get(k, d)
mock_request.form = AsyncMock(return_value=mock_form)
mock_conn = AsyncMock()
@ -534,11 +510,8 @@ class TestStreamsUpdate:
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
with patch("central.gui.routes.get_pool", return_value=mock_pool):
result = await streams_update(mock_request, "nonexistent", mock_csrf)
result = await streams_update(mock_request, "nonexistent")
assert result.status_code == 404
@ -554,8 +527,12 @@ class TestStreamsAudit:
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_request.state.csrf_token = "test_csrf_token"
mock_form = MagicMock()
mock_form.get.return_value = "1209600" # 14 days
mock_form.get.side_effect = lambda k, d="": {
"csrf_token": "test_csrf_token",
"max_age_s": "1209600",
}.get(k, d)
mock_request.form = AsyncMock(return_value=mock_form)
mock_conn = AsyncMock()
@ -566,9 +543,6 @@ class TestStreamsAudit:
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
mock_csrf = MagicMock()
mock_csrf.validate_csrf = AsyncMock()
captured_audit = {}
async def capture_audit(conn, action, operator_id=None, target=None, before=None, after=None):
@ -580,7 +554,7 @@ class TestStreamsAudit:
with patch("central.gui.routes.get_pool", return_value=mock_pool):
with patch("central.gui.routes.write_audit", side_effect=capture_audit):
await streams_update(mock_request, "CENTRAL_QUAKE", mock_csrf)
await streams_update(mock_request, "CENTRAL_QUAKE")
assert captured_audit["action"] == "stream.update"
assert captured_audit["operator_id"] == 1

665
tests/test_wizard.py Normal file
View file

@ -0,0 +1,665 @@
"""Tests for the first-run setup wizard."""
from datetime import datetime, timezone
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from central.gui.routes import (
setup_operator_form,
setup_operator_submit,
setup_system_form,
setup_system_submit,
setup_keys_form,
setup_keys_submit,
setup_adapters_form,
setup_adapters_submit,
setup_finish_form,
setup_finish_submit,
)
from central.gui.middleware import SetupGateMiddleware, _get_wizard_redirect_step
class TestWizardStepRedirect:
"""Test wizard step redirect logic."""
@pytest.mark.asyncio
async def test_no_operators_redirects_to_operator(self):
"""When no operators exist, redirect to /setup/operator."""
mock_conn = AsyncMock()
mock_conn.fetchval.side_effect = [0] # No operators
result = await _get_wizard_redirect_step(mock_conn)
assert result == "/setup/operator"
@pytest.mark.asyncio
async def test_default_tile_url_redirects_to_system(self):
"""When map_tile_url is default, redirect to /setup/system."""
mock_conn = AsyncMock()
mock_conn.fetchval.side_effect = [1] # Has operator
mock_conn.fetchrow.return_value = {
"map_tile_url": "https://tile.openstreetmap.org/{z}/{x}/{y}.png"
}
result = await _get_wizard_redirect_step(mock_conn)
assert result == "/setup/system"
@pytest.mark.asyncio
async def test_no_adapters_touched_redirects_to_keys(self):
"""When no adapters have been updated, redirect to /setup/keys."""
mock_conn = AsyncMock()
mock_conn.fetchval.side_effect = [1, 0] # Has operator, no adapters touched
mock_conn.fetchrow.return_value = {
"map_tile_url": "https://custom.example.com/{z}/{x}/{y}.png"
}
result = await _get_wizard_redirect_step(mock_conn)
assert result == "/setup/keys"
@pytest.mark.asyncio
async def test_all_steps_complete_redirects_to_finish(self):
"""When all steps done, redirect to /setup/finish."""
mock_conn = AsyncMock()
mock_conn.fetchval.side_effect = [1, 1] # Has operator, adapters touched
mock_conn.fetchrow.return_value = {
"map_tile_url": "https://custom.example.com/{z}/{x}/{y}.png"
}
result = await _get_wizard_redirect_step(mock_conn)
assert result == "/setup/finish"
class TestSetupOperatorForm:
"""Test operator creation form (step 1)."""
@pytest.mark.asyncio
async def test_get_returns_form(self):
"""GET /setup/operator returns the form when no operator exists."""
mock_request = MagicMock()
mock_templates = MagicMock()
mock_templates.TemplateResponse.return_value = MagicMock()
mock_conn = AsyncMock()
mock_conn.fetchrow.return_value = None # No operator exists
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
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.routes.get_settings") as mock_settings:
mock_settings.return_value.csrf_secret = "testsecret"
with patch("central.gui.routes.generate_pre_auth_csrf", return_value=("test_token", "signed_token")):
result = await setup_operator_form(mock_request)
mock_templates.TemplateResponse.assert_called_once()
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
assert "csrf_token" in context and context["csrf_token"]
assert context["error"] is None
assert context["existing_operator"] is None
@pytest.mark.asyncio
async def test_get_returns_confirmation_when_operator_exists(self):
"""GET /setup/operator shows confirmation when operator already exists."""
mock_request = MagicMock()
mock_templates = MagicMock()
mock_response = MagicMock()
mock_response.body = b"Operator Already Configured"
mock_templates.TemplateResponse.return_value = mock_response
mock_conn = AsyncMock()
mock_conn.fetchrow.return_value = {"username": "admin"} # Operator exists
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
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.routes.get_settings") as mock_settings:
mock_settings.return_value.csrf_secret = "testsecret"
with patch("central.gui.routes.generate_pre_auth_csrf", return_value=("test_token", "signed_token")):
result = await setup_operator_form(mock_request)
mock_templates.TemplateResponse.assert_called_once()
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
assert context["existing_operator"] == {"username": "admin"}
assert context["error"] is None
class TestSetupOperatorSubmit:
"""Test operator creation submission."""
@pytest.mark.asyncio
async def test_password_mismatch_shows_error(self):
"""POST with password mismatch re-renders with error."""
mock_request = MagicMock()
mock_request.state.csrf_token = "test_csrf"
mock_request.form = AsyncMock(return_value={
"csrf_token": "test_csrf",
"username": "testuser",
"password": "password1",
"confirm_password": "password2", # Mismatch
})
mock_templates = MagicMock()
mock_templates.TemplateResponse.return_value = MagicMock()
mock_conn = AsyncMock()
mock_conn.fetchval.return_value = 0 # No existing operators
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
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.routes.validate_pre_auth_csrf", return_value=True):
with patch("central.gui.routes.get_settings") as mock_settings:
mock_settings.return_value.csrf_secret = "testsecret"
result = await setup_operator_submit(
mock_request,
username="testuser",
password="password1",
confirm_password="password2",
)
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
assert context["error"] == "Passwords do not match"
@pytest.mark.asyncio
async def test_valid_creates_operator_and_redirects(self):
"""POST with valid data creates operator and redirects to /setup/system."""
mock_request = MagicMock()
mock_request.state.csrf_token = "test_csrf"
mock_request.form = AsyncMock(return_value={
"csrf_token": "test_csrf",
"username": "testuser",
"password": "password123",
"confirm_password": "password123",
})
mock_conn = AsyncMock()
mock_conn.fetchval.return_value = 0 # No existing operators
mock_conn.fetchrow.side_effect = [
{"id": 1}, # INSERT RETURNING id
{"session_lifetime_days": 90}, # system settings
]
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
with patch("central.gui.routes.get_pool", return_value=mock_pool):
with patch("central.gui.routes.validate_pre_auth_csrf", return_value=True):
with patch("central.gui.routes.get_settings") as mock_settings:
mock_settings.return_value.csrf_secret = "testsecret"
with patch("central.gui.routes.hash_password", return_value="hashed"):
with patch("central.gui.routes.create_session", new_callable=AsyncMock) as mock_session:
mock_session.return_value = ("session_token", datetime.now(), "csrf_token")
with patch("central.gui.routes.write_audit", new_callable=AsyncMock):
result = await setup_operator_submit(
mock_request,
username="testuser",
password="password123",
confirm_password="password123",
)
assert result.status_code == 302
assert result.headers["location"] == "/setup/system"
@pytest.mark.asyncio
async def test_post_when_operator_exists_shows_confirmation(self):
"""POST when operator exists returns 200 with confirmation, no insert."""
mock_request = MagicMock()
mock_request.form = AsyncMock(return_value={
"csrf_token": "test_csrf",
"username": "testuser",
"password": "password123",
"confirm_password": "password123",
})
mock_templates = MagicMock()
mock_response = MagicMock()
mock_response.status_code = 200
mock_templates.TemplateResponse.return_value = mock_response
mock_conn = AsyncMock()
mock_conn.fetchval.return_value = 1 # Operator already exists
mock_conn.fetchrow.return_value = {"username": "existing_admin"}
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
mock_request.state.csrf_token = "test_csrf"
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.routes.validate_pre_auth_csrf", return_value=True):
with patch("central.gui.routes.get_settings") as mock_settings:
mock_settings.return_value.csrf_secret = "testsecret"
with patch("central.gui.routes.write_audit", new_callable=AsyncMock) as mock_audit:
result = await setup_operator_submit(
mock_request,
username="testuser",
password="password123",
confirm_password="password123",
)
# Should return 200, not 500 or redirect
assert result.status_code == 200
# Should render confirmation state
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
assert context["existing_operator"] == {"username": "existing_admin"}
# Should NOT call write_audit (no insert happened)
mock_audit.assert_not_called()
class TestSetupSystemForm:
"""Test system settings form (step 2)."""
@pytest.mark.asyncio
async def test_unauthenticated_redirects_to_operator(self):
"""GET /setup/system without auth redirects to /setup/operator."""
mock_request = MagicMock()
mock_request.state.operator = None
result = await setup_system_form(mock_request)
assert result.status_code == 302
assert result.headers["location"] == "/setup/operator"
@pytest.mark.asyncio
async def test_authenticated_returns_form(self):
"""GET /setup/system with auth returns the form."""
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="admin")
mock_templates = MagicMock()
mock_templates.TemplateResponse.return_value = MagicMock()
mock_conn = AsyncMock()
mock_conn.fetchrow.return_value = {
"map_tile_url": "https://tile.openstreetmap.org/{z}/{x}/{y}.png",
"map_attribution": "&copy; OpenStreetMap contributors",
}
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
result = await setup_system_form(mock_request)
mock_templates.TemplateResponse.assert_called_once()
class TestSetupSystemSubmit:
"""Test system settings submission."""
@pytest.mark.asyncio
async def test_missing_placeholders_shows_error(self):
"""POST without {z},{x},{y} placeholders shows error."""
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="admin")
mock_request.state.csrf_token = "test_csrf_token"
form_data = MagicMock()
form_data.get = lambda k, default="": {
"csrf_token": "test_csrf_token",
"map_tile_url": "https://example.com/tiles",
"map_attribution": "Test",
}.get(k, default)
mock_request.form = AsyncMock(return_value=form_data)
mock_templates = MagicMock()
mock_templates.TemplateResponse.return_value = MagicMock()
mock_conn = AsyncMock()
mock_conn.fetchrow.return_value = {
"map_tile_url": "",
"map_attribution": "",
}
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
result = await setup_system_submit(mock_request)
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
assert "map_tile_url" in context["errors"]
@pytest.mark.asyncio
async def test_valid_updates_and_redirects(self):
"""POST with valid data updates system and redirects to /setup/keys."""
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="admin")
mock_request.state.csrf_token = "test_csrf_token"
form_data = MagicMock()
form_data.get = lambda k, default="": {
"csrf_token": "test_csrf_token",
"map_tile_url": "https://example.com/{z}/{x}/{y}.png",
"map_attribution": "Test Attribution",
}.get(k, default)
mock_request.form = AsyncMock(return_value=form_data)
mock_conn = AsyncMock()
mock_conn.fetchrow.return_value = {
"map_tile_url": "old_url",
"map_attribution": "old_attr",
}
mock_conn.execute = AsyncMock()
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
with patch("central.gui.routes.get_pool", return_value=mock_pool):
with patch("central.gui.routes.write_audit", new_callable=AsyncMock):
result = await setup_system_submit(mock_request)
assert result.status_code == 302
assert result.headers["location"] == "/setup/keys"
class TestSetupKeysForm:
"""Test API keys form (step 3)."""
@pytest.mark.asyncio
async def test_unauthenticated_redirects_to_operator(self):
"""GET /setup/keys without auth redirects to /setup/operator."""
mock_request = MagicMock()
mock_request.state.operator = None
result = await setup_keys_form(mock_request)
assert result.status_code == 302
assert result.headers["location"] == "/setup/operator"
class TestSetupKeysSubmit:
"""Test API keys submission."""
@pytest.mark.asyncio
async def test_next_action_redirects_to_adapters(self):
"""POST with action=next redirects to /setup/adapters."""
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="admin")
mock_request.state.csrf_token = "test_csrf_token"
form_data = MagicMock()
form_data.get = lambda k, default="": {
"csrf_token": "test_csrf_token",
"action": "next",
}.get(k, default)
mock_request.form = AsyncMock(return_value=form_data)
# No need to mock get_pool since action="next" returns before it's called
result = await setup_keys_submit(mock_request)
assert result.status_code == 302
assert result.headers["location"] == "/setup/adapters"
@pytest.mark.asyncio
async def test_add_key_creates_and_rerenders(self):
"""POST with action=add creates key and re-renders with success."""
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="admin")
mock_request.state.csrf_token = "test_csrf_token"
form_data = MagicMock()
form_data.get = lambda k, default="": {
"csrf_token": "test_csrf_token",
"action": "add",
"alias": "testkey",
"plaintext_key": "secret123",
}.get(k, default)
mock_request.form = AsyncMock(return_value=form_data)
mock_templates = MagicMock()
mock_templates.TemplateResponse.return_value = MagicMock()
mock_conn = AsyncMock()
mock_conn.fetchrow.side_effect = [
None, # No existing key
{"created_at": datetime(2026, 5, 18, 12, 0, tzinfo=timezone.utc)},
]
mock_conn.fetch.side_effect = [
[], # First list
[{"alias": "testkey", "created_at": datetime(2026, 5, 18, 12, 0, tzinfo=timezone.utc)}], # After insert
]
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
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.crypto.encrypt", return_value=b"encrypted"):
with patch("central.gui.routes.write_audit", new_callable=AsyncMock):
result = await setup_keys_submit(mock_request)
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
assert context["success"] == "API key 'testkey' added successfully."
class TestSetupAdaptersForm:
"""Test adapters configuration form (step 4)."""
@pytest.mark.asyncio
async def test_unauthenticated_redirects_to_operator(self):
"""GET /setup/adapters without auth redirects to /setup/operator."""
mock_request = MagicMock()
mock_request.state.operator = None
result = await setup_adapters_form(mock_request)
assert result.status_code == 302
assert result.headers["location"] == "/setup/operator"
class TestSetupFinishForm:
"""Test finish page (step 5)."""
@pytest.mark.asyncio
async def test_unauthenticated_redirects_to_operator(self):
"""GET /setup/finish without auth redirects to /setup/operator."""
mock_request = MagicMock()
mock_request.state.operator = None
result = await setup_finish_form(mock_request)
assert result.status_code == 302
assert result.headers["location"] == "/setup/operator"
@pytest.mark.asyncio
async def test_authenticated_shows_summary(self):
"""GET /setup/finish with auth shows summary."""
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="admin")
mock_templates = MagicMock()
mock_templates.TemplateResponse.return_value = MagicMock()
mock_conn = AsyncMock()
mock_conn.fetchval.side_effect = [1, 2] # 1 operator, 2 keys
mock_conn.fetchrow.return_value = {"map_tile_url": "https://example.com/{z}/{x}/{y}.png"}
mock_conn.fetch.return_value = [
{"name": "nws", "enabled": True, "cadence_s": 300},
{"name": "firms", "enabled": False, "cadence_s": 600},
]
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
with patch("central.gui.routes._get_templates", return_value=mock_templates):
with patch("central.gui.routes.get_pool", return_value=mock_pool):
result = await setup_finish_form(mock_request)
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
assert context["operator_count"] == 1
assert context["key_count"] == 2
assert len(context["adapters"]) == 2
class TestSetupFinishSubmit:
"""Test setup completion."""
@pytest.mark.asyncio
async def test_marks_setup_complete_and_redirects(self):
"""POST /setup/finish marks setup_complete=true and redirects to /."""
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="admin")
mock_request.state.csrf_token = "test_csrf_token"
# Mock form with CSRF token
form_data = MagicMock()
form_data.get = lambda k, default="": {"csrf_token": "test_csrf_token"}.get(k, default)
mock_request.form = AsyncMock(return_value=form_data)
mock_conn = AsyncMock()
mock_conn.execute = AsyncMock()
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_pool.acquire.return_value.__aexit__.return_value = None
with patch("central.gui.routes.get_pool", return_value=mock_pool):
with patch("central.gui.routes.write_audit", new_callable=AsyncMock) as mock_audit:
result = await setup_finish_submit(mock_request)
assert result.status_code == 302
assert result.headers["location"] == "/"
mock_conn.execute.assert_called_once()
mock_audit.assert_called_once()
class TestSetupGateMiddlewareWizard:
"""Test SetupGateMiddleware with wizard paths."""
@pytest.mark.asyncio
async def test_allows_setup_operator_when_incomplete(self):
"""SetupGateMiddleware allows /setup/operator when setup_complete=False."""
from starlette.testclient import TestClient
from fastapi import FastAPI
mock_pool = MagicMock()
mock_conn = MagicMock()
mock_conn.fetchrow = AsyncMock(return_value={"setup_complete": False})
mock_conn.__aenter__ = AsyncMock(return_value=mock_conn)
mock_conn.__aexit__ = AsyncMock()
mock_pool.acquire = MagicMock(return_value=mock_conn)
with patch("central.gui.middleware.get_pool", return_value=mock_pool):
app = FastAPI()
@app.get("/setup/operator")
async def setup_operator():
return {"message": "operator form"}
app.add_middleware(SetupGateMiddleware)
client = TestClient(app)
response = client.get("/setup/operator")
assert response.status_code == 200
@pytest.mark.asyncio
async def test_redirects_base_setup_to_wizard_step(self):
"""SetupGateMiddleware redirects /setup to appropriate wizard step."""
from starlette.testclient import TestClient
from fastapi import FastAPI
mock_pool = MagicMock()
mock_conn = MagicMock()
mock_conn.fetchrow = AsyncMock(return_value={"setup_complete": False})
mock_conn.fetchval = AsyncMock(return_value=0) # No operators
mock_conn.__aenter__ = AsyncMock(return_value=mock_conn)
mock_conn.__aexit__ = AsyncMock()
mock_pool.acquire = MagicMock(return_value=mock_conn)
with patch("central.gui.middleware.get_pool", return_value=mock_pool):
app = FastAPI()
@app.get("/setup")
async def setup():
return {"message": "base setup"}
@app.get("/setup/operator")
async def setup_operator():
return {"message": "operator"}
app.add_middleware(SetupGateMiddleware)
client = TestClient(app, follow_redirects=False)
response = client.get("/setup")
assert response.status_code == 302
assert response.headers["location"] == "/setup/operator"
@pytest.mark.asyncio
async def test_redirects_login_to_setup_when_incomplete(self):
"""SetupGateMiddleware redirects /login to /setup when setup_complete=False."""
from starlette.testclient import TestClient
from fastapi import FastAPI
mock_pool = MagicMock()
mock_conn = MagicMock()
mock_conn.fetchrow = AsyncMock(return_value={"setup_complete": False})
mock_conn.__aenter__ = AsyncMock(return_value=mock_conn)
mock_conn.__aexit__ = AsyncMock()
mock_pool.acquire = MagicMock(return_value=mock_conn)
with patch("central.gui.middleware.get_pool", return_value=mock_pool):
app = FastAPI()
@app.get("/login")
async def login():
return {"message": "login"}
@app.get("/setup")
async def setup():
return {"message": "setup"}
app.add_middleware(SetupGateMiddleware)
client = TestClient(app, follow_redirects=False)
response = client.get("/login")
assert response.status_code == 302
assert response.headers["location"] == "/setup"
@pytest.mark.asyncio
async def test_redirects_all_setup_paths_when_complete(self):
"""SetupGateMiddleware redirects /setup/* to / when setup_complete=True."""
from starlette.testclient import TestClient
from fastapi import FastAPI
mock_pool = MagicMock()
mock_conn = MagicMock()
mock_conn.fetchrow = AsyncMock(return_value={"setup_complete": True})
mock_conn.__aenter__ = AsyncMock(return_value=mock_conn)
mock_conn.__aexit__ = AsyncMock()
mock_pool.acquire = MagicMock(return_value=mock_conn)
with patch("central.gui.middleware.get_pool", return_value=mock_pool):
app = FastAPI()
@app.get("/")
async def index():
return {"message": "home"}
@app.get("/setup/operator")
async def setup_operator():
return {"message": "operator"}
app.add_middleware(SetupGateMiddleware)
client = TestClient(app, follow_redirects=False)
response = client.get("/setup/operator")
assert response.status_code == 302
assert response.headers["location"] == "/"