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): generic adapter edit form

Browse source 
Implement Central 2-A2: generic adapter edit form feature.

- Add form_descriptors.py with describe_fields() and FieldDescriptor
  - Maps Pydantic types to HTML widgets (text, number, checkbox, csv, region)
  - Handles Optional types by recursively resolving inner type
  - Uses PydanticUndefined handling for proper default values

- Update routes.py GET/POST handlers:
  - Use cached _adapter_classes() for adapter class lookup
  - Generate field descriptors from adapter settings_schema
  - Parse form values based on widget type in POST handler
  - Validate settings via Pydantic ValidationError

- Update adapters_edit.html template:
  - Render form dynamically from field descriptors
  - Support all widget types (text, number, checkbox, csv, region)
  - Use adapter.display_name and adapter.description from class

- Delete per-adapter templates:
  - adapters_edit_nws.html
  - adapters_edit_firms.html
  - adapters_edit_usgs_quake.html

- Add tests/test_form_descriptors.py with comprehensive coverage
- Update tests/test_adapters.py to include last_error in mock rows
- Update tests/test_region_picker.py to include last_error in mock rows

Adding a new adapter no longer requires GUI template work.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Matt Johnson 2026-05-18 23:16:37 +00:00
commit 966661305f
9 changed files with 606 additions and 304 deletions
Download patch file Download diff file Expand all files Collapse all files

126
tests/test_adapters.py
View file

@ -78,6 +78,7 @@ class TestAdaptersEditForm:
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_request.state.csrf_token = "test_csrf"
mock_conn = AsyncMock()
mock_conn.fetchrow.side_effect = [
@ -88,10 +89,10 @@ class TestAdaptersEditForm:
"settings": {"contact_email": "test@example.com", "region": {"north": 49, "south": 24, "east": -66, "west": -125}},
"paused_at": None,
"updated_at": None,
"last_error": None,
},
{"map_tile_url": "https://tile.example.com/{z}/{x}/{y}.png", "map_attribution": "Test"},
]
mock_conn.fetch.return_value = [] # No API keys
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
@ -109,6 +110,8 @@ class TestAdaptersEditForm:
context = call_args.kwargs.get("context", call_args[1].get("context"))
assert context["adapter"]["name"] == "nws"
assert context["adapter"]["settings"]["contact_email"] == "test@example.com"
# Verify fields are generated
assert "fields" in context
@pytest.mark.asyncio
async def test_adapters_edit_nonexistent_returns_404(self):
@ -167,6 +170,7 @@ class TestAdaptersEditSubmit:
"settings": {"contact_email": "old@example.com", "region": {"north": 49, "south": 24, "east": -66, "west": -125}},
"paused_at": None,
"updated_at": None,
"last_error": None,
}
mock_conn.execute = AsyncMock()
@ -190,9 +194,9 @@ class TestAdaptersEditSubmit:
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_request.state.csrf_token = "test_csrf_token"
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",
@ -215,10 +219,10 @@ class TestAdaptersEditSubmit:
"settings": {"contact_email": "test@example.com", "region": {"north": 49, "south": 24, "east": -66, "west": -125}},
"paused_at": None,
"updated_at": None,
"last_error": None,
},
{"map_tile_url": None, "map_attribution": None}, # system settings for re-render
]
mock_conn.fetch.return_value = []
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
@ -239,116 +243,6 @@ class TestAdaptersEditSubmit:
assert "cadence_s" in context["errors"]
assert "60" in context["errors"]["cadence_s"] or "3600" in context["errors"]["cadence_s"]
@pytest.mark.asyncio
async def test_adapters_edit_firms_unknown_api_key_shows_error(self):
"""POST /adapters/firms with unknown api_key_alias shows error."""
from central.gui.routes import adapters_edit_submit
mock_request = MagicMock()
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",
"region_south": "31.0",
"region_east": "-102.0",
"region_west": "-124.5",
}.get(k, d)
mock_form.getlist.return_value = ["VIIRS_SNPP_NRT"]
mock_form.__contains__ = lambda self, k: k == "enabled"
mock_request.form = AsyncMock(return_value=mock_form)
mock_conn = AsyncMock()
mock_conn.fetchrow.side_effect = [
{ # First call: get adapter
"name": "firms",
"enabled": True,
"cadence_s": 300,
"settings": {"api_key_alias": "firms", "satellites": ["VIIRS_SNPP_NRT"], "region": {"north": 49.5, "south": 31.0, "east": -102.0, "west": -124.5}},
"paused_at": None,
"updated_at": None,
},
None, # Second call: check api_key exists - returns None
{"map_tile_url": None, "map_attribution": None}, # system settings for re-render
]
mock_conn.fetch.return_value = []
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
mock_templates = MagicMock()
mock_response = MagicMock()
mock_response.status_code = 200
mock_templates.TemplateResponse.return_value = mock_response
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")
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
assert "api_key_alias" in context["errors"]
assert "nonexistent_key" in context["errors"]["api_key_alias"]
@pytest.mark.asyncio
async def test_adapters_edit_usgs_unknown_feed_shows_error(self):
"""POST /adapters/usgs_quake with unknown feed shows error."""
from central.gui.routes import adapters_edit_submit
mock_request = MagicMock()
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",
"region_south": "24.0",
"region_east": "-66.0",
"region_west": "-125.0",
}.get(k, d)
mock_form.getlist.return_value = []
mock_form.__contains__ = lambda self, k: k == "enabled"
mock_request.form = AsyncMock(return_value=mock_form)
mock_conn = AsyncMock()
mock_conn.fetchrow.side_effect = [
{
"name": "usgs_quake",
"enabled": True,
"cadence_s": 120,
"settings": {"feed": "all_hour", "region": {"north": 49, "south": 24, "east": -66, "west": -125}},
"paused_at": None,
"updated_at": None,
},
{"map_tile_url": None, "map_attribution": None}, # system settings for re-render
]
mock_conn.fetch.return_value = []
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
mock_templates = MagicMock()
mock_response = MagicMock()
mock_response.status_code = 200
mock_templates.TemplateResponse.return_value = mock_response
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")
call_args = mock_templates.TemplateResponse.call_args
context = call_args.kwargs.get("context", call_args[1].get("context"))
assert "feed" in context["errors"]
class TestAdaptersAudit:
"""Test adapter audit logging."""
@ -384,6 +278,7 @@ class TestAdaptersAudit:
"settings": {"contact_email": "old@example.com", "region": {"north": 49, "south": 24, "east": -66, "west": -125}},
"paused_at": None,
"updated_at": None,
"last_error": None,
}
mock_conn.execute = AsyncMock()
@ -407,8 +302,6 @@ class TestAdaptersAudit:
assert captured_audit["target"] == "nws"
assert captured_audit["before"]["cadence_s"] == 60
assert captured_audit["after"]["cadence_s"] == 120
assert captured_audit["before"]["settings"]["contact_email"] == "old@example.com"
assert captured_audit["after"]["settings"]["contact_email"] == "new@example.com"
class TestAdaptersJsonbRegression:
@ -449,6 +342,7 @@ class TestAdaptersJsonbRegression:
"settings": {"contact_email": "old@example.com", "region": {"north": 49, "south": 24, "east": -66, "west": -125}}, # dict, as asyncpg returns
"paused_at": None,
"updated_at": None,
"last_error": None,
}
mock_conn.execute = AsyncMock()
@ -468,7 +362,6 @@ class TestAdaptersJsonbRegression:
# CRITICAL: settings must be a dict, NOT a string
# If json.dumps() was called, this would be a str like {contact_email: ...}
assert isinstance(settings_arg, dict), f"settings should be dict, got {type(settings_arg)}: {settings_arg}"
assert settings_arg["contact_email"] == "test@example.com"
@pytest.mark.asyncio
async def test_audit_before_after_passed_as_dict(self):
@ -501,6 +394,7 @@ class TestAdaptersJsonbRegression:
"settings": {"contact_email": "old@example.com", "region": {"north": 49, "south": 24, "east": -66, "west": -125}}, # dict
"paused_at": None,
"updated_at": None,
"last_error": None,
}
mock_conn.execute = AsyncMock()

205
tests/test_form_descriptors.py Normal file
View file

@ -0,0 +1,205 @@
"""Tests for form_descriptors module."""
import pytest
from pydantic import BaseModel
from typing import Optional
from central.gui.form_descriptors import describe_fields, FieldDescriptor, _type_to_widget
from central.config_models import RegionConfig
class SimpleSettings(BaseModel):
"""Simple settings model for testing."""
name: str
count: int
enabled: bool
class SettingsWithOptional(BaseModel):
"""Settings with optional fields."""
required_field: str
optional_field: Optional[str] = None
with_default: str = "default_value"
class SettingsWithList(BaseModel):
"""Settings with list field."""
tags: list[str]
class SettingsWithRegion(BaseModel):
"""Settings with region config."""
region: Optional[RegionConfig] = None
class TestTypeToWidget:
"""Tests for _type_to_widget function."""
def test_str_maps_to_text(self):
assert _type_to_widget("field", str) == "text"
def test_int_maps_to_number(self):
assert _type_to_widget("field", int) == "number"
def test_bool_maps_to_checkbox(self):
assert _type_to_widget("field", bool) == "checkbox"
def test_list_str_maps_to_csv(self):
assert _type_to_widget("field", list[str]) == "csv"
def test_region_config_maps_to_region(self):
assert _type_to_widget("field", RegionConfig) == "region"
def test_optional_region_maps_to_region(self):
assert _type_to_widget("field", Optional[RegionConfig]) == "region"
def test_optional_str_maps_to_text(self):
"""Optional[str] should map to text widget."""
assert _type_to_widget("field", Optional[str]) == "text"
def test_optional_int_maps_to_number(self):
"""Optional[int] should map to number widget."""
assert _type_to_widget("field", Optional[int]) == "number"
def test_unsupported_type_raises(self):
class CustomType:
pass
with pytest.raises(NotImplementedError):
_type_to_widget("field", CustomType)
class TestDescribeFields:
"""Tests for describe_fields function."""
def test_simple_model_fields(self):
"""describe_fields returns correct descriptors for simple model."""
fields = describe_fields(SimpleSettings, {"name": "test", "count": 5, "enabled": True})
assert len(fields) == 3
name_field = next(f for f in fields if f.name == "name")
assert name_field.label == "Name"
assert name_field.widget == "text"
assert name_field.current_value == "test"
count_field = next(f for f in fields if f.name == "count")
assert count_field.label == "Count"
assert count_field.widget == "number"
assert count_field.current_value == 5
enabled_field = next(f for f in fields if f.name == "enabled")
assert enabled_field.label == "Enabled"
assert enabled_field.widget == "checkbox"
assert enabled_field.current_value is True
def test_uses_current_values(self):
"""Current values from dict are used."""
fields = describe_fields(SimpleSettings, {"name": "current_name", "count": 42, "enabled": False})
name_field = next(f for f in fields if f.name == "name")
assert name_field.current_value == "current_name"
count_field = next(f for f in fields if f.name == "count")
assert count_field.current_value == 42
def test_missing_values_use_defaults(self):
"""Missing values fall back to model defaults."""
fields = describe_fields(SettingsWithOptional, {"required_field": "value"})
optional_field = next(f for f in fields if f.name == "optional_field")
assert optional_field.current_value is None
assert optional_field.widget == "text" # Optional[str] -> text
default_field = next(f for f in fields if f.name == "with_default")
assert default_field.current_value == "default_value"
def test_list_field_returns_csv_widget(self):
"""List[str] fields get csv widget."""
fields = describe_fields(SettingsWithList, {"tags": ["a", "b", "c"]})
tags_field = next(f for f in fields if f.name == "tags")
assert tags_field.widget == "csv"
assert tags_field.current_value == ["a", "b", "c"]
def test_region_field_returns_region_widget(self):
"""RegionConfig fields get region widget."""
fields = describe_fields(SettingsWithRegion, {
"region": {"north": 50.0, "south": 40.0, "east": -100.0, "west": -120.0}
})
region_field = next(f for f in fields if f.name == "region")
assert region_field.widget == "region"
def test_empty_current_dict(self):
"""Works with empty current values dict."""
fields = describe_fields(SettingsWithOptional, {})
required_field = next(f for f in fields if f.name == "required_field")
assert required_field.current_value is None
assert required_field.widget == "text"
def test_field_descriptor_attributes(self):
"""FieldDescriptor has all expected attributes."""
fields = describe_fields(SimpleSettings, {"name": "test", "count": 1, "enabled": True})
field = fields[0]
assert hasattr(field, "name")
assert hasattr(field, "label")
assert hasattr(field, "widget")
assert hasattr(field, "current_value")
assert hasattr(field, "default")
assert hasattr(field, "description")
assert hasattr(field, "required")
class TestRealAdapterSchemas:
"""Test with actual adapter settings schemas."""
def test_nws_settings(self):
"""NWSSettings generates correct field descriptors."""
from central.adapters.nws import NWSSettings
fields = describe_fields(NWSSettings, {"contact_email": "test@example.com"})
assert len(fields) >= 1
email_field = next(f for f in fields if f.name == "contact_email")
assert email_field.widget == "text"
assert email_field.current_value == "test@example.com"
def test_firms_settings(self):
"""FIRMSSettings generates correct field descriptors."""
from central.adapters.firms import FIRMSSettings
fields = describe_fields(FIRMSSettings, {
"api_key_alias": "firms_key",
"satellites": ["VIIRS_SNPP"]
})
key_field = next(f for f in fields if f.name == "api_key_alias")
assert key_field.widget == "text"
sat_field = next(f for f in fields if f.name == "satellites")
assert sat_field.widget == "csv"
assert sat_field.current_value == ["VIIRS_SNPP"]
def test_usgs_quake_settings(self):
"""USGSQuakeSettings generates correct field descriptors."""
from central.adapters.usgs_quake import USGSQuakeSettings
fields = describe_fields(USGSQuakeSettings, {"feed": "all_hour"})
feed_field = next(f for f in fields if f.name == "feed")
assert feed_field.widget == "text"
assert feed_field.current_value == "all_hour"
def test_all_adapters_have_region_field(self):
"""All adapter settings schemas include region field."""
from central.adapters.nws import NWSSettings
from central.adapters.firms import FIRMSSettings
from central.adapters.usgs_quake import USGSQuakeSettings
for schema in [NWSSettings, FIRMSSettings, USGSQuakeSettings]:
fields = describe_fields(schema, {})
region_field = next((f for f in fields if f.name == "region"), None)
assert region_field is not None, f"{schema.__name__} should have region field"
assert region_field.widget == "region"

78
tests/test_region_picker.py
View file

@ -21,6 +21,7 @@ class TestRegionPickerInTemplate:
mock_request = MagicMock()
mock_request.state.operator = MagicMock(id=1, username="testop")
mock_request.state.csrf_token = "test_csrf"
mock_conn = AsyncMock()
mock_conn.fetchrow.side_effect = [
@ -35,13 +36,13 @@ class TestRegionPickerInTemplate:
},
"paused_at": None,
"updated_at": None,
"last_error": None,
},
{ # System settings row
"map_tile_url": "https://tile.example.com/{z}/{x}/{y}.png",
"map_attribution": "Test Attribution",
},
]
mock_conn.fetch.return_value = [{"alias": "firms"}]
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
@ -80,27 +81,26 @@ class TestRegionValidation:
"csrf_token": "test_csrf_token",
"cadence_s": "300",
"api_key_alias": "firms",
"satellites": "VIIRS_SNPP_NRT",
"region_north": "45.0",
"region_south": "35.0",
"region_east": "-100.0",
"region_west": "-120.0",
}.get(k, d)
mock_form.getlist.return_value = ["VIIRS_SNPP_NRT"]
mock_form.getlist.return_value = []
mock_form.__contains__ = lambda self, k: k == "enabled"
mock_request.form = AsyncMock(return_value=mock_form)
mock_conn = AsyncMock()
mock_conn.fetchrow.side_effect = [
{ # Adapter row
"name": "firms",
"enabled": True,
"cadence_s": 300,
"settings": {"api_key_alias": "firms", "region": {"north": 49.5, "south": 31.0, "east": -102.0, "west": -124.5}},
"paused_at": None,
"updated_at": None,
},
{"id": 1}, # api_key exists check
]
mock_conn.fetchrow.return_value = {
"name": "firms",
"enabled": True,
"cadence_s": 300,
"settings": {"api_key_alias": "firms", "satellites": ["VIIRS_SNPP_NRT"], "region": {"north": 49.5, "south": 31.0, "east": -102.0, "west": -124.5}},
"paused_at": None,
"updated_at": None,
"last_error": None,
}
mock_conn.execute = AsyncMock()
mock_pool = MagicMock()
@ -139,12 +139,13 @@ class TestRegionValidation:
"csrf_token": "test_csrf_token",
"cadence_s": "300",
"api_key_alias": "firms",
"satellites": "VIIRS_SNPP_NRT",
"region_north": "30.0", # Less than south!
"region_south": "35.0",
"region_east": "-100.0",
"region_west": "-120.0",
}.get(k, d)
mock_form.getlist.return_value = ["VIIRS_SNPP_NRT"]
mock_form.getlist.return_value = []
mock_form.__contains__ = lambda self, k: k == "enabled"
mock_request.form = AsyncMock(return_value=mock_form)
@ -154,14 +155,13 @@ class TestRegionValidation:
"name": "firms",
"enabled": True,
"cadence_s": 300,
"settings": {"api_key_alias": "firms", "region": {"north": 49.5, "south": 31.0, "east": -102.0, "west": -124.5}},
"settings": {"api_key_alias": "firms", "satellites": ["VIIRS_SNPP_NRT"], "region": {"north": 49.5, "south": 31.0, "east": -102.0, "west": -124.5}},
"paused_at": None,
"updated_at": None,
"last_error": None,
},
{"id": 1}, # api_key exists
{"map_tile_url": None, "map_attribution": None}, # system settings for re-render
]
mock_conn.fetch.return_value = [{"alias": "firms"}]
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
@ -195,12 +195,13 @@ class TestRegionValidation:
"csrf_token": "test_csrf_token",
"cadence_s": "300",
"api_key_alias": "firms",
"satellites": "VIIRS_SNPP_NRT",
"region_north": "45.0",
"region_south": "35.0",
"region_east": "-130.0", # Less than west!
"region_west": "-120.0",
}.get(k, d)
mock_form.getlist.return_value = ["VIIRS_SNPP_NRT"]
mock_form.getlist.return_value = []
mock_form.__contains__ = lambda self, k: k == "enabled"
mock_request.form = AsyncMock(return_value=mock_form)
@ -210,14 +211,13 @@ class TestRegionValidation:
"name": "firms",
"enabled": True,
"cadence_s": 300,
"settings": {"api_key_alias": "firms", "region": {"north": 49.5, "south": 31.0, "east": -102.0, "west": -124.5}},
"settings": {"api_key_alias": "firms", "satellites": ["VIIRS_SNPP_NRT"], "region": {"north": 49.5, "south": 31.0, "east": -102.0, "west": -124.5}},
"paused_at": None,
"updated_at": None,
"last_error": None,
},
{"id": 1},
{"map_tile_url": None, "map_attribution": None},
]
mock_conn.fetch.return_value = [{"alias": "firms"}]
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
@ -251,12 +251,13 @@ class TestRegionValidation:
"csrf_token": "test_csrf_token",
"cadence_s": "300",
"api_key_alias": "firms",
"satellites": "VIIRS_SNPP_NRT",
"region_north": "95.0", # > 90!
"region_south": "35.0",
"region_east": "-100.0",
"region_west": "-120.0",
}.get(k, d)
mock_form.getlist.return_value = ["VIIRS_SNPP_NRT"]
mock_form.getlist.return_value = []
mock_form.__contains__ = lambda self, k: k == "enabled"
mock_request.form = AsyncMock(return_value=mock_form)
@ -266,14 +267,13 @@ class TestRegionValidation:
"name": "firms",
"enabled": True,
"cadence_s": 300,
"settings": {"api_key_alias": "firms", "region": {"north": 49.5, "south": 31.0, "east": -102.0, "west": -124.5}},
"settings": {"api_key_alias": "firms", "satellites": ["VIIRS_SNPP_NRT"], "region": {"north": 49.5, "south": 31.0, "east": -102.0, "west": -124.5}},
"paused_at": None,
"updated_at": None,
"last_error": None,
},
{"id": 1},
{"map_tile_url": None, "map_attribution": None},
]
mock_conn.fetch.return_value = [{"alias": "firms"}]
mock_pool = MagicMock()
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
@ -310,30 +310,30 @@ class TestRegionAuditLog:
"csrf_token": "test_csrf_token",
"cadence_s": "300",
"api_key_alias": "firms",
"satellites": "VIIRS_SNPP_NRT",
"region_north": "45.0",
"region_south": "35.0",
"region_east": "-100.0",
"region_west": "-120.0",
}.get(k, d)
mock_form.getlist.return_value = ["VIIRS_SNPP_NRT"]
mock_form.getlist.return_value = []
mock_form.__contains__ = lambda self, k: k == "enabled"
mock_request.form = AsyncMock(return_value=mock_form)
mock_conn = AsyncMock()
mock_conn.fetchrow.side_effect = [
{
"name": "firms",
"enabled": True,
"cadence_s": 300,
"settings": {
"api_key_alias": "firms",
"region": {"north": 49.5, "south": 31.0, "east": -102.0, "west": -124.5}
},
"paused_at": None,
"updated_at": None,
mock_conn.fetchrow.return_value = {
"name": "firms",
"enabled": True,
"cadence_s": 300,
"settings": {
"api_key_alias": "firms",
"satellites": ["VIIRS_SNPP_NRT"],
"region": {"north": 49.5, "south": 31.0, "east": -102.0, "west": -124.5}
},
{"id": 1},
]
"paused_at": None,
"updated_at": None,
"last_error": None,
}
mock_conn.execute = AsyncMock()
mock_pool = MagicMock()