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"]