henry3260 commented on code in PR #68250:
URL: https://github.com/apache/airflow/pull/68250#discussion_r3382299271
##########
providers/google/tests/unit/google/cloud/log/test_stackdriver_task_handler.py:
##########
@@ -280,6 +280,43 @@ def test_processors_skips_non_task_logger(self,
mock_client, mock_get_creds_and_
assert result is event
mock_transport_type.return_value.send.assert_not_called()
+ @pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="airflow.sdk.log only
exists in Airflow 3+")
+
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
+
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.gcp_logging.Client")
+ def test_processors_survives_transport_send_failure(self, mock_client,
mock_get_creds_and_project_id, caplog):
+ """``proc()`` must not let a transport.send() failure crash the
process.
+
+ In AF3's supervisor model REMOTE_TASK_LOG applies to ALL supervised
+ components (scheduler, dag-processor, triggerer, workers). A single
+ IAM misconfiguration or gRPC error would crash the entire process
+ without this guard. The fix degrades gracefully: log a warning and
+ continue processing.
+ """
+ mock_get_creds_and_project_id.return_value = ("creds", "project_id")
+
+ mock_transport_type = mock.MagicMock()
+ mock_transport_type.return_value.send.side_effect = RuntimeError("IAM
permission denied")
+ with mock.patch("airflow.sdk.log.relative_path_from_logger",
return_value="dag/task/1.log"):
+ io = StackdriverRemoteLogIO(
+ base_log_folder=self.local_log_location,
+ gcp_log_name="airflow",
+ transport_type=mock_transport_type,
+ )
+ proc = io.processors[0]
+
+ event = {
+ "event": "hello world",
+ "logger_name": "airflow.task",
+ "timestamp": "2026-01-15T10:30:00+00:00",
+ }
+ with caplog.at_level(logging.WARNING):
+ result = proc(mock.MagicMock(), "info", event)
+
+ # Must still return the event — crash would mean no return at all
+ assert result is event
+ # Must log the failure so the operator can see it
+ assert "Failed to send log to Cloud Logging" in caplog.text
Review Comment:
Could we avoid using `caplog `here? It can modify the logging configuration
and make tests flaky.
##########
providers/google/tests/unit/google/cloud/log/test_stackdriver_task_handler.py:
##########
@@ -280,6 +280,43 @@ def test_processors_skips_non_task_logger(self,
mock_client, mock_get_creds_and_
assert result is event
mock_transport_type.return_value.send.assert_not_called()
+ @pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="airflow.sdk.log only
exists in Airflow 3+")
+
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.get_credentials_and_project_id")
+
@mock.patch("airflow.providers.google.cloud.log.stackdriver_task_handler.gcp_logging.Client")
+ def test_processors_survives_transport_send_failure(self, mock_client,
mock_get_creds_and_project_id, caplog):
+ """``proc()`` must not let a transport.send() failure crash the
process.
+
+ In AF3's supervisor model REMOTE_TASK_LOG applies to ALL supervised
+ components (scheduler, dag-processor, triggerer, workers). A single
+ IAM misconfiguration or gRPC error would crash the entire process
+ without this guard. The fix degrades gracefully: log a warning and
+ continue processing.
+ """
+ mock_get_creds_and_project_id.return_value = ("creds", "project_id")
+
+ mock_transport_type = mock.MagicMock()
Review Comment:
```suggestion
mock_transport_type = mock.create_autospec(Transport)
```
This ensures the mock follows the `Transport `interface and catches invalid
method calls or arguments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
