This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 8bf71ce1069 Parametrize Spark sensor waiting-state tests (#75879)
8bf71ce1069 is described below
commit 8bf71ce106907b2469582f194034a7cf7a1dd809
Author: Hussein Awala <[email protected]>
AuthorDate: Thu Oct 29 06:16:04 2026 +0100
Parametrize Spark sensor waiting-state tests (#75879)
Five waiting-state tests repeat the same assertions. Parametrization keeps
all five distinct application responses and fresh mocks for every case.
---
.../kubernetes/sensors/test_spark_kubernetes.py | 78 ++++------------------
1 file changed, 13 insertions(+), 65 deletions(-)
diff --git
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/sensors/test_spark_kubernetes.py
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/sensors/test_spark_kubernetes.py
index bd6dd73f210..6ccf8938e9d 100644
---
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/sensors/test_spark_kubernetes.py
+++
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/sensors/test_spark_kubernetes.py
@@ -621,79 +621,27 @@ class TestSparkKubernetesSensor:
version="v1beta2",
)
- @patch(
-
"kubernetes.client.api.custom_objects_api.CustomObjectsApi.get_namespaced_custom_object",
- return_value=TEST_NOT_PROCESSED_APPLICATION,
- )
- def test_not_processed_application(self, mock_get_namespaced_crd,
mock_kubernetes_hook):
- sensor = SparkKubernetesSensor(application_name="spark_pi",
dag=self.dag, task_id="test_task_id")
- assert not sensor.poke({})
- mock_kubernetes_hook.assert_called_once_with()
- mock_get_namespaced_crd.assert_called_once_with(
- group="sparkoperator.k8s.io",
- name="spark_pi",
- namespace="default",
- plural="sparkapplications",
- version="v1beta2",
- )
-
- @patch(
-
"kubernetes.client.api.custom_objects_api.CustomObjectsApi.get_namespaced_custom_object",
- return_value=TEST_NEW_APPLICATION,
+ @pytest.mark.parametrize(
+ "application",
+ [
+ pytest.param(TEST_NOT_PROCESSED_APPLICATION, id="not-processed"),
+ pytest.param(TEST_NEW_APPLICATION, id="new"),
+ pytest.param(TEST_RUNNING_APPLICATION, id="running"),
+ pytest.param(TEST_SUBMITTED_APPLICATION, id="submitted"),
+ pytest.param(TEST_PENDING_RERUN_APPLICATION, id="pending-rerun"),
+ ],
)
- def test_new_application(self, mock_get_namespaced_crd,
mock_kubernetes_hook):
- sensor = SparkKubernetesSensor(application_name="spark_pi",
dag=self.dag, task_id="test_task_id")
- assert not sensor.poke({})
- mock_kubernetes_hook.assert_called_once_with()
- mock_get_namespaced_crd.assert_called_once_with(
- group="sparkoperator.k8s.io",
- name="spark_pi",
- namespace="default",
- plural="sparkapplications",
- version="v1beta2",
- )
-
- @patch(
-
"kubernetes.client.api.custom_objects_api.CustomObjectsApi.get_namespaced_custom_object",
- return_value=TEST_RUNNING_APPLICATION,
- )
- def test_running_application(self, mock_get_namespaced_crd,
mock_kubernetes_hook):
- sensor = SparkKubernetesSensor(application_name="spark_pi",
dag=self.dag, task_id="test_task_id")
- assert not sensor.poke({})
- mock_kubernetes_hook.assert_called_once_with()
- mock_get_namespaced_crd.assert_called_once_with(
- group="sparkoperator.k8s.io",
- name="spark_pi",
- namespace="default",
- plural="sparkapplications",
- version="v1beta2",
- )
-
- @patch(
-
"kubernetes.client.api.custom_objects_api.CustomObjectsApi.get_namespaced_custom_object",
- return_value=TEST_SUBMITTED_APPLICATION,
- )
- def test_submitted_application(self, mock_get_namespaced_crd,
mock_kubernetes_hook):
- sensor = SparkKubernetesSensor(application_name="spark_pi",
dag=self.dag, task_id="test_task_id")
- assert not sensor.poke({})
- mock_kubernetes_hook.assert_called_once_with()
- mock_get_namespaced_crd.assert_called_once_with(
- group="sparkoperator.k8s.io",
- name="spark_pi",
- namespace="default",
- plural="sparkapplications",
- version="v1beta2",
- )
-
@patch(
"kubernetes.client.api.custom_objects_api.CustomObjectsApi.get_namespaced_custom_object",
- return_value=TEST_PENDING_RERUN_APPLICATION,
+ autospec=True,
)
- def test_pending_rerun_application(self, mock_get_namespaced_crd,
mock_kubernetes_hook):
+ def test_waiting_application(self, mock_get_namespaced_crd,
mock_kubernetes_hook, application):
+ mock_get_namespaced_crd.return_value = application
sensor = SparkKubernetesSensor(application_name="spark_pi",
dag=self.dag, task_id="test_task_id")
assert not sensor.poke({})
mock_kubernetes_hook.assert_called_once_with()
mock_get_namespaced_crd.assert_called_once_with(
+ ANY,
group="sparkoperator.k8s.io",
name="spark_pi",
namespace="default",
