From d6e670b0b9c6da06d92156b1a216b78a0efcbbfc Mon Sep 17 00:00:00 2001 From: Jens Scheffler Date: Wed, 29 Oct 2025 22:11:45 +0100 Subject: [PATCH 1/4] Enable ruff PLW0129 rule --- .../aws/executors/batch/test_batch_executor.py | 5 ++++- .../unit/apache/livy/operators/test_livy.py | 5 ++++- .../spark/operators/test_spark_submit.py | 5 ++++- .../unit/cncf/kubernetes/operators/test_pod.py | 7 ++++++- .../unit/cncf/kubernetes/triggers/test_pod.py | 10 +++++----- .../google/cloud/triggers/test_bigquery_dts.py | 4 ++-- .../google/cloud/triggers/test_dataflow.py | 4 ++-- .../google/cloud/triggers/test_dataplex.py | 5 ++++- .../cloud/triggers/test_kubernetes_engine.py | 18 +++++++++--------- pyproject.toml | 6 ++++-- 10 files changed, 44 insertions(+), 25 deletions(-) diff --git a/providers/amazon/tests/unit/amazon/aws/executors/batch/test_batch_executor.py b/providers/amazon/tests/unit/amazon/aws/executors/batch/test_batch_executor.py index 1098efac777c4..548e82a1e9e41 100644 --- a/providers/amazon/tests/unit/amazon/aws/executors/batch/test_batch_executor.py +++ b/providers/amazon/tests/unit/amazon/aws/executors/batch/test_batch_executor.py @@ -461,7 +461,10 @@ def test_task_retry_on_api_failure(self, _, mock_executor, caplog): mock_executor.attempt_submit_jobs() mock_executor.sync_running_jobs() for i in range(2): - assert f"Airflow task {airflow_keys[i]} has failed a maximum of {mock_executor.MAX_SUBMIT_JOB_ATTEMPTS} times. Marking as failed" + assert ( + f"Airflow task {airflow_keys[i]} has failed a maximum of {mock_executor.MAX_SUBMIT_JOB_ATTEMPTS} times. Marking as failed" + in caplog.text + ) @mock.patch("airflow.providers.amazon.aws.executors.batch.batch_executor.exponential_backoff_retry") def test_sync_unhealthy_boto_connection(self, mock_exponentional_backoff_retry, mock_executor): diff --git a/providers/apache/livy/tests/unit/apache/livy/operators/test_livy.py b/providers/apache/livy/tests/unit/apache/livy/operators/test_livy.py index 2987796dcd362..da69f22cfa34a 100644 --- a/providers/apache/livy/tests/unit/apache/livy/operators/test_livy.py +++ b/providers/apache/livy/tests/unit/apache/livy/operators/test_livy.py @@ -554,7 +554,10 @@ def test_inject_openlineage_simple_config_wrong_transport_to_spark( operator.hook.TERMINAL_STATES = [BatchState.SUCCESS] operator.execute(MagicMock()) - assert "OpenLineage transport type `console` does not support automatic injection of OpenLineage transport information into Spark properties." + assert ( + "OpenLineage transport type `console` does not support automatic injection of OpenLineage transport information into Spark properties." + in caplog.text + ) assert operator.spark_params["conf"] == {} diff --git a/providers/apache/spark/tests/unit/apache/spark/operators/test_spark_submit.py b/providers/apache/spark/tests/unit/apache/spark/operators/test_spark_submit.py index 7acffb447d07c..c7010a0fb835f 100644 --- a/providers/apache/spark/tests/unit/apache/spark/operators/test_spark_submit.py +++ b/providers/apache/spark/tests/unit/apache/spark/operators/test_spark_submit.py @@ -467,7 +467,10 @@ def test_inject_openlineage_simple_config_wrong_transport_to_spark( ) operator.execute(MagicMock()) - assert "OpenLineage transport type `console` does not support automatic injection of OpenLineage transport information into Spark properties." + assert ( + "OpenLineage transport type `console` does not support automatic injection of OpenLineage transport information into Spark properties." + in caplog.text + ) assert operator.conf == { "parquet.compression": "SNAPPY", } diff --git a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py index abb503c8aa477..3a64fe1f9b193 100644 --- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py +++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py @@ -2545,6 +2545,7 @@ def test_async_get_logs_should_execute_successfully( @patch(KUB_OP_PATH.format("extract_xcom")) @patch(HOOK_CLASS) @patch(KUB_OP_PATH.format("pod_manager")) + @pytest.mark.xfail def test_async_write_logs_should_execute_successfully( self, mock_manager, mocked_hook, mock_extract_xcom, post_complete_action, get_logs ): @@ -2565,8 +2566,12 @@ def test_async_write_logs_should_execute_successfully( self.run_pod_async(k) if get_logs: - assert f"Container logs: {test_logs}" + # Note: the test below is broken and failing. Either the mock is wrong + # or the mocked container is not in a state that logging methods are called at-all. + # See https://github.com/apache/airflow/issues/57515 + assert f"Container logs: {test_logs}" # noqa: PLW0129 post_complete_action.assert_called_once() + mock_manager.return_value.read_pod_logs.assert_called() else: mock_manager.return_value.read_pod_logs.assert_not_called() diff --git a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_pod.py b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_pod.py index fc477fe0cf763..adee8959f9477 100644 --- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_pod.py +++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_pod.py @@ -162,8 +162,8 @@ async def test_run_loop_return_waiting_event( await asyncio.sleep(0.5) assert not task.done() - assert "Container is not completed and still working." - assert f"Sleeping for {POLL_INTERVAL} seconds." + assert "Container is not completed and still working." in caplog.text + assert f"Sleeping for {POLL_INTERVAL} seconds." in caplog.text @pytest.mark.asyncio @mock.patch(f"{TRIGGER_PATH}._wait_for_pod_start") @@ -181,8 +181,8 @@ async def test_run_loop_return_running_event( await asyncio.sleep(0.5) assert not task.done() - assert "Container is not completed and still working." - assert f"Sleeping for {POLL_INTERVAL} seconds." + assert "Container is not completed and still working." in caplog.text + assert f"Sleeping for {POLL_INTERVAL} seconds." in caplog.text @pytest.mark.asyncio @mock.patch(f"{TRIGGER_PATH}._wait_for_pod_start") @@ -250,7 +250,7 @@ async def test_logging_in_trigger_when_fail_should_execute_successfully( generator = trigger.run() await generator.asend(None) - assert "Container logs:" + assert "Container logs:" in caplog.text @pytest.mark.asyncio @pytest.mark.parametrize( diff --git a/providers/google/tests/unit/google/cloud/triggers/test_bigquery_dts.py b/providers/google/tests/unit/google/cloud/triggers/test_bigquery_dts.py index 9ce383c53c901..5e83201a007e2 100644 --- a/providers/google/tests/unit/google/cloud/triggers/test_bigquery_dts.py +++ b/providers/google/tests/unit/google/cloud/triggers/test_bigquery_dts.py @@ -153,5 +153,5 @@ async def test_run_loop_is_still_running(self, mock_hook, trigger, caplog): await asyncio.sleep(0.5) assert not task.done() - assert f"Current job status is: {TransferState.RUNNING}" - assert f"Sleeping for {POLL_INTERVAL} seconds." + assert f"Current job status is: {TransferState.RUNNING}" in caplog.text + assert f"Sleeping for {POLL_INTERVAL} seconds." in caplog.text diff --git a/providers/google/tests/unit/google/cloud/triggers/test_dataflow.py b/providers/google/tests/unit/google/cloud/triggers/test_dataflow.py index 1312a1dfb2aca..d337da15fc4e8 100644 --- a/providers/google/tests/unit/google/cloud/triggers/test_dataflow.py +++ b/providers/google/tests/unit/google/cloud/triggers/test_dataflow.py @@ -229,8 +229,8 @@ async def test_run_loop_is_still_running(self, mock_job_status, template_job_sta await asyncio.sleep(0.5) assert not task.done() - assert f"Current job status is: {JobState.JOB_STATE_RUNNING}" - assert f"Sleeping for {POLL_SLEEP} seconds." + assert f"Current job status is: {JobState.JOB_STATE_RUNNING}" in caplog.text + assert f"Sleeping for {POLL_SLEEP} seconds." in caplog.text # cancel the task to suppress test warnings task.cancel() diff --git a/providers/google/tests/unit/google/cloud/triggers/test_dataplex.py b/providers/google/tests/unit/google/cloud/triggers/test_dataplex.py index 27ee12bb43629..9b3e38427b707 100644 --- a/providers/google/tests/unit/google/cloud/triggers/test_dataplex.py +++ b/providers/google/tests/unit/google/cloud/triggers/test_dataplex.py @@ -132,4 +132,7 @@ async def test_async_dataplex_job_run_loop_is_still_running( await asyncio.sleep(0.5) assert not task.done() - assert f"Current state is: {DataScanJob.State.RUNNING}, sleeping for {TEST_POLL_INTERVAL} seconds." + assert ( + f"Current state is: {DataScanJob.State.RUNNING}, sleeping for {TEST_POLL_INTERVAL} seconds." + in caplog.text + ) diff --git a/providers/google/tests/unit/google/cloud/triggers/test_kubernetes_engine.py b/providers/google/tests/unit/google/cloud/triggers/test_kubernetes_engine.py index 3704c5c320624..21ec8fa895ee5 100644 --- a/providers/google/tests/unit/google/cloud/triggers/test_kubernetes_engine.py +++ b/providers/google/tests/unit/google/cloud/triggers/test_kubernetes_engine.py @@ -206,8 +206,8 @@ async def test_run_loop_return_waiting_event_should_execute_successfully( await asyncio.sleep(0.5) assert not task.done() - assert "Container is not completed and still working." - assert f"Sleeping for {POLL_INTERVAL} seconds." + assert "Container is not completed and still working." in caplog.text + assert f"Sleeping for {POLL_INTERVAL} seconds." in caplog.text @pytest.mark.asyncio @mock.patch(f"{TRIGGER_KUB_POD_PATH}._wait_for_pod_start") @@ -225,8 +225,8 @@ async def test_run_loop_return_running_event_should_execute_successfully( await asyncio.sleep(0.5) assert not task.done() - assert "Container is not completed and still working." - assert f"Sleeping for {POLL_INTERVAL} seconds." + assert "Container is not completed and still working." in caplog.text + assert f"Sleeping for {POLL_INTERVAL} seconds." in caplog.text @pytest.mark.asyncio @mock.patch(f"{TRIGGER_KUB_POD_PATH}._wait_for_pod_start") @@ -265,7 +265,7 @@ async def test_logging_in_trigger_when_fail_should_execute_successfully( generator = trigger.run() await generator.asend(None) - assert "Container logs:" + assert "Container logs:" in caplog.text @pytest.mark.parametrize( "container_state, expected_state", @@ -447,8 +447,8 @@ async def test_run_loop_return_waiting_event_pending_status( await asyncio.sleep(0.5) assert not task.done() - assert "Operation is still running." - assert f"Sleeping for {POLL_INTERVAL}s..." + assert "Operation is still running." in caplog.text + assert f"Sleeping for {POLL_INTERVAL}s..." in caplog.text @pytest.mark.asyncio @mock.patch(f"{TRIGGER_PATH}._get_hook") @@ -470,8 +470,8 @@ async def test_run_loop_return_waiting_event_running_status( await asyncio.sleep(0.5) assert not task.done() - assert "Operation is still running." - assert f"Sleeping for {POLL_INTERVAL}s..." + assert "Operation is still running." in caplog.text + assert f"Sleeping for {POLL_INTERVAL}s..." in caplog.text class TestGKEStartJobTrigger: diff --git a/pyproject.toml b/pyproject.toml index 8a134826de0b8..f6417fa4c3291 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -591,8 +591,11 @@ extend-select = [ "E", # pycodestyle rules "W", # pycodestyle rules # Warning (PLW) re-implemented in ruff from Pylint - "PLW0127", # Self-assignment of variable "PLW0120", # else clause on loop without a break statement; remove the else and dedent its contents + "PLW0127", # Self-assignment of variable + "PLW0128", # Redeclared variable {name} in assignment + "PLW0129", # Asserting on an empty string literal will never pass + "PLW0133", # Missing raise statement on exception # Per rule enables "RUF006", # Checks for asyncio dangling task "RUF015", # Checks for unnecessary iterable allocation for first element @@ -625,7 +628,6 @@ extend-select = [ "RET506", # Unnecessary {branch} after raise statement "RET507", # Unnecessary {branch} after continue statement "RET508", # Unnecessary {branch} after break statement - "PLW0133", # Missing raise statement on exception ] ignore = [ "D100", # Unwanted; Docstring at the top of every file. From 3b3408bb67e93dc07a29076951e332e3eaaa0b31 Mon Sep 17 00:00:00 2001 From: Jens Scheffler Date: Thu, 30 Oct 2025 22:13:22 +0100 Subject: [PATCH 2/4] Fix pytests --- .../tests/unit/cncf/kubernetes/triggers/test_pod.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_pod.py b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_pod.py index adee8959f9477..ce57d0f5e96d9 100644 --- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_pod.py +++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_pod.py @@ -156,7 +156,7 @@ async def test_run_loop_return_waiting_event( mock_hook.get_pod.return_value = self._mock_pod_result(mock.AsyncMock()) mock_method.return_value = ContainerState.WAITING - caplog.set_level(logging.INFO) + caplog.set_level(logging.DEBUG) task = asyncio.create_task(trigger.run().__anext__()) await asyncio.sleep(0.5) @@ -175,7 +175,7 @@ async def test_run_loop_return_running_event( mock_hook.get_pod.return_value = self._mock_pod_result(mock.AsyncMock()) mock_method.return_value = ContainerState.RUNNING - caplog.set_level(logging.INFO) + caplog.set_level(logging.DEBUG) task = asyncio.create_task(trigger.run().__anext__()) await asyncio.sleep(0.5) @@ -250,7 +250,7 @@ async def test_logging_in_trigger_when_fail_should_execute_successfully( generator = trigger.run() await generator.asend(None) - assert "Container logs:" in caplog.text + assert "Waiting until 120s to get the POD scheduled..." in caplog.text @pytest.mark.asyncio @pytest.mark.parametrize( From bbf6a90f7d672d156e0cdc3b2e34ed5131334795 Mon Sep 17 00:00:00 2001 From: Jens Scheffler Date: Thu, 30 Oct 2025 23:37:42 +0100 Subject: [PATCH 3/4] Fix pytests as working in logs changed --- .../tests/unit/google/cloud/triggers/test_bigquery_dts.py | 4 ++-- .../tests/unit/google/cloud/triggers/test_dataflow.py | 2 +- .../tests/unit/google/cloud/triggers/test_dataplex.py | 5 +---- .../unit/google/cloud/triggers/test_kubernetes_engine.py | 6 +++--- 4 files changed, 7 insertions(+), 10 deletions(-) diff --git a/providers/google/tests/unit/google/cloud/triggers/test_bigquery_dts.py b/providers/google/tests/unit/google/cloud/triggers/test_bigquery_dts.py index 5e83201a007e2..a012ae0877232 100644 --- a/providers/google/tests/unit/google/cloud/triggers/test_bigquery_dts.py +++ b/providers/google/tests/unit/google/cloud/triggers/test_bigquery_dts.py @@ -153,5 +153,5 @@ async def test_run_loop_is_still_running(self, mock_hook, trigger, caplog): await asyncio.sleep(0.5) assert not task.done() - assert f"Current job status is: {TransferState.RUNNING}" in caplog.text - assert f"Sleeping for {POLL_INTERVAL} seconds." in caplog.text + assert f"Current state is {TransferState.RUNNING}" in caplog.text + assert f"Waiting for {POLL_INTERVAL} seconds" in caplog.text diff --git a/providers/google/tests/unit/google/cloud/triggers/test_dataflow.py b/providers/google/tests/unit/google/cloud/triggers/test_dataflow.py index d337da15fc4e8..bf8cfd05d6041 100644 --- a/providers/google/tests/unit/google/cloud/triggers/test_dataflow.py +++ b/providers/google/tests/unit/google/cloud/triggers/test_dataflow.py @@ -229,7 +229,7 @@ async def test_run_loop_is_still_running(self, mock_job_status, template_job_sta await asyncio.sleep(0.5) assert not task.done() - assert f"Current job status is: {JobState.JOB_STATE_RUNNING}" in caplog.text + assert "Current job status is: JOB_STATE_RUNNING" in caplog.text assert f"Sleeping for {POLL_SLEEP} seconds." in caplog.text # cancel the task to suppress test warnings task.cancel() diff --git a/providers/google/tests/unit/google/cloud/triggers/test_dataplex.py b/providers/google/tests/unit/google/cloud/triggers/test_dataplex.py index 9b3e38427b707..084d1175f8d49 100644 --- a/providers/google/tests/unit/google/cloud/triggers/test_dataplex.py +++ b/providers/google/tests/unit/google/cloud/triggers/test_dataplex.py @@ -132,7 +132,4 @@ async def test_async_dataplex_job_run_loop_is_still_running( await asyncio.sleep(0.5) assert not task.done() - assert ( - f"Current state is: {DataScanJob.State.RUNNING}, sleeping for {TEST_POLL_INTERVAL} seconds." - in caplog.text - ) + assert f"Current state is: RUNNING, sleeping for {TEST_POLL_INTERVAL} seconds." in caplog.text diff --git a/providers/google/tests/unit/google/cloud/triggers/test_kubernetes_engine.py b/providers/google/tests/unit/google/cloud/triggers/test_kubernetes_engine.py index 21ec8fa895ee5..05e194a3f9735 100644 --- a/providers/google/tests/unit/google/cloud/triggers/test_kubernetes_engine.py +++ b/providers/google/tests/unit/google/cloud/triggers/test_kubernetes_engine.py @@ -200,7 +200,7 @@ async def test_run_loop_return_waiting_event_should_execute_successfully( mock_hook.get_pod.return_value = self._mock_pod_result(mock.MagicMock()) mock_method.return_value = ContainerState.WAITING - caplog.set_level(logging.INFO) + caplog.set_level(logging.DEBUG) task = asyncio.create_task(trigger.run().__anext__()) await asyncio.sleep(0.5) @@ -219,7 +219,7 @@ async def test_run_loop_return_running_event_should_execute_successfully( mock_hook.get_pod.return_value = self._mock_pod_result(mock.MagicMock()) mock_method.return_value = ContainerState.RUNNING - caplog.set_level(logging.INFO) + caplog.set_level(logging.DEBUG) task = asyncio.create_task(trigger.run().__anext__()) await asyncio.sleep(0.5) @@ -265,7 +265,7 @@ async def test_logging_in_trigger_when_fail_should_execute_successfully( generator = trigger.run() await generator.asend(None) - assert "Container logs:" in caplog.text + assert "Waiting until 120s to get the POD scheduled..." in caplog.text @pytest.mark.parametrize( "container_state, expected_state", From 33e6215f4eded3cfbe7bb0d9793db5cab24feea5 Mon Sep 17 00:00:00 2001 From: Jens Scheffler Date: Fri, 31 Oct 2025 00:13:30 +0100 Subject: [PATCH 4/4] Fix pytests as working in logs changed --- .../tests/unit/google/cloud/triggers/test_bigquery_dts.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/providers/google/tests/unit/google/cloud/triggers/test_bigquery_dts.py b/providers/google/tests/unit/google/cloud/triggers/test_bigquery_dts.py index a012ae0877232..cba7cb58993b4 100644 --- a/providers/google/tests/unit/google/cloud/triggers/test_bigquery_dts.py +++ b/providers/google/tests/unit/google/cloud/triggers/test_bigquery_dts.py @@ -153,5 +153,8 @@ async def test_run_loop_is_still_running(self, mock_hook, trigger, caplog): await asyncio.sleep(0.5) assert not task.done() - assert f"Current state is {TransferState.RUNNING}" in caplog.text + assert ( + f"Current state is {TransferState.RUNNING}" in caplog.text + or "Current state is TransferState.RUNNING" in caplog.text + ) assert f"Waiting for {POLL_INTERVAL} seconds" in caplog.text