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

feat: GUI create/delete for adapter instances (v0.15.0 PR3)

Browse source 
Completes the CRUD loop for operator-creatable adapters.  Operators can
now create generic_http instances (GET+POST /adapters/new) and delete
operator instances (POST /adapters/{name}/delete) without editing Python.

Key changes:
- SourceAdapter.operator_creatable class attr (default False); set True on
  GenericHttpAdapter so it appears in the kind select
- ADAPTER_CREATE / ADAPTER_DELETE audit constants
- _parse_adapter_settings() shared helper extracted from adapters_edit_submit
  (single place for widget parsing, region handling, Pydantic validation,
  quota blocking) — edit_submit refactored to call it
- GET/POST /adapters/new: kind select → name/cadence/enabled → settings fields;
  INSERT with kind supplied explicitly; 302 → edit page on success
- POST /adapters/{name}/delete: primary guard (name in adapter_classes → 403);
  second guard (kind not operator_creatable → 403); DELETE + audit; 302 → list
- adapters_list.html: "New adapter" button; per-row Delete form (operator
  instances only; built-ins have no button); deletable flag passed from route
- adapters_new.html: new template reusing same field-rendering blocks as
  adapters_edit.html; enabled checkbox unchecked by default
- 46 new tests (all pass; mock-DB pattern mirrors test_gui_adapter_edit.py)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Ubuntu 2026-06-30 17:28:32 +00:00
commit a068df8f5d
7 changed files with 1249 additions and 137 deletions
Download patch file Download diff file Expand all files Collapse all files

6
src/central/adapter.py
View file

@ -66,6 +66,12 @@ class SourceAdapter(ABC):
set in ``central.archive`` -- the two MUST stay in sync (enforced by
``tests/test_bypass_bbox_consistency.py``)."""
operator_creatable: bool = False
"""True for adapters that operators may instantiate many times from the
GUI (one config.adapters row per instance, each with a unique name and
kind=<class_name>). False (default) for singleton built-ins where the
name equals the kind and the row is seeded by migrations."""
@abstractmethod
async def poll(self) -> AsyncIterator[Event]:
"""

1
src/central/adapters/generic_http.py
View file

@ -193,6 +193,7 @@ class GenericHttpAdapter(SourceAdapter):
wizard_order = None # not in the setup wizard; created via operator GUI
enrichment_locations = []
bypass_bbox_filter = False
operator_creatable = True # GUI allows operators to create multiple instances
def __init__(
self,

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

@ -9,7 +9,9 @@ AUTH_LOGIN_FAILED = "auth.login_failed"
AUTH_LOGOUT = "auth.logout"
AUTH_PASSWORD_CHANGE = "auth.password_change"
OPERATOR_CREATE = "operator.create"
ADAPTER_CREATE = "adapter.create"
ADAPTER_UPDATE = "adapter.update"
ADAPTER_DELETE = "adapter.delete"
STREAM_UPDATE = "stream.update"
API_KEY_CREATE = "api_key.create"
API_KEY_ROTATE = "api_key.rotate"

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

@ -36,6 +36,8 @@ from central.gui.auth import (
verify_password,
)
from central.gui.audit import (
ADAPTER_CREATE,
ADAPTER_DELETE,
ADAPTER_UPDATE,
API_KEY_CREATE,
API_KEY_DELETE,
@ -100,6 +102,10 @@ ALIAS_REGEX = re.compile(r"^[a-zA-Z0-9_]+$")
# Email validation regex (simple but effective)
EMAIL_REGEX = re.compile(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$")
# Adapter instance-name regex: must start with a lowercase letter, followed by
# 1–63 lowercase letters, digits, or underscores (total 2–64 chars).
ADAPTER_NAME_REGEX = re.compile(r"^[a-z][a-z0-9_]{1,63}$")
def _get_templates():
"""Get templates instance (deferred import to avoid circular)."""
@ -1437,6 +1443,164 @@ async def change_password_submit(
# =============================================================================
def _parse_adapter_settings(
form,
adapter_cls,
current_settings: dict,
cadence_s: int,
) -> tuple[dict, dict[str, str]]:
"""Parse and validate adapter settings from a form submission.
Shared by ``adapters_edit_submit`` and ``adapters_create_submit`` so that
field parsing, region handling, Pydantic validation, and quota-blocking live
in exactly one place.
Args:
form: Starlette ``FormData`` from ``await request.form()``.
adapter_cls: The resolved ``SourceAdapter`` subclass, or ``None``.
current_settings: Existing settings dict (``{}`` for a new adapter).
Used by ``describe_fields`` to populate ``field.current_value``.
cadence_s: Validated cadence for quota estimation.
Returns:
``(new_settings, errors)`` — on success ``errors`` is empty and
``new_settings`` is the Pydantic-validated dict; on failure
``new_settings`` is ``{}`` and ``errors`` maps field names to messages.
"""
errors: dict[str, str] = {}
if not (adapter_cls and hasattr(adapter_cls, "settings_schema")):
# No schema — preserve existing settings unchanged.
return dict(current_settings), errors
schema = adapter_cls.settings_schema
fields = describe_fields(schema, current_settings)
parsed_values: dict = {}
for field in fields:
raw = form.get(field.name, "")
if field.widget == "text":
parsed_values[field.name] = raw.strip() if raw else None
elif field.widget == "number":
try:
parsed_values[field.name] = int(raw) if raw else None
except ValueError:
errors[field.name] = f"{field.label} must be a number"
elif field.widget == "checkbox":
parsed_values[field.name] = field.name in form
elif field.widget == "csv":
if raw.strip():
parsed_values[field.name] = [v.strip() for v in raw.split(",") if v.strip()]
else:
parsed_values[field.name] = []
elif field.widget == "csv_int":
parsed_ints: list[int] = []
if raw.strip():
for tok in raw.split(","):
tok = tok.strip()
if not tok:
continue
try:
parsed_ints.append(int(tok))
except ValueError:
logger.warning(
"csv_int: dropped non-numeric token",
extra={"field": field.name, "token": tok},
)
parsed_values[field.name] = parsed_ints
elif field.widget == "select":
value = raw.strip() if raw else None
if value and field.options and value not in field.options:
errors[field.name] = f"Invalid {field.label.lower()}"
else:
parsed_values[field.name] = value
elif field.widget == "checkboxes":
values = form.getlist(field.name)
if field.options:
invalid = [v for v in values if v not in field.options]
if invalid:
errors[field.name] = f"Invalid values: {', '.join(invalid)}"
else:
parsed_values[field.name] = values
else:
parsed_values[field.name] = values
elif field.widget == "api_key_select":
value = raw.strip() if raw else None
parsed_values[field.name] = value
elif field.widget == "model_list":
rows = _parse_model_list(form, field)
parsed_values[field.name] = rows
elif field.widget == "region":
pass # handled in the region block below
# Region fields (common to adapters that expose a bounding-box region).
region_north_str = form.get("region_north", "").strip()
region_south_str = form.get("region_south", "").strip()
region_east_str = form.get("region_east", "").strip()
region_west_str = form.get("region_west", "").strip()
has_region = any([region_north_str, region_south_str, region_east_str, region_west_str])
if has_region:
try:
region_north = float(region_north_str)
region_south = float(region_south_str)
region_east = float(region_east_str)
region_west = float(region_west_str)
if not (-90 <= region_south < region_north <= 90):
errors["region"] = (
"Invalid latitude: south must be less than north, "
"both between -90 and 90"
)
elif not (-180 <= region_west < region_east <= 180):
errors["region"] = (
"Invalid longitude: west must be less than east, "
"both between -180 and 180"
)
else:
parsed_values["region"] = {
"north": region_north,
"south": region_south,
"east": region_east,
"west": region_west,
}
except ValueError:
errors["region"] = "Region coordinates must be valid numbers"
else:
parsed_values["region"] = None
if errors:
return {}, errors
# Pydantic validation + quota check.
try:
validated_data = {k: v for k, v in parsed_values.items() if v is not None}
validated = schema(**validated_data)
new_settings = validated.model_dump(mode="json")
q = adapter_cls.quota_estimate(validated, cadence_s)
if q and q.get("blocked"):
ml = next((f.name for f in fields if f.widget == "model_list"), "quota")
errors[ml] = (
f"Estimated {q['calls_per_month']:,} calls/month exceeds the "
f"{q['cap']:,}/month free-tier cap — raise cadence or remove rows."
)
return {}, errors
except ValidationError as e:
ml_name = next((f.name for f in fields if f.widget == "model_list"), None)
for err in e.errors():
loc = err["loc"]
key = str(loc[0]) if loc else (ml_name or "unknown")
if len(loc) >= 2 and isinstance(loc[1], int):
errors[key] = f"Row {loc[1] + 1}: {err['msg']}"
else:
errors[key] = err["msg"]
return {}, errors
return new_settings, errors
@router.get("/adapters", response_class=HTMLResponse)
async def adapters_list(
request: Request,
@ -1471,6 +1635,10 @@ async def adapters_list(
)
api_key_missing = not has_key
# Operator instances have a name that is NOT a registered kind key.
# Built-ins always have name == kind which IS in the registry.
deletable = row["name"] not in adapter_classes
adapters.append({
"name": row["name"],
"display_name": getattr(adapter_cls, "display_name", row["name"]) if adapter_cls else row["name"],
@ -1482,6 +1650,7 @@ async def adapters_list(
"last_error": row["last_error"],
"api_key_missing": api_key_missing,
"requires_api_key_alias": requires_api_key_alias,
"deletable": deletable,
})
csrf_token = request.state.csrf_token
@ -1530,6 +1699,202 @@ def _parse_model_list(form, field) -> list[dict]:
return out
@router.get("/adapters/new", response_class=HTMLResponse)
async def adapters_create_form(request: Request) -> Response:
"""Render the create-adapter form.
Lists only adapter kinds where ``operator_creatable is True`` so operators
can instantiate them freely without touching Python code.
"""
templates = _get_templates()
pool = get_pool()
operator = request.state.operator
csrf_token = request.state.csrf_token
adapter_classes = _adapter_classes()
creatable_kinds = {
kind: cls
for kind, cls in adapter_classes.items()
if getattr(cls, "operator_creatable", False)
}
if not creatable_kinds:
return Response(status_code=404, content="No operator-creatable adapter kinds are registered.")
# NOTE: single creatable kind today; multi-kind HTMX field-swap is a future enhancement.
first_kind, first_cls = next(iter(creatable_kinds.items()))
fields = []
if hasattr(first_cls, "settings_schema"):
fields = describe_fields(first_cls.settings_schema, {})
if first_cls.api_key_field is not None:
for f in fields:
if f.name == first_cls.api_key_field:
f.widget = "api_key_select"
async with pool.acquire() as conn:
api_key_rows = await conn.fetch("SELECT alias FROM config.api_keys ORDER BY alias")
api_keys = [{"alias": r["alias"]} for r in api_key_rows]
return templates.TemplateResponse(
request=request,
name="adapters_new.html",
context={
"operator": operator,
"csrf_token": csrf_token,
"creatable_kinds": [
{"kind": kind, "display_name": getattr(cls, "display_name", kind)}
for kind, cls in creatable_kinds.items()
],
"selected_kind": first_kind,
"default_cadence_s": first_cls.default_cadence_s,
"fields": fields,
"api_keys": api_keys,
"errors": None,
"form_data": None,
},
)
@router.post("/adapters/new")
async def adapters_create_submit(request: Request) -> Response:
"""Process the create-adapter form (first INSERT in the codebase)."""
templates = _get_templates()
pool = get_pool()
operator = request.state.operator
form = await request.form()
form_csrf = form.get("csrf_token", "")
if not form_csrf or form_csrf != request.state.csrf_token:
raise CsrfValidationError("Invalid CSRF token")
adapter_classes = _adapter_classes()
creatable_kinds = {
kind: cls
for kind, cls in adapter_classes.items()
if getattr(cls, "operator_creatable", False)
}
kind = (form.get("kind") or "").strip()
name = (form.get("name") or "").strip()
enabled = "enabled" in form
cadence_s_str = form.get("cadence_s", "")
errors: dict[str, str] = {}
form_data: dict[str, Any] = {
"kind": kind,
"name": name,
"enabled": enabled,
"cadence_s": cadence_s_str,
}
# Validate kind — must be operator-creatable.
kind_cls = creatable_kinds.get(kind)
if kind not in creatable_kinds:
errors["kind"] = f"'{kind}' is not a valid operator-creatable adapter kind."
# Validate instance name.
if "kind" not in errors:
if not ADAPTER_NAME_REGEX.match(name):
errors["name"] = (
"Name must start with a lowercase letter followed by 1–63 "
"lowercase letters, digits, or underscores."
)
elif name in adapter_classes:
errors["name"] = (
f"'{name}' is a reserved kind name and cannot be used as an "
"instance name."
)
# Validate cadence_s.
cadence_s = 0
try:
cadence_s = int(cadence_s_str)
if cadence_s < 10:
errors["cadence_s"] = "Input should be greater than or equal to 10"
except ValueError:
errors["cadence_s"] = "Cadence must be a valid integer"
# Check for duplicate name in DB (only when name passed format + kind checks).
if "name" not in errors and "kind" not in errors:
async with pool.acquire() as conn:
existing = await conn.fetchval(
"SELECT 1 FROM config.adapters WHERE name = $1", name
)
if existing:
return Response(
status_code=409,
content=f"An adapter named '{name}' already exists.",
)
# Parse + validate settings via the shared helper.
new_settings: dict = {}
if not errors and kind_cls:
new_settings, settings_errors = _parse_adapter_settings(
form, kind_cls, {}, cadence_s
)
errors.update(settings_errors)
# Re-render on error.
if errors:
fields = []
if kind_cls and hasattr(kind_cls, "settings_schema"):
fields = describe_fields(kind_cls.settings_schema, {})
if kind_cls.api_key_field is not None:
for f in fields:
if f.name == kind_cls.api_key_field:
f.widget = "api_key_select"
# Populate form_data for settings fields so inputs restore values.
for field in fields:
form_data.setdefault(field.name, form.get(field.name, ""))
async with pool.acquire() as conn:
api_key_rows = await conn.fetch("SELECT alias FROM config.api_keys ORDER BY alias")
api_keys = [{"alias": r["alias"]} for r in api_key_rows]
selected_kind = kind if kind in creatable_kinds else (next(iter(creatable_kinds)) if creatable_kinds else "")
return templates.TemplateResponse(
request=request,
name="adapters_new.html",
context={
"operator": operator,
"csrf_token": request.state.csrf_token,
"creatable_kinds": [
{"kind": k, "display_name": getattr(c, "display_name", k)}
for k, c in creatable_kinds.items()
],
"selected_kind": selected_kind,
"default_cadence_s": getattr(kind_cls, "default_cadence_s", 300) if kind_cls else 300,
"fields": fields,
"api_keys": api_keys,
"errors": errors,
"form_data": form_data,
},
status_code=422,
)
# INSERT INTO config.adapters — kind supplied explicitly (migration 043 has no DEFAULT).
async with pool.acquire() as conn:
await conn.execute(
"""
INSERT INTO config.adapters (name, kind, enabled, cadence_s, settings, updated_at)
VALUES ($1, $2, $3, $4, $5, now())
""",
name,
kind,
enabled,
cadence_s,
new_settings,
)
await write_audit(
conn,
ADAPTER_CREATE,
operator_id=operator.id,
target=name,
after={"kind": kind, "enabled": enabled, "cadence_s": cadence_s, "settings": new_settings},
)
return RedirectResponse(url=f"/adapters/{name}", status_code=302)
@router.get("/adapters/{name}", response_class=HTMLResponse)
async def adapters_edit_form(
request: Request,
@ -1715,144 +2080,29 @@ async def adapters_edit_submit(
current_settings = row["settings"] or {}
# Parse and validate settings via Pydantic if we have the adapter class
new_settings = {}
# Collect raw form values into form_data for error re-renders.
if adapter_cls and hasattr(adapter_cls, "settings_schema"):
schema = adapter_cls.settings_schema
fields = describe_fields(schema, current_settings)
for _f in describe_fields(adapter_cls.settings_schema, current_settings):
if _f.widget == "checkboxes":
form_data[_f.name] = form.getlist(_f.name)
elif _f.widget == "model_list":
form_data[_f.name] = _parse_model_list(form, _f)
else:
form_data[_f.name] = form.get(_f.name, "")
form_data["region_north"] = form.get("region_north", "").strip()
form_data["region_south"] = form.get("region_south", "").strip()
form_data["region_east"] = form.get("region_east", "").strip()
form_data["region_west"] = form.get("region_west", "").strip()
# Parse form values based on widget type
parsed_values = {}
for field in fields:
raw = form.get(field.name, "")
form_data[field.name] = raw
if field.widget == "text":
parsed_values[field.name] = raw.strip() if raw else None
elif field.widget == "number":
try:
parsed_values[field.name] = int(raw) if raw else None
except ValueError:
errors[field.name] = f"{field.label} must be a number"
elif field.widget == "checkbox":
parsed_values[field.name] = field.name in form
elif field.widget == "csv":
if raw.strip():
parsed_values[field.name] = [v.strip() for v in raw.split(",") if v.strip()]
else:
parsed_values[field.name] = []
elif field.widget == "csv_int":
# v0.11.3: parallel to "csv" but coerces each token through
# int(), dropping non-numeric entries with a warning.
parsed_ints: list[int] = []
if raw.strip():
for tok in raw.split(","):
tok = tok.strip()
if not tok:
continue
try:
parsed_ints.append(int(tok))
except ValueError:
logger.warning(
"csv_int: dropped non-numeric token",
extra={"field": field.name, "token": tok},
)
parsed_values[field.name] = parsed_ints
elif field.widget == "select":
value = raw.strip() if raw else None
if value and field.options and value not in field.options:
errors[field.name] = f"Invalid {field.label.lower()}"
else:
parsed_values[field.name] = value
elif field.widget == "checkboxes":
# Use getlist for checkbox groups
values = form.getlist(field.name)
form_data[field.name] = values # Override raw value
if field.options:
invalid = [v for v in values if v not in field.options]
if invalid:
errors[field.name] = f"Invalid values: {', '.join(invalid)}"
else:
parsed_values[field.name] = values
else:
parsed_values[field.name] = values
elif field.widget == "api_key_select":
# API key select - validate against existing keys
value = raw.strip() if raw else None
parsed_values[field.name] = value
elif field.widget == "model_list":
rows = _parse_model_list(form, field)
form_data[field.name] = rows
parsed_values[field.name] = rows
elif field.widget == "region":
# Region handled separately below
pass
# Handle region fields (common pattern)
region_north_str = form.get("region_north", "").strip()
region_south_str = form.get("region_south", "").strip()
region_east_str = form.get("region_east", "").strip()
region_west_str = form.get("region_west", "").strip()
form_data["region_north"] = region_north_str
form_data["region_south"] = region_south_str
form_data["region_east"] = region_east_str
form_data["region_west"] = region_west_str
# Check if any region field has a value
has_region = any([region_north_str, region_south_str, region_east_str, region_west_str])
if has_region:
try:
region_north = float(region_north_str)
region_south = float(region_south_str)
region_east = float(region_east_str)
region_west = float(region_west_str)
if not (-90 <= region_south < region_north <= 90):
errors["region"] = "Invalid latitude: south must be less than north, both between -90 and 90"
elif not (-180 <= region_west < region_east <= 180):
errors["region"] = "Invalid longitude: west must be less than east, both between -180 and 180"
else:
parsed_values["region"] = {
"north": region_north,
"south": region_south,
"east": region_east,
"west": region_west,
}
except ValueError:
errors["region"] = "Region coordinates must be valid numbers"
else:
parsed_values["region"] = None
# Only validate with Pydantic if no parse errors
if not errors:
try:
# Filter out None values for optional fields without defaults
validated_data = {k: v for k, v in parsed_values.items() if v is not None}
validated = schema(**validated_data)
new_settings = validated.model_dump(mode="json")
# Hard-block a save that would blow the provider free tier.
q = adapter_cls.quota_estimate(validated, cadence_s)
if q and q.get("blocked"):
ml = next((f.name for f in fields if f.widget == "model_list"), "quota")
errors[ml] = (
f"Estimated {q['calls_per_month']:,} calls/month exceeds the "
f"{q['cap']:,}/month free-tier cap — raise cadence or remove rows."
)
except ValidationError as e:
ml_name = next((f.name for f in fields if f.widget == "model_list"), None)
for err in e.errors():
loc = err["loc"]
key = str(loc[0]) if loc else (ml_name or "unknown")
if len(loc) >= 2 and isinstance(loc[1], int):
errors[key] = f"Row {loc[1] + 1}: {err['msg']}"
else:
errors[key] = err["msg"]
else:
# No schema - just preserve existing settings
new_settings = dict(current_settings)
# Parse + validate settings via the shared helper.
# Mirror the old behavior: skip Pydantic validation when upstream
# checks (e.g. cadence) already failed, same as the old
# "if not errors: try: validated = schema(...)" guard.
if not errors:
new_settings, settings_errors = _parse_adapter_settings(
form, adapter_cls, current_settings, cadence_s
)
errors.update(settings_errors)
# If there are errors, re-render the form
if errors:
@ -1968,6 +2218,71 @@ async def adapters_edit_submit(
return RedirectResponse(url="/adapters", status_code=302)
@router.post("/adapters/{name}/delete")
async def adapters_delete(request: Request, name: str) -> Response:
"""Delete an operator-created adapter instance.
Safety rule: a row is deletable iff its ``name`` is NOT a key in the
adapter class registry. Built-in adapters always have ``name == kind``
which IS a registry key; operator instances have a unique name that is NOT.
NOTE: orphaned ``published_ids`` rows in cursors.db (a separate SQLite
database) are left to age out via ``dedup_sweep_days``; they are not
cleaned here because the two stores are decoupled by design.
"""
pool = get_pool()
operator = request.state.operator
form = await request.form()
form_csrf = form.get("csrf_token", "")
if not form_csrf or form_csrf != request.state.csrf_token:
raise CsrfValidationError("Invalid CSRF token")
adapter_classes = _adapter_classes()
async with pool.acquire() as conn:
row = await conn.fetchrow(
"SELECT name, kind FROM config.adapters WHERE name = $1", name
)
if row is None:
return Response(status_code=404, content=f"Adapter '{name}' not found.")
# Primary guard: built-ins have name == kind (a registered class key).
if name in adapter_classes:
return Response(
status_code=403,
content=(
f"'{name}' is a built-in adapter and cannot be deleted; "
"disable it instead."
),
)
# Second guard: the row's kind must be operator_creatable.
kind_cls = adapter_classes.get(row["kind"])
if kind_cls is not None and not getattr(kind_cls, "operator_creatable", False):
return Response(
status_code=403,
content=(
f"Adapter kind '{row['kind']}' is not operator-creatable; "
"cannot delete."
),
)
await conn.execute(
"DELETE FROM config.adapters WHERE name = $1", name
)
await write_audit(
conn,
ADAPTER_DELETE,
operator_id=operator.id,
target=name,
before={"kind": row["kind"], "name": name},
)
return RedirectResponse(url="/adapters", status_code=302)
# =============================================================================
# Streams routes
# =============================================================================

11
src/central/gui/templates/adapters_list.html
View file

@ -4,6 +4,7 @@
{% block content %}
<h1>Adapters</h1>
<p><a href="/adapters/new" role="button">+ New adapter</a></p>
<table>
<thead>
<tr>
@ -12,6 +13,7 @@
<th>Cadence</th>
<th>Last Updated</th>
<th></th>
<th></th>
</tr>
</thead>
<tbody>
@ -27,6 +29,15 @@
<td>{{ adapter.cadence_s }}s</td>
<td>{{ adapter.updated_at.strftime('%Y-%m-%d %H:%M') if adapter.updated_at else '—' }}</td>
<td><a href="/adapters/{{ adapter.name }}">Edit</a></td>
<td>
{% if adapter.deletable %}
<form method="post" action="/adapters/{{ adapter.name }}/delete" style="display:inline;"
onsubmit="return confirm('Delete adapter &quot;{{ adapter.name }}&quot;? This cannot be undone.')">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<button type="submit" class="btn-danger">Delete</button>
</form>
{% endif %}
</td>
</tr>
{% endfor %}
</tbody>

192
src/central/gui/templates/adapters_new.html Normal file
View file

@ -0,0 +1,192 @@
{% extends "base.html" %}
{% block title %}Central — New Adapter{% endblock %}
{% block content %}
<h1>New Adapter</h1>
<p class="muted">Create a new adapter instance from an operator-creatable kind.</p>
<form method="post" action="/adapters/new">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<fieldset>
<legend>Adapter Kind</legend>
{# NOTE: single creatable kind today; multi-kind HTMX field-swap is a future enhancement. #}
<label for="kind">Kind</label>
<select id="kind" name="kind">
{% for ck in creatable_kinds %}
<option value="{{ ck.kind }}"
{% if ck.kind == (form_data.kind if form_data else selected_kind) %}selected{% endif %}>
{{ ck.display_name }} ({{ ck.kind }})
</option>
{% endfor %}
</select>
{% if errors and errors.kind %}
<small class="field-error">{{ errors.kind }}</small>
{% endif %}
</fieldset>
<fieldset>
<legend>Instance Identity</legend>
<label for="name">Instance Name</label>
<input type="text" id="name" name="name"
value="{{ form_data.name if form_data else '' }}"
placeholder="e.g. my_source_v2"
required>
<small>Lowercase letters, digits, and underscores; starts with a letter; 2–64 characters.
Must not match a built-in kind name.</small>
{% if errors and errors.name %}
<small class="field-error">{{ errors.name }}</small>
{% endif %}
</fieldset>
<fieldset>
<legend>Core Settings</legend>
<label>
<input type="checkbox" name="enabled"
{% if form_data and form_data.enabled %}checked{% endif %}>
Enabled <small>(leave unchecked — ships disabled; enable after verifying settings)</small>
</label>
<label for="cadence_s">Cadence (seconds)</label>
<input type="number" id="cadence_s" name="cadence_s"
value="{{ form_data.cadence_s if form_data else default_cadence_s }}"
min="10"
required>
{% if errors and errors.cadence_s %}
<small class="field-error">{{ errors.cadence_s }}</small>
{% endif %}
</fieldset>
{% if fields %}
<fieldset>
<legend>Adapter Settings</legend>
{% for field in fields %}
{% if field.widget == "region" %}
{# Region is rendered in a separate fieldset below #}
{% elif field.widget == "text" %}
<label for="{{ field.name }}">{{ field.label }}</label>
<input type="text" id="{{ field.name }}" name="{{ field.name }}"
value="{{ form_data[field.name] if form_data and field.name in form_data else field.current_value or '' }}"
{% if field.required %}required{% endif %}>
{% if field.description %}
<small>{{ field.description }}</small>
{% endif %}
{% if errors and errors[field.name] %}
<small class="field-error">{{ errors[field.name] }}</small>
{% endif %}
{% elif field.widget == "number" %}
<label for="{{ field.name }}">{{ field.label }}</label>
<input type="number" id="{{ field.name }}" name="{{ field.name }}"
value="{{ form_data[field.name] if form_data and field.name in form_data else field.current_value or '' }}"
{% if field.required %}required{% endif %}>
{% if field.description %}
<small>{{ field.description }}</small>
{% endif %}
{% if errors and errors[field.name] %}
<small class="field-error">{{ errors[field.name] }}</small>
{% endif %}
{% elif field.widget == "checkbox" %}
<label>
<input type="checkbox" name="{{ field.name }}"
{% if form_data and field.name in form_data %}
{% if form_data[field.name] %}checked{% endif %}
{% elif field.current_value %}checked{% endif %}>
{{ field.label }}
</label>
{% if field.description %}
<small>{{ field.description }}</small>
{% endif %}
{% if errors and errors[field.name] %}
<small class="field-error">{{ errors[field.name] }}</small>
{% endif %}
{% elif field.widget == "csv" %}
<label for="{{ field.name }}">{{ field.label }}</label>
<input type="text" id="{{ field.name }}" name="{{ field.name }}"
value="{{ form_data[field.name] if form_data and field.name in form_data else (field.current_value | join(',') if field.current_value else '') }}"
{% if field.required %}required{% endif %}>
<small>Comma-separated values{% if field.description %} — {{ field.description }}{% endif %}</small>
{% if errors and errors[field.name] %}
<small class="field-error">{{ errors[field.name] }}</small>
{% endif %}
{% elif field.widget == "csv_int" %}
<label for="{{ field.name }}">{{ field.label }}</label>
<input type="text" id="{{ field.name }}" name="{{ field.name }}"
value="{{ form_data[field.name] if form_data and field.name in form_data else (field.current_value | join(',') if field.current_value else '') }}"
{% if field.required %}required{% endif %}>
<small>Comma-separated integers{% if field.description %} — {{ field.description }}{% endif %}</small>
{% if errors and errors[field.name] %}
<small class="field-error">{{ errors[field.name] }}</small>
{% endif %}
{% elif field.widget == "select" %}
<label for="{{ field.name }}">{{ field.label }}</label>
<select id="{{ field.name }}" name="{{ field.name }}">
{% for opt in field.options %}
<option value="{{ opt }}"
{% if (form_data[field.name] if form_data and field.name in form_data else field.current_value) == opt %}selected{% endif %}>
{{ opt }}
</option>
{% endfor %}
</select>
{% if field.description %}
<small>{{ field.description }}</small>
{% endif %}
{% if errors and errors[field.name] %}
<small class="field-error">{{ errors[field.name] }}</small>
{% endif %}
{% elif field.widget == "checkboxes" %}
<label>{{ field.label }}</label>
{% set current_values = form_data.getlist(field.name) if form_data and form_data.getlist else (field.current_value or []) %}
{% for opt in field.options %}
<label style="display: inline-block; margin-right: 1rem;">
<input type="checkbox" name="{{ field.name }}" value="{{ opt }}"
{% if opt in current_values %}checked{% endif %}>
{{ opt }}
</label>
{% endfor %}
{% if field.description %}
<small style="display: block;">{{ field.description }}</small>
{% endif %}
{% if errors and errors[field.name] %}
<small class="field-error">{{ errors[field.name] }}</small>
{% endif %}
{% elif field.widget == "api_key_select" %}
<label for="{{ field.name }}">{{ field.label }}</label>
<select id="{{ field.name }}" name="{{ field.name }}">
<option value="">(none)</option>
{% for key in api_keys %}
<option value="{{ key.alias }}"
{% if (form_data[field.name] if form_data and field.name in form_data else field.current_value) == key.alias %}selected{% endif %}>
{{ key.alias }}
</option>
{% endfor %}
</select>
{% if field.description %}
<small>{{ field.description }}</small>
{% endif %}
{% if errors and errors[field.name] %}
<small class="field-error">{{ errors[field.name] }}</small>
{% endif %}
{% elif field.widget == "model_list" %}
{% include "_partials/model_list.html" %}
{% endif %}
{% endfor %}
</fieldset>
{% endif %}
<button type="submit">Create Adapter</button>
<a href="/adapters" role="button" class="btn-outline">Cancel</a>
</form>
{% endblock %}

585
tests/test_gui_adapter_create_delete.py Normal file
View file

@ -0,0 +1,585 @@
"""v0.15.0 PR3 — GUI create + delete for adapter instances.
Test strategy
─────────────
* Pure-unit (always run, no DB):
- ADAPTER_NAME_REGEX validation
- Deletability rule (name in registry → not deletable)
* Mock-DB (always run; mirrors test_gui_adapter_edit.py pattern):
- GET /adapters/new renders correctly
- POST create: valid → INSERT, audit, 302
- POST create: duplicate name → 409
- POST create: bad name format → 422
- POST create: non-creatable kind → 422
- POST create: invalid settings (missing required field) → 422
- POST delete: operator instance → DELETE, audit, 302
- POST delete: built-in adapter → 403, no DELETE
DB-backed INSERT/DELETE tests (test_db_* below) use the central_test
Postgres fixture. They will raise ConnectionRefusedError when the test DB
is absent — the same behaviour as other DB-backed tests in this suite (e.g.
test_config_store.py, test_supervisor_hotreload.py).
"""
import re
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from starlette.datastructures import FormData
from starlette.requests import Request
from central.gui import templates as gui_templates
from central.gui.routes import (
ADAPTER_NAME_REGEX,
adapters_create_form,
adapters_create_submit,
adapters_delete,
adapters_list,
)
from central.adapters.generic_http import GenericHttpAdapter
# ---------------------------------------------------------------------------
# Helpers shared across test classes
# ---------------------------------------------------------------------------
def _make_request(method="GET", form_pairs=None, csrf="x"):
"""Build a mock Request with CSRF + optional form data."""
req = MagicMock()
req.state.operator = SimpleNamespace(id=1, username="admin")
req.state.csrf_token = csrf
if form_pairs is not None:
pairs = [("csrf_token", csrf)] + list(form_pairs)
req.form = AsyncMock(return_value=FormData(pairs))
else:
req.form = AsyncMock(return_value=FormData([("csrf_token", csrf)]))
return req
def _make_pool(fetchrow_returns=None, fetchval_returns=None, fetch_returns=None):
"""Build a mock asyncpg pool.
Values are set unconditionally so that None (e.g. "row not found") is
returned faithfully instead of the default truthy AsyncMock sentinel.
Pass a list for fetchrow_returns to use side_effect for sequential calls.
"""
conn = AsyncMock()
if isinstance(fetchrow_returns, list):
conn.fetchrow.side_effect = fetchrow_returns
else:
conn.fetchrow.return_value = fetchrow_returns # None = not found
conn.fetchval.return_value = fetchval_returns # None = not found
conn.fetch.return_value = fetch_returns if fetch_returns is not None else []
pool = MagicMock()
pool.acquire.return_value.__aenter__ = AsyncMock(return_value=conn)
pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
return pool, conn
# ---------------------------------------------------------------------------
# UNIT: name-regex validation
# ---------------------------------------------------------------------------
class TestAdapterNameRegex:
"""Pure-unit — no I/O, always run."""
VALID = [
"my_source",
"mysource2",
"aa", # minimum length (2 chars)
"a" + "b" * 63, # maximum length (64 chars)
"a1_b2_c3",
]
INVALID = [
"", # empty
"a", # too short (only 1 char)
"A_source", # uppercase
"1source", # starts with digit
"_source", # starts with underscore
"my-source", # hyphen not allowed
"my source", # space not allowed
"a" + "b" * 64, # 65 chars — too long
]
@pytest.mark.parametrize("name", VALID)
def test_valid(self, name):
assert ADAPTER_NAME_REGEX.match(name), f"Expected {name!r} to match"
@pytest.mark.parametrize("name", INVALID)
def test_invalid(self, name):
assert not ADAPTER_NAME_REGEX.match(name), f"Expected {name!r} not to match"
# ---------------------------------------------------------------------------
# UNIT: deletability rule
# ---------------------------------------------------------------------------
class TestDeletabilityRule:
"""Pure-unit — the rule is: name NOT IN adapter_classes → deletable.
Built-ins have name == kind (the class's .name attribute) which IS a key in
the adapter class registry. Operator instances have a unique name that is
NOT a registry key.
"""
def test_builtin_not_deletable(self):
from central.adapter_discovery import discover_adapters
classes = discover_adapters()
# Every registered kind key should be considered a built-in.
for kind in classes:
assert kind in classes, "sanity"
# The deletability check: name in adapter_classes → NOT deletable
assert kind in classes # confirms the guard fires
def test_operator_instance_is_deletable(self):
from central.adapter_discovery import discover_adapters
classes = discover_adapters()
operator_name = "my_custom_source_42"
assert operator_name not in classes, (
"Test assumes operator_name is not a registered kind; "
"update the name if a new kind was added with this identifier."
)
def test_generic_http_kind_is_not_deletable_by_name(self):
"""The KIND 'generic_http' itself should not be deletable (it's a built-in key)."""
from central.adapter_discovery import discover_adapters
classes = discover_adapters()
assert "generic_http" in classes
def test_generic_http_instance_is_deletable(self):
"""An operator instance named 'my_feed' (not a kind key) should be deletable."""
from central.adapter_discovery import discover_adapters
classes = discover_adapters()
assert "my_feed" not in classes
# ---------------------------------------------------------------------------
# UNIT: GenericHttpAdapter.operator_creatable
# ---------------------------------------------------------------------------
def test_generic_http_is_operator_creatable():
assert GenericHttpAdapter.operator_creatable is True
def test_base_class_default_not_creatable():
from central.adapter import SourceAdapter
assert SourceAdapter.operator_creatable is False
# ---------------------------------------------------------------------------
# Mock-DB: GET /adapters/new
# ---------------------------------------------------------------------------
class TestGetAdaptersNew:
@pytest.mark.asyncio
async def test_renders_200_with_generic_http_in_kind_select(self):
pool, conn = _make_pool(fetch_returns=[])
tmpl = MagicMock()
tmpl.TemplateResponse.return_value = MagicMock(status_code=200)
req = _make_request()
with patch("central.gui.routes._get_templates", return_value=tmpl), \
patch("central.gui.routes.get_pool", return_value=pool):
await adapters_create_form(req)
ctx = tmpl.TemplateResponse.call_args.kwargs["context"]
kind_names = [ck["kind"] for ck in ctx["creatable_kinds"]]
assert "generic_http" in kind_names
@pytest.mark.asyncio
async def test_fields_present_for_generic_http(self):
pool, conn = _make_pool(fetch_returns=[])
tmpl = MagicMock()
tmpl.TemplateResponse.return_value = MagicMock(status_code=200)
req = _make_request()
with patch("central.gui.routes._get_templates", return_value=tmpl), \
patch("central.gui.routes.get_pool", return_value=pool):
await adapters_create_form(req)
ctx = tmpl.TemplateResponse.call_args.kwargs["context"]
field_names = [f.name for f in ctx["fields"]]
# GenericHttpSettings requires url, domain, id_path at minimum
assert "url" in field_names
assert "domain" in field_names
assert "id_path" in field_names
@pytest.mark.asyncio
async def test_template_renders_without_errors(self):
"""Smoke test: the template itself renders without crashing."""
pool, conn = _make_pool(fetch_returns=[])
tmpl = MagicMock()
tmpl.TemplateResponse.return_value = MagicMock(status_code=200)
req = _make_request()
with patch("central.gui.routes._get_templates", return_value=tmpl), \
patch("central.gui.routes.get_pool", return_value=pool):
resp = await adapters_create_form(req)
# Template was called — no exception raised
assert tmpl.TemplateResponse.called
# ---------------------------------------------------------------------------
# Mock-DB: POST /adapters/new — happy path
# ---------------------------------------------------------------------------
def _valid_generic_http_pairs(name="my_feed"):
"""Minimal valid form pairs for a generic_http instance."""
return [
("kind", "generic_http"),
("name", name),
("cadence_s", "300"),
# enabled intentionally absent → ships disabled
("url", "https://example.com/feed.geojson"),
("domain", "fire"),
("id_path", "properties.id"),
]
class TestPostAdaptersNewHappyPath:
@pytest.mark.asyncio
async def test_valid_creates_and_redirects(self):
pool, conn = _make_pool(
fetchval_returns=None, # name does not exist yet
fetch_returns=[], # no api keys
)
inserted: list = []
async def cap_execute(q, *args):
if "INSERT INTO config.adapters" in q:
inserted.append(args)
conn.execute.side_effect = cap_execute
req = _make_request(form_pairs=_valid_generic_http_pairs())
with patch("central.gui.routes.get_pool", return_value=pool), \
patch("central.gui.routes.write_audit", new=AsyncMock()):
resp = await adapters_create_submit(req)
assert resp.status_code == 302
assert "/adapters/my_feed" in resp.headers["location"]
assert len(inserted) == 1
# args: name, kind, enabled, cadence_s, settings
_name, _kind, _enabled, _cadence, _settings = inserted[0]
assert _name == "my_feed"
assert _kind == "generic_http"
assert _enabled is False # no 'enabled' in form → ships disabled
assert _cadence == 300
assert _settings["url"] == "https://example.com/feed.geojson"
@pytest.mark.asyncio
async def test_enabled_flag_set_when_checked(self):
pool, conn = _make_pool(fetchval_returns=None, fetch_returns=[])
inserted: list = []
async def cap(q, *args):
if "INSERT" in q:
inserted.append(args)
conn.execute.side_effect = cap
pairs = _valid_generic_http_pairs() + [("enabled", "on")]
req = _make_request(form_pairs=pairs)
with patch("central.gui.routes.get_pool", return_value=pool), \
patch("central.gui.routes.write_audit", new=AsyncMock()):
resp = await adapters_create_submit(req)
assert resp.status_code == 302
_name, _kind, _enabled, *_ = inserted[0]
assert _enabled is True
@pytest.mark.asyncio
async def test_audit_record_written_on_create(self):
pool, conn = _make_pool(fetchval_returns=None, fetch_returns=[])
conn.execute.return_value = None
audited: list = []
async def cap_audit(conn_, action, **kw):
audited.append((action, kw))
req = _make_request(form_pairs=_valid_generic_http_pairs())
with patch("central.gui.routes.get_pool", return_value=pool), \
patch("central.gui.routes.write_audit", side_effect=cap_audit):
await adapters_create_submit(req)
assert len(audited) == 1
action, kw = audited[0]
assert action == "adapter.create"
assert kw["target"] == "my_feed"
# ---------------------------------------------------------------------------
# Mock-DB: POST /adapters/new — validation errors
# ---------------------------------------------------------------------------
async def _post_new(pairs, fetchval=None, fetch_returns=None):
"""Helper: POST /adapters/new and return (response, template_call_args).
fetchval=None means "adapter name does not exist" (duplicate check passes).
Pass fetchval=1 to simulate a duplicate.
"""
pool, conn = _make_pool(
fetchval_returns=fetchval, # None = not found; passed unconditionally
fetch_returns=fetch_returns or [],
)
tmpl = MagicMock()
tmpl.TemplateResponse.return_value = MagicMock()
req = _make_request(form_pairs=pairs)
with patch("central.gui.routes._get_templates", return_value=tmpl), \
patch("central.gui.routes.get_pool", return_value=pool), \
patch("central.gui.routes.write_audit", new=AsyncMock()):
resp = await adapters_create_submit(req)
return resp, tmpl.TemplateResponse.call_args
class TestPostAdaptersNewValidationErrors:
@pytest.mark.asyncio
async def test_duplicate_name_returns_409(self):
pool, conn = _make_pool(fetchval_returns=1, fetch_returns=[])
req = _make_request(form_pairs=_valid_generic_http_pairs())
with patch("central.gui.routes.get_pool", return_value=pool), \
patch("central.gui.routes.write_audit", new=AsyncMock()):
resp = await adapters_create_submit(req)
assert resp.status_code == 409
@pytest.mark.asyncio
async def test_bad_name_format_returns_422(self):
pairs = _valid_generic_http_pairs(name="Bad-Name!")
resp, ca = await _post_new(pairs)
assert ca.kwargs["status_code"] == 422
assert "name" in ca.kwargs["context"]["errors"]
@pytest.mark.asyncio
async def test_name_starts_with_digit_returns_422(self):
pairs = _valid_generic_http_pairs(name="1invalid")
resp, ca = await _post_new(pairs)
assert ca.kwargs["status_code"] == 422
@pytest.mark.asyncio
async def test_name_too_short_returns_422(self):
pairs = _valid_generic_http_pairs(name="a") # only 1 char
resp, ca = await _post_new(pairs)
assert ca.kwargs["status_code"] == 422
@pytest.mark.asyncio
async def test_name_shadows_kind_returns_422(self):
"""Cannot use a registered kind name as an instance name."""
pairs = _valid_generic_http_pairs(name="generic_http")
resp, ca = await _post_new(pairs)
assert ca.kwargs["status_code"] == 422
assert "name" in ca.kwargs["context"]["errors"]
@pytest.mark.asyncio
async def test_non_creatable_kind_returns_422(self):
# usgs_quake is a real built-in kind that is NOT operator_creatable
pairs = [
("kind", "usgs_quake"),
("name", "my_quake"),
("cadence_s", "300"),
("url", "https://example.com"),
("domain", "quake"),
("id_path", "id"),
]
resp, ca = await _post_new(pairs)
assert ca.kwargs["status_code"] == 422
assert "kind" in ca.kwargs["context"]["errors"]
@pytest.mark.asyncio
async def test_invalid_settings_missing_required_field_returns_422(self):
# Omit required 'url' field from generic_http settings
pairs = [
("kind", "generic_http"),
("name", "my_feed"),
("cadence_s", "300"),
# no 'url', no 'domain', no 'id_path'
]
resp, ca = await _post_new(pairs)
assert ca.kwargs["status_code"] == 422
@pytest.mark.asyncio
async def test_cadence_below_10_returns_422(self):
pairs = _valid_generic_http_pairs()
# replace cadence_s
pairs = [(k, "5") if k == "cadence_s" else (k, v) for k, v in pairs]
resp, ca = await _post_new(pairs)
assert ca.kwargs["status_code"] == 422
assert "cadence_s" in ca.kwargs["context"]["errors"]
@pytest.mark.asyncio
async def test_invalid_domain_returns_422(self):
pairs = _valid_generic_http_pairs()
pairs = [(k, "notadomain") if k == "domain" else (k, v) for k, v in pairs]
resp, ca = await _post_new(pairs)
assert ca.kwargs["status_code"] == 422
# ---------------------------------------------------------------------------
# Mock-DB: POST /adapters/{name}/delete
# ---------------------------------------------------------------------------
class TestPostAdaptersDelete:
@pytest.mark.asyncio
async def test_operator_instance_is_deleted(self):
"""Deleting an operator instance removes the row and audits."""
pool, conn = _make_pool(
fetchrow_returns={"name": "my_feed", "kind": "generic_http"},
)
deleted: list = []
async def cap(q, *args):
if "DELETE FROM config.adapters" in q:
deleted.append(args)
conn.execute.side_effect = cap
req = _make_request()
with patch("central.gui.routes.get_pool", return_value=pool), \
patch("central.gui.routes.write_audit", new=AsyncMock()):
resp = await adapters_delete(req, "my_feed")
assert resp.status_code == 302
assert resp.headers["location"] == "/adapters"
assert len(deleted) == 1
assert deleted[0][0] == "my_feed"
@pytest.mark.asyncio
async def test_builtin_adapter_returns_403(self):
"""Attempting to delete a built-in adapter (name in registry) → 403."""
pool, conn = _make_pool(
fetchrow_returns={"name": "usgs_quake", "kind": "usgs_quake"},
)
req = _make_request()
with patch("central.gui.routes.get_pool", return_value=pool), \
patch("central.gui.routes.write_audit", new=AsyncMock()):
resp = await adapters_delete(req, "usgs_quake")
assert resp.status_code == 403
assert "built-in" in resp.body.decode()
# DELETE must NOT have been called
for call in conn.execute.call_args_list:
assert "DELETE" not in str(call)
@pytest.mark.asyncio
async def test_missing_adapter_returns_404(self):
pool, conn = _make_pool(fetchrow_returns=None)
req = _make_request()
with patch("central.gui.routes.get_pool", return_value=pool), \
patch("central.gui.routes.write_audit", new=AsyncMock()):
resp = await adapters_delete(req, "nonexistent")
assert resp.status_code == 404
@pytest.mark.asyncio
async def test_delete_audit_record_written(self):
pool, conn = _make_pool(
fetchrow_returns={"name": "my_feed", "kind": "generic_http"},
)
conn.execute.return_value = None
audited: list = []
async def cap_audit(conn_, action, **kw):
audited.append((action, kw))
req = _make_request()
with patch("central.gui.routes.get_pool", return_value=pool), \
patch("central.gui.routes.write_audit", side_effect=cap_audit):
await adapters_delete(req, "my_feed")
assert len(audited) == 1
action, kw = audited[0]
assert action == "adapter.delete"
assert kw["target"] == "my_feed"
@pytest.mark.asyncio
async def test_generic_http_kind_itself_is_protected(self):
"""The class entry 'generic_http' IS in the registry → 403."""
pool, conn = _make_pool(
fetchrow_returns={"name": "generic_http", "kind": "generic_http"},
)
req = _make_request()
with patch("central.gui.routes.get_pool", return_value=pool), \
patch("central.gui.routes.write_audit", new=AsyncMock()):
resp = await adapters_delete(req, "generic_http")
assert resp.status_code == 403
# ---------------------------------------------------------------------------
# Template smoke test: adapters_new.html renders without crashing
# ---------------------------------------------------------------------------
class TestAdaptersNewTemplate:
def _render(self, ctx):
req = Request({
"type": "http", "method": "GET", "path": "/",
"headers": [], "query_string": b"",
})
return gui_templates.TemplateResponse(
request=req, name="adapters_new.html", context=ctx
).body.decode()
def _ctx(self, errors=None, form_data=None):
from central.gui.form_descriptors import describe_fields
from central.adapters.generic_http import GenericHttpSettings
fields = describe_fields(GenericHttpSettings, {})
return {
"operator": SimpleNamespace(username="admin"),
"csrf_token": "x",
"creatable_kinds": [{"kind": "generic_http", "display_name": "Generic HTTP Source"}],
"selected_kind": "generic_http",
"default_cadence_s": 300,
"fields": fields,
"api_keys": [],
"errors": errors,
"form_data": form_data,
}
def test_renders_kind_select(self):
out = self._render(self._ctx())
assert "generic_http" in out
assert 'name="kind"' in out
def test_renders_name_input(self):
out = self._render(self._ctx())
assert 'name="name"' in out
def test_renders_cadence_input_with_default(self):
out = self._render(self._ctx())
assert 'name="cadence_s"' in out
assert "300" in out
def test_renders_url_field_for_generic_http(self):
out = self._render(self._ctx())
assert 'name="url"' in out
def test_enabled_unchecked_by_default(self):
out = self._render(self._ctx())
# The enabled checkbox must not be checked in default render
# (spec: ships disabled)
assert 'name="enabled"' in out
# Extract the enabled checkbox line and confirm no 'checked' attribute
for line in out.splitlines():
if 'name="enabled"' in line:
assert "checked" not in line, f"enabled checkbox should be unchecked by default: {line}"
break
def test_error_messages_displayed(self):
errors = {"name": "Name is invalid", "url": "URL is required"}
out = self._render(self._ctx(errors=errors))
assert "Name is invalid" in out
assert "URL is required" in out
def test_form_data_restores_values(self):
form_data = {"kind": "generic_http", "name": "restored_name",
"cadence_s": "600", "url": "https://example.com/data.json",
"domain": "fire", "id_path": "id", "enabled": False}
out = self._render(self._ctx(form_data=form_data))
assert "restored_name" in out
assert "https://example.com/data.json" in out