374 lines
13 KiB
Python
374 lines
13 KiB
Python
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": [],
|
|
"in_progress_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": [],
|
|
"in_progress_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"]] == [
|
|
"metis",
|
|
"ariadne",
|
|
]
|
|
assert [item["job"] for item in signals["in_progress_builds"]] == ["soteria"]
|
|
assert attached == ["metis", "ariadne"]
|