Add JenkinsBuildSensor (#22421)

* Add `JenkinsBuildSensor`


Co-authored-by: eladkal <45845474+eladkal@users.noreply.github.com>

airflow/providers/jenkins/hooks/jenkins.py

@@ -17,8 +17,11 @@
 # under the License.
 #
 
+from typing import Optional
+
 import jenkins
 
+from airflow import AirflowException
 from airflow.hooks.base import BaseHook
 from airflow.utils.strings import to_boolean
 
@@ -46,3 +49,20 @@ def __init__(self, conn_id: str = default_conn_name) -> None:
     def get_jenkins_server(self) -> jenkins.Jenkins:
         """Get jenkins server"""
         return self.jenkins_server
+
+    def get_build_building_state(self, job_name: str, build_number: Optional[int]) -> 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
+
+            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}')

airflow/providers/jenkins/provider.yaml

@@ -52,6 +52,11 @@ hooks:
     python-modules:
       - airflow.providers.jenkins.hooks.jenkins
 
+sensors:
+  - integration-name: Jenkins
+    python-modules:
+      - 'airflow.providers.jenkins.sensors.jenkins'
+
 hook-class-names:  # deprecated - to be removed after providers add dependency on Airflow 2.2.0+
   - airflow.providers.jenkins.hooks.jenkins.JenkinsHook
 

airflow/providers/jenkins/sensors/__init__.py

@@ -0,0 +1,17 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.

airflow/providers/jenkins/sensors/jenkins.py

@@ -0,0 +1,61 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from typing import TYPE_CHECKING, Optional
+
+from airflow.providers.jenkins.hooks.jenkins import JenkinsHook
+from airflow.sensors.base import BaseSensorOperator
+
+if TYPE_CHECKING:
+    from airflow.utils.context import Context
+
+
+class JenkinsBuildSensor(BaseSensorOperator):
+    """
+    Monitor a jenkins job and pass when it is finished building. This is regardless of the build outcome.
+    This sensor depend on python-jenkins library,
+
+    :param jenkins_connection_id: The jenkins connection to use for this job
+    :param job_name: The name of the job to check
+    :param build_number: Build number to check - if None, the latest build will be used
+    """
+
+    def __init__(
+        self,
+        *,
+        jenkins_connection_id: str,
+        job_name: str,
+        build_number: Optional[int] = None,
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+        self.job_name = job_name
+        self.build_number = build_number
+        self.jenkins_connection_id = jenkins_connection_id
+
+    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)
+
+        if is_building:
+            self.log.info("Build still ongoing!")
+            return False
+        else:
+            self.log.info("Build is finished.")
+            return True

tests/providers/jenkins/hooks/test_jenkins.py

@@ -19,6 +19,8 @@
 import unittest
 from unittest import mock
 
+from parameterized import parameterized
+
 from airflow.providers.jenkins.hooks.jenkins import JenkinsHook
 
 
@@ -65,3 +67,16 @@ def test_client_created_default_https(self, get_connection_mock):
         hook = JenkinsHook(default_connection_id)
         assert hook.jenkins_server is not None
         assert hook.jenkins_server.server == complete_url
+
+    @parameterized.expand([(True,), (False,)])
+    @mock.patch('airflow.hooks.base.BaseHook.get_connection')
+    @mock.patch('jenkins.Jenkins.get_job_info')
+    @mock.patch('jenkins.Jenkins.get_build_info')
+    def test_get_build_building_state(
+        self, param_building, mock_get_build_info, mock_get_job_info, get_connection_mock
+    ):
+        mock_get_build_info.return_value = {'building': param_building}
+
+        hook = JenkinsHook('none_connection_id')
+        result = hook.get_build_building_state('some_job', 1)
+        assert result == param_building

tests/providers/jenkins/sensors/__init__.py

@@ -0,0 +1,17 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.

tests/providers/jenkins/sensors/test_jenkins.py

@@ -0,0 +1,71 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import unittest
+from unittest.mock import MagicMock, patch
+
+from parameterized import parameterized
+
+from airflow.providers.jenkins.hooks.jenkins import JenkinsHook
+from airflow.providers.jenkins.sensors.jenkins import JenkinsBuildSensor
+
+
+class TestJenkinsBuildSensor(unittest.TestCase):
+    @parameterized.expand(
+        [
+            (
+                1,
+                False,
+            ),
+            (
+                None,
+                True,
+            ),
+            (
+                3,
+                True,
+            ),
+        ]
+    )
+    @patch('jenkins.Jenkins')
+    def test_poke(self, build_number, build_state, mock_jenkins):
+        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,
+        }
+        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,
+            )
+
+            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)