check jenkins result state (#28659)

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,27 +39,30 @@ 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)
 
     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

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}."
+            )

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,

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,10 +60,56 @@ 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)
 
             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