Extract HTML5 operation helpers

services/api-server/src/api_server/html5_operations.py

@@ -124,6 +124,40 @@ def render_html5_operation_detail(job: object | None) -> str:
     """
 
 
+def filter_html5_operation_jobs(
+    jobs: Iterable[object],
+    *,
+    project_id: str = "",
+    status: str = "",
+    kind: str = "",
+    limit: int = 50,
+) -> list[object]:
+    normalized_project = project_id.strip().casefold()
+    normalized_status = status.strip().casefold()
+    normalized_kind = kind.strip().casefold()
+    filtered = []
+    for job in jobs:
+        payload = getattr(job, "payload", {}) or {}
+        if normalized_project and str(payload.get("project_id") or "").casefold() != normalized_project:
+            continue
+        if normalized_status and _operation_value(getattr(job, "status", "")).casefold() != normalized_status:
+            continue
+        if normalized_kind and _operation_value(getattr(job, "kind", "")).casefold() != normalized_kind:
+            continue
+        filtered.append(job)
+    return sorted(filtered, key=lambda job: getattr(job, "updated_at", ""), reverse=True)[:limit]
+
+
+def latest_html5_import_job(jobs: Iterable[object], project_id: str) -> object | None:
+    import_jobs = [
+        job
+        for job in jobs
+        if (getattr(job, "payload", {}) or {}).get("project_id") == project_id
+        and _operation_value(getattr(job, "kind", "")) == "SERVER_IMPORT"
+    ]
+    return max(import_jobs, key=lambda job: getattr(job, "updated_at", "")) if import_jobs else None
+
+
 def _operation_row(job: object) -> str:
     job_id = str(getattr(job, "job_id", ""))
     kind = str(getattr(job, "kind", ""))
@@ -183,3 +217,7 @@ def _operation_filter_query(*, project_id: str, status: str, kind: str) -> str:
     if kind:
         params.append(f"kind={quote(kind)}")
     return f"?{'&'.join(params)}" if params else ""
+
+
+def _operation_value(value: object) -> str:
+    return str(getattr(value, "value", value))

services/api-server/src/api_server/main.py

@@ -77,6 +77,8 @@ from api_server.html5_editor import (
     render_html5_symbols,
 )
 from api_server.html5_operations import (
+    filter_html5_operation_jobs,
+    latest_html5_import_job,
     render_html5_operation_detail,
     render_html5_operation_rows,
     render_html5_operation_summary,
@@ -8333,33 +8335,16 @@ def _current_import_source(project_id: str) -> ImportSourceKind:
 
 
 def _html5_operation_jobs(project_id: str = "", status: str = "", kind: str = "") -> list[OperationJob]:
-    normalized_project = project_id.strip().casefold()
+    return filter_html5_operation_jobs(
-    normalized_status = status.strip().casefold()
+        _operations.jobs.values(),
-    normalized_kind = kind.strip().casefold()
+        project_id=project_id,
-    jobs = []
+        status=status,
-    for job in _operations.jobs.values():
+        kind=kind,
-        payload = job.payload or {}
+    )
-        if normalized_project and str(payload.get("project_id") or "").casefold() != normalized_project:
-            continue
-        if normalized_status and _operation_value(getattr(job, "status", "")).casefold() != normalized_status:
-            continue
-        if normalized_kind and _operation_value(getattr(job, "kind", "")).casefold() != normalized_kind:
-            continue
-        jobs.append(job)
-    return sorted(jobs, key=lambda job: job.updated_at, reverse=True)[:50]
 
 
 def _html5_latest_import_job(project_id: str) -> OperationJob | None:
-    jobs = [
+    return latest_html5_import_job(_operations.jobs.values(), project_id)
-        job
-        for job in _operations.jobs.values()
-        if job.payload.get("project_id") == project_id and _operation_value(getattr(job, "kind", "")) == "SERVER_IMPORT"
-    ]
-    return max(jobs, key=lambda job: job.updated_at) if jobs else None
-
-
-def _operation_value(value: object) -> str:
-    return str(getattr(value, "value", value))
 
 
 def _project_summaries() -> list[ProjectSummaryResponse]:

services/api-server/tests/test_api.py

@@ -1,6 +1,7 @@
 from pathlib import Path
 import re
 import time
+from types import SimpleNamespace
 from uuid import uuid4
 import zipfile
 
@@ -13,6 +14,7 @@ from api_server.html5_forms import (
     html5_metadata_payload,
     html5_metadata_request_payload,
 )
+from api_server.html5_operations import filter_html5_operation_jobs, latest_html5_import_job
 from api_server.html5_sse import html5_sse_comment, html5_sse_event, html5_sse_if_changed
 from api_server.main import app
 from one_c_normalizer import ConfigurationRoot, MetadataGroup, MetadataObject, Module, NormalizedProject
@@ -120,6 +122,54 @@ def test_html5_form_helpers_normalize_metadata_payloads():
     assert "_raw_attributes" not in html5_metadata_request_payload(payload)
 
 
+def test_html5_operation_helpers_filter_sort_and_find_latest_import_job():
+    jobs = [
+        SimpleNamespace(
+            job_id="old",
+            kind="SERVER_IMPORT",
+            status="SUCCEEDED",
+            payload={"project_id": "demo"},
+            updated_at=10,
+        ),
+        SimpleNamespace(
+            job_id="new",
+            kind="SERVER_IMPORT",
+            status="RUNNING",
+            payload={"project_id": "demo"},
+            updated_at=30,
+        ),
+        SimpleNamespace(
+            job_id="other-kind",
+            kind="REINDEX",
+            status="RUNNING",
+            payload={"project_id": "demo"},
+            updated_at=40,
+        ),
+        SimpleNamespace(
+            job_id="other-project",
+            kind="SERVER_IMPORT",
+            status="RUNNING",
+            payload={"project_id": "other"},
+            updated_at=50,
+        ),
+    ]
+
+    filtered = filter_html5_operation_jobs(
+        jobs,
+        project_id=" DEMO ",
+        status="running",
+        kind="server_import",
+    )
+
+    assert [job.job_id for job in filtered] == ["new"]
+    assert [job.job_id for job in filter_html5_operation_jobs(jobs, project_id="demo")] == [
+        "other-kind",
+        "new",
+        "old",
+    ]
+    assert latest_html5_import_job(jobs, "demo").job_id == "new"
+
+
 def test_cors_allows_lan_panel_origin():
     client = TestClient(app)
     response = client.options(