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 890c20b6ecb Fix Cloud Monitoring list operators with explicit output
formats (#73434)
890c20b6ecb is described below
commit 890c20b6ecbd6061ba8dcab6c274c2c626e865ce
Author: ylv01 <[email protected]>
AuthorDate: Tue Sep 22 22:39:13 2026 +0800
Fix Cloud Monitoring list operators with explicit output formats (#73434)
---
.../google/cloud/operators/cloud_monitoring.py | 30 ++++----
.../cloud/operators/test_cloud_monitoring.py | 87 ++++++++++++++++++++++
2 files changed, 103 insertions(+), 14 deletions(-)
diff --git
a/providers/google/src/airflow/providers/google/cloud/operators/cloud_monitoring.py
b/providers/google/src/airflow/providers/google/cloud/operators/cloud_monitoring.py
index 29143ef4e5c..85e586913b6 100644
---
a/providers/google/src/airflow/providers/google/cloud/operators/cloud_monitoring.py
+++
b/providers/google/src/airflow/providers/google/cloud/operators/cloud_monitoring.py
@@ -41,17 +41,16 @@ class
CloudMonitoringListAlertPoliciesOperator(GoogleCloudBaseOperator):
"""
Fetches all the Alert Policies identified by the filter passed as filter
parameter.
- The desired return type can be specified by the format parameter, the
supported
- formats are "dict", "json" and None which returns python dictionary,
stringified
- JSON and protobuf respectively.
+ Returns a list of dictionaries by default, or JSON strings when
``format_="json"``.
.. seealso::
For more information on how to use this operator, take a look at the
guide:
:ref:`howto/operator:CloudMonitoringListAlertPoliciesOperator`
- :param format_: (Optional) Desired output format of the result. The
- supported formats are "dict", "json" and None which returns
- python dictionary, stringified JSON and protobuf respectively.
+ :param format_: (Optional) Desired output format. ``"dict"`` and
``"json"`` return
+ the hook's list of dictionaries or JSON strings unchanged,
respectively.
+ When ``None`` (the default) or any other value, protobuf objects are
converted to dictionaries
+ for XCom serialization.
:param filter_: If provided, this field specifies the criteria that must
be met by alert
policies to be included in the response.
For more details, see
https://cloud.google.com/monitoring/api/v3/sorting-and-filtering.
@@ -148,7 +147,9 @@ class
CloudMonitoringListAlertPoliciesOperator(GoogleCloudBaseOperator):
context=context,
project_id=self.project_id or self.hook.project_id,
)
- return [AlertPolicy.to_dict(policy) for policy in result]
+ if self.format_ not in ("dict", "json"):
+ return [AlertPolicy.to_dict(policy) for policy in result]
+ return result
class CloudMonitoringEnableAlertPoliciesOperator(GoogleCloudBaseOperator):
@@ -477,17 +478,16 @@ class
CloudMonitoringListNotificationChannelsOperator(GoogleCloudBaseOperator):
"""
Fetches all the Notification Channels identified by the filter passed as
filter parameter.
- The desired return type can be specified by the format parameter, the
- supported formats are "dict", "json" and None which returns python
- dictionary, stringified JSON and protobuf respectively.
+ Returns a list of dictionaries by default, or JSON strings when
``format_="json"``.
.. seealso::
For more information on how to use this operator, take a look at the
guide:
:ref:`howto/operator:CloudMonitoringListNotificationChannelsOperator`
- :param format_: (Optional) Desired output format of the result. The
- supported formats are "dict", "json" and None which returns
- python dictionary, stringified JSON and protobuf respectively.
+ :param format_: (Optional) Desired output format. ``"dict"`` and
``"json"`` return
+ the hook's list of dictionaries or JSON strings unchanged,
respectively.
+ When ``None`` (the default) or any other value, protobuf objects are
converted to dictionaries
+ for XCom serialization.
:param filter_: If provided, this field specifies the criteria that
must be met by notification channels to be included in the response.
For more details, see
https://cloud.google.com/monitoring/api/v3/sorting-and-filtering.
@@ -584,7 +584,9 @@ class
CloudMonitoringListNotificationChannelsOperator(GoogleCloudBaseOperator):
context=context,
project_id=self.project_id or self.hook.project_id,
)
- return [NotificationChannel.to_dict(channel) for channel in channels]
+ if self.format_ not in ("dict", "json"):
+ return [NotificationChannel.to_dict(channel) for channel in
channels]
+ return channels
class
CloudMonitoringEnableNotificationChannelsOperator(GoogleCloudBaseOperator):
diff --git
a/providers/google/tests/unit/google/cloud/operators/test_cloud_monitoring.py
b/providers/google/tests/unit/google/cloud/operators/test_cloud_monitoring.py
index 8ef3bf7ac08..660fd69af3b 100644
---
a/providers/google/tests/unit/google/cloud/operators/test_cloud_monitoring.py
+++
b/providers/google/tests/unit/google/cloud/operators/test_cloud_monitoring.py
@@ -20,6 +20,7 @@ from __future__ import annotations
import json
from unittest import mock
+import pytest
from google.api_core.gapic_v1.method import DEFAULT
from google.cloud.monitoring_v3 import AlertPolicy, NotificationChannel
@@ -89,7 +90,63 @@ TEST_NOTIFICATION_CHANNEL_2 = {
}
[email protected]("format_", ["Dict", "protobuf", ""])
[email protected](
+ ("operator_class", "hook_method", "resource"),
+ [
+ (CloudMonitoringListAlertPoliciesOperator, "list_alert_policies",
AlertPolicy(name="test-policy")),
+ (
+ CloudMonitoringListNotificationChannelsOperator,
+ "list_notification_channels",
+ NotificationChannel(name="test-channel"),
+ ),
+ ],
+)
[email protected]("airflow.providers.google.cloud.operators.cloud_monitoring.CloudMonitoringHook",
autospec=True)
+def test_list_operator_converts_unrecognized_format(
+ mock_hook, operator_class, hook_method, resource, format_
+):
+ operator = operator_class(task_id=TEST_TASK_ID, format_=format_)
+ list_resources = getattr(mock_hook.return_value, hook_method)
+ list_resources.return_value = iter([resource])
+
+ result = operator.execute(context=mock.MagicMock(spec=dict))
+
+ assert list_resources.call_args.kwargs["format_"] == format_
+ assert result == [type(resource).to_dict(resource)]
+
+
class TestCloudMonitoringListAlertPoliciesOperator:
+ @pytest.mark.parametrize(
+ ("format_", "policies"),
+ [
+ ("dict", [TEST_ALERT_POLICY_1, TEST_ALERT_POLICY_2]),
+ ("json", [json.dumps(TEST_ALERT_POLICY_1),
json.dumps(TEST_ALERT_POLICY_2)]),
+ ],
+ )
+ @mock.patch(
+
"airflow.providers.google.cloud.operators.cloud_monitoring.CloudMonitoringHook",
autospec=True
+ )
+ def test_execute_preserves_formatted_result(self, mock_hook, format_,
policies):
+ operator = CloudMonitoringListAlertPoliciesOperator(
+ task_id=TEST_TASK_ID, filter_=TEST_FILTER, format_=format_
+ )
+ mock_hook.return_value.list_alert_policies.return_value = policies
+
+ result = operator.execute(context=mock.MagicMock(spec=dict))
+
+ mock_hook.return_value.list_alert_policies.assert_called_once_with(
+ project_id=None,
+ filter_=TEST_FILTER,
+ format_=format_,
+ order_by=None,
+ page_size=None,
+ retry=DEFAULT,
+ timeout=None,
+ metadata=(),
+ )
+ assert result is policies
+
@mock.patch("airflow.providers.google.cloud.operators.cloud_monitoring.CloudMonitoringHook")
def test_execute(self, mock_hook):
operator =
CloudMonitoringListAlertPoliciesOperator(task_id=TEST_TASK_ID,
filter_=TEST_FILTER)
@@ -168,6 +225,36 @@ class TestCloudMonitoringDeleteAlertOperator:
class TestCloudMonitoringListNotificationChannelsOperator:
+ @pytest.mark.parametrize(
+ ("format_", "channels"),
+ [
+ ("dict", [TEST_NOTIFICATION_CHANNEL_1,
TEST_NOTIFICATION_CHANNEL_2]),
+ ("json", [json.dumps(TEST_NOTIFICATION_CHANNEL_1),
json.dumps(TEST_NOTIFICATION_CHANNEL_2)]),
+ ],
+ )
+ @mock.patch(
+
"airflow.providers.google.cloud.operators.cloud_monitoring.CloudMonitoringHook",
autospec=True
+ )
+ def test_execute_preserves_formatted_result(self, mock_hook, format_,
channels):
+ operator = CloudMonitoringListNotificationChannelsOperator(
+ task_id=TEST_TASK_ID, filter_=TEST_FILTER, format_=format_
+ )
+ mock_hook.return_value.list_notification_channels.return_value =
channels
+
+ result = operator.execute(context=mock.MagicMock(spec=dict))
+
+
mock_hook.return_value.list_notification_channels.assert_called_once_with(
+ project_id=None,
+ filter_=TEST_FILTER,
+ format_=format_,
+ order_by=None,
+ page_size=None,
+ retry=DEFAULT,
+ timeout=None,
+ metadata=(),
+ )
+ assert result is channels
+
@mock.patch("airflow.providers.google.cloud.operators.cloud_monitoring.CloudMonitoringHook")
def test_execute(self, mock_hook):
operator =
CloudMonitoringListNotificationChannelsOperator(task_id=TEST_TASK_ID,
filter_=TEST_FILTER)
