diff --git a/airflow/providers/jenkins/hooks/jenkins.py b/airflow/providers/jenkins/hooks/jenkins.py index db1284a59a53b..f061d96e6cf47 100644 --- a/airflow/providers/jenkins/hooks/jenkins.py +++ b/airflow/providers/jenkins/hooks/jenkins.py @@ -19,7 +19,6 @@ import jenkins -from airflow import AirflowException from airflow.hooks.base import BaseHook from airflow.utils.strings import to_boolean @@ -40,7 +39,7 @@ def __init__(self, conn_id: str = default_conn_name) -> None: # connection.extra contains info about using https (true) or http (false) if to_boolean(connection.extra): connection_prefix = "https" - url = f"{connection_prefix}://{connection.host}:{connection.port}" + url = f"{connection_prefix}://{connection.host}:{connection.port}/{connection.schema}" self.log.info("Trying to connect to %s", url) self.jenkins_server = jenkins.Jenkins(url, connection.login, connection.password) @@ -48,19 +47,22 @@ def get_jenkins_server(self) -> jenkins.Jenkins: """Get jenkins server""" return self.jenkins_server + def get_latest_build_number(self, job_name) -> int: + self.log.info("Build number not specified, getting latest build info from Jenkins") + job_info = self.jenkins_server.get_job_info(job_name) + return job_info["lastBuild"]["number"] + + def get_build_result(self, job_name: str, build_number) -> str: + build_info = self.jenkins_server.get_build_info(job_name, build_number) + return build_info["result"] + def get_build_building_state(self, job_name: str, build_number: int | None) -> bool: - """Get build building state""" - try: - if not build_number: - self.log.info("Build number not specified, getting latest build info from Jenkins") - job_info = self.jenkins_server.get_job_info(job_name) - build_number_to_check = job_info["lastBuild"]["number"] - else: - build_number_to_check = build_number + if not build_number: + build_number_to_check = self.get_latest_build_number(job_name) + else: + build_number_to_check = build_number - self.log.info("Getting build info for %s build number: #%s", job_name, build_number_to_check) - build_info = self.jenkins_server.get_build_info(job_name, build_number_to_check) - building = build_info["building"] - return building - except jenkins.JenkinsException as err: - raise AirflowException(f"Jenkins call failed with error : {err}") + self.log.info("Getting build info for %s build number: #%s", job_name, build_number_to_check) + build_info = self.jenkins_server.get_build_info(job_name, build_number_to_check) + building = build_info["building"] + return building diff --git a/airflow/providers/jenkins/sensors/jenkins.py b/airflow/providers/jenkins/sensors/jenkins.py index dbfc9932d69e6..14110975a0c75 100644 --- a/airflow/providers/jenkins/sensors/jenkins.py +++ b/airflow/providers/jenkins/sensors/jenkins.py @@ -17,8 +17,9 @@ # under the License. from __future__ import annotations -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Iterable +from airflow import AirflowException from airflow.providers.jenkins.hooks.jenkins import JenkinsHook from airflow.sensors.base import BaseSensorOperator @@ -42,21 +43,31 @@ def __init__( jenkins_connection_id: str, job_name: str, build_number: int | None = None, + target_states: Iterable[str] | None = None, **kwargs, ): super().__init__(**kwargs) self.job_name = job_name self.build_number = build_number self.jenkins_connection_id = jenkins_connection_id + self.target_states = target_states or ["SUCCESS", "FAILED"] def poke(self, context: Context) -> bool: self.log.info("Poking jenkins job %s", self.job_name) hook = JenkinsHook(self.jenkins_connection_id) - is_building = hook.get_build_building_state(self.job_name, self.build_number) + build_number = self.build_number or hook.get_latest_build_number(self.job_name) + is_building = hook.get_build_building_state(self.job_name, build_number) if is_building: self.log.info("Build still ongoing!") return False - else: - self.log.info("Build is finished.") + + build_result = hook.get_build_result(self.job_name, build_number) + self.log.info("Build is finished, result is %s", "build_result") + if build_result in self.target_states: return True + else: + raise AirflowException( + f"Build {build_number} finished with a result {build_result}, " + f"which does not meet the target state {self.target_states}." + ) diff --git a/tests/providers/jenkins/hooks/test_jenkins.py b/tests/providers/jenkins/hooks/test_jenkins.py index 0ad0ba7603304..6a3ab376929d0 100644 --- a/tests/providers/jenkins/hooks/test_jenkins.py +++ b/tests/providers/jenkins/hooks/test_jenkins.py @@ -36,6 +36,7 @@ def test_client_created_default_http(self, get_connection_mock): connection_id=default_connection_id, login="test", password="test", + schema="", extra="", host=connection_host, port=connection_port, @@ -58,6 +59,7 @@ def test_client_created_default_https(self, get_connection_mock): connection_id=default_connection_id, login="test", password="test", + schema="", extra="true", host=connection_host, port=connection_port, diff --git a/tests/providers/jenkins/sensors/test_jenkins.py b/tests/providers/jenkins/sensors/test_jenkins.py index 8f69c41510d85..1faafdf08c734 100644 --- a/tests/providers/jenkins/sensors/test_jenkins.py +++ b/tests/providers/jenkins/sensors/test_jenkins.py @@ -21,37 +21,34 @@ import pytest +from airflow import AirflowException from airflow.providers.jenkins.hooks.jenkins import JenkinsHook from airflow.providers.jenkins.sensors.jenkins import JenkinsBuildSensor class TestJenkinsBuildSensor: @pytest.mark.parametrize( - "build_number, build_state", + "build_number, build_state, result", [ - ( - 1, - False, - ), ( None, True, + "", ), ( 3, True, + "", ), ], ) @patch("jenkins.Jenkins") - def test_poke(self, mock_jenkins, build_number, build_state): + def test_poke_buliding(self, mock_jenkins, build_number, build_state, result): target_build_number = build_number if build_number else 10 jenkins_mock = MagicMock() jenkins_mock.get_job_info.return_value = {"lastBuild": {"number": target_build_number}} - jenkins_mock.get_build_info.return_value = { - "building": build_state, - } + jenkins_mock.get_build_info.return_value = {"building": build_state} mock_jenkins.return_value = jenkins_mock with patch.object(JenkinsHook, "get_connection") as mock_get_connection: @@ -63,6 +60,7 @@ def test_poke(self, mock_jenkins, build_number, build_state): task_id="sensor_test", job_name="a_job_on_jenkins", build_number=target_build_number, + target_states=["SUCCESS"], ) output = sensor.poke(None) @@ -70,3 +68,48 @@ def test_poke(self, mock_jenkins, build_number, build_state): assert output == (not build_state) assert jenkins_mock.get_job_info.call_count == 0 if build_number else 1 jenkins_mock.get_build_info.assert_called_once_with("a_job_on_jenkins", target_build_number) + + @pytest.mark.parametrize( + "build_number, build_state, result", + [ + ( + 1, + False, + "SUCCESS", + ), + ( + 2, + False, + "FAILED", + ), + ], + ) + @patch("jenkins.Jenkins") + def test_poke_finish_building(self, mock_jenkins, build_number, build_state, result): + target_build_number = build_number if build_number else 10 + + jenkins_mock = MagicMock() + jenkins_mock.get_job_info.return_value = {"lastBuild": {"number": target_build_number}} + jenkins_mock.get_build_info.return_value = {"building": build_state, "result": result} + mock_jenkins.return_value = jenkins_mock + + with patch.object(JenkinsHook, "get_connection") as mock_get_connection: + mock_get_connection.return_value = MagicMock() + + sensor = JenkinsBuildSensor( + dag=None, + jenkins_connection_id="fake_jenkins_connection", + task_id="sensor_test", + job_name="a_job_on_jenkins", + build_number=target_build_number, + target_states=["SUCCESS"], + ) + if result not in sensor.target_states: + with pytest.raises(AirflowException): + sensor.poke(None) + assert jenkins_mock.get_build_info.call_count == 2 + else: + output = sensor.poke(None) + assert output == (not build_state) + assert jenkins_mock.get_job_info.call_count == 0 if build_number else 1 + assert jenkins_mock.get_build_info.call_count == 2