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 48d0ad5f89f Add tests for links/base and links/bigquery in Google
provider (#68066)
48d0ad5f89f is described below
commit 48d0ad5f89f0f7c9af876016dd83a7445af6190f
Author: Chao-Hung Wan <[email protected]>
AuthorDate: Wed Jun 17 15:07:08 2026 +0800
Add tests for links/base and links/bigquery in Google provider (#68066)
---
.../tests/unit/always/test_project_structure.py | 2 -
.../tests/unit/google/cloud/links/test_base.py | 167 +++++++++++++++++++++
.../tests/unit/google/cloud/links/test_bigquery.py | 143 ++++++++++++++++++
3 files changed, 310 insertions(+), 2 deletions(-)
diff --git a/airflow-core/tests/unit/always/test_project_structure.py
b/airflow-core/tests/unit/always/test_project_structure.py
index 808d62ff5fc..750966a3202 100644
--- a/airflow-core/tests/unit/always/test_project_structure.py
+++ b/airflow-core/tests/unit/always/test_project_structure.py
@@ -136,8 +136,6 @@ class TestProjectStructure:
"providers/fab/tests/unit/fab/www/test_security_manager.py",
"providers/fab/tests/unit/fab/www/test_session.py",
"providers/fab/tests/unit/fab/www/test_views.py",
- "providers/google/tests/unit/google/cloud/links/test_base.py",
- "providers/google/tests/unit/google/cloud/links/test_bigquery.py",
"providers/google/tests/unit/google/cloud/links/test_bigquery_dts.py",
"providers/google/tests/unit/google/cloud/links/test_bigtable.py",
"providers/google/tests/unit/google/cloud/links/test_cloud_build.py",
diff --git a/providers/google/tests/unit/google/cloud/links/test_base.py
b/providers/google/tests/unit/google/cloud/links/test_base.py
new file mode 100644
index 00000000000..48e93986af9
--- /dev/null
+++ b/providers/google/tests/unit/google/cloud/links/test_base.py
@@ -0,0 +1,167 @@
+#
+# 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 __future__ import annotations
+
+from unittest import mock
+
+from airflow.providers.google.cloud.links.base import BASE_LINK, BaseGoogleLink
+
+# ---------------------------------------------------------------------------
+# Concrete subclass used throughout the tests
+# ---------------------------------------------------------------------------
+TEST_KEY = "test_link"
+TEST_NAME = "Test Link"
+TEST_FORMAT_STR = "/test/{project_id}/{resource_id}"
+TEST_PROJECT_ID = "test-project"
+TEST_RESOURCE_ID = "test-resource"
+
+
+class ConcreteGoogleLink(BaseGoogleLink):
+ key = TEST_KEY
+ name = TEST_NAME
+ format_str = TEST_FORMAT_STR
+
+
+class TestBaseGoogleLinkXcomKey:
+ def test_xcom_key_equals_key(self):
+ link = ConcreteGoogleLink()
+ assert link.xcom_key == TEST_KEY
+
+
+class TestBaseGoogleLinkPersist:
+ def test_persist_pushes_to_xcom(self):
+ mock_context = mock.MagicMock()
+ mock_context["ti"] = mock.MagicMock()
+ mock_context["task"] = mock.MagicMock(spec=[]) # no extra_links_params
+
+ ConcreteGoogleLink.persist(
+ context=mock_context,
+ project_id=TEST_PROJECT_ID,
+ resource_id=TEST_RESOURCE_ID,
+ )
+
+ mock_context["ti"].xcom_push.assert_called_once_with(
+ key=TEST_KEY,
+ value={"project_id": TEST_PROJECT_ID, "resource_id":
TEST_RESOURCE_ID},
+ )
+
+
+class TestBaseGoogleLinkGetConfig:
+ def test_returns_none_when_no_config(self):
+ link = ConcreteGoogleLink()
+ operator = mock.MagicMock(spec=[]) # no extra_links_params
+ ti_key = mock.MagicMock()
+
+ with
mock.patch("airflow.providers.google.cloud.links.base.XCom.get_value",
return_value=None):
+ result = link.get_config(operator, ti_key)
+
+ assert result is None
+
+ def test_returns_merged_config(self):
+ link = ConcreteGoogleLink()
+ operator = mock.MagicMock(extra_links_params={"project_id":
"from-operator"})
+ ti_key = mock.MagicMock()
+
+ with mock.patch(
+ "airflow.providers.google.cloud.links.base.XCom.get_value",
+ return_value={"resource_id": "from-xcom"},
+ ):
+ result = link.get_config(operator, ti_key)
+
+ assert result == {
+ "project_id": "from-operator",
+ "resource_id": "from-xcom",
+ "namespace": "default", # default injected for datafusion
back-compat
+ }
+
+ def test_xcom_overrides_operator_params(self):
+ link = ConcreteGoogleLink()
+ operator = mock.MagicMock(extra_links_params={"project_id":
"from-operator"})
+ ti_key = mock.MagicMock()
+
+ with mock.patch(
+ "airflow.providers.google.cloud.links.base.XCom.get_value",
+ return_value={"project_id": "from-xcom", "resource_id":
TEST_RESOURCE_ID},
+ ):
+ result = link.get_config(operator, ti_key)
+
+ assert result["project_id"] == "from-xcom"
+
+
+class TestBaseGoogleLinkFormatLink:
+ def test_formats_relative_path_with_base_link(self):
+ link = ConcreteGoogleLink()
+ result = link._format_link(project_id=TEST_PROJECT_ID,
resource_id=TEST_RESOURCE_ID)
+ assert result == BASE_LINK +
f"/test/{TEST_PROJECT_ID}/{TEST_RESOURCE_ID}"
+
+ def test_returns_empty_string_on_missing_key(self):
+ link = ConcreteGoogleLink()
+ result = link._format_link(project_id=TEST_PROJECT_ID) # missing
resource_id
+ assert result == ""
+
+ def test_returns_absolute_url_unchanged(self):
+ class AbsoluteLink(BaseGoogleLink):
+ key = "abs_link"
+ name = "Abs Link"
+ format_str = "https://example.com/{resource_id}";
+
+ link = AbsoluteLink()
+ result = link._format_link(resource_id="res")
+ assert result == "https://example.com/res";
+
+
+class TestBaseGoogleLinkGetLink:
+ def test_returns_empty_string_when_no_config(self):
+ link = ConcreteGoogleLink()
+ operator = mock.MagicMock(spec=[], extra_links_params={})
+ ti_key = mock.MagicMock()
+
+ with
mock.patch("airflow.providers.google.cloud.links.base.XCom.get_value",
return_value=None):
+ result = link.get_link(operator=operator, ti_key=ti_key)
+
+ assert result == ""
+
+ def test_returns_http_string_directly_without_formatting(self):
+ """If XCom already holds a full http URL it should be returned
as-is."""
+ link = ConcreteGoogleLink()
+ operator = mock.MagicMock(spec=[])
+ ti_key = mock.MagicMock()
+ stored_url = "https://console.cloud.google.com/already/formatted";
+
+ with mock.patch(
+ "airflow.providers.google.cloud.links.base.XCom.get_value",
+ return_value=stored_url,
+ ):
+ result = link.get_link(operator=operator, ti_key=ti_key)
+
+ assert result == stored_url
+
+ def test_formats_link_from_config(self):
+ link = ConcreteGoogleLink()
+ operator = mock.MagicMock(spec=[], extra_links_params={})
+ ti_key = mock.MagicMock()
+ xcom_value = {"project_id": TEST_PROJECT_ID, "resource_id":
TEST_RESOURCE_ID}
+
+ with mock.patch(
+ "airflow.providers.google.cloud.links.base.XCom.get_value",
+ return_value=xcom_value,
+ ):
+ result = link.get_link(operator=operator, ti_key=ti_key)
+
+ expected = BASE_LINK + f"/test/{TEST_PROJECT_ID}/{TEST_RESOURCE_ID}"
+ assert result == expected
diff --git a/providers/google/tests/unit/google/cloud/links/test_bigquery.py
b/providers/google/tests/unit/google/cloud/links/test_bigquery.py
new file mode 100644
index 00000000000..8822b46c60d
--- /dev/null
+++ b/providers/google/tests/unit/google/cloud/links/test_bigquery.py
@@ -0,0 +1,143 @@
+#
+# 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 for BigQuery links."""
+
+from __future__ import annotations
+
+from unittest import mock
+
+from airflow.providers.google.cloud.links.base import BASE_LINK
+from airflow.providers.google.cloud.links.bigquery import (
+ BIGQUERY_DATASET_LINK,
+ BIGQUERY_JOB_DETAIL_LINK,
+ BIGQUERY_TABLE_LINK,
+ BigQueryDatasetLink,
+ BigQueryJobDetailLink,
+ BigQueryTableLink,
+)
+
+TEST_PROJECT_ID = "test-project"
+TEST_DATASET_ID = "test-dataset"
+TEST_TABLE_ID = "test-table"
+TEST_JOB_ID = "test-job-id"
+TEST_LOCATION = "US"
+
+
+class TestBigQueryDatasetLink:
+ def test_class_attributes(self):
+ assert BigQueryDatasetLink.key == "bigquery_dataset"
+ assert BigQueryDatasetLink.name == "BigQuery Dataset"
+ assert BigQueryDatasetLink.format_str == BIGQUERY_DATASET_LINK
+
+ def test_persist(self):
+ mock_context = mock.MagicMock()
+ mock_context["ti"] = mock.MagicMock()
+ mock_context["task"] = mock.MagicMock(spec=[])
+
+ BigQueryDatasetLink.persist(
+ context=mock_context,
+ project_id=TEST_PROJECT_ID,
+ dataset_id=TEST_DATASET_ID,
+ )
+
+ mock_context["ti"].xcom_push.assert_called_once_with(
+ key="bigquery_dataset",
+ value={"project_id": TEST_PROJECT_ID, "dataset_id":
TEST_DATASET_ID},
+ )
+
+ def test_format_link(self):
+ link = BigQueryDatasetLink()
+ result = link._format_link(project_id=TEST_PROJECT_ID,
dataset_id=TEST_DATASET_ID)
+ expected = BASE_LINK + BIGQUERY_DATASET_LINK.format(
+ project_id=TEST_PROJECT_ID, dataset_id=TEST_DATASET_ID
+ )
+ assert result == expected
+
+
+class TestBigQueryTableLink:
+ def test_class_attributes(self):
+ assert BigQueryTableLink.key == "bigquery_table"
+ assert BigQueryTableLink.name == "BigQuery Table"
+ assert BigQueryTableLink.format_str == BIGQUERY_TABLE_LINK
+
+ def test_persist(self):
+ mock_context = mock.MagicMock()
+ mock_context["ti"] = mock.MagicMock()
+ mock_context["task"] = mock.MagicMock(spec=[])
+
+ BigQueryTableLink.persist(
+ context=mock_context,
+ project_id=TEST_PROJECT_ID,
+ dataset_id=TEST_DATASET_ID,
+ table_id=TEST_TABLE_ID,
+ )
+
+ mock_context["ti"].xcom_push.assert_called_once_with(
+ key="bigquery_table",
+ value={
+ "project_id": TEST_PROJECT_ID,
+ "dataset_id": TEST_DATASET_ID,
+ "table_id": TEST_TABLE_ID,
+ },
+ )
+
+ def test_format_link(self):
+ link = BigQueryTableLink()
+ result = link._format_link(
+ project_id=TEST_PROJECT_ID, dataset_id=TEST_DATASET_ID,
table_id=TEST_TABLE_ID
+ )
+ expected = BASE_LINK + BIGQUERY_TABLE_LINK.format(
+ project_id=TEST_PROJECT_ID, dataset_id=TEST_DATASET_ID,
table_id=TEST_TABLE_ID
+ )
+ assert result == expected
+
+
+class TestBigQueryJobDetailLink:
+ def test_class_attributes(self):
+ assert BigQueryJobDetailLink.key == "bigquery_job_detail"
+ assert BigQueryJobDetailLink.name == "BigQuery Job Detail"
+ assert BigQueryJobDetailLink.format_str == BIGQUERY_JOB_DETAIL_LINK
+
+ def test_persist(self):
+ mock_context = mock.MagicMock()
+ mock_context["ti"] = mock.MagicMock()
+ mock_context["task"] = mock.MagicMock(spec=[])
+
+ BigQueryJobDetailLink.persist(
+ context=mock_context,
+ project_id=TEST_PROJECT_ID,
+ job_id=TEST_JOB_ID,
+ location=TEST_LOCATION,
+ )
+
+ mock_context["ti"].xcom_push.assert_called_once_with(
+ key="bigquery_job_detail",
+ value={
+ "project_id": TEST_PROJECT_ID,
+ "job_id": TEST_JOB_ID,
+ "location": TEST_LOCATION,
+ },
+ )
+
+ def test_format_link(self):
+ link = BigQueryJobDetailLink()
+ result = link._format_link(project_id=TEST_PROJECT_ID,
job_id=TEST_JOB_ID, location=TEST_LOCATION)
+ expected = BASE_LINK + BIGQUERY_JOB_DETAIL_LINK.format(
+ project_id=TEST_PROJECT_ID, job_id=TEST_JOB_ID,
location=TEST_LOCATION
+ )
+ assert result == expected
