triage: split testing scope helpers

ariadne/services/testing_triage.py

@@ -17,6 +17,12 @@ from .testing_triage_diagnosis import (
     latest_testing_triage_diagnosis,  # noqa: F401 - re-exported for app routes.
     model_diagnosis_enabled,
 )
+from .testing_triage_scope import (
+    jenkins_suites,
+    metric_suite_names,
+    quality_items_in_scope,
+    suite_in_scope,
+)
 
 
 logger = get_logger(__name__)
@@ -31,24 +37,6 @@ _JENKINS_TREE = (
 _MAX_JENKINS_LOG_LINES = 80
 _MAX_JENKINS_LOG_CHARS = 12000
 _MAX_EVIDENCE_ITEMS = 12
-_IN_SCOPE_TEST_SUITES = frozenset(
-    {
-        "ananke",
-        "ariadne",
-        "atlasbot",
-        "bstein_home",
-        "data_prepper",
-        "metis",
-        "pegasus",
-        "soteria",
-        "titan_iac",
-    }
-)
-_TEST_SUITE_ALIASES = {
-    "bstein-dev-home": "bstein_home",
-    "data-prepper": "data_prepper",
-    "titan-iac": "titan_iac",
-}
 
 
 @dataclass(frozen=True)
@@ -235,7 +223,7 @@ def _quality_signals(errors: list[str]) -> dict[str, Any]:
     return {
         name: {
             "query": query,
-            "items": _scope_quality_items(_vm_items(query, errors)),
+            "items": quality_items_in_scope(_vm_items(query, errors)),
         }
         for name, query in queries.items()
     }
@@ -280,7 +268,7 @@ def _jenkins_signals(errors: list[str]) -> dict[str, Any]:
     except Exception as exc:
         errors.append(f"jenkins: {exc}")
         return {"failed_builds": []}
-    scoped_jobs = [job for job in jobs if _suite_in_scope(job.get("job"))]
+    scoped_jobs = [job for job in jobs if suite_in_scope(job.get("job"))]
     failed = [job for job in scoped_jobs if job.get("status") in {"failure", "running", "unknown"}]
     failed.sort(key=lambda item: -(item.get("last_run_ts") or 0))
     for job in failed[:3]:
@@ -391,7 +379,7 @@ def _summary(
     jenkins: dict[str, Any],
     errors: list[str],
 ) -> dict[str, Any]:
-    failed_suites = sorted(_failed_suites(quality) | _jenkins_suites(jenkins))
+    failed_suites = sorted(metric_suite_names(quality) | jenkins_suites(jenkins))
     problem_count = (
         len(cluster.get("flux_not_ready") or [])
         + len(cluster.get("pod_issues") or [])
@@ -410,52 +398,7 @@ def _summary(
 
 
 def _failed_suites(quality: dict[str, Any]) -> set[str]:
-    suites: set[str] = set()
+    return metric_suite_names(quality)
-    for bucket in quality.values():
-        if not isinstance(bucket, dict):
-            continue
-        for item in bucket.get("items") or []:
-            labels = item.get("labels") if isinstance(item, dict) else {}
-            suite = labels.get("suite") if isinstance(labels, dict) else None
-            if isinstance(suite, str) and suite:
-                canonical = _canonical_suite_name(suite)
-                if canonical in _IN_SCOPE_TEST_SUITES:
-                    suites.add(canonical)
-    return suites
-
-
-def _jenkins_suites(jenkins: dict[str, Any]) -> set[str]:
-    suites: set[str] = set()
-    for item in jenkins.get("failed_builds") or []:
-        if not isinstance(item, dict):
-            continue
-        canonical = _canonical_suite_name(item.get("job"))
-        if canonical in _IN_SCOPE_TEST_SUITES:
-            suites.add(canonical)
-    return suites
-
-
-def _scope_quality_items(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    return [item for item in items if _quality_item_in_scope(item)]
-
-
-def _quality_item_in_scope(item: dict[str, Any]) -> bool:
-    labels = item.get("labels") if isinstance(item, dict) else {}
-    if not isinstance(labels, dict):
-        return True
-    raw_suite = labels.get("suite") or labels.get("exported_job") or labels.get("job")
-    return _suite_in_scope(raw_suite) if raw_suite else True
-
-
-def _suite_in_scope(value: Any) -> bool:
-    return _canonical_suite_name(value) in _IN_SCOPE_TEST_SUITES
-
-
-def _canonical_suite_name(value: Any) -> str:
-    name = str(value or "").strip().lower()
-    if "/" in name:
-        name = name.rsplit("/", 1)[-1]
-    return _TEST_SUITE_ALIASES.get(name, name.replace("-", "_"))
 
 
 def _render_markdown(bundle: dict[str, Any]) -> str:

ariadne/services/testing_triage_scope.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+IN_SCOPE_TEST_SUITES = frozenset(
+    {
+        "ananke",
+        "ariadne",
+        "atlasbot",
+        "bstein_home",
+        "data_prepper",
+        "metis",
+        "pegasus",
+        "soteria",
+        "titan_iac",
+    }
+)
+_TEST_SUITE_ALIASES = {
+    "bstein-dev-home": "bstein_home",
+    "data-prepper": "data_prepper",
+    "titan-iac": "titan_iac",
+}
+
+
+def metric_suite_names(quality: dict[str, Any]) -> set[str]:
+    suites: set[str] = set()
+    for bucket in quality.values():
+        if not isinstance(bucket, dict):
+            continue
+        for item in bucket.get("items") or []:
+            labels = item.get("labels") if isinstance(item, dict) else {}
+            suite = labels.get("suite") if isinstance(labels, dict) else None
+            if isinstance(suite, str) and suite:
+                canonical = canonical_suite_name(suite)
+                if canonical in IN_SCOPE_TEST_SUITES:
+                    suites.add(canonical)
+    return suites
+
+
+def jenkins_suites(jenkins: dict[str, Any]) -> set[str]:
+    suites: set[str] = set()
+    for item in jenkins.get("failed_builds") or []:
+        if not isinstance(item, dict):
+            continue
+        canonical = canonical_suite_name(item.get("job"))
+        if canonical in IN_SCOPE_TEST_SUITES:
+            suites.add(canonical)
+    return suites
+
+
+def quality_items_in_scope(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    return [item for item in items if _quality_item_in_scope(item)]
+
+
+def suite_in_scope(value: Any) -> bool:
+    return canonical_suite_name(value) in IN_SCOPE_TEST_SUITES
+
+
+def canonical_suite_name(value: Any) -> str:
+    name = str(value or "").strip().lower()
+    if "/" in name:
+        name = name.rsplit("/", 1)[-1]
+    return _TEST_SUITE_ALIASES.get(name, name.replace("-", "_"))
+
+
+def _quality_item_in_scope(item: dict[str, Any]) -> bool:
+    labels = item.get("labels") if isinstance(item, dict) else {}
+    if not isinstance(labels, dict):
+        return True
+    raw_suite = labels.get("suite") or labels.get("exported_job") or labels.get("job")
+    return suite_in_scope(raw_suite) if raw_suite else True

tests/test_testing_triage.py

@@ -436,51 +436,6 @@ def test_jenkins_signals_attaches_recent_failed_builds(monkeypatch) -> None:
     assert attached == ["soteria", "metis", "ariadne"]
 
 
-def test_jenkins_signals_filters_to_in_scope_suite_jobs(monkeypatch) -> None:
-    jobs = [
-        {"job": "lesavka", "status": "running", "last_run_ts": 50},
-        {"job": "harbor-arm-build", "status": "failure", "last_run_ts": 40},
-        {"job": "data-prepper", "status": "running", "last_run_ts": 30},
-        {"job": "folder/ariadne", "status": "failure", "last_run_ts": 20},
-        {"job": "bstein-dev-home", "status": "unknown", "last_run_ts": 10},
-    ]
-    attached: list[str] = []
-    monkeypatch.setattr(testing_triage, "settings", SettingsStub(jenkins_base_url="http://jenkins"))
-    monkeypatch.setattr(testing_triage, "_fetch_jenkins_jobs", lambda base_url: jobs)
-    monkeypatch.setattr(testing_triage, "_attach_jenkins_log_tail", lambda job, errors: attached.append(job["job"]))
-
-    signals = testing_triage._jenkins_signals([])  # noqa: SLF001
-
-    assert [item["job"] for item in signals["failed_builds"]] == [
-        "data-prepper",
-        "folder/ariadne",
-        "bstein-dev-home",
-    ]
-    assert attached == ["data-prepper", "folder/ariadne", "bstein-dev-home"]
-
-
-def test_quality_signals_filters_to_in_scope_suites(monkeypatch) -> None:
-    rows = [
-        {"labels": {"suite": "ariadne"}, "value": 1.0},
-        {"labels": {"suite": "lesavka"}, "value": 1.0},
-        {"labels": {"suite": "typhon"}, "value": 1.0},
-        {"labels": {"exported_job": "titan-iac"}, "value": 1.0},
-        {"labels": {"exported_job": "harbor-arm-build"}, "value": 1.0},
-        {"labels": {}, "value": 1.0},
-    ]
-    monkeypatch.setattr(testing_triage, "settings", SettingsStub(vm_url="http://vm"))
-    monkeypatch.setattr(testing_triage, "_vm_items", lambda query, errors: rows)
-
-    quality = testing_triage._quality_signals([])  # noqa: SLF001
-
-    assert quality["failed_runs_24h"]["items"] == [
-        {"labels": {"suite": "ariadne"}, "value": 1.0},
-        {"labels": {"exported_job": "titan-iac"}, "value": 1.0},
-        {"labels": {}, "value": 1.0},
-    ]
-    assert testing_triage._failed_suites(quality) == {"ariadne"}  # noqa: SLF001
-
-
 def test_summary_and_markdown_helpers() -> None:
     quality = {
         "failed_runs_24h": {"items": [{"labels": {"suite": "ariadne"}, "value": 2}]},

tests/test_testing_triage_scope.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+from ariadne.services import testing_triage
+
+
+class SettingsStub:
+    def __init__(self, **overrides) -> None:  # type: ignore[no-untyped-def]
+        self.vm_url = ""
+        self.cluster_state_vm_timeout_sec = 1.0
+        self.jenkins_base_url = ""
+        self.jenkins_api_user = ""
+        self.jenkins_api_token = ""
+        self.jenkins_api_timeout_sec = 1.0
+        for key, value in overrides.items():
+            setattr(self, key, value)
+
+
+def test_jenkins_signals_filters_to_in_scope_suite_jobs(monkeypatch) -> None:
+    jobs = [
+        {"job": "lesavka", "status": "running", "last_run_ts": 50},
+        {"job": "harbor-arm-build", "status": "failure", "last_run_ts": 40},
+        {"job": "data-prepper", "status": "running", "last_run_ts": 30},
+        {"job": "folder/ariadne", "status": "failure", "last_run_ts": 20},
+        {"job": "bstein-dev-home", "status": "unknown", "last_run_ts": 10},
+    ]
+    attached: list[str] = []
+    monkeypatch.setattr(testing_triage, "settings", SettingsStub(jenkins_base_url="http://jenkins"))
+    monkeypatch.setattr(testing_triage, "_fetch_jenkins_jobs", lambda base_url: jobs)
+    monkeypatch.setattr(testing_triage, "_attach_jenkins_log_tail", lambda job, errors: attached.append(job["job"]))
+
+    signals = testing_triage._jenkins_signals([])  # noqa: SLF001
+
+    assert [item["job"] for item in signals["failed_builds"]] == [
+        "data-prepper",
+        "folder/ariadne",
+        "bstein-dev-home",
+    ]
+    assert attached == ["data-prepper", "folder/ariadne", "bstein-dev-home"]
+
+
+def test_quality_signals_filters_to_in_scope_suites(monkeypatch) -> None:
+    rows = [
+        {"labels": {"suite": "ariadne"}, "value": 1.0},
+        {"labels": {"suite": "lesavka"}, "value": 1.0},
+        {"labels": {"suite": "typhon"}, "value": 1.0},
+        {"labels": {"exported_job": "titan-iac"}, "value": 1.0},
+        {"labels": {"exported_job": "harbor-arm-build"}, "value": 1.0},
+        {"labels": {}, "value": 1.0},
+    ]
+    monkeypatch.setattr(testing_triage, "settings", SettingsStub(vm_url="http://vm"))
+    monkeypatch.setattr(testing_triage, "_vm_items", lambda query, errors: rows)
+
+    quality = testing_triage._quality_signals([])  # noqa: SLF001
+
+    assert quality["failed_runs_24h"]["items"] == [
+        {"labels": {"suite": "ariadne"}, "value": 1.0},
+        {"labels": {"exported_job": "titan-iac"}, "value": 1.0},
+        {"labels": {}, "value": 1.0},
+    ]
+    assert testing_triage._failed_suites(quality) == {"ariadne"}  # noqa: SLF001