Files
wehub-resource-sync 4b6817381b
CI (OpenClaw E2E) / openclaw test (push) Has been cancelled
CI / coverage-report (push) Has been cancelled
CI / test-kubernetes (push) Has been cancelled
CI / should-run-thorough (push) Has been cancelled
CI / test-thorough (cloudwatch-demo) (push) Has been cancelled
CI / test-thorough (flink-ecs) (push) Has been cancelled
CI / test-thorough (upstream-lambda) (push) Has been cancelled
CI / test-thorough (prefect-ecs-fargate) (push) Has been cancelled
Release / build-binaries (zip, opensre.exe, onefile, windows-latest, windows-x64) (push) Has been cancelled
Benchmark image — build + push to ECR (any adapter) / build + push (push) Has been cancelled
CI / quality (ubuntu-latest) (push) Has been cancelled
CI / test (tools-runtime) (push) Has been cancelled
CI / test (e2e-general) (push) Has been cancelled
CI / test (cli-runtime) (push) Has been cancelled
CI / test (e2e-provider-and-openclaw) (push) Has been cancelled
CI / test (integrations-and-misc) (push) Has been cancelled
Release / verify (push) Has been cancelled
Release / build-python-dist (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-15-intel, darwin-x64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-latest, darwin-arm64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04, linux-x64) (push) Has been cancelled
Release / publish-release (push) Has been cancelled
Release / publish-main-release (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-checks (no-LLM) (push) Has been cancelled
CodeQL / Analyze (python) (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-live shard ${{ matrix.shard_index }} (push) Has been cancelled
Release / prepare (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04-arm, linux-arm64) (push) Has been cancelled
Synthetic Deterministic Tests / Synthetic offline (deterministic) (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:10:45 +08:00

699 lines
28 KiB
Python

"""Unit tests for the Jenkins integration.
Covers the config layer, the REST client (against ``httpx.MockTransport`` —
no live Jenkins), the response-shaping helpers, and the investigation tools.
"""
from __future__ import annotations
from collections.abc import Callable
from typing import Any
import httpx
import pytest
from pydantic import ValidationError
from integrations import jenkins as jenkins_module
from integrations.jenkins import (
JenkinsConfig,
build_jenkins_config,
jenkins_config_from_env,
validate_jenkins_config,
)
from integrations.jenkins.client import (
JenkinsClient,
_iso_from_ms,
_job_api_path,
_safe_job_name,
_status_from_color,
make_jenkins_client,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
Handler = Callable[[httpx.Request], httpx.Response]
class _FakeResponse:
"""Minimal stand-in for httpx.Response used to mock module-level httpx.request."""
def __init__(self, payload: Any, status_code: int = 200) -> None:
self._payload = payload
self.status_code = status_code
self.text = ""
def raise_for_status(self) -> None:
if self.status_code >= 400:
raise httpx.HTTPStatusError(
"error",
request=httpx.Request("GET", "http://jenkins.local"),
response=httpx.Response(self.status_code),
)
def json(self) -> Any:
return self._payload
def _client_with_handler(handler: Handler, monkeypatch: pytest.MonkeyPatch) -> JenkinsClient:
"""Build a JenkinsClient whose HTTP calls route through a MockTransport."""
config = JenkinsConfig(base_url="http://jenkins.local", username="u", api_token="t")
client = JenkinsClient(config)
mock = httpx.Client(
base_url=config.api_base_url,
transport=httpx.MockTransport(handler),
)
monkeypatch.setattr(client, "_get_client", lambda: mock)
return client
# ---------------------------------------------------------------------------
# Config layer
# ---------------------------------------------------------------------------
class TestJenkinsConfig:
def test_api_base_url_strips_trailing_slash(self) -> None:
cfg = JenkinsConfig(base_url="http://jenkins.local/")
assert cfg.api_base_url == "http://jenkins.local"
def test_base_url_whitespace_normalized(self) -> None:
cfg = JenkinsConfig(base_url=" http://jenkins.local ")
assert cfg.api_base_url == "http://jenkins.local"
def test_auth_is_username_token_tuple(self) -> None:
cfg = JenkinsConfig(base_url="http://x", username="alice", api_token="secret")
assert cfg.auth == ("alice", "secret")
def test_is_configured_requires_url_username_and_token(self) -> None:
assert JenkinsConfig(base_url="http://x", username="u", api_token="t").is_configured
assert not JenkinsConfig(base_url="http://x", api_token="t").is_configured # no username
assert not JenkinsConfig(base_url="http://x", username="u").is_configured # no token
assert not JenkinsConfig(username="u", api_token="t").is_configured # no url
def test_timeout_zero_raises(self) -> None:
with pytest.raises(ValidationError):
JenkinsConfig(base_url="http://x", timeout_seconds=0)
def test_unknown_field_rejected(self) -> None:
with pytest.raises(ValidationError):
JenkinsConfig.model_validate({"base_url": "http://x", "bad_field": 1})
class TestBuildAndEnvConfig:
def test_build_from_empty(self) -> None:
cfg = build_jenkins_config(None)
assert cfg.base_url == ""
def test_env_returns_none_without_url(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("JENKINS_URL", raising=False)
monkeypatch.setenv("JENKINS_API_TOKEN", "t")
assert jenkins_config_from_env() is None
def test_env_returns_none_without_token(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("JENKINS_URL", "http://jenkins.local")
monkeypatch.delenv("JENKINS_API_TOKEN", raising=False)
assert jenkins_config_from_env() is None
def test_env_loads_full_config(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("JENKINS_URL", "http://jenkins.local")
monkeypatch.setenv("JENKINS_USER", "alice")
monkeypatch.setenv("JENKINS_API_TOKEN", "tok")
cfg = jenkins_config_from_env()
assert cfg is not None
assert cfg.api_base_url == "http://jenkins.local"
assert cfg.username == "alice"
assert cfg.api_token == "tok"
class TestValidateConfig:
def test_fails_without_base_url(self) -> None:
result = validate_jenkins_config(JenkinsConfig(base_url="", api_token="t"))
assert not result.ok
assert "base URL" in result.detail
def test_fails_without_token(self) -> None:
result = validate_jenkins_config(JenkinsConfig(base_url="http://x", username="u"))
assert not result.ok
assert "token" in result.detail
def test_fails_without_username(self) -> None:
result = validate_jenkins_config(
JenkinsConfig(base_url="http://x", username="", api_token="t")
)
assert not result.ok
assert "username" in result.detail
def test_fails_on_missing_scheme(self) -> None:
result = validate_jenkins_config(
JenkinsConfig(base_url="localhost:8080", username="u", api_token="t")
)
assert not result.ok
assert "http" in result.detail.lower()
def test_passes_on_success(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
jenkins_module.httpx,
"request",
lambda *_, **__: _FakeResponse({"nodeName": "controller"}),
)
result = validate_jenkins_config(
JenkinsConfig(base_url="http://jenkins.local", username="u", api_token="t")
)
assert result.ok
assert "controller" in result.detail
def test_fails_on_http_error(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
jenkins_module.httpx,
"request",
lambda *_, **__: _FakeResponse({}, status_code=401),
)
result = validate_jenkins_config(
JenkinsConfig(base_url="http://jenkins.local", api_token="bad")
)
assert not result.ok
# ---------------------------------------------------------------------------
# Shaping helpers
# ---------------------------------------------------------------------------
class TestHelpers:
def test_iso_from_ms_converts(self) -> None:
# 1780150032692 ms -> 2026-05-30 (verifies ms, not seconds)
assert _iso_from_ms(1780150032692).startswith("2026-05-30")
def test_iso_from_ms_invalid_returns_empty(self) -> None:
assert _iso_from_ms(None) == ""
assert _iso_from_ms("not-a-number") == ""
assert _iso_from_ms(0) == ""
def test_status_from_color(self) -> None:
assert _status_from_color("blue") == ("SUCCESS", False)
assert _status_from_color("red") == ("FAILURE", False)
assert _status_from_color("yellow") == ("UNSTABLE", False)
def test_status_from_color_anime_means_building(self) -> None:
status, building = _status_from_color("blue_anime")
assert status == "SUCCESS"
assert building is True
def test_status_from_color_unknown(self) -> None:
assert _status_from_color("") == ("UNKNOWN", False)
def test_safe_job_name_validates(self) -> None:
assert _safe_job_name("demo-fail") == "demo-fail"
assert _safe_job_name("team/payment-service") == "team/payment-service" # folder
assert _safe_job_name(" team / svc ") == "team/svc" # trims segments
assert _safe_job_name("../etc") is None
assert _safe_job_name("a\\b") is None
assert _safe_job_name("a//b") is None # empty segment
assert _safe_job_name("team/ /svc") is None # whitespace-only segment
assert _safe_job_name("") is None
def test_coerce_build_number(self) -> None:
from integrations.jenkins.client import _coerce_build_number
assert _coerce_build_number(4) == 4
assert _coerce_build_number("7") == 7
assert _coerce_build_number(0) is None
assert _coerce_build_number(-1) is None
assert _coerce_build_number("abc") is None
assert _coerce_build_number(None) is None
assert _coerce_build_number(True) is None # bool is not a build number
def test_job_api_path_maps_folders(self) -> None:
assert _job_api_path("demo") == "job/demo"
assert _job_api_path("team/payment-service") == "job/team/job/payment-service"
# ---------------------------------------------------------------------------
# Service client
# ---------------------------------------------------------------------------
_BUILDS_PAYLOAD = {
"builds": [
{"number": 4, "result": "FAILURE", "timestamp": 1780150032692, "duration": 11, "url": "u4"},
{"number": 3, "result": "SUCCESS", "timestamp": 1780150031599, "duration": 10, "url": "u3"},
{"number": 2, "result": None, "timestamp": 1780150030571, "building": True, "url": "u2"},
]
}
class TestListBuilds:
def test_returns_shaped_builds(self, monkeypatch: pytest.MonkeyPatch) -> None:
def handler(request: httpx.Request) -> httpx.Response:
assert request.url.path == "/job/demo/api/json"
return httpx.Response(200, json=_BUILDS_PAYLOAD)
client = _client_with_handler(handler, monkeypatch)
result = client.list_builds("demo")
assert result["success"]
assert result["total"] == 3
assert result["builds"][0]["status"] == "FAILURE"
assert result["builds"][2]["status"] == "RUNNING" # null result -> RUNNING
assert len(result["failed_builds"]) == 1
def test_status_filter(self, monkeypatch: pytest.MonkeyPatch) -> None:
client = _client_with_handler(
lambda _: httpx.Response(200, json=_BUILDS_PAYLOAD), monkeypatch
)
result = client.list_builds("demo", status="success")
assert result["total"] == 1
assert result["builds"][0]["status"] == "SUCCESS"
def test_invalid_job_name(self, monkeypatch: pytest.MonkeyPatch) -> None:
client = _client_with_handler(lambda _: httpx.Response(200, json={}), monkeypatch)
result = client.list_builds("../evil")
assert not result["success"]
assert "invalid job name" in result["error"]
def test_folder_job_maps_to_nested_path(self, monkeypatch: pytest.MonkeyPatch) -> None:
def handler(request: httpx.Request) -> httpx.Response:
assert request.url.path == "/job/team/job/svc/api/json"
return httpx.Response(200, json=_BUILDS_PAYLOAD)
client = _client_with_handler(handler, monkeypatch)
result = client.list_builds("team/svc")
assert result["success"]
assert result["job"] == "team/svc"
def test_server_side_cap_in_tree_query(self, monkeypatch: pytest.MonkeyPatch) -> None:
captured: dict[str, str] = {}
def handler(request: httpx.Request) -> httpx.Response:
captured["tree"] = request.url.params.get("tree", "")
return httpx.Response(200, json=_BUILDS_PAYLOAD)
client = _client_with_handler(handler, monkeypatch)
client.list_builds("demo", limit=5)
# range specifier caps the server response (no full-history transfer)
assert "{0,5}" in captured["tree"]
def test_http_error_returns_failure(self, monkeypatch: pytest.MonkeyPatch) -> None:
client = _client_with_handler(lambda _: httpx.Response(404, text="nope"), monkeypatch)
result = client.list_builds("demo")
assert not result["success"]
assert "404" in result["error"]
def test_handles_non_dict_response(self, monkeypatch: pytest.MonkeyPatch) -> None:
client = _client_with_handler(
lambda _: httpx.Response(200, json=["unexpected"]), monkeypatch
)
result = client.list_builds("demo")
assert result["success"] is True
assert result["builds"] == []
def test_handles_null_builds_field(self, monkeypatch: pytest.MonkeyPatch) -> None:
client = _client_with_handler(
lambda _: httpx.Response(200, json={"builds": None}), monkeypatch
)
result = client.list_builds("demo")
assert result["success"] is True
assert result["builds"] == []
class TestGetBuildLog:
def test_returns_log(self, monkeypatch: pytest.MonkeyPatch) -> None:
def handler(request: httpx.Request) -> httpx.Response:
assert request.url.path == "/job/demo/4/consoleText"
return httpx.Response(200, text="ERROR: boom\nFinished: FAILURE")
client = _client_with_handler(handler, monkeypatch)
result = client.get_build_log("demo", 4)
assert result["success"]
assert "ERROR: boom" in result["log"]
assert result["truncated"] is False
def test_tail_truncation(self, monkeypatch: pytest.MonkeyPatch) -> None:
body = "x" * 10 + "TAIL_MARKER"
client = _client_with_handler(lambda _: httpx.Response(200, text=body), monkeypatch)
result = client.get_build_log("demo", 1, max_chars=5)
assert result["truncated"] is True
assert result["log"] == "ARKER" # keeps the tail
def test_invalid_build_number(self, monkeypatch: pytest.MonkeyPatch) -> None:
client = _client_with_handler(lambda _: httpx.Response(200, text=""), monkeypatch)
result = client.get_build_log("demo", "abc") # type: ignore[arg-type]
assert not result["success"]
def test_rejects_nonpositive_build_number(self, monkeypatch: pytest.MonkeyPatch) -> None:
client = _client_with_handler(lambda _: httpx.Response(200, text=""), monkeypatch)
zero = client.get_build_log("demo", 0)
assert not zero["success"]
assert "invalid build number" in zero["error"]
assert not client.get_build_log("demo", -3)["success"]
class TestGetPipelineStages:
def test_parses_stages_for_pipeline_build(self, monkeypatch: pytest.MonkeyPatch) -> None:
payload = {
"name": "#5",
"status": "FAILED",
"stages": [
{"name": "Build", "status": "SUCCESS", "durationMillis": 1200},
{"name": "Deploy", "status": "FAILED", "durationMillis": 800},
],
}
def handler(request: httpx.Request) -> httpx.Response:
assert request.url.path == "/job/pipe/5/wfapi/describe"
return httpx.Response(200, json=payload)
client = _client_with_handler(handler, monkeypatch)
result = client.get_pipeline_stages("pipe", 5)
assert result["success"]
assert result["is_pipeline"] is True
assert [s["name"] for s in result["stages"]] == ["Build", "Deploy"]
assert result["stages"][1]["status"] == "FAILED"
def test_freestyle_404_returns_empty_not_error(self, monkeypatch: pytest.MonkeyPatch) -> None:
client = _client_with_handler(lambda _: httpx.Response(404, text="Not Found"), monkeypatch)
result = client.get_pipeline_stages("demo-fail", 4)
assert result["success"] is True
assert result["is_pipeline"] is False
assert result["stages"] == []
def test_invalid_build_number(self, monkeypatch: pytest.MonkeyPatch) -> None:
client = _client_with_handler(lambda _: httpx.Response(200, json={}), monkeypatch)
result = client.get_pipeline_stages("pipe", "abc") # type: ignore[arg-type]
assert not result["success"]
class TestListJobs:
def test_tree_descends_folders(self, monkeypatch: pytest.MonkeyPatch) -> None:
captured: dict[str, str] = {}
def handler(request: httpx.Request) -> httpx.Response:
captured["tree"] = request.url.params.get("tree", "")
return httpx.Response(200, json={"jobs": []})
client = _client_with_handler(handler, monkeypatch)
result = client.list_jobs()
# nested jobs[...] in the tree means folders are descended
assert captured["tree"].count("jobs[") > 1
assert result["truncated"] is False
def test_flattens_folder_jobs(self, monkeypatch: pytest.MonkeyPatch) -> None:
payload = {
"jobs": [
{"name": "top", "url": "ut", "color": "blue", "lastBuild": {"number": 2}},
{
"name": "team",
"jobs": [
{
"name": "payment-service",
"url": "up",
"color": "red",
"lastBuild": {"number": 7},
}
],
},
]
}
client = _client_with_handler(lambda _: httpx.Response(200, json=payload), monkeypatch)
result = client.list_jobs()
statuses = {j["name"]: j["status"] for j in result["jobs"]}
# folder job reported by its full folder/job path
assert statuses == {"top": "SUCCESS", "team/payment-service": "FAILURE"}
def test_decodes_color_status(self, monkeypatch: pytest.MonkeyPatch) -> None:
payload = {
"jobs": [
{
"name": "demo-fail",
"url": "uf",
"color": "red",
"lastBuild": {"number": 4, "timestamp": 1780150032692},
},
{"name": "demo-pass", "url": "up", "color": "blue", "lastBuild": {"number": 3}},
]
}
client = _client_with_handler(lambda _: httpx.Response(200, json=payload), monkeypatch)
result = client.list_jobs()
assert result["success"]
statuses = {j["name"]: j["status"] for j in result["jobs"]}
assert statuses == {"demo-fail": "FAILURE", "demo-pass": "SUCCESS"}
class TestListRunningBuilds:
def test_filters_building(self, monkeypatch: pytest.MonkeyPatch) -> None:
payload = {
"jobs": [
{
"name": "demo",
"builds": [
{"number": 5, "building": True, "timestamp": 1780150032692, "url": "u5"},
{"number": 4, "building": False, "result": "SUCCESS", "url": "u4"},
],
}
]
}
captured: dict[str, str] = {}
def handler(request: httpx.Request) -> httpx.Response:
captured["tree"] = request.url.params.get("tree", "")
return httpx.Response(200, json=payload)
client = _client_with_handler(handler, monkeypatch)
result = client.list_running_builds()
assert result["total"] == 1
assert result["running_builds"][0]["number"] == 5
assert result["running_builds"][0]["status"] == "RUNNING"
# per-job build cap present, and the tree descends folders
assert "{0,5}" in captured["tree"]
assert captured["tree"].count("jobs[") > 1
def test_running_builds_in_folder(self, monkeypatch: pytest.MonkeyPatch) -> None:
payload = {
"jobs": [
{
"name": "team",
"jobs": [
{
"name": "deploy",
"builds": [
{
"number": 9,
"building": True,
"timestamp": 1780150032692,
"url": "u9",
}
],
}
],
}
]
}
client = _client_with_handler(lambda _: httpx.Response(200, json=payload), monkeypatch)
result = client.list_running_builds()
assert result["total"] == 1
assert result["running_builds"][0]["job"] == "team/deploy"
assert result["running_builds"][0]["number"] == 9
class TestClientErrorPaths:
@pytest.mark.parametrize(
("method_name", "args"),
[
("list_builds", ("demo",)),
("get_build_log", ("demo", 1)),
("get_pipeline_stages", ("demo", 1)),
("list_jobs", ()),
("list_running_builds", ()),
],
)
def test_client_methods_return_failure_on_timeout(
self,
method_name: str,
args: tuple[Any, ...],
monkeypatch: pytest.MonkeyPatch,
) -> None:
def handler(_: httpx.Request) -> httpx.Response:
raise httpx.TimeoutException("jenkins timed out")
client = _client_with_handler(handler, monkeypatch)
result = getattr(client, method_name)(*args)
assert result["success"] is False
assert "jenkins timed out" in result["error"]
@pytest.mark.parametrize(
("method_name", "args"),
[
("list_builds", ("demo",)),
("get_pipeline_stages", ("demo", 1)),
("list_jobs", ()),
("list_running_builds", ()),
],
)
def test_json_methods_return_failure_on_invalid_json(
self,
method_name: str,
args: tuple[Any, ...],
monkeypatch: pytest.MonkeyPatch,
) -> None:
def handler(_: httpx.Request) -> httpx.Response:
return httpx.Response(200, content=b"not-json")
client = _client_with_handler(handler, monkeypatch)
result = getattr(client, method_name)(*args)
assert result["success"] is False
assert "expecting value" in result["error"].lower()
class TestMakeClient:
def test_returns_none_without_creds(self) -> None:
assert make_jenkins_client("", api_token="") is None
assert make_jenkins_client("http://x", "u", "") is None # no token
assert make_jenkins_client("", "u", "t") is None # no url
assert make_jenkins_client("http://x", "", "t") is None # no username
assert make_jenkins_client("http://x", None, "t") is None # username defaults to None
def test_builds_client_with_creds(self) -> None:
client = make_jenkins_client("http://jenkins.local", "alice", "tok")
assert isinstance(client, JenkinsClient)
assert client.config.auth == ("alice", "tok")
# ---------------------------------------------------------------------------
# Tools
# ---------------------------------------------------------------------------
class _FakeToolClient:
"""Context-managed fake client returning canned method results for tool tests."""
def __init__(self, **results: Any) -> None:
self._results = results
def __enter__(self) -> _FakeToolClient:
return self
def __exit__(self, *_: object) -> None:
pass
def list_builds(self, *_: Any, **__: Any) -> dict[str, Any]:
return self._results["list_builds"]
def list_jobs(self, *_: Any, **__: Any) -> dict[str, Any]:
return self._results["list_jobs"]
class TestTools:
def test_availability_requires_verified_connection(self) -> None:
import integrations.jenkins.tools as jenkins_tool
assert not jenkins_tool._jenkins_available({"jenkins": {}})
assert not jenkins_tool._jenkins_available({"jenkins": {"connection_verified": False}})
assert jenkins_tool._jenkins_available({"jenkins": {"connection_verified": True}})
def test_build_tool_extract_params_soft_defaults_job_name(self) -> None:
import integrations.jenkins.tools as jenkins_tool
# job_name absent from sources -> empty default (LLM supplies it as a tool arg)
params = jenkins_tool._list_jenkins_builds_extract_params(
{"jenkins": {"connection_verified": True}}
)
assert params["job_name"] == ""
def test_creds_mapping_from_source_dict(self) -> None:
import integrations.jenkins.tools as jenkins_tool
creds = jenkins_tool._jenkins_creds(
{"base_url": "http://x", "username": "u", "api_token": "t"}
)
assert creds == {"jenkins_url": "http://x", "jenkins_user": "u", "jenkins_token": "t"}
def test_resolve_client_needs_both_url_and_token_explicitly(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
import integrations.jenkins.tools as jenkins_tool
seen: list[tuple] = []
monkeypatch.setattr(jenkins_tool, "jenkins_config_from_env", lambda: None)
monkeypatch.setattr(
jenkins_tool,
"make_jenkins_client",
lambda url, user, token: seen.append((url, user, token)) or "client",
)
# only url, no token -> falls through to env (None here), explicit path skipped
assert jenkins_tool._resolve_client("http://x", None, None) is None
assert seen == []
# both present -> explicit path
assert jenkins_tool._resolve_client("http://x", "u", "t") == "client"
assert seen[-1] == ("http://x", "u", "t")
def test_resolve_client_env_path_requires_complete_config(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
import integrations.jenkins.tools as jenkins_tool
# env has url+token but no username -> jenkins_config_from_env returns a
# config, but it is not is_configured, so no client is built.
incomplete = JenkinsConfig(base_url="http://x", api_token="t")
assert not incomplete.is_configured
monkeypatch.setattr(jenkins_tool, "jenkins_config_from_env", lambda: incomplete)
called: list = []
monkeypatch.setattr(
jenkins_tool, "make_jenkins_client", lambda *a: called.append(a) or "client"
)
assert jenkins_tool._resolve_client(None, None, None) is None
assert called == []
def test_resolve_client_explicit_path_without_username_returns_none(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
import integrations.jenkins.tools as jenkins_tool
# url + token explicitly present but no username (and no env) -> the
# factory refuses to build an empty-username client -> None.
monkeypatch.setattr(jenkins_tool, "jenkins_config_from_env", lambda: None)
assert jenkins_tool._resolve_client("http://x", None, "t") is None
def test_not_configured_when_no_client(self, monkeypatch: pytest.MonkeyPatch) -> None:
import integrations.jenkins.tools as jenkins_tool
monkeypatch.setattr(jenkins_tool, "_resolve_client", lambda *_a, **_k: None)
result = jenkins_tool.list_jenkins_builds("demo")
assert result["available"] is False
assert "not configured" in result["error"]
def test_list_builds_tool_shapes_result(self, monkeypatch: pytest.MonkeyPatch) -> None:
import integrations.jenkins.tools as jenkins_tool
fake = _FakeToolClient(
list_builds={
"success": True,
"job": "demo",
"builds": [{"number": 4, "status": "FAILURE"}],
"failed_builds": [{"number": 4, "status": "FAILURE"}],
"total": 1,
}
)
monkeypatch.setattr(jenkins_tool, "_resolve_client", lambda *_a, **_k: fake)
result = jenkins_tool.list_jenkins_builds("demo", jenkins_url="http://x", jenkins_token="t")
assert result["available"] is True
assert result["source"] == "jenkins"
assert result["total"] == 1
assert result["failed_builds"][0]["number"] == 4
def test_tools_registered_in_registry(self) -> None:
from tools.registry import get_registered_tools
names = {t.name for t in get_registered_tools() if t.source == "jenkins"}
assert names == {
"list_jenkins_builds",
"get_jenkins_build_log",
"get_jenkins_pipeline_stages",
"list_jenkins_jobs",
"list_jenkins_running_builds",
}