From d72754cba6076953dcad75238e5a20a82c46fb21 Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Tue, 21 Jul 2026 07:06:26 +0300 Subject: [PATCH 1/6] db: await PyMongo async operations PyMongo's native async collection methods return awaitables for index creation and aggregation. Calling them without await leaves indexes uncreated and makes aggregation endpoints fail when they try to consume a coroutine as a cursor. Await both operations and add focused tests that verify the collection calls and cursor conversion are awaited. Signed-off-by: Denys Fedoryshchenko --- api/db.py | 4 ++-- tests/unit_tests/test_db.py | 47 +++++++++++++++++++++++++++++++++++++ 2 files changed, 49 insertions(+), 2 deletions(-) create mode 100644 tests/unit_tests/test_db.py diff --git a/api/db.py b/api/db.py index 9172af4c..a158eb35 100644 --- a/api/db.py +++ b/api/db.py @@ -102,7 +102,7 @@ async def create_indexes(self): continue col = self._get_collection(model) for index in indexes: - col.create_index(index.field, **index.attributes) + await col.create_index(index.field, **index.attributes) async def find_one(self, model, **kwargs): """Find one object with matching attributes @@ -291,7 +291,7 @@ async def update(self, obj): async def aggregate(self, model, pipeline): """Run an aggregation pipeline on a model's collection""" col = self._get_collection(model) - cursor = col.aggregate(pipeline) + cursor = await col.aggregate(pipeline) return await cursor.to_list(length=None) async def delete_by_id(self, model, obj_id): diff --git a/tests/unit_tests/test_db.py b/tests/unit_tests/test_db.py new file mode 100644 index 00000000..36233df1 --- /dev/null +++ b/tests/unit_tests/test_db.py @@ -0,0 +1,47 @@ +# SPDX-License-Identifier: LGPL-2.1-or-later +# +# Copyright (C) 2026 Collabora Limited + +"""Unit tests for the database abstraction.""" + +from unittest.mock import AsyncMock, Mock + +import pytest +from kernelci.api.models import TelemetryEvent + +from api.db import Database + + +@pytest.mark.asyncio +async def test_create_indexes_awaits_pymongo(mocker): + """Every PyMongo AsyncCollection.create_index call is awaited.""" + collection = Mock() + collection.create_index = AsyncMock() + database = Database.__new__(Database) + database.COLLECTIONS = {TelemetryEvent: "telemetry"} + mocker.patch.object(database, "_get_collection", return_value=collection) + + await database.create_indexes() + + assert collection.create_index.await_count == len( + TelemetryEvent.get_indexes() + ) + + +@pytest.mark.asyncio +async def test_aggregate_awaits_pymongo_cursor(mocker): + """PyMongo AsyncCollection.aggregate is a coroutine.""" + expected = [{"_id": {"runtime": "lava-test"}, "total": 3}] + cursor = Mock() + cursor.to_list = AsyncMock(return_value=expected) + collection = Mock() + collection.aggregate = AsyncMock(return_value=cursor) + database = Database.__new__(Database) + mocker.patch.object(database, "_get_collection", return_value=collection) + pipeline = [{"$group": {"_id": "$runtime", "total": {"$sum": 1}}}] + + result = await database.aggregate(TelemetryEvent, pipeline) + + assert result == expected + collection.aggregate.assert_awaited_once_with(pipeline) + cursor.to_list.assert_awaited_once_with(length=None) From d608c1f16969b5b00cbaebd8700ec10ad1c0cea0 Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Tue, 21 Jul 2026 07:07:10 +0300 Subject: [PATCH 2/6] telemetry: detect failing physical devices Device-type aggregation can hide a single dead board behind healthy peers, while mixing job and test events dilutes boot failures with many per-test results. Add a physical-device anomaly scope backed only by job results, expose device_id in stats, and return constant-failure flags plus the latest error context. Document the public query and cover grouping, filtering, classification, and validation. Signed-off-by: Denys Fedoryshchenko --- api/main.py | 67 ++++++++--- doc/api-details.md | 28 +++++ tests/unit_tests/test_telemetry_handler.py | 130 +++++++++++++++++++++ 3 files changed, 212 insertions(+), 13 deletions(-) create mode 100644 tests/unit_tests/test_telemetry_handler.py diff --git a/api/main.py b/api/main.py index b733f300..7adfd9de 100644 --- a/api/main.py +++ b/api/main.py @@ -1255,6 +1255,7 @@ async def get_telemetry(request: Request): TELEMETRY_STATS_GROUP_FIELDS = { "runtime", "device_type", + "device_id", "job_name", "tree", "branch", @@ -1277,7 +1278,7 @@ async def get_telemetry_stats(request: Request): Query parameters: - group_by: Comma-separated fields to group by - (runtime, device_type, job_name, tree, branch, arch, + (runtime, device_type, device_id, job_name, tree, branch, arch, kind, error_type) - kind: Filter by event kind before aggregating - runtime: Filter by runtime name @@ -1309,6 +1310,7 @@ async def get_telemetry_stats(request: Request): "kind", "runtime", "device_type", + "device_id", "job_name", "tree", "branch", @@ -1395,6 +1397,13 @@ async def get_telemetry_anomalies( min_total: int = Query( 3, ge=1, description="Min events in window to consider (avoids noise)" ), + scope: str = Query( + "device_type", + pattern="^(device_type|device)$", + description=( + "Group results by device type (default) or physical device ID" + ), + ), ): """Detect anomalies in telemetry data. @@ -1402,6 +1411,10 @@ async def get_telemetry_anomalies( rate or failure rate exceeds the threshold within the given time window. Also detects runtimes with submission errors. + Use ``scope=device`` to identify individual devices with repeated + failures. Physical-device health uses job results only, so per-test + events do not dilute boot and infrastructure failure rates. + Returns a list sorted by severity (highest error rate first). """ metrics.add("http_requests_total", 1) @@ -1412,23 +1425,38 @@ async def get_telemetry_anomalies( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid window '{window}'. Use: {', '.join(ANOMALY_WINDOW_MAP.keys())}", ) - since = datetime.utcnow() - timedelta(hours=hours) + since = datetime.now(timezone.utc) - timedelta(hours=hours) + + result_match = { + "kind": ( + "job_result" + if scope == "device" + else {"$in": ["job_result", "test_result"]} + ), + "ts": {"$gte": since}, + } + result_group = { + "runtime": "$runtime", + "device_type": "$device_type", + } + if scope == "device": + result_match["device_id"] = {"$nin": [None, ""]} + result_group["device_id"] = "$device_id" - # Anomaly 1: High infra error / failure rate per runtime+device_type + # Anomaly 1: High infra error / failure rate per selected device scope result_pipeline = [ - { - "$match": { - "kind": {"$in": ["job_result", "test_result"]}, - "ts": {"$gte": since}, - } - }, + {"$match": result_match}, + {"$sort": {"ts": -1}}, { "$group": { - "_id": { - "runtime": "$runtime", - "device_type": "$device_type", - }, + "_id": result_group, + "last_seen": {"$first": "$ts"}, + "latest_error_type": {"$first": "$error_type"}, + "latest_error_msg": {"$first": "$error_msg"}, "total": {"$sum": 1}, + "pass": { + "$sum": {"$cond": [{"$eq": ["$result", "pass"]}, 1, 0]} + }, "fail": { "$sum": {"$cond": [{"$eq": ["$result", "fail"]}, 1, 0]} }, @@ -1437,6 +1465,9 @@ async def get_telemetry_anomalies( "$cond": [{"$eq": ["$result", "incomplete"]}, 1, 0] } }, + "skip": { + "$sum": {"$cond": [{"$eq": ["$result", "skip"]}, 1, 0]} + }, "infra_error": {"$sum": {"$cond": ["$is_infra_error", 1, 0]}}, } }, @@ -1488,6 +1519,7 @@ async def get_telemetry_anomalies( "window": window, "threshold": threshold, "min_total": min_total, + "scope": scope, "since": since.isoformat(), "result_anomalies": [], "error_anomalies": [], @@ -1496,11 +1528,20 @@ async def get_telemetry_anomalies( for doc in result_anomalies: row = doc["_id"].copy() row["total"] = doc["total"] + row["pass"] = doc["pass"] row["fail"] = doc["fail"] row["incomplete"] = doc["incomplete"] + row["skip"] = doc["skip"] row["infra_error"] = doc["infra_error"] row["infra_rate"] = round(doc["infra_rate"], 3) row["fail_rate"] = round(doc["fail_rate"], 3) + row["constant_failure"] = ( + doc["pass"] == 0 and doc["fail"] + doc["incomplete"] == doc["total"] + ) + row["constant_infra_failure"] = doc["infra_error"] == doc["total"] + row["last_seen"] = doc.get("last_seen") + row["latest_error_type"] = doc.get("latest_error_type") + row["latest_error_msg"] = doc.get("latest_error_msg") output["result_anomalies"].append(row) for doc in error_anomalies: diff --git a/doc/api-details.md b/doc/api-details.md index e3e489fb..6621ece4 100644 --- a/doc/api-details.md +++ b/doc/api-details.md @@ -1284,3 +1284,31 @@ full node object: } ] ``` + +## Telemetry device health + +The telemetry anomaly endpoint can aggregate either device types or physical +devices. To find devices whose recent jobs have all failed, use physical-device +scope with a 100% threshold: + +```shell +curl 'https://api.kernelci.org/latest/telemetry/anomalies?scope=device&window=48h&threshold=1.0&min_total=3' +``` + +Physical-device scope uses only `job_result` events. This avoids counting the +many test-case results emitted by one boot as independent device-health +samples. Each result includes `device_id`, `device_type`, `runtime`, result +counts and rates, the latest error and timestamp, plus these classifications: + +- `constant_failure`: every job in the window was `fail` or `incomplete`. +- `constant_infra_failure`: every job was classified as an infrastructure + error. This is the strongest signal that a device is not booting or that its + lab connection is broken; inspect its raw telemetry errors to distinguish + those cases. + +The statistics endpoint also accepts `device_id` as a filter and a `group_by` +field for custom analysis: + +```shell +curl 'https://api.kernelci.org/latest/telemetry/stats?group_by=runtime,device_id&kind=job_result&since=2026-07-20T00:00:00' +``` diff --git a/tests/unit_tests/test_telemetry_handler.py b/tests/unit_tests/test_telemetry_handler.py new file mode 100644 index 00000000..f51f856f --- /dev/null +++ b/tests/unit_tests/test_telemetry_handler.py @@ -0,0 +1,130 @@ +# SPDX-License-Identifier: LGPL-2.1-or-later +# +# Copyright (C) 2026 Collabora Limited + +"""Unit tests for KernelCI API telemetry aggregation handlers.""" + +from unittest.mock import AsyncMock + +from kernelci.api.models import TelemetryEvent + + +def test_stats_groups_and_filters_by_device_id(mocker, test_client): + """Physical device IDs can be used in telemetry statistics.""" + aggregate = AsyncMock( + return_value=[ + { + "_id": { + "runtime": "lava-test", + "device_id": "board-01", + }, + "total": 4, + "pass": 0, + "fail": 0, + "incomplete": 4, + "skip": 0, + "infra_error": 4, + } + ] + ) + mocker.patch("api.db.Database.aggregate", side_effect=aggregate) + + response = test_client.get( + "telemetry/stats", + params={ + "group_by": "runtime,device_id", + "kind": "job_result", + "device_id": "board-01", + }, + ) + + assert response.status_code == 200 + assert response.json()[0]["device_id"] == "board-01" + model, pipeline = aggregate.call_args.args + assert model is TelemetryEvent + assert pipeline[0]["$match"] == { + "kind": "job_result", + "device_id": "board-01", + } + assert pipeline[1]["$group"]["_id"]["device_id"] == "$device_id" + + +def test_device_anomalies_use_job_results_and_physical_ids(mocker, test_client): + """Device scope reports physical devices without per-test dilution.""" + aggregate = AsyncMock( + side_effect=[ + [ + { + "_id": { + "runtime": "lava-test", + "device_type": "board", + "device_id": "board-01", + }, + "total": 3, + "pass": 0, + "fail": 0, + "incomplete": 3, + "skip": 0, + "infra_error": 3, + "infra_rate": 1.0, + "fail_rate": 1.0, + "last_seen": "2026-07-21T01:24:01.322000", + "latest_error_type": "Infrastructure", + "latest_error_msg": ( + "bootloader-commands timed out after 180 seconds" + ), + } + ], + [], + ] + ) + mocker.patch("api.db.Database.aggregate", side_effect=aggregate) + + response = test_client.get( + "telemetry/anomalies", + params={ + "scope": "device", + "window": "48h", + "threshold": 1.0, + "min_total": 3, + }, + ) + + assert response.status_code == 200 + data = response.json() + assert data["scope"] == "device" + assert data["result_anomalies"] == [ + { + "runtime": "lava-test", + "device_type": "board", + "device_id": "board-01", + "total": 3, + "pass": 0, + "fail": 0, + "incomplete": 3, + "skip": 0, + "infra_error": 3, + "infra_rate": 1.0, + "fail_rate": 1.0, + "constant_failure": True, + "constant_infra_failure": True, + "last_seen": "2026-07-21T01:24:01.322000", + "latest_error_type": "Infrastructure", + "latest_error_msg": ( + "bootloader-commands timed out after 180 seconds" + ), + } + ] + + model, pipeline = aggregate.call_args_list[0].args + assert model is TelemetryEvent + assert pipeline[0]["$match"]["kind"] == "job_result" + assert pipeline[0]["$match"]["device_id"] == {"$nin": [None, ""]} + assert pipeline[1] == {"$sort": {"ts": -1}} + assert pipeline[2]["$group"]["_id"]["device_id"] == "$device_id" + + +def test_device_anomalies_reject_invalid_scope(test_client): + response = test_client.get("telemetry/anomalies?scope=lab") + + assert response.status_code == 422 From 68b674e1c9a42ff5afd2ee1cd2f3f17f061d207d Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Tue, 21 Jul 2026 07:14:33 +0300 Subject: [PATCH 3/6] telemetry: expose lab-aware device health LAVA runtime names currently identify labs, and device anomalies already use runtime as part of their grouping key. Exposing only the runtime field makes that isolation easy for API consumers to miss. Add lab as an explicit stats filter and grouping alias, include it in anomaly rows, reject conflicting lab/runtime filters, and document that results from different labs remain separate. Signed-off-by: Denys Fedoryshchenko --- api/main.py | 20 ++++++++++++-- doc/api-details.md | 13 +++++---- tests/unit_tests/test_telemetry_handler.py | 32 +++++++++++++++++++--- 3 files changed, 54 insertions(+), 11 deletions(-) diff --git a/api/main.py b/api/main.py index 7adfd9de..acd69f60 100644 --- a/api/main.py +++ b/api/main.py @@ -1254,6 +1254,7 @@ async def get_telemetry(request: Request): TELEMETRY_STATS_GROUP_FIELDS = { "runtime", + "lab", "device_type", "device_id", "job_name", @@ -1278,7 +1279,7 @@ async def get_telemetry_stats(request: Request): Query parameters: - group_by: Comma-separated fields to group by - (runtime, device_type, device_id, job_name, tree, branch, arch, + (runtime, lab, device_type, device_id, job_name, tree, branch, arch, kind, error_type) - kind: Filter by event kind before aggregating - runtime: Filter by runtime name @@ -1318,6 +1319,15 @@ async def get_telemetry_stats(request: Request): ) if query_params.get(key) } + lab = query_params.pop("lab", None) + if lab: + runtime = match_stage.get("runtime") + if runtime and runtime != lab: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="'lab' and 'runtime' must match when both are provided", + ) + match_stage["runtime"] = lab since = query_params.pop("since", None) until = query_params.pop("until", None) @@ -1331,7 +1341,9 @@ async def get_telemetry_stats(request: Request): pipeline.append( { "$group": { - "_id": {f: f"${f}" for f in group_by}, + "_id": { + f: "$runtime" if f == "lab" else f"${f}" for f in group_by + }, "total": {"$sum": 1}, "pass": { "$sum": {"$cond": [{"$eq": ["$result", "pass"]}, 1, 0]} @@ -1437,6 +1449,10 @@ async def get_telemetry_anomalies( } result_group = { "runtime": "$runtime", + # LAVA runtimes are configured one-to-one with labs. Keep the + # existing runtime field and expose the same grouping key as lab so + # clients can reason about cross-lab results explicitly. + "lab": "$runtime", "device_type": "$device_type", } if scope == "device": diff --git a/doc/api-details.md b/doc/api-details.md index 6621ece4..476bc669 100644 --- a/doc/api-details.md +++ b/doc/api-details.md @@ -1297,8 +1297,11 @@ curl 'https://api.kernelci.org/latest/telemetry/anomalies?scope=device&window=48 Physical-device scope uses only `job_result` events. This avoids counting the many test-case results emitted by one boot as independent device-health -samples. Each result includes `device_id`, `device_type`, `runtime`, result -counts and rates, the latest error and timestamp, plus these classifications: +samples. Each result includes `device_id`, `device_type`, `lab`, `runtime`, +result counts and rates, the latest error and timestamp, plus these +classifications. For LAVA device events, `lab` is the explicit name for the +existing runtime grouping key. Thus, a platform passing in one lab does not +hide the same platform or device identifier failing in another lab. - `constant_failure`: every job in the window was `fail` or `incomplete`. - `constant_infra_failure`: every job was classified as an infrastructure @@ -1306,9 +1309,9 @@ counts and rates, the latest error and timestamp, plus these classifications: lab connection is broken; inspect its raw telemetry errors to distinguish those cases. -The statistics endpoint also accepts `device_id` as a filter and a `group_by` -field for custom analysis: +The statistics endpoint accepts both `lab` and `device_id` as filters and +`group_by` fields for custom analysis: ```shell -curl 'https://api.kernelci.org/latest/telemetry/stats?group_by=runtime,device_id&kind=job_result&since=2026-07-20T00:00:00' +curl 'https://api.kernelci.org/latest/telemetry/stats?group_by=lab,device_id&kind=job_result&since=2026-07-20T00:00:00' ``` diff --git a/tests/unit_tests/test_telemetry_handler.py b/tests/unit_tests/test_telemetry_handler.py index f51f856f..02f46e3e 100644 --- a/tests/unit_tests/test_telemetry_handler.py +++ b/tests/unit_tests/test_telemetry_handler.py @@ -9,13 +9,13 @@ from kernelci.api.models import TelemetryEvent -def test_stats_groups_and_filters_by_device_id(mocker, test_client): - """Physical device IDs can be used in telemetry statistics.""" +def test_stats_groups_and_filters_by_lab_and_device_id(mocker, test_client): + """Lab and physical device IDs can be used in telemetry statistics.""" aggregate = AsyncMock( return_value=[ { "_id": { - "runtime": "lava-test", + "lab": "lava-test", "device_id": "board-01", }, "total": 4, @@ -32,20 +32,24 @@ def test_stats_groups_and_filters_by_device_id(mocker, test_client): response = test_client.get( "telemetry/stats", params={ - "group_by": "runtime,device_id", + "group_by": "lab,device_id", "kind": "job_result", + "lab": "lava-test", "device_id": "board-01", }, ) assert response.status_code == 200 + assert response.json()[0]["lab"] == "lava-test" assert response.json()[0]["device_id"] == "board-01" model, pipeline = aggregate.call_args.args assert model is TelemetryEvent assert pipeline[0]["$match"] == { "kind": "job_result", + "runtime": "lava-test", "device_id": "board-01", } + assert pipeline[1]["$group"]["_id"]["lab"] == "$runtime" assert pipeline[1]["$group"]["_id"]["device_id"] == "$device_id" @@ -57,6 +61,7 @@ def test_device_anomalies_use_job_results_and_physical_ids(mocker, test_client): { "_id": { "runtime": "lava-test", + "lab": "lava-test", "device_type": "board", "device_id": "board-01", }, @@ -96,6 +101,7 @@ def test_device_anomalies_use_job_results_and_physical_ids(mocker, test_client): assert data["result_anomalies"] == [ { "runtime": "lava-test", + "lab": "lava-test", "device_type": "board", "device_id": "board-01", "total": 3, @@ -121,9 +127,27 @@ def test_device_anomalies_use_job_results_and_physical_ids(mocker, test_client): assert pipeline[0]["$match"]["kind"] == "job_result" assert pipeline[0]["$match"]["device_id"] == {"$nin": [None, ""]} assert pipeline[1] == {"$sort": {"ts": -1}} + assert pipeline[2]["$group"]["_id"]["lab"] == "$runtime" assert pipeline[2]["$group"]["_id"]["device_id"] == "$device_id" +def test_stats_rejects_conflicting_lab_and_runtime(mocker, test_client): + aggregate = AsyncMock(return_value=[]) + mocker.patch("api.db.Database.aggregate", side_effect=aggregate) + + response = test_client.get( + "telemetry/stats", + params={ + "group_by": "lab,device_id", + "lab": "lava-collabora", + "runtime": "lava-cip", + }, + ) + + assert response.status_code == 400 + aggregate.assert_not_awaited() + + def test_device_anomalies_reject_invalid_scope(test_client): response = test_client.get("telemetry/anomalies?scope=lab") From f7c6354c7452d94811f94ce238f3336eaa7ccc23 Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Tue, 21 Jul 2026 07:35:17 +0300 Subject: [PATCH 4/6] api: centralize HTML template responses The HTML page handlers duplicate template path resolution, file reads, response construction, metrics, and cache headers. Some also use PlainTextResponse while manually overriding the content type. Introduce an allowlisted template helper that consistently returns HTMLResponse, centralize the cache policy and template directory, and reuse it across public pages. Update the stale root test and cover page content types, cache headers, and rejection of non-allowlisted templates. Signed-off-by: Denys Fedoryshchenko --- api/main.py | 125 ++++++++++---------------- tests/unit_tests/test_root_handler.py | 52 +++++++++-- 2 files changed, 93 insertions(+), 84 deletions(-) diff --git a/api/main.py b/api/main.py index acd69f60..a3782066 100644 --- a/api/main.py +++ b/api/main.py @@ -269,6 +269,46 @@ async def ensure_initial_admin_user(): ) from exc +TEMPLATES_DIR = os.path.join( + os.path.dirname(os.path.abspath(__file__)), "templates" +) + +# Allowlist of HTML templates that may be served to clients. Serving is +# restricted to these names so a helper can never be tricked into reading +# other files from the templates directory (e.g. email templates). +ALLOWED_HTML_TEMPLATES = { + "index.html", + "invite.html", + "accept-invite.html", + "viewer.html", + "dashboard.html", + "manage.html", + "analytics.html", + "stats.html", +} + +NO_CACHE_HEADERS = { + "Cache-Control": "no-cache, no-store, must-revalidate", + "Pragma": "no-cache", + "Expires": "0", +} + + +def _serve_template(name: str, no_cache: bool = True) -> HTMLResponse: + """Serve an HTML template page from the allowlist""" + metrics.add("http_requests_total", 1) + if name not in ALLOWED_HTML_TEMPLATES: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Template not found: {name}", + ) + page_path = os.path.join(TEMPLATES_DIR, name) + with open(page_path, "r", encoding="utf-8") as file: + return HTMLResponse( + file.read(), headers=NO_CACHE_HEADERS if no_cache else None + ) + + @app.exception_handler(ValueError) async def value_error_exception_handler(request: Request, exc: ValueError): """Global exception handler for 'ValueError'""" @@ -291,11 +331,7 @@ async def invalid_id_exception_handler(request: Request, exc: errors.InvalidId): @app.get("/") async def root(): """Root endpoint handler""" - metrics.add("http_requests_total", 1) - root_dir = os.path.dirname(os.path.abspath(__file__)) - index_path = os.path.join(root_dir, "templates", "index.html") - with open(index_path, "r", encoding="utf-8") as file: - return HTMLResponse(file.read()) + return _serve_template("index.html", no_cache=False) # ----------------------------------------------------------------------------- @@ -537,11 +573,7 @@ async def register( @app.get("/user/invite", response_class=HTMLResponse, include_in_schema=False) async def invite_user_page(): """Web UI for inviting a user (admin token required)""" - metrics.add("http_requests_total", 1) - root_dir = os.path.dirname(os.path.abspath(__file__)) - page_path = os.path.join(root_dir, "templates", "invite.html") - with open(page_path, "r", encoding="utf-8") as file: - return HTMLResponse(file.read()) + return _serve_template("invite.html", no_cache=False) @app.post( @@ -602,11 +634,7 @@ async def invite_user( ) async def accept_invite_page(): """Web UI for accepting an invite (sets password + verifies)""" - metrics.add("http_requests_total", 1) - root_dir = os.path.dirname(os.path.abspath(__file__)) - page_path = os.path.join(root_dir, "templates", "accept-invite.html") - with open(page_path, "r", encoding="utf-8") as file: - return HTMLResponse(file.read()) + return _serve_template("accept-invite.html", no_cache=False) @app.get("/user/invite/url", response_model=InviteUrlResponse, tags=["user"]) @@ -2207,98 +2235,43 @@ async def stats(user: User = Depends(get_current_superuser)): async def viewer(): """Serve simple HTML page to view the API /static/viewer.html Set various no-cache tag we might update it often""" - metrics.add("http_requests_total", 1) - root_dir = os.path.dirname(os.path.abspath(__file__)) - viewer_path = os.path.join(root_dir, "templates", "viewer.html") - with open(viewer_path, "r", encoding="utf-8") as file: - # set header to text/html and no-cache stuff - hdr = { - "Content-Type": "text/html", - "Cache-Control": "no-cache, no-store, must-revalidate", - "Pragma": "no-cache", - "Expires": "0", - } - return PlainTextResponse(file.read(), headers=hdr) + return _serve_template("viewer.html") @app.get("/dashboard") async def dashboard(): """Serve simple HTML page to view the API dashboard.html Set various no-cache tag we might update it often""" - metrics.add("http_requests_total", 1) - root_dir = os.path.dirname(os.path.abspath(__file__)) - dashboard_path = os.path.join(root_dir, "templates", "dashboard.html") - with open(dashboard_path, "r", encoding="utf-8") as file: - # set header to text/html and no-cache stuff - hdr = { - "Content-Type": "text/html", - "Cache-Control": "no-cache, no-store, must-revalidate", - "Pragma": "no-cache", - "Expires": "0", - } - return PlainTextResponse(file.read(), headers=hdr) + return _serve_template("dashboard.html") @app.get("/manage") async def manage(): """Serve simple HTML page to submit custom nodes""" - metrics.add("http_requests_total", 1) - root_dir = os.path.dirname(os.path.abspath(__file__)) - manage_path = os.path.join(root_dir, "templates", "manage.html") - with open(manage_path, "r", encoding="utf-8") as file: - # set header to text/html and no-cache stuff - hdr = { - "Content-Type": "text/html", - "Cache-Control": "no-cache, no-store, must-revalidate", - "Pragma": "no-cache", - "Expires": "0", - } - return PlainTextResponse(file.read(), headers=hdr) + return _serve_template("manage.html") @app.get("/analytics") async def analytics_page(): """Serve pipeline analytics dashboard with telemetry data""" - metrics.add("http_requests_total", 1) - root_dir = os.path.dirname(os.path.abspath(__file__)) - analytics_path = os.path.join(root_dir, "templates", "analytics.html") - with open(analytics_path, "r", encoding="utf-8") as file: - hdr = { - "Content-Type": "text/html", - "Cache-Control": "no-cache, no-store, must-revalidate", - "Pragma": "no-cache", - "Expires": "0", - } - return PlainTextResponse(file.read(), headers=hdr) + return _serve_template("analytics.html") @app.get("/stats") async def stats_page(): """Serve simple HTML page to view infrastructure statistics""" - metrics.add("http_requests_total", 1) - root_dir = os.path.dirname(os.path.abspath(__file__)) - stats_path = os.path.join(root_dir, "templates", "stats.html") - with open(stats_path, "r", encoding="utf-8") as file: - # set header to text/html and no-cache stuff - hdr = { - "Content-Type": "text/html", - "Cache-Control": "no-cache, no-store, must-revalidate", - "Pragma": "no-cache", - "Expires": "0", - } - return PlainTextResponse(file.read(), headers=hdr) + return _serve_template("stats.html") @app.get("/icons/{icon_name}") async def icons(icon_name: str): """Serve icons from /static/icons""" metrics.add("http_requests_total", 1) - root_dir = os.path.dirname(os.path.abspath(__file__)) if not re.match(r"^[A-Za-z0-9_.-]+\.png$", icon_name): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid icon name" ) - icon_path = os.path.join(root_dir, "templates", icon_name) + icon_path = os.path.join(TEMPLATES_DIR, icon_name) return FileResponse(icon_path) diff --git a/tests/unit_tests/test_root_handler.py b/tests/unit_tests/test_root_handler.py index f74955c5..5ca2d188 100644 --- a/tests/unit_tests/test_root_handler.py +++ b/tests/unit_tests/test_root_handler.py @@ -3,16 +3,52 @@ # Copyright (C) 2022 Jeny Sadadia # Author: Jeny Sadadia -"""Unit test function for KernelCI API root handler""" +"""Unit tests for KernelCI API HTML page handlers.""" + +import pytest +from fastapi import HTTPException + +from api.main import _serve_template def test_root_endpoint(test_client): - """ - Test Case : Test KernelCI API root endpoint - Expected Result : - HTTP Response Code 200 OK - JSON with 'message' key - """ + """The root endpoint serves the index as HTML without forced no-cache.""" response = test_client.get("/") + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/html") + assert "KernelCI API Server" in response.text + assert "cache-control" not in response.headers + + +@pytest.mark.parametrize( + "path, marker", + [ + ("viewer", "Maestro API Viewer"), + ("dashboard", "KernelCI Dashboard"), + ("manage", "Maestro API Manage"), + ("analytics", "KernelCI Pipeline Analytics"), + ("stats", "KernelCI API Statistics"), + ], +) +def test_dynamic_html_pages_disable_caching(test_client, path, marker): + """Frequently updated HTML tools use one consistent cache policy.""" + response = test_client.get(path) + assert response.status_code == 200 - assert response.json() == {"message": "KernelCI API"} + assert response.headers["content-type"].startswith("text/html") + assert marker in response.text + assert response.headers["cache-control"] == ( + "no-cache, no-store, must-revalidate" + ) + assert response.headers["pragma"] == "no-cache" + assert response.headers["expires"] == "0" + + +@pytest.mark.parametrize("name", ["invite-email.jinja2", "../index.html"]) +def test_template_helper_rejects_non_allowlisted_files(name): + """The shared helper cannot expose email templates or traverse paths.""" + with pytest.raises(HTTPException) as error: + _serve_template(name) + + assert error.value.status_code == 404 From 35ef89e30d3e587fa0d8ec7baf3c48d49fc245c2 Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Tue, 21 Jul 2026 07:35:36 +0300 Subject: [PATCH 5/6] tests: add reproducible local check runner Running the focused API checks currently requires manually creating isolated MongoDB and Redis services, assembling API and core dependencies in a disposable Python environment, and cleaning up each resource. Add a configurable Docker-based runner that performs the targeted pytest suite and pinned Ruff checks with read-only source mounts and guaranteed cleanup. It accepts custom pytest arguments and supports alternate core checkout and image paths through environment variables. Signed-off-by: Denys Fedoryshchenko --- scripts/run_local_checks.sh | 92 +++++++++++++++++++++++++++++++++++++ 1 file changed, 92 insertions(+) create mode 100755 scripts/run_local_checks.sh diff --git a/scripts/run_local_checks.sh b/scripts/run_local_checks.sh new file mode 100755 index 00000000..24a1221d --- /dev/null +++ b/scripts/run_local_checks.sh @@ -0,0 +1,92 @@ +#!/usr/bin/env bash +# SPDX-License-Identifier: LGPL-2.1-or-later +# +# Copyright (C) 2026 Collabora Limited + +set -euo pipefail + +script_dir=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd) +api_dir=$(cd -- "$script_dir/.." && pwd) +stack_dir=$(cd -- "$api_dir/.." && pwd) +core_dir=${KERNELCI_CORE_DIR:-"$stack_dir/kernelci-core"} + +python_image=${KERNELCI_TEST_PYTHON_IMAGE:-python:3.12-slim} +redis_image=${KERNELCI_TEST_REDIS_IMAGE:-redis:6.2} +mongo_image=${KERNELCI_TEST_MONGO_IMAGE:-mongo:5.0} +run_id="$$-${RANDOM}" +network="kernelci-api-checks-$run_id" +redis_container="kernelci-api-redis-$run_id" +mongo_container="kernelci-api-mongo-$run_id" + +if [[ ! -f "$core_dir/pyproject.toml" ]]; then + echo "kernelci-core not found at $core_dir" >&2 + echo "Set KERNELCI_CORE_DIR to its checkout path." >&2 + exit 2 +fi + +if [[ $# -gt 0 ]]; then + pytest_args=("$@") +else + pytest_args=( + -q + tests/unit_tests/test_db.py + tests/unit_tests/test_telemetry_handler.py + tests/unit_tests/test_root_handler.py + ) +fi + +cleanup() { + local exit_code=$? + trap - EXIT INT TERM + docker stop "$redis_container" "$mongo_container" >/dev/null 2>&1 || true + docker network rm "$network" >/dev/null 2>&1 || true + exit "$exit_code" +} +trap cleanup EXIT INT TERM + +echo "Creating isolated test services" +docker network create "$network" >/dev/null +docker run -d --rm \ + --name "$redis_container" \ + --network "$network" \ + --network-alias redis \ + "$redis_image" >/dev/null +docker run -d --rm \ + --name "$mongo_container" \ + --network "$network" \ + --network-alias db \ + "$mongo_image" >/dev/null + +echo "Running pytest and Ruff in $python_image" +docker run --rm \ + --network "$network" \ + --volume "$api_dir:/workspace/kernelci-api:ro" \ + --volume "$core_dir:/workspace/kernelci-core:ro" \ + --workdir /tmp \ + --env SECRET_KEY=test-secret \ + --env KCI_INITIAL_PASSWORD=test-password \ + --env PYTEST_ADDOPTS=--asyncio-mode=auto \ + --env PIP_ROOT_USER_ACTION=ignore \ + --env RUFF_CACHE_DIR=/tmp/ruff-cache \ + "$python_image" \ + sh -lc ' + cp -a /workspace/kernelci-api /tmp/kernelci-api + cp -a /workspace/kernelci-core /tmp/kernelci-core + pip install --disable-pip-version-check -q \ + "/tmp/kernelci-api[tests]" requests pyyaml jinja2 ruff==0.15.12 + export PYTHONPATH="/tmp/kernelci-core${PYTHONPATH:+:$PYTHONPATH}" + cd /tmp/kernelci-api + pytest "$@" + ruff check \ + api/db.py \ + api/main.py \ + tests/unit_tests/test_db.py \ + tests/unit_tests/test_telemetry_handler.py \ + tests/unit_tests/test_root_handler.py + ruff format --check \ + api/db.py \ + api/main.py \ + tests/unit_tests/test_db.py \ + tests/unit_tests/test_telemetry_handler.py \ + tests/unit_tests/test_root_handler.py + ' sh "${pytest_args[@]}" From 304ce55f5f1ea08248f2e46d8e5995aa48b6d390 Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Tue, 21 Jul 2026 08:31:55 +0300 Subject: [PATCH 6/6] db: reuse equivalent existing indexes MongoDB rejects creation when the same index definition already exists under a different name. Staging retains the event-history TTL definition as ttl_timestamp, so startup fails with IndexOptionsConflict when the API requests the generated timestamp_1 name. Inspect existing index keys and behavior-affecting options before creation, reuse only equivalent definitions regardless of name, and recover from equivalent indexes created concurrently. Preserve startup failures for genuinely different TTL, uniqueness, or filter options. Signed-off-by: Denys Fedoryshchenko --- api/db.py | 85 ++++++++++++++++++++++++++++++++++- tests/unit_tests/test_db.py | 90 +++++++++++++++++++++++++++++++++++++ 2 files changed, 173 insertions(+), 2 deletions(-) diff --git a/api/db.py b/api/db.py index a158eb35..b9ccee89 100644 --- a/api/db.py +++ b/api/db.py @@ -6,6 +6,8 @@ """Database abstraction""" +import logging + from beanie import init_beanie from bson import ObjectId from fastapi_pagination.ext.pymongo import apaginate as paginate @@ -16,11 +18,14 @@ TelemetryEvent, parse_node_obj, ) -from pymongo import AsyncMongoClient +from pymongo import AsyncMongoClient, IndexModel +from pymongo.errors import OperationFailure from redis import asyncio as aioredis from .models import User, UserGroup +logger = logging.getLogger(__name__) + class Database: """Database abstraction class @@ -51,6 +56,21 @@ class Database: BOOL_VALUE_MAP = {"true": True, "false": False} + _INDEX_OPTIONS = ( + "unique", + "sparse", + "expireAfterSeconds", + "partialFilterExpression", + "collation", + "wildcardProjection", + "hidden", + ) + _INDEX_OPTION_DEFAULTS = { + "unique": False, + "sparse": False, + "hidden": False, + } + def __init__(self, service="mongodb://db:27017", db_name="kernelci"): self._mongo = AsyncMongoClient(service) # TBD: Make redis host configurable @@ -94,6 +114,35 @@ async def del_kv(self, namespace, key): keyname = ":".join([namespace, key]) return await self._redis.delete(keyname) + @staticmethod + def _normalize_index_keys(field): + if isinstance(field, str): + return [(field, 1)] + return [tuple(key) for key in field] + + @classmethod + def _find_equivalent_index(cls, index_information, index): + desired_keys = cls._normalize_index_keys(index.field) + options = set(cls._INDEX_OPTIONS) | set(index.attributes) + options.discard("name") + options.discard("background") + + for name, existing in index_information.items(): + existing_keys = [tuple(key) for key in existing.get("key", [])] + if existing_keys != desired_keys: + continue + + for option in options: + default = cls._INDEX_OPTION_DEFAULTS.get(option) + if existing.get(option, default) != index.attributes.get( + option, default + ): + break + else: + return name + + return None + async def create_indexes(self): """Create indexes for models""" for model in self.COLLECTIONS: @@ -101,8 +150,40 @@ async def create_indexes(self): if not indexes: continue col = self._get_collection(model) + existing_indexes = await col.index_information() for index in indexes: - await col.create_index(index.field, **index.attributes) + requested_name = IndexModel( + index.field, **index.attributes + ).document["name"] + equivalent = self._find_equivalent_index( + existing_indexes, index + ) + if equivalent: + if equivalent != requested_name: + logger.warning( + "Reusing index %s for %s instead of equivalent %s", + equivalent, + model.__name__, + requested_name, + ) + continue + + try: + await col.create_index(index.field, **index.attributes) + except OperationFailure as exc: + if exc.code != 85: + raise + existing_indexes = await col.index_information() + equivalent = self._find_equivalent_index( + existing_indexes, index + ) + if not equivalent: + raise + logger.info( + "Reusing concurrently created index %s for %s", + equivalent, + model.__name__, + ) async def find_one(self, model, **kwargs): """Find one object with matching attributes diff --git a/tests/unit_tests/test_db.py b/tests/unit_tests/test_db.py index 36233df1..6c967275 100644 --- a/tests/unit_tests/test_db.py +++ b/tests/unit_tests/test_db.py @@ -4,18 +4,38 @@ """Unit tests for the database abstraction.""" +from types import SimpleNamespace from unittest.mock import AsyncMock, Mock import pytest from kernelci.api.models import TelemetryEvent +from pymongo.errors import OperationFailure from api.db import Database +EVENT_INDEX = SimpleNamespace( + field="timestamp", attributes={"expireAfterSeconds": 604800} +) + + +class EventModel: + @classmethod + def get_indexes(cls): + return [EVENT_INDEX] + + +def _database_with_collection(mocker, collection): + database = Database.__new__(Database) + database.COLLECTIONS = {EventModel: "eventhistory"} + mocker.patch.object(database, "_get_collection", return_value=collection) + return database + @pytest.mark.asyncio async def test_create_indexes_awaits_pymongo(mocker): """Every PyMongo AsyncCollection.create_index call is awaited.""" collection = Mock() + collection.index_information = AsyncMock(return_value={}) collection.create_index = AsyncMock() database = Database.__new__(Database) database.COLLECTIONS = {TelemetryEvent: "telemetry"} @@ -28,6 +48,76 @@ async def test_create_indexes_awaits_pymongo(mocker): ) +@pytest.mark.asyncio +async def test_create_indexes_reuses_equivalent_named_index(mocker): + collection = Mock() + collection.index_information = AsyncMock( + return_value={ + "ttl_timestamp": { + "v": 2, + "key": [("timestamp", 1)], + "expireAfterSeconds": 604800, + } + } + ) + collection.create_index = AsyncMock() + database = _database_with_collection(mocker, collection) + + await database.create_indexes() + + collection.create_index.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_create_indexes_rejects_conflicting_options(mocker): + collection = Mock() + collection.index_information = AsyncMock( + return_value={ + "ttl_timestamp": { + "v": 2, + "key": [("timestamp", 1)], + "expireAfterSeconds": 86400, + } + } + ) + conflict = OperationFailure("index options conflict", code=85) + collection.create_index = AsyncMock(side_effect=conflict) + database = _database_with_collection(mocker, collection) + + with pytest.raises(OperationFailure) as exc_info: + await database.create_indexes() + + assert exc_info.value is conflict + + +@pytest.mark.asyncio +async def test_create_indexes_accepts_concurrent_equivalent_index(mocker): + collection = Mock() + collection.index_information = AsyncMock( + side_effect=[ + {}, + { + "ttl_timestamp": { + "v": 2, + "key": [("timestamp", 1)], + "expireAfterSeconds": 604800, + } + }, + ] + ) + collection.create_index = AsyncMock( + side_effect=OperationFailure("index options conflict", code=85) + ) + database = _database_with_collection(mocker, collection) + + await database.create_indexes() + + collection.create_index.assert_awaited_once_with( + "timestamp", expireAfterSeconds=604800 + ) + assert collection.index_information.await_count == 2 + + @pytest.mark.asyncio async def test_aggregate_awaits_pymongo_cursor(mocker): """PyMongo AsyncCollection.aggregate is a coroutine."""