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 0d5f53297a0 Add unit tests for Dataproc, GKE and Translate provider
links (#75694)
0d5f53297a0 is described below
commit 0d5f53297a0a7dc745c66637a4e96edffda887d3
Author: Vinayak <[email protected]>
AuthorDate: Thu Nov 5 11:27:42 2026 +0530
Add unit tests for Dataproc, GKE and Translate provider links (#75694)
---
.../tests/unit/always/test_project_structure.py | 3 -
.../tests/unit/google/cloud/links/test_dataproc.py | 199 +++++++++++++++++++
.../google/cloud/links/test_kubernetes_engine.py | 152 +++++++++++++++
.../unit/google/cloud/links/test_translate.py | 212 +++++++++++++++++++++
4 files changed, 563 insertions(+), 3 deletions(-)
diff --git a/airflow-core/tests/unit/always/test_project_structure.py
b/airflow-core/tests/unit/always/test_project_structure.py
index babd38bb9ce..570393b33a7 100644
--- a/airflow-core/tests/unit/always/test_project_structure.py
+++ b/airflow-core/tests/unit/always/test_project_structure.py
@@ -87,9 +87,6 @@ class TestProjectStructure:
"providers/fab/tests/unit/fab/www/test_security_appless.py",
"providers/fab/tests/unit/fab/www/test_security_manager.py",
"providers/fab/tests/unit/fab/www/test_session.py",
- "providers/google/tests/unit/google/cloud/links/test_dataproc.py",
-
"providers/google/tests/unit/google/cloud/links/test_kubernetes_engine.py",
- "providers/google/tests/unit/google/cloud/links/test_translate.py",
"providers/google/tests/unit/google/cloud/operators/vertex_ai/test_auto_ml.py",
"providers/google/tests/unit/google/cloud/operators/vertex_ai/test_batch_prediction_job.py",
"providers/google/tests/unit/google/cloud/operators/vertex_ai/test_custom_job.py",
diff --git a/providers/google/tests/unit/google/cloud/links/test_dataproc.py
b/providers/google/tests/unit/google/cloud/links/test_dataproc.py
new file mode 100644
index 00000000000..40e0c2d9a97
--- /dev/null
+++ b/providers/google/tests/unit/google/cloud/links/test_dataproc.py
@@ -0,0 +1,199 @@
+#
+# 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 Google Cloud Dataproc links.
+
+Expected URLs are fully hard-coded: the path and query parts are never derived
from the
+constants under test, so a broken or hand-edited ``format_str`` always fails
the test.
+"""
+
+from __future__ import annotations
+
+from unittest import mock
+
+import pytest
+
+from airflow.exceptions import AirflowProviderDeprecationWarning
+from airflow.providers.google.cloud.links import dataproc
+from airflow.providers.google.cloud.links.base import BASE_LINK
+from airflow.providers.google.cloud.links.dataproc import (
+ DataprocBatchesListLink,
+ DataprocBatchLink,
+ DataprocClusterLink,
+ DataprocJobLink,
+ DataprocLink,
+ DataprocListLink,
+ DataprocWorkflowLink,
+ DataprocWorkflowTemplateLink,
+)
+
+TEST_PROJECT_ID = "test-project"
+TEST_REGION = "us-central1"
+
+# The legacy links store the whole URL template in the XCom, not just the
identifiers.
+LEGACY_JOB_URL_TEMPLATE = (
+
"https://console.cloud.google.com/dataproc/jobs/{resource}?region={region}&project={project_id}";
+)
+LEGACY_BATCHES_URL_TEMPLATE =
"https://console.cloud.google.com/dataproc/batches?project={project_id}";
+
+
+def _legacy_link_instance_with_warning():
+ with pytest.warns(AirflowProviderDeprecationWarning):
+ return DataprocListLink()
+
+
+class TestDataprocLink:
+ def test_persist(self):
+ mock_context = mock.MagicMock()
+
+ DataprocLink.persist(
+ context=mock_context,
+ url=LEGACY_JOB_URL_TEMPLATE,
+ resource="test-job-1",
+ region=TEST_REGION,
+ project_id=TEST_PROJECT_ID,
+ )
+
+ mock_context["task_instance"].xcom_push.assert_called_once_with(
+ key="conf",
+ value={
+ "region": TEST_REGION,
+ "project_id": TEST_PROJECT_ID,
+ "url": LEGACY_JOB_URL_TEMPLATE,
+ "resource": "test-job-1",
+ },
+ )
+
+ def test_get_link_formats_the_xcom_url(self):
+ link = DataprocLink()
+ conf = {
+ "url": LEGACY_JOB_URL_TEMPLATE,
+ "region": TEST_REGION,
+ "project_id": TEST_PROJECT_ID,
+ "resource": "test-job-1",
+ }
+
+ with
mock.patch("airflow.providers.google.cloud.links.dataproc.XCom.get_value",
return_value=conf):
+ result = link.get_link(operator=mock.MagicMock(),
ti_key=mock.MagicMock())
+
+ assert (
+ result
+ ==
"https://console.cloud.google.com/dataproc/jobs/test-job-1?region=us-central1&project=test-project";
+ )
+
+ def test_get_link_returns_empty_string_when_xcom_missing(self):
+ link = DataprocLink()
+
+ with
mock.patch("airflow.providers.google.cloud.links.dataproc.XCom.get_value",
return_value=None):
+ result = link.get_link(operator=mock.MagicMock(),
ti_key=mock.MagicMock())
+
+ assert result == ""
+
+
+class TestDataprocListLink:
+ def test_instantiation_warns_about_deprecation(self):
+ with pytest.warns(AirflowProviderDeprecationWarning):
+ DataprocListLink()
+
+ def test_persist(self):
+ mock_context = mock.MagicMock()
+
+ DataprocListLink.persist(
+ context=mock_context,
+ url=LEGACY_BATCHES_URL_TEMPLATE,
+ project_id=TEST_PROJECT_ID,
+ )
+
+ mock_context["task_instance"].xcom_push.assert_called_once_with(
+ key="list_conf",
+ value={"project_id": TEST_PROJECT_ID, "url":
LEGACY_BATCHES_URL_TEMPLATE},
+ )
+
+ def test_get_link_formats_the_xcom_url(self):
+ link = _legacy_link_instance_with_warning()
+ conf = {"url": LEGACY_BATCHES_URL_TEMPLATE, "project_id":
TEST_PROJECT_ID}
+
+ with
mock.patch("airflow.providers.google.cloud.links.dataproc.XCom.get_value",
return_value=conf):
+ result = link.get_link(operator=mock.MagicMock(),
ti_key=mock.MagicMock())
+
+ assert result ==
"https://console.cloud.google.com/dataproc/batches?project=test-project";
+
+
+class TestDataprocClusterLink:
+ def test_get_link(self):
+ result = DataprocClusterLink()._format_link(
+ cluster_id="test-cluster", region=TEST_REGION,
project_id=TEST_PROJECT_ID
+ )
+ assert result == (
+ BASE_LINK +
"/dataproc/clusters/test-cluster/monitoring?region=us-central1&project=test-project"
+ )
+
+
+class TestDataprocJobLink:
+ def test_get_link(self):
+ result = DataprocJobLink()._format_link(
+ job_id="test-job-1", region=TEST_REGION, project_id=TEST_PROJECT_ID
+ )
+ assert result == BASE_LINK +
"/dataproc/jobs/test-job-1?region=us-central1&project=test-project"
+
+
+class TestDataprocWorkflowLink:
+ def test_get_link(self):
+ result = DataprocWorkflowLink()._format_link(
+ region=TEST_REGION, workflow_id="test-workflow-1",
project_id=TEST_PROJECT_ID
+ )
+ assert result == (
+ BASE_LINK +
"/dataproc/workflows/instances/us-central1/test-workflow-1?project=test-project"
+ )
+
+
+class TestDataprocWorkflowTemplateLink:
+ def test_get_link(self):
+ result = DataprocWorkflowTemplateLink()._format_link(
+ region=TEST_REGION, workflow_template_id="test-template-1",
project_id=TEST_PROJECT_ID
+ )
+ assert result == (
+ BASE_LINK +
"/dataproc/workflows/templates/us-central1/test-template-1?project=test-project"
+ )
+
+
+class TestDataprocBatchLink:
+ def test_get_link(self):
+ # Note the template orders the path elements as "region, batch", not
"batch, region".
+ result = DataprocBatchLink()._format_link(
+ region=TEST_REGION, batch_id="test-batch-1",
project_id=TEST_PROJECT_ID
+ )
+ assert result == (
+ BASE_LINK +
"/dataproc/batches/us-central1/test-batch-1/monitoring?project=test-project"
+ )
+
+
+class TestDataprocBatchesListLink:
+ def test_get_link(self):
+ result =
DataprocBatchesListLink()._format_link(project_id=TEST_PROJECT_ID)
+ assert result == BASE_LINK + "/dataproc/batches?project=test-project"
+
+
+class TestDeprecatedModuleAlias:
+ def test_legacy_alias_warns_and_resolves_to_new_constant(self):
+ with pytest.warns(AirflowProviderDeprecationWarning):
+ alias = getattr(dataproc, "DATAPROC_JOB_LOG_LINK")
+ assert alias == dataproc.DATAPROC_JOB_LINK
+
+ def test_unknown_attribute_raises(self):
+ with pytest.raises(AttributeError):
+ getattr(dataproc, "DATAPROC_UNKNOWN_LINK")
diff --git
a/providers/google/tests/unit/google/cloud/links/test_kubernetes_engine.py
b/providers/google/tests/unit/google/cloud/links/test_kubernetes_engine.py
new file mode 100644
index 00000000000..b21f50f60a9
--- /dev/null
+++ b/providers/google/tests/unit/google/cloud/links/test_kubernetes_engine.py
@@ -0,0 +1,152 @@
+#
+# 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 Google Cloud Kubernetes Engine links.
+
+Expected URLs are fully hard-coded: the path and query parts are never derived
from the
+constants under test, so a broken or hand-edited ``format_str`` always fails
the test.
+"""
+
+from __future__ import annotations
+
+from unittest import mock
+
+import pytest
+from google.cloud.container_v1.types import Cluster
+
+from airflow.providers.google.cloud.links.base import BASE_LINK
+from airflow.providers.google.cloud.links.kubernetes_engine import (
+ KubernetesEngineClusterLink,
+ KubernetesEngineJobLink,
+ KubernetesEnginePodLink,
+ KubernetesEngineWorkloadsLink,
+)
+
+TEST_PROJECT_ID = "test-project"
+TEST_LOCATION = "us-central1"
+TEST_CLUSTER_NAME = "test-cluster"
+TEST_NAMESPACE = "test-namespace"
+
+
+def _mock_context():
+ mock_context = mock.MagicMock()
+ mock_context["ti"] = mock.MagicMock()
+ mock_context["task"] = mock.MagicMock(spec=[])
+ return mock_context
+
+
+class TestKubernetesEngineClusterLink:
+ def test_get_link(self):
+ result = KubernetesEngineClusterLink()._format_link(
+ location=TEST_LOCATION, cluster_name=TEST_CLUSTER_NAME,
project_id=TEST_PROJECT_ID
+ )
+ assert result == (
+ BASE_LINK +
"/kubernetes/clusters/details/us-central1/test-cluster/details?project=test-project"
+ )
+
+ def test_persist_accepts_a_cluster_dict(self):
+ mock_context = _mock_context()
+
+ KubernetesEngineClusterLink.persist(mock_context, cluster={"name":
TEST_CLUSTER_NAME})
+
+ mock_context["ti"].xcom_push.assert_called_once_with(
+ key="kubernetes_cluster_conf",
+ value={"cluster_name": TEST_CLUSTER_NAME},
+ )
+
+ def test_persist_accepts_a_cluster_object(self):
+ mock_context = _mock_context()
+
+ KubernetesEngineClusterLink.persist(mock_context,
cluster=Cluster(name=TEST_CLUSTER_NAME))
+
+ mock_context["ti"].xcom_push.assert_called_once_with(
+ key="kubernetes_cluster_conf",
+ value={"cluster_name": TEST_CLUSTER_NAME},
+ )
+
+ def test_persist_without_cluster_raises(self):
+ mock_context = _mock_context()
+
+ with pytest.raises(ValueError, match="Cluster must be provided for
KubernetesEngineClusterLink."):
+ KubernetesEngineClusterLink.persist(mock_context)
+
+
+class TestKubernetesEnginePodLink:
+ def test_get_link(self):
+ result = KubernetesEnginePodLink()._format_link(
+ location=TEST_LOCATION,
+ cluster_name=TEST_CLUSTER_NAME,
+ namespace=TEST_NAMESPACE,
+ pod_name="test-pod-1",
+ project_id=TEST_PROJECT_ID,
+ )
+ assert result == (
+ BASE_LINK
+ +
"/kubernetes/pod/us-central1/test-cluster/test-namespace/test-pod-1/details?project=test-project"
+ )
+
+ def test_persist(self):
+ mock_context = _mock_context()
+
+ KubernetesEnginePodLink.persist(
+ mock_context,
+ location=TEST_LOCATION,
+ cluster_name=TEST_CLUSTER_NAME,
+ namespace=TEST_NAMESPACE,
+ pod_name="test-pod-1",
+ )
+
+ mock_context["ti"].xcom_push.assert_called_once_with(
+ key="kubernetes_pod_conf",
+ value={
+ "location": TEST_LOCATION,
+ "cluster_name": TEST_CLUSTER_NAME,
+ "namespace": TEST_NAMESPACE,
+ "pod_name": "test-pod-1",
+ },
+ )
+
+
+class TestKubernetesEngineJobLink:
+ def test_get_link(self):
+ result = KubernetesEngineJobLink()._format_link(
+ location=TEST_LOCATION,
+ cluster_name=TEST_CLUSTER_NAME,
+ namespace=TEST_NAMESPACE,
+ job_name="test-job-1",
+ project_id=TEST_PROJECT_ID,
+ )
+ assert result == (
+ BASE_LINK
+ +
"/kubernetes/job/us-central1/test-cluster/test-namespace/test-job-1/details?project=test-project"
+ )
+
+
+class TestKubernetesEngineWorkloadsLink:
+ def test_get_link_builds_the_console_deep_link(self):
+ result = KubernetesEngineWorkloadsLink()._format_link(
+ location=TEST_LOCATION,
+ cluster_name=TEST_CLUSTER_NAME,
+ namespace=TEST_NAMESPACE,
+ project_id=TEST_PROJECT_ID,
+ )
+ expected = (
+ BASE_LINK
+ +
'/kubernetes/workload/overview?project=test-project&pageState=("savedViews":'
+ +
'("c":%5B"gke%2Fus-central1%2Ftest-cluster"%5D,"n":%5B"test-namespace"%5D))'
+ )
+ assert result == expected
diff --git a/providers/google/tests/unit/google/cloud/links/test_translate.py
b/providers/google/tests/unit/google/cloud/links/test_translate.py
new file mode 100644
index 00000000000..2197b77d10b
--- /dev/null
+++ b/providers/google/tests/unit/google/cloud/links/test_translate.py
@@ -0,0 +1,212 @@
+#
+# 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 Google Cloud Translate links.
+
+Expected URLs are fully hard-coded: the path and query parts are never derived
from the
+constants under test, so a broken or hand-edited ``format_str`` always fails
the test.
+"""
+
+from __future__ import annotations
+
+from unittest import mock
+
+from airflow.providers.google.cloud.links.base import BASE_LINK
+from airflow.providers.google.cloud.links.translate import (
+ TranslateResultByOutputConfigLink,
+ TranslateTextBatchLink,
+ TranslationDatasetListLink,
+ TranslationDatasetsListLink,
+ TranslationGlossariesListLink,
+ TranslationLegacyDatasetLink,
+ TranslationLegacyModelLink,
+ TranslationLegacyModelPredictLink,
+ TranslationLegacyModelTrainLink,
+ TranslationModelLink,
+ TranslationModelsListLink,
+ TranslationNativeDatasetLink,
+)
+
+TEST_PROJECT_ID = "test-project"
+TEST_DATASET_ID = "test-dataset"
+TEST_MODEL_ID = "test-model"
+TEST_LOCATION = "europe-west1"
+TEST_OUTPUT_URI_PREFIX = "test-bucket/test-results/run-1"
+
+
+class TestTranslationLegacyDatasetLink:
+ def test_get_link(self):
+ result = TranslationLegacyDatasetLink()._format_link(
+ location=TEST_LOCATION, dataset_id=TEST_DATASET_ID,
project_id=TEST_PROJECT_ID
+ )
+ assert result == (
+ BASE_LINK
+ +
"/translation/locations/europe-west1/datasets/test-dataset/sentences?project=test-project"
+ )
+
+
+class TestTranslationDatasetListLink:
+ def test_get_link(self):
+ result =
TranslationDatasetListLink()._format_link(project_id=TEST_PROJECT_ID)
+ assert result == BASE_LINK +
"/translation/datasets?project=test-project"
+
+
+class TestTranslationLegacyModelLink:
+ def test_get_link(self):
+ result = TranslationLegacyModelLink()._format_link(
+ location=TEST_LOCATION,
+ dataset_id=TEST_DATASET_ID,
+ model_id=TEST_MODEL_ID,
+ project_id=TEST_PROJECT_ID,
+ )
+ assert result == (
+ BASE_LINK
+ + "/translation/locations/europe-west1/datasets/test-dataset/"
+ + "evaluate;modelId=test-model?project=test-project"
+ )
+
+
+class TestTranslationLegacyModelTrainLink:
+ def test_get_link(self):
+ result = TranslationLegacyModelTrainLink()._format_link(
+ location=TEST_LOCATION, dataset_id=TEST_DATASET_ID,
project_id=TEST_PROJECT_ID
+ )
+ assert result == (
+ BASE_LINK +
"/translation/locations/europe-west1/datasets/test-dataset/train?project=test-project"
+ )
+
+
+class TestTranslationLegacyModelPredictLink:
+ def test_get_link(self):
+ result = TranslationLegacyModelPredictLink()._format_link(
+ location=TEST_LOCATION,
+ dataset_id=TEST_DATASET_ID,
+ model_id=TEST_MODEL_ID,
+ project_id=TEST_PROJECT_ID,
+ )
+ assert result == (
+ BASE_LINK
+ + "/translation/locations/europe-west1/datasets/test-dataset/"
+ + "predict;modelId=test-model?project=test-project"
+ )
+
+
+class TestTranslationNativeDatasetLink:
+ def test_get_link(self):
+ result = TranslationNativeDatasetLink()._format_link(
+ location=TEST_LOCATION, dataset_id=TEST_DATASET_ID,
project_id=TEST_PROJECT_ID
+ )
+ assert result == (
+ BASE_LINK
+ +
"/translation/locations/europe-west1/datasets/test-dataset/sentences?project=test-project"
+ )
+
+
+class TestTranslationDatasetsListLink:
+ def test_get_link(self):
+ result =
TranslationDatasetsListLink()._format_link(project_id=TEST_PROJECT_ID)
+ assert result == BASE_LINK +
"/translation/datasets?project=test-project"
+
+
+class TestTranslationModelLink:
+ def test_get_link(self):
+ result = TranslationModelLink()._format_link(
+ location=TEST_LOCATION,
+ dataset_id=TEST_DATASET_ID,
+ model_id=TEST_MODEL_ID,
+ project_id=TEST_PROJECT_ID,
+ )
+ assert result == (
+ BASE_LINK
+ + "/translation/locations/europe-west1/datasets/test-dataset/"
+ + "evaluate;modelId=test-model?project=test-project"
+ )
+
+
+class TestTranslationModelsListLink:
+ def test_get_link(self):
+ result =
TranslationModelsListLink()._format_link(project_id=TEST_PROJECT_ID)
+ assert result == BASE_LINK +
"/translation/models/list?project=test-project"
+
+
+class TestTranslationGlossariesListLink:
+ def test_get_link(self):
+ result =
TranslationGlossariesListLink()._format_link(project_id=TEST_PROJECT_ID)
+ assert result == BASE_LINK +
"/translation/hub/resources?project=test-project"
+
+
+class TestTranslateTextBatchLink:
+ def test_extract_output_uri_prefix_strips_gs_scheme(self):
+ output_config = {"gcs_destination": {"output_uri_prefix": "gs://" +
TEST_OUTPUT_URI_PREFIX}}
+ assert TranslateTextBatchLink.extract_output_uri_prefix(output_config)
== TEST_OUTPUT_URI_PREFIX
+
+ def test_extract_output_uri_prefix_keeps_relative_prefix(self):
+ long_prefix = {"gcs_destination": {"output_uri_prefix":
TEST_OUTPUT_URI_PREFIX}}
+ assert TranslateTextBatchLink.extract_output_uri_prefix(long_prefix)
== TEST_OUTPUT_URI_PREFIX
+
+ def test_get_link(self):
+ result = TranslateTextBatchLink()._format_link(
+ project_id=TEST_PROJECT_ID,
output_uri_prefix=TEST_OUTPUT_URI_PREFIX
+ )
+ assert result == BASE_LINK +
"/storage/browser/test-bucket/test-results/run-1?project=test-project"
+
+ def test_persist(self):
+ mock_context = mock.MagicMock()
+ mock_context["ti"] = mock.MagicMock()
+ mock_context["task"] = mock.MagicMock(spec=[])
+
+ TranslateTextBatchLink.persist(
+ context=mock_context,
+ project_id=TEST_PROJECT_ID,
+ output_config={"gcs_destination": {"output_uri_prefix": "gs://" +
TEST_OUTPUT_URI_PREFIX}},
+ )
+
+ mock_context["ti"].xcom_push.assert_called_once_with(
+ key="translate_text_batch",
+ value={"project_id": TEST_PROJECT_ID, "output_uri_prefix":
TEST_OUTPUT_URI_PREFIX},
+ )
+
+
+class TestTranslateResultByOutputConfigLink:
+ def test_extract_output_uri_prefix(self):
+ output_config = {"gcs_destination": {"output_uri_prefix": "gs://" +
TEST_OUTPUT_URI_PREFIX}}
+ assert (
+
TranslateResultByOutputConfigLink.extract_output_uri_prefix(output_config)
+ == TEST_OUTPUT_URI_PREFIX
+ )
+
+ def test_persist(self):
+ mock_context = mock.MagicMock()
+ mock_context["ti"] = mock.MagicMock()
+ mock_context["task"] = mock.MagicMock(spec=[])
+
+ TranslateResultByOutputConfigLink.persist(
+ context=mock_context,
+ project_id=TEST_PROJECT_ID,
+ output_config={"gcs_destination": {"output_uri_prefix": "gs://" +
TEST_OUTPUT_URI_PREFIX}},
+ )
+
+ mock_context["ti"].xcom_push.assert_called_once_with(
+ key="translate_results_by_output_config",
+ value={"project_id": TEST_PROJECT_ID, "output_uri_prefix":
TEST_OUTPUT_URI_PREFIX},
+ )
+
+ def test_get_link(self):
+ result = TranslateResultByOutputConfigLink()._format_link(
+ project_id=TEST_PROJECT_ID,
output_uri_prefix=TEST_OUTPUT_URI_PREFIX
+ )
+ assert result == BASE_LINK +
"/storage/browser/test-bucket/test-results/run-1?project=test-project"
