diff --git a/ariadne/services/testing_triage.py b/ariadne/services/testing_triage.py index c2de36d..02b7e7e 100644 --- a/ariadne/services/testing_triage.py +++ b/ariadne/services/testing_triage.py @@ -17,11 +17,11 @@ from .testing_triage_diagnosis import ( latest_testing_triage_diagnosis, # noqa: F401 - re-exported for app routes. model_diagnosis_enabled, ) +from .testing_triage_jenkins import jenkins_signals as _jenkins_signals from .testing_triage_scope import ( jenkins_suites, metric_suite_names, quality_items_in_scope, - suite_in_scope, ) @@ -29,13 +29,6 @@ logger = get_logger(__name__) TRIAGE_EVENT_TYPE = "testing_triage_bundle" _SUCCESS_STATUS = "ok|passed|success|not_applicable|skipped|na|n/a" -_JENKINS_TREE = ( - "jobs[name,url,color,lastBuild[number,result,timestamp,duration,url]," - "lastFailedBuild[number,timestamp,url],jobs[name,url,color," - "lastBuild[number,result,timestamp,duration,url],lastFailedBuild[number,timestamp,url]]]" -) -_MAX_JENKINS_LOG_LINES = 80 -_MAX_JENKINS_LOG_CHARS = 12000 _MAX_EVIDENCE_ITEMS = 12 @@ -158,7 +151,9 @@ def collect_testing_triage(storage: Storage | None = None) -> dict[str, Any]: return bundle -def _latest_cluster_snapshot(storage: Storage | None, errors: list[str]) -> dict[str, Any]: +def _latest_cluster_snapshot( + storage: Storage | None, errors: list[str] +) -> dict[str, Any]: if storage is not None: try: snapshot = storage.latest_cluster_state() @@ -176,12 +171,22 @@ def _latest_cluster_snapshot(storage: Storage | None, errors: list[str]) -> dict def _cluster_evidence(snapshot: dict[str, Any]) -> dict[str, Any]: - summary = snapshot.get("summary") if isinstance(snapshot.get("summary"), dict) else {} + summary = ( + snapshot.get("summary") if isinstance(snapshot.get("summary"), dict) else {} + ) flux = snapshot.get("flux") if isinstance(snapshot.get("flux"), dict) else {} - pod_issues = snapshot.get("pod_issues") if isinstance(snapshot.get("pod_issues"), dict) else {} + pod_issues = ( + snapshot.get("pod_issues") + if isinstance(snapshot.get("pod_issues"), dict) + else {} + ) jobs = snapshot.get("jobs") if isinstance(snapshot.get("jobs"), dict) else {} events = snapshot.get("events") if isinstance(snapshot.get("events"), dict) else {} - nodes = snapshot.get("nodes_summary") if isinstance(snapshot.get("nodes_summary"), dict) else {} + nodes = ( + snapshot.get("nodes_summary") + if isinstance(snapshot.get("nodes_summary"), dict) + else {} + ) return { "collected_at": snapshot.get("collected_at") or "", "health_bullets": _limit(summary.get("health_bullets")), @@ -204,7 +209,7 @@ def _cluster_evidence(snapshot: dict[str, Any]) -> dict[str, Any]: def _quality_signals(errors: list[str]) -> dict[str, Any]: queries = { "failed_runs_24h": ( - 'topk(12, sum by (suite) (increase(platform_quality_gate_runs_total' + "topk(12, sum by (suite) (increase(platform_quality_gate_runs_total" f'{{exported_job="platform-quality-ci",status!~"{_SUCCESS_STATUS}"}}[24h])))' ), "failing_checks_24h": ( @@ -212,7 +217,7 @@ def _quality_signals(errors: list[str]) -> dict[str, Any]: f'exported_job="platform-quality-ci",status!~"{_SUCCESS_STATUS}"}}[24h])))' ), "problem_tests_24h": ( - 'topk(20, sum by (suite,test,status) (increase(platform_quality_gate_test_case_result' + "topk(20, sum by (suite,test,status) (increase(platform_quality_gate_test_case_result" '{exported_job="platform-quality-ci",test!="",test!="__no_test_cases__",status="failed"}[24h])))' ), "jenkins_weather_failures": ( @@ -246,7 +251,9 @@ def _vm_items(query: str, errors: list[str]) -> list[dict[str, Any]]: return [] result = payload.get("data", {}).get("result") rows = result if isinstance(result, list) else [] - return [_vm_item(row) for row in rows[:_MAX_EVIDENCE_ITEMS] if isinstance(row, dict)] + return [ + _vm_item(row) for row in rows[:_MAX_EVIDENCE_ITEMS] if isinstance(row, dict) + ] def _vm_item(row: dict[str, Any]) -> dict[str, Any]: @@ -259,120 +266,6 @@ def _vm_item(row: dict[str, Any]) -> dict[str, Any]: } -def _jenkins_signals(errors: list[str]) -> dict[str, Any]: - base_url = settings.jenkins_base_url.strip().rstrip("/") - if not base_url: - return {"failed_builds": []} - try: - jobs = _fetch_jenkins_jobs(base_url) - 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"))] - 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]: - _attach_jenkins_log_tail(job, errors) - return {"failed_builds": failed[:_MAX_EVIDENCE_ITEMS]} - - -def _fetch_jenkins_jobs(base_url: str) -> list[dict[str, Any]]: - auth = _jenkins_auth() - kwargs: dict[str, Any] = {"timeout": settings.jenkins_api_timeout_sec, "follow_redirects": True} - if auth is not None: - kwargs["auth"] = auth - with httpx.Client(**kwargs) as client: - response = client.get(f"{base_url}/api/json", params={"tree": _JENKINS_TREE}) - response.raise_for_status() - payload = response.json() - items = payload.get("jobs") if isinstance(payload, dict) and isinstance(payload.get("jobs"), list) else [] - jobs: list[dict[str, Any]] = [] - for row in _flatten_jobs(items): - job = _jenkins_job(row) - if job is not None: - jobs.append(job) - return jobs - - -def _flatten_jobs(items: list[Any], prefix: str = "") -> list[dict[str, Any]]: - output: list[dict[str, Any]] = [] - for item in items: - if not isinstance(item, dict): - continue - name = item.get("name") - if not isinstance(name, str) or not name: - continue - full_name = f"{prefix}/{name}" if prefix else name - children = item.get("jobs") if isinstance(item.get("jobs"), list) else [] - if children: - output.extend(_flatten_jobs(children, full_name)) - if isinstance(item.get("lastBuild"), dict): - entry = dict(item) - entry["name"] = full_name - output.append(entry) - return output - - -def _jenkins_job(raw: dict[str, Any]) -> dict[str, Any] | None: - name = raw.get("name") - url = raw.get("url") - if not isinstance(name, str) or not isinstance(url, str): - return None - last_build = raw.get("lastBuild") if isinstance(raw.get("lastBuild"), dict) else {} - result = str(last_build.get("result") or "").upper() - status = _jenkins_status(raw, result) - return { - "job": name, - "job_url": url, - "status": status, - "result": result or "UNKNOWN", - "last_build_number": last_build.get("number"), - "last_run_ts": _millis_to_seconds(last_build.get("timestamp")), - "last_duration_seconds": _millis_to_seconds(last_build.get("duration")), - "console_url": str(last_build.get("url") or url).rstrip("/") + "/consoleText", - } - - -def _jenkins_status(raw: dict[str, Any], result: str) -> str: - color = str(raw.get("color") or "").lower() - if color.endswith("_anime"): - return "running" - if result == "SUCCESS": - return "success" - if result in {"FAILURE", "ABORTED", "UNSTABLE", "NOT_BUILT"}: - return "failure" - if color.startswith(("blue", "green")): - return "success" - if color.startswith(("red", "yellow")): - return "failure" - return "unknown" - - -def _attach_jenkins_log_tail(job: dict[str, Any], errors: list[str]) -> None: - url = job.get("console_url") - if not isinstance(url, str) or not url: - return - auth = _jenkins_auth() - kwargs: dict[str, Any] = {"timeout": settings.jenkins_api_timeout_sec, "follow_redirects": True} - if auth is not None: - kwargs["auth"] = auth - try: - with httpx.Client(**kwargs) as client: - response = client.get(url) - response.raise_for_status() - job["log_tail"] = _tail_text(response.text) - except Exception as exc: - errors.append(f"jenkins_log:{job.get('job')}: {exc}") - - -def _tail_text(text: str) -> str: - lines = text.splitlines()[-_MAX_JENKINS_LOG_LINES:] - tail = "\n".join(lines) - if len(tail) <= _MAX_JENKINS_LOG_CHARS: - return tail - return tail[-_MAX_JENKINS_LOG_CHARS:] - - def _summary( cluster: dict[str, Any], quality: dict[str, Any], @@ -403,10 +296,18 @@ def _failed_suites(quality: dict[str, Any]) -> set[str]: def _render_markdown(bundle: dict[str, Any]) -> str: summary = bundle.get("summary") if isinstance(bundle.get("summary"), dict) else {} - evidence = bundle.get("evidence") if isinstance(bundle.get("evidence"), dict) else {} - cluster = evidence.get("cluster") if isinstance(evidence.get("cluster"), dict) else {} - quality = evidence.get("quality") if isinstance(evidence.get("quality"), dict) else {} - jenkins = evidence.get("jenkins") if isinstance(evidence.get("jenkins"), dict) else {} + evidence = ( + bundle.get("evidence") if isinstance(bundle.get("evidence"), dict) else {} + ) + cluster = ( + evidence.get("cluster") if isinstance(evidence.get("cluster"), dict) else {} + ) + quality = ( + evidence.get("quality") if isinstance(evidence.get("quality"), dict) else {} + ) + jenkins = ( + evidence.get("jenkins") if isinstance(evidence.get("jenkins"), dict) else {} + ) lines = [ "# Testing Triage Evidence", "", @@ -426,7 +327,9 @@ def _render_markdown(bundle: dict[str, Any]) -> str: "## Jenkins", *_markdown_named_items("Failed builds", jenkins.get("failed_builds"), "job"), ] - unknowns = bundle.get("unknowns") if isinstance(bundle.get("unknowns"), list) else [] + unknowns = ( + bundle.get("unknowns") if isinstance(bundle.get("unknowns"), list) else [] + ) if unknowns: lines.extend(["", "## Unknowns", *_markdown_items(unknowns)]) return "\n".join(lines).strip() + "\n" @@ -476,15 +379,3 @@ def _float_value(value: Any) -> float: return float(value) except (TypeError, ValueError): return 0.0 - - -def _millis_to_seconds(value: Any) -> float: - return _float_value(value) / 1000.0 - - -def _jenkins_auth() -> tuple[str, str] | None: - username = settings.jenkins_api_user.strip() - token = settings.jenkins_api_token.strip() - if username and token: - return username, token - return None diff --git a/ariadne/services/testing_triage_jenkins.py b/ariadne/services/testing_triage_jenkins.py new file mode 100644 index 0000000..fabec10 --- /dev/null +++ b/ariadne/services/testing_triage_jenkins.py @@ -0,0 +1,251 @@ +from __future__ import annotations + +from typing import Any +from urllib.parse import quote + +import httpx + +from ..settings import settings +from .testing_triage_scope import suite_in_scope + + +_JENKINS_TREE = ( + "jobs[name,url,color,lastBuild[number,result,timestamp,duration,url,artifacts[fileName,relativePath]]," + "lastFailedBuild[number,timestamp,url],jobs[name,url,color," + "lastBuild[number,result,timestamp,duration,url,artifacts[fileName,relativePath]]," + "lastFailedBuild[number,timestamp,url]]]" +) +_MAX_JENKINS_LOG_LINES = 80 +_MAX_JENKINS_LOG_CHARS = 12000 +_MAX_JENKINS_ARTIFACTS = 20 +_MAX_JENKINS_ARTIFACT_CONTENTS = 6 +_MAX_JENKINS_ARTIFACT_CHARS = 6000 +_MAX_EVIDENCE_ITEMS = 12 +_JENKINS_EVIDENCE_ARTIFACTS = frozenset( + { + "coverage-summary.json", + "ironbank-compliance.json", + "loc-summary.json", + "quality-summary.json", + "sonarqube-quality-gate.json", + "test-summary.json", + } +) + + +def jenkins_signals(errors: list[str]) -> dict[str, Any]: + """Collect recent failed build, console, and retained artifact evidence.""" + + base_url = settings.jenkins_base_url.strip().rstrip("/") + if not base_url: + return {"failed_builds": []} + try: + jobs = _fetch_jenkins_jobs(base_url) + 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"))] + 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]: + _attach_jenkins_log_tail(job, errors) + _attach_jenkins_artifact_evidence(job, errors) + return {"failed_builds": failed[:_MAX_EVIDENCE_ITEMS]} + + +def _fetch_jenkins_jobs(base_url: str) -> list[dict[str, Any]]: + auth = _jenkins_auth() + kwargs: dict[str, Any] = { + "timeout": settings.jenkins_api_timeout_sec, + "follow_redirects": True, + } + if auth is not None: + kwargs["auth"] = auth + with httpx.Client(**kwargs) as client: + response = client.get(f"{base_url}/api/json", params={"tree": _JENKINS_TREE}) + response.raise_for_status() + payload = response.json() + items = ( + payload.get("jobs") + if isinstance(payload, dict) and isinstance(payload.get("jobs"), list) + else [] + ) + jobs: list[dict[str, Any]] = [] + for row in _flatten_jobs(items): + job = _jenkins_job(row) + if job is not None: + jobs.append(job) + return jobs + + +def _flatten_jobs(items: list[Any], prefix: str = "") -> list[dict[str, Any]]: + output: list[dict[str, Any]] = [] + for item in items: + if not isinstance(item, dict): + continue + name = item.get("name") + if not isinstance(name, str) or not name: + continue + full_name = f"{prefix}/{name}" if prefix else name + children = item.get("jobs") if isinstance(item.get("jobs"), list) else [] + if children: + output.extend(_flatten_jobs(children, full_name)) + if isinstance(item.get("lastBuild"), dict): + entry = dict(item) + entry["name"] = full_name + output.append(entry) + return output + + +def _jenkins_job(raw: dict[str, Any]) -> dict[str, Any] | None: + name = raw.get("name") + url = raw.get("url") + if not isinstance(name, str) or not isinstance(url, str): + return None + last_build = raw.get("lastBuild") if isinstance(raw.get("lastBuild"), dict) else {} + result = str(last_build.get("result") or "").upper() + status = _jenkins_status(raw, result) + build_url = str(last_build.get("url") or url).rstrip("/") + return { + "job": name, + "job_url": url, + "status": status, + "result": result or "UNKNOWN", + "last_build_number": last_build.get("number"), + "last_run_ts": _millis_to_seconds(last_build.get("timestamp")), + "last_duration_seconds": _millis_to_seconds(last_build.get("duration")), + "console_url": build_url + "/consoleText", + "artifacts": _jenkins_artifacts(last_build.get("artifacts"), build_url), + } + + +def _jenkins_artifacts(raw: Any, build_url: str) -> list[dict[str, Any]]: + """Return bounded Jenkins artifact metadata with safe evidence URLs.""" + + rows = raw if isinstance(raw, list) else [] + artifacts: list[dict[str, Any]] = [] + for row in rows[:_MAX_JENKINS_ARTIFACTS]: + if not isinstance(row, dict): + continue + file_name = row.get("fileName") + relative_path = row.get("relativePath") + if not isinstance(file_name, str) or not isinstance(relative_path, str): + continue + artifacts.append( + { + "file_name": file_name, + "relative_path": relative_path, + "url": f"{build_url}/artifact/{quote(relative_path, safe='/')}", + } + ) + return artifacts + + +def _jenkins_status(raw: dict[str, Any], result: str) -> str: + color = str(raw.get("color") or "").lower() + if color.endswith("_anime"): + return "running" + if result == "SUCCESS": + return "success" + if result in {"FAILURE", "ABORTED", "UNSTABLE", "NOT_BUILT"}: + return "failure" + if color.startswith(("blue", "green")): + return "success" + if color.startswith(("red", "yellow")): + return "failure" + return "unknown" + + +def _attach_jenkins_log_tail(job: dict[str, Any], errors: list[str]) -> None: + url = job.get("console_url") + if not isinstance(url, str) or not url: + return + kwargs = _client_kwargs() + try: + with httpx.Client(**kwargs) as client: + response = client.get(url) + response.raise_for_status() + job["log_tail"] = _tail_text(response.text) + except Exception as exc: + errors.append(f"jenkins_log:{job.get('job')}: {exc}") + + +def _attach_jenkins_artifact_evidence(job: dict[str, Any], errors: list[str]) -> None: + """Attach bounded text from known quality artifacts using Ariadne's read credential.""" + + raw_artifacts = job.get("artifacts") + artifacts = raw_artifacts if isinstance(raw_artifacts, list) else [] + selected = [ + artifact + for artifact in artifacts + if isinstance(artifact, dict) + and artifact.get("file_name") in _JENKINS_EVIDENCE_ARTIFACTS + ][:_MAX_JENKINS_ARTIFACT_CONTENTS] + if not selected: + return + + try: + with httpx.Client(**_client_kwargs()) as client: + for artifact in selected: + _attach_artifact_content(client, job, artifact, errors) + except Exception as exc: + errors.append(f"jenkins_artifacts:{job.get('job')}: {exc}") + + +def _attach_artifact_content( + client: httpx.Client, + job: dict[str, Any], + artifact: dict[str, Any], + errors: list[str], +) -> None: + url = artifact.get("url") + if not isinstance(url, str) or not url: + return + try: + response = client.get(url) + response.raise_for_status() + content = response.text + artifact["content"] = content[:_MAX_JENKINS_ARTIFACT_CHARS] + artifact["content_truncated"] = len(content) > _MAX_JENKINS_ARTIFACT_CHARS + except Exception as exc: + errors.append( + f"jenkins_artifact:{job.get('job')}:{artifact.get('relative_path')}: {exc}" + ) + + +def _client_kwargs() -> dict[str, Any]: + kwargs: dict[str, Any] = { + "timeout": settings.jenkins_api_timeout_sec, + "follow_redirects": True, + } + auth = _jenkins_auth() + if auth is not None: + kwargs["auth"] = auth + return kwargs + + +def _tail_text(text: str) -> str: + lines = text.splitlines()[-_MAX_JENKINS_LOG_LINES:] + tail = "\n".join(lines) + if len(tail) <= _MAX_JENKINS_LOG_CHARS: + return tail + return tail[-_MAX_JENKINS_LOG_CHARS:] + + +def _millis_to_seconds(value: Any) -> float: + try: + return float(value) / 1000.0 + except (TypeError, ValueError): + return 0.0 + + +def _jenkins_auth() -> tuple[str, str] | None: + username = settings.jenkins_api_user.strip() + token = settings.jenkins_api_token.strip() + if username and token: + return username, token + return None diff --git a/tests/test_testing_triage.py b/tests/test_testing_triage.py index 742c766..7276bd1 100644 --- a/tests/test_testing_triage.py +++ b/tests/test_testing_triage.py @@ -19,7 +19,9 @@ class DummyStorage: "nodes_summary": {"total": 3, "ready": 3, "not_ready": 0}, "flux": {"items": [{"namespace": "flux-system", "name": "monitoring"}]}, "pod_issues": { - "items": [{"namespace": "jenkins", "pod": "agent-1", "phase": "Pending"}], + "items": [ + {"namespace": "jenkins", "pod": "agent-1", "phase": "Pending"} + ], "pending_oldest": [{"namespace": "jenkins", "pod": "agent-1"}], }, "jobs": {"failing": [], "active_oldest": []}, @@ -62,7 +64,9 @@ def test_collect_testing_triage_builds_bundle(monkeypatch) -> None: if "platform_quality_gate_runs_total" in query else [], ) - monkeypatch.setattr(testing_triage, "_jenkins_signals", lambda errors: {"failed_builds": []}) + monkeypatch.setattr( + testing_triage, "_jenkins_signals", lambda errors: {"failed_builds": []} + ) bundle = testing_triage.collect_testing_triage(storage) @@ -70,7 +74,9 @@ def test_collect_testing_triage_builds_bundle(monkeypatch) -> None: assert bundle["summary"]["status"] == "needs_attention" assert bundle["summary"]["failed_suites"] == ["ariadne"] assert "Testing Triage Evidence" in bundle["markdown"] - assert bundle["openclaw"]["ariadne_latest_url"].endswith("/api/internal/testing/triage/latest") + assert bundle["openclaw"]["ariadne_latest_url"].endswith( + "/api/internal/testing/triage/latest" + ) def test_run_testing_triage_stores_latest(monkeypatch) -> None: @@ -91,6 +97,26 @@ def test_run_testing_triage_stores_latest(monkeypatch) -> None: assert latest["summary"]["status"] == "ok" +def test_run_testing_triage_diagnosis_stores_bundle_and_result(monkeypatch) -> None: + storage = DummyStorage() + bundle = {"summary": {"status": "needs_attention"}} + diagnosis = {"status": "ok", "model": "test-model"} + monkeypatch.setattr( + testing_triage, "collect_testing_triage", lambda _storage: bundle + ) + monkeypatch.setattr( + testing_triage, "diagnose_testing_triage", lambda _bundle: diagnosis + ) + + result = testing_triage.run_testing_triage_diagnosis(storage) + + assert result == diagnosis + assert storage.events == [ + (testing_triage.TRIAGE_EVENT_TYPE, bundle), + (testing_triage.TRIAGE_DIAGNOSIS_EVENT_TYPE, diagnosis), + ] + + def test_latest_testing_triage_bundle_handles_json_strings() -> None: class JsonStorage: def list_events(self, limit: int = 1, event_type: str | None = None): # type: ignore[no-untyped-def] @@ -160,7 +186,12 @@ def test_cluster_evidence_limits_and_defaults() -> None: "health_bullets": list(range(20)), "attention_ranked": [{"item": i} for i in range(20)], }, - "nodes_summary": {"total": 2, "ready": 1, "not_ready": 1, "not_ready_names": ["titan-06"]}, + "nodes_summary": { + "total": 2, + "ready": 1, + "not_ready": 1, + "not_ready_names": ["titan-06"], + }, "flux": {"items": [{"name": str(i)} for i in range(20)]}, "pod_issues": { "items": [{"pod": str(i)} for i in range(20)], @@ -206,7 +237,10 @@ def test_vm_items_handles_success_failure_and_bad_values(monkeypatch) -> None: "status": "success", "data": { "result": [ - {"metric": {"suite": "ariadne", "__name__": "ignored"}, "value": [1, "2.5"]}, + { + "metric": {"suite": "ariadne", "__name__": "ignored"}, + "value": [1, "2.5"], + }, {"metric": {"suite": "metis"}, "value": [1, "bad"]}, ] }, @@ -277,165 +311,6 @@ def test_vm_items_handles_disabled_and_query_failure(monkeypatch) -> None: assert errors == ["victoria_metrics: query failed"] -def test_jenkins_flatten_status_and_log_tail(monkeypatch) -> None: - rows = [ - "skip", - {"name": "", "lastBuild": {}}, - { - "name": "folder", - "jobs": [ - { - "name": "child", - "url": "http://jenkins/job/folder/job/child/", - "color": "red", - "lastBuild": { - "number": 7, - "result": "FAILURE", - "timestamp": 2000, - "duration": 3000, - "url": "http://jenkins/build/7/", - }, - } - ], - }, - {"name": "empty", "jobs": []}, - ] - - flattened = testing_triage._flatten_jobs(rows) # noqa: SLF001 - job = testing_triage._jenkins_job(flattened[0]) # noqa: SLF001 - - assert flattened[0]["name"] == "folder/child" - assert job is not None - assert job["status"] == "failure" - assert job["last_run_ts"] == 2.0 - assert job["last_duration_seconds"] == 3.0 - assert job["console_url"] == "http://jenkins/build/7/consoleText" - assert testing_triage._jenkins_status({"color": "blue_anime"}, "") == "running" # noqa: SLF001 - assert testing_triage._jenkins_status({}, "SUCCESS") == "success" # noqa: SLF001 - assert testing_triage._jenkins_status({"color": "green"}, "") == "success" # noqa: SLF001 - assert testing_triage._jenkins_status({"color": "yellow"}, "") == "failure" # noqa: SLF001 - assert testing_triage._jenkins_status({"color": "grey"}, "") == "unknown" # noqa: SLF001 - assert testing_triage._jenkins_job({"name": 1, "url": "u"}) is None # noqa: SLF001 - long_tail = testing_triage._tail_text("x" * (testing_triage._MAX_JENKINS_LOG_CHARS + 10)) # noqa: SLF001 - assert len(long_tail) == testing_triage._MAX_JENKINS_LOG_CHARS # noqa: SLF001 - assert testing_triage._tail_text("\n".join(str(i) for i in range(100))).startswith("20\n") # noqa: SLF001 - - -def test_attach_jenkins_log_tail_ignores_missing_url_and_records_errors(monkeypatch) -> None: - class BrokenClient: - def __init__(self, **kwargs) -> None: # type: ignore[no-untyped-def] - return None - - def __enter__(self): - return self - - def __exit__(self, *args) -> None: # type: ignore[no-untyped-def] - return None - - def get(self, url): # type: ignore[no-untyped-def] - raise RuntimeError("log gone") - - errors: list[str] = [] - testing_triage._attach_jenkins_log_tail({"job": "missing"}, errors) # noqa: SLF001 - assert errors == [] - - monkeypatch.setattr(testing_triage, "settings", SettingsStub(jenkins_api_timeout_sec=1)) - monkeypatch.setattr(testing_triage.httpx, "Client", BrokenClient) - testing_triage._attach_jenkins_log_tail({"job": "ariadne", "console_url": "http://jenkins/log"}, errors) # noqa: SLF001 - - assert errors == ["jenkins_log:ariadne: log gone"] - - -def test_fetch_jenkins_jobs_and_log_tail(monkeypatch) -> None: - class FakeResponse: - def __init__(self, payload=None, text="") -> None: # type: ignore[no-untyped-def] - self.payload = payload - self.text = text - - def raise_for_status(self) -> None: - return None - - def json(self): # type: ignore[no-untyped-def] - return self.payload - - class FakeClient: - def __init__(self, **kwargs) -> None: # type: ignore[no-untyped-def] - assert kwargs["auth"] == ("jenkins", "token") - - def __enter__(self): - return self - - def __exit__(self, *args) -> None: # type: ignore[no-untyped-def] - return None - - def get(self, url, params=None): # type: ignore[no-untyped-def] - if url.endswith("/api/json"): - assert "tree" in params - return FakeResponse( - { - "jobs": [ - { - "name": "ariadne", - "url": "http://jenkins/job/ariadne/", - "color": "red", - "lastBuild": {"number": 8, "result": "FAILURE", "timestamp": 1000, "duration": 1000}, - } - ] - } - ) - return FakeResponse(text="line1\nline2") - - monkeypatch.setattr( - testing_triage, - "settings", - SettingsStub(jenkins_api_user="jenkins", jenkins_api_token="token", jenkins_api_timeout_sec=3), - ) - monkeypatch.setattr(testing_triage.httpx, "Client", FakeClient) - - jobs = testing_triage._fetch_jenkins_jobs("http://jenkins") # noqa: SLF001 - errors: list[str] = [] - testing_triage._attach_jenkins_log_tail(jobs[0], errors) # noqa: SLF001 - - assert jobs[0]["job"] == "ariadne" - assert jobs[0]["status"] == "failure" - assert jobs[0]["log_tail"] == "line1\nline2" - assert errors == [] - - -def test_jenkins_signals_handles_disabled_and_failures(monkeypatch) -> None: - monkeypatch.setattr(testing_triage, "settings", SettingsStub(jenkins_base_url="")) - assert testing_triage._jenkins_signals([]) == {"failed_builds": []} # noqa: SLF001 - - monkeypatch.setattr(testing_triage, "settings", SettingsStub(jenkins_base_url="http://jenkins")) - monkeypatch.setattr( - testing_triage, - "_fetch_jenkins_jobs", - lambda base_url: (_ for _ in ()).throw(RuntimeError("boom")), - ) - errors: list[str] = [] - - assert testing_triage._jenkins_signals(errors) == {"failed_builds": []} # noqa: SLF001 - assert errors == ["jenkins: boom"] - - -def test_jenkins_signals_attaches_recent_failed_builds(monkeypatch) -> None: - jobs = [ - {"job": "ariadne", "status": "failure", "last_run_ts": 1}, - {"job": "pegasus", "status": "success", "last_run_ts": 5}, - {"job": "soteria", "status": "running", "last_run_ts": 10}, - {"job": "metis", "status": "unknown", "last_run_ts": 3}, - ] - 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"]] == ["soteria", "metis", "ariadne"] - assert attached == ["soteria", "metis", "ariadne"] - - def test_summary_and_markdown_helpers() -> None: quality = { "failed_runs_24h": {"items": [{"labels": {"suite": "ariadne"}, "value": 2}]}, @@ -465,5 +340,6 @@ def test_summary_and_markdown_helpers() -> None: assert "- Flux: monitoring" in markdown assert "- failed_runs_24h: {'suite': 'ariadne'} value=2" in markdown assert "- missing vm" in markdown - assert testing_triage._markdown_named_items("Pods", ["bad"], "pod") == ["- Pods: none"] # noqa: SLF001 - assert testing_triage._jenkins_auth() is None # noqa: SLF001 + assert testing_triage._markdown_named_items("Pods", ["bad"], "pod") == [ + "- Pods: none" + ] # noqa: SLF001 diff --git a/tests/test_testing_triage_jenkins.py b/tests/test_testing_triage_jenkins.py new file mode 100644 index 0000000..e75bd98 --- /dev/null +++ b/tests/test_testing_triage_jenkins.py @@ -0,0 +1,367 @@ +from __future__ import annotations + +from ariadne.services import testing_triage_jenkins + + +class SettingsStub: + def __init__(self, **overrides) -> None: # type: ignore[no-untyped-def] + 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_flatten_status_and_log_tail(monkeypatch) -> None: + rows = [ + "skip", + {"name": "", "lastBuild": {}}, + { + "name": "folder", + "jobs": [ + { + "name": "child", + "url": "http://jenkins/job/folder/job/child/", + "color": "red", + "lastBuild": { + "number": 7, + "result": "FAILURE", + "timestamp": 2000, + "duration": 3000, + "url": "http://jenkins/build/7/", + }, + } + ], + }, + {"name": "empty", "jobs": []}, + ] + + flattened = testing_triage_jenkins._flatten_jobs(rows) # noqa: SLF001 + job = testing_triage_jenkins._jenkins_job(flattened[0]) # noqa: SLF001 + + assert flattened[0]["name"] == "folder/child" + assert job is not None + assert job["status"] == "failure" + assert job["last_run_ts"] == 2.0 + assert job["last_duration_seconds"] == 3.0 + assert job["console_url"] == "http://jenkins/build/7/consoleText" + assert ( + testing_triage_jenkins._jenkins_status({"color": "blue_anime"}, "") == "running" + ) # noqa: SLF001 + assert testing_triage_jenkins._jenkins_status({}, "SUCCESS") == "success" # noqa: SLF001 + assert testing_triage_jenkins._jenkins_status({"color": "green"}, "") == "success" # noqa: SLF001 + assert testing_triage_jenkins._jenkins_status({"color": "yellow"}, "") == "failure" # noqa: SLF001 + assert testing_triage_jenkins._jenkins_status({"color": "grey"}, "") == "unknown" # noqa: SLF001 + assert testing_triage_jenkins._jenkins_job({"name": 1, "url": "u"}) is None # noqa: SLF001 + long_tail = testing_triage_jenkins._tail_text( + "x" * (testing_triage_jenkins._MAX_JENKINS_LOG_CHARS + 10) + ) # noqa: SLF001 + assert len(long_tail) == testing_triage_jenkins._MAX_JENKINS_LOG_CHARS # noqa: SLF001 + assert testing_triage_jenkins._tail_text( + "\n".join(str(i) for i in range(100)) + ).startswith("20\n") # noqa: SLF001 + + +def test_attach_jenkins_log_tail_ignores_missing_url_and_records_errors( + monkeypatch, +) -> None: + class BrokenClient: + def __init__(self, **kwargs) -> None: # type: ignore[no-untyped-def] + return None + + def __enter__(self): + return self + + def __exit__(self, *args) -> None: # type: ignore[no-untyped-def] + return None + + def get(self, url): # type: ignore[no-untyped-def] + raise RuntimeError("log gone") + + errors: list[str] = [] + testing_triage_jenkins._attach_jenkins_log_tail({"job": "missing"}, errors) # noqa: SLF001 + assert errors == [] + + monkeypatch.setattr( + testing_triage_jenkins, "settings", SettingsStub(jenkins_api_timeout_sec=1) + ) + monkeypatch.setattr(testing_triage_jenkins.httpx, "Client", BrokenClient) + testing_triage_jenkins._attach_jenkins_log_tail( + {"job": "ariadne", "console_url": "http://jenkins/log"}, errors + ) # noqa: SLF001 + + assert errors == ["jenkins_log:ariadne: log gone"] + + +def test_fetch_jenkins_jobs_and_log_tail(monkeypatch) -> None: + class FakeResponse: + def __init__(self, payload=None, text="") -> None: # type: ignore[no-untyped-def] + self.payload = payload + self.text = text + + def raise_for_status(self) -> None: + return None + + def json(self): # type: ignore[no-untyped-def] + return self.payload + + class FakeClient: + def __init__(self, **kwargs) -> None: # type: ignore[no-untyped-def] + assert kwargs["auth"] == ("jenkins", "token") + + def __enter__(self): + return self + + def __exit__(self, *args) -> None: # type: ignore[no-untyped-def] + return None + + def get(self, url, params=None): # type: ignore[no-untyped-def] + if url.endswith("/api/json"): + assert "tree" in params + return FakeResponse( + { + "jobs": [ + { + "name": "ariadne", + "url": "http://jenkins/job/ariadne/", + "color": "red", + "lastBuild": { + "number": 8, + "result": "FAILURE", + "timestamp": 1000, + "duration": 1000, + "url": "http://jenkins/build/8/", + "artifacts": [ + { + "fileName": "quality-summary.json", + "relativePath": "build/quality summary.json", + }, + { + "fileName": "binary.zip", + "relativePath": "build/binary.zip", + }, + ], + }, + } + ] + } + ) + if url.endswith("/consoleText"): + return FakeResponse(text="line1\nline2") + return FakeResponse(text='{"status":"failed"}') + + monkeypatch.setattr( + testing_triage_jenkins, + "settings", + SettingsStub( + jenkins_api_user="jenkins", + jenkins_api_token="token", + jenkins_api_timeout_sec=3, + ), + ) + monkeypatch.setattr(testing_triage_jenkins.httpx, "Client", FakeClient) + + jobs = testing_triage_jenkins._fetch_jenkins_jobs("http://jenkins") # noqa: SLF001 + errors: list[str] = [] + testing_triage_jenkins._attach_jenkins_log_tail(jobs[0], errors) # noqa: SLF001 + testing_triage_jenkins._attach_jenkins_artifact_evidence(jobs[0], errors) # noqa: SLF001 + + assert jobs[0]["job"] == "ariadne" + assert jobs[0]["status"] == "failure" + assert jobs[0]["log_tail"] == "line1\nline2" + assert jobs[0]["artifacts"] == [ + { + "file_name": "quality-summary.json", + "relative_path": "build/quality summary.json", + "url": "http://jenkins/build/8/artifact/build/quality%20summary.json", + "content": '{"status":"failed"}', + "content_truncated": False, + }, + { + "file_name": "binary.zip", + "relative_path": "build/binary.zip", + "url": "http://jenkins/build/8/artifact/build/binary.zip", + }, + ] + assert errors == [] + + +def test_jenkins_artifacts_are_bounded_and_ignore_bad_rows() -> None: + rows = [ + "bad", + {"fileName": 1, "relativePath": "bad"}, + {"fileName": "test-summary.json", "relativePath": "build/test-summary.json"}, + ] + + artifacts = testing_triage_jenkins._jenkins_artifacts( + rows, "http://jenkins/build/1" + ) # noqa: SLF001 + + assert artifacts == [ + { + "file_name": "test-summary.json", + "relative_path": "build/test-summary.json", + "url": "http://jenkins/build/1/artifact/build/test-summary.json", + } + ] + assert ( + testing_triage_jenkins._jenkins_artifacts(None, "http://jenkins/build/1") == [] + ) # noqa: SLF001 + + +def test_attach_jenkins_artifacts_records_fetch_errors_and_truncates( + monkeypatch, +) -> None: + class FakeResponse: + def __init__(self, text: str, broken: bool = False) -> None: + self.text = text + self.broken = broken + + def raise_for_status(self) -> None: + if self.broken: + raise RuntimeError("artifact gone") + + class FakeClient: + def __init__(self, **kwargs) -> None: # type: ignore[no-untyped-def] + return None + + def __enter__(self): + return self + + def __exit__(self, *args) -> None: # type: ignore[no-untyped-def] + return None + + def get(self, url): # type: ignore[no-untyped-def] + if url.endswith("missing.json"): + return FakeResponse("", broken=True) + return FakeResponse( + "x" * (testing_triage_jenkins._MAX_JENKINS_ARTIFACT_CHARS + 1) + ) # noqa: SLF001 + + monkeypatch.setattr( + testing_triage_jenkins, "settings", SettingsStub(jenkins_api_timeout_sec=1) + ) + monkeypatch.setattr(testing_triage_jenkins.httpx, "Client", FakeClient) + job = { + "job": "soteria", + "artifacts": [ + { + "file_name": "quality-summary.json", + "relative_path": "build/quality-summary.json", + "url": "http://jenkins/artifact/quality-summary.json", + }, + { + "file_name": "test-summary.json", + "relative_path": "build/missing.json", + "url": "http://jenkins/artifact/missing.json", + }, + ], + } + errors: list[str] = [] + + testing_triage_jenkins._attach_jenkins_artifact_evidence(job, errors) # noqa: SLF001 + + assert ( + len(job["artifacts"][0]["content"]) + == testing_triage_jenkins._MAX_JENKINS_ARTIFACT_CHARS + ) # noqa: SLF001 + assert job["artifacts"][0]["content_truncated"] is True + assert "content" not in job["artifacts"][1] + assert errors == ["jenkins_artifact:soteria:build/missing.json: artifact gone"] + + +def test_attach_jenkins_artifacts_records_client_failure(monkeypatch) -> None: + class BrokenClient: + def __init__(self, **kwargs) -> None: # type: ignore[no-untyped-def] + return None + + def __enter__(self): + raise RuntimeError("client unavailable") + + def __exit__(self, *args) -> None: # type: ignore[no-untyped-def] + return None + + monkeypatch.setattr(testing_triage_jenkins.httpx, "Client", BrokenClient) + job = { + "job": "soteria", + "artifacts": [ + { + "file_name": "quality-summary.json", + "relative_path": "build/quality-summary.json", + "url": "http://jenkins/artifact/quality-summary.json", + } + ], + } + errors: list[str] = [] + + testing_triage_jenkins._attach_jenkins_artifact_evidence(job, errors) # noqa: SLF001 + + assert errors == ["jenkins_artifacts:soteria: client unavailable"] + + +def test_jenkins_helper_edges() -> None: + errors: list[str] = [] + testing_triage_jenkins._attach_artifact_content( # noqa: SLF001 + None, # type: ignore[arg-type] + {"job": "soteria"}, + {"url": None}, + errors, + ) + + assert errors == [] + assert testing_triage_jenkins._millis_to_seconds("bad") == 0.0 # noqa: SLF001 + + +def test_jenkins_signals_handles_disabled_and_failures(monkeypatch) -> None: + monkeypatch.setattr( + testing_triage_jenkins, "settings", SettingsStub(jenkins_base_url="") + ) + assert testing_triage_jenkins.jenkins_signals([]) == {"failed_builds": []} + + monkeypatch.setattr( + testing_triage_jenkins, + "settings", + SettingsStub(jenkins_base_url="http://jenkins"), + ) + monkeypatch.setattr( + testing_triage_jenkins, + "_fetch_jenkins_jobs", + lambda base_url: (_ for _ in ()).throw(RuntimeError("boom")), + ) + errors: list[str] = [] + + assert testing_triage_jenkins.jenkins_signals(errors) == {"failed_builds": []} + assert errors == ["jenkins: boom"] + + +def test_jenkins_signals_attaches_recent_failed_builds(monkeypatch) -> None: + jobs = [ + {"job": "ariadne", "status": "failure", "last_run_ts": 1}, + {"job": "pegasus", "status": "success", "last_run_ts": 5}, + {"job": "soteria", "status": "running", "last_run_ts": 10}, + {"job": "metis", "status": "unknown", "last_run_ts": 3}, + ] + attached: list[str] = [] + monkeypatch.setattr( + testing_triage_jenkins, + "settings", + SettingsStub(jenkins_base_url="http://jenkins"), + ) + monkeypatch.setattr( + testing_triage_jenkins, "_fetch_jenkins_jobs", lambda base_url: jobs + ) + monkeypatch.setattr( + testing_triage_jenkins, + "_attach_jenkins_log_tail", + lambda job, errors: attached.append(job["job"]), + ) + + signals = testing_triage_jenkins.jenkins_signals([]) + + assert [item["job"] for item in signals["failed_builds"]] == [ + "soteria", + "metis", + "ariadne", + ] + assert attached == ["soteria", "metis", "ariadne"] diff --git a/tests/test_testing_triage_scope.py b/tests/test_testing_triage_scope.py index 147a3da..4e7fc95 100644 --- a/tests/test_testing_triage_scope.py +++ b/tests/test_testing_triage_scope.py @@ -1,6 +1,6 @@ from __future__ import annotations -from ariadne.services import testing_triage +from ariadne.services import testing_triage, testing_triage_jenkins class SettingsStub: @@ -24,11 +24,21 @@ def test_jenkins_signals_filters_to_in_scope_suite_jobs(monkeypatch) -> None: {"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"])) + monkeypatch.setattr( + testing_triage_jenkins, + "settings", + SettingsStub(jenkins_base_url="http://jenkins"), + ) + monkeypatch.setattr( + testing_triage_jenkins, "_fetch_jenkins_jobs", lambda base_url: jobs + ) + monkeypatch.setattr( + testing_triage_jenkins, + "_attach_jenkins_log_tail", + lambda job, errors: attached.append(job["job"]), + ) - signals = testing_triage._jenkins_signals([]) # noqa: SLF001 + signals = testing_triage_jenkins.jenkins_signals([]) assert [item["job"] for item in signals["failed_builds"]] == [ "data-prepper",