From 940d7e884ba6faa4262c8031db925cbcbda5e5ed Mon Sep 17 00:00:00 2001 From: Mikhail Date: Sun, 17 May 2026 12:36:56 +0300 Subject: [PATCH] Extract HTML5 operation helpers --- .../src/api_server/html5_operations.py | 38 ++++++++++++++ services/api-server/src/api_server/main.py | 33 ++++-------- services/api-server/tests/test_api.py | 50 +++++++++++++++++++ 3 files changed, 97 insertions(+), 24 deletions(-) diff --git a/services/api-server/src/api_server/html5_operations.py b/services/api-server/src/api_server/html5_operations.py index a32224c..84a4e3c 100644 --- a/services/api-server/src/api_server/html5_operations.py +++ b/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)) diff --git a/services/api-server/src/api_server/main.py b/services/api-server/src/api_server/main.py index a2ab238..11d35d7 100644 --- a/services/api-server/src/api_server/main.py +++ b/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() - normalized_status = status.strip().casefold() - normalized_kind = kind.strip().casefold() - jobs = [] - for job in _operations.jobs.values(): - 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] + return filter_html5_operation_jobs( + _operations.jobs.values(), + project_id=project_id, + status=status, + kind=kind, + ) def _html5_latest_import_job(project_id: str) -> OperationJob | None: - jobs = [ - 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)) + return latest_html5_import_job(_operations.jobs.values(), project_id) def _project_summaries() -> list[ProjectSummaryResponse]: diff --git a/services/api-server/tests/test_api.py b/services/api-server/tests/test_api.py index 93f9133..ecce6b7 100644 --- a/services/api-server/tests/test_api.py +++ b/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(