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 e5eda218ef4 Validate Cloud Function deploy body after template
rendering (#70531)
e5eda218ef4 is described below
commit e5eda218ef4bad62097e058bfbecf3615262b849
Author: Dr Alex Mitre <[email protected]>
AuthorDate: Wed Sep 23 00:06:10 2026 -0600
Validate Cloud Function deploy body after template rendering (#70531)
Co-authored-by: Shahar Epstein <[email protected]>
---
.../providers/google/cloud/operators/functions.py | 19 ++--
.../unit/google/cloud/operators/test_functions.py | 121 ++++++++++++++++++---
.../ci/prek/validate_operators_init_exemptions.txt | 1 -
3 files changed, 116 insertions(+), 25 deletions(-)
diff --git
a/providers/google/src/airflow/providers/google/cloud/operators/functions.py
b/providers/google/src/airflow/providers/google/cloud/operators/functions.py
index 6b29f121c14..bfdc4791168 100644
--- a/providers/google/src/airflow/providers/google/cloud/operators/functions.py
+++ b/providers/google/src/airflow/providers/google/cloud/operators/functions.py
@@ -171,20 +171,18 @@ class
CloudFunctionDeployFunctionOperator(GoogleCloudBaseOperator):
self.gcp_conn_id = gcp_conn_id
self.api_version = api_version
self.zip_path = zip_path
- self.zip_path_preprocessor = ZipPathPreprocessor(body, zip_path)
+ self.validate_body = validate_body
+ self.zip_path_preprocessor: ZipPathPreprocessor | None = None
self._field_validator: GcpBodyFieldValidator | None = None
self.impersonation_chain = impersonation_chain
- if validate_body:
- self._field_validator =
GcpBodyFieldValidator(CLOUD_FUNCTION_VALIDATION, api_version=api_version)
- self._validate_inputs()
super().__init__(**kwargs)
- def _validate_inputs(self) -> None:
+ def _validate_inputs(self, zip_path_preprocessor: ZipPathPreprocessor) ->
None:
if not self.location:
raise AirflowException("The required parameter 'location' is
missing")
if not self.body:
raise AirflowException("The required parameter 'body' is missing")
- self.zip_path_preprocessor.preprocess_body()
+ zip_path_preprocessor.preprocess_body()
def _validate_all_body_fields(self) -> None:
if self._field_validator:
@@ -227,12 +225,19 @@ class
CloudFunctionDeployFunctionOperator(GoogleCloudBaseOperator):
}
def execute(self, context: Context):
+ zip_path_preprocessor = ZipPathPreprocessor(self.body, self.zip_path)
+ self.zip_path_preprocessor = zip_path_preprocessor
+ if self.validate_body:
+ self._field_validator = GcpBodyFieldValidator(
+ CLOUD_FUNCTION_VALIDATION, api_version=self.api_version
+ )
+ self._validate_inputs(zip_path_preprocessor)
hook = CloudFunctionsHook(
gcp_conn_id=self.gcp_conn_id,
api_version=self.api_version,
impersonation_chain=self.impersonation_chain,
)
- if self.zip_path_preprocessor.should_upload_function():
+ if zip_path_preprocessor.should_upload_function():
self.body[GCF_SOURCE_UPLOAD_URL] = self._upload_source_code(hook)
self._validate_all_body_fields()
self._set_airflow_version_label()
diff --git
a/providers/google/tests/unit/google/cloud/operators/test_functions.py
b/providers/google/tests/unit/google/cloud/operators/test_functions.py
index 9a08d695c68..b168721558b 100644
--- a/providers/google/tests/unit/google/cloud/operators/test_functions.py
+++ b/providers/google/tests/unit/google/cloud/operators/test_functions.py
@@ -80,10 +80,94 @@ class TestGcfFunctionDeploy:
op.execute(None)
def test_body_empty(self):
+ op = CloudFunctionDeployFunctionOperator(
+ project_id="test_project_id", location="test_region", body={},
task_id="id"
+ )
with pytest.raises(AirflowException):
- CloudFunctionDeployFunctionOperator(
- project_id="test_project_id", location="test_region", body={},
task_id="id"
- )
+ op.execute(None)
+
+
@mock.patch("airflow.providers.google.cloud.operators.functions.CloudFunctionsHook")
+ def test_templated_body_with_zip_path_uploads_after_rendering(self,
mock_hook):
+ """A templated ``body`` combined with ``zip_path`` must be
preprocessed after rendering.
+
+ Before this rendered-body preprocessing, the pre-render ``body`` was
still a Jinja
+ string when ``ZipPathPreprocessor`` ran (at ``__init__`` time), so its
``x in self.body``
+ membership checks silently evaluated as substring tests and always
returned ``False``.
+ ``upload_function`` was then left unset and defaulted to ``False``, so
the zip was never
+ uploaded and ``sourceUploadUrl`` was never populated -- a silent wrong
result, not a
+ crash. Validating after rendering (in ``execute``) makes the
preprocessor see the
+ rendered dict and actually upload the source.
+ """
+ mock_hook.return_value.get_function.side_effect = mock.Mock(
+ side_effect=HttpError(resp=MOCK_RESP_404, content=b"not found")
+ )
+ mock_hook.return_value.upload_function_zip.return_value =
"https://uploadUrl";
+ mock_hook.return_value.create_new_function.return_value = True
+ body = deepcopy(VALID_BODY)
+ body.pop("sourceArchiveUrl", None)
+ body["sourceUploadUrl"] = None
+ op = CloudFunctionDeployFunctionOperator(
+ project_id=GCP_PROJECT_ID,
+ location=GCP_LOCATION,
+ body="{{ var.value.body }}",
+ zip_path="/path/to/file.zip",
+ validate_body=False,
+ task_id="id",
+ )
+ # Template rendering replaces the Jinja expression with the resolved
value before execute.
+ op.body = body
+ op.execute(context=mock.MagicMock())
+ mock_hook.return_value.upload_function_zip.assert_called_once_with(
+ project_id=GCP_PROJECT_ID, location=GCP_LOCATION,
zip_path="/path/to/file.zip"
+ )
+ assert op.body["sourceUploadUrl"] == "https://uploadUrl";
+
+
@mock.patch("airflow.providers.google.cloud.operators.functions.CloudFunctionsHook")
+ def test_templated_location_rendered_empty_raises(self, mock_hook):
+ """A templated ``location`` must be validated after rendering, not
before.
+
+ Before validating after rendering, the truthiness check in
``_validate_inputs`` ran
+ against the truthy, un-rendered ``"{{ ... }}"`` string at ``__init__``
time and passed,
+ so an empty rendered value was never caught and ``execute`` proceeded
to deploy with an
+ empty location.
+ """
+ op = CloudFunctionDeployFunctionOperator(
+ project_id=GCP_PROJECT_ID,
+ location="{{ var.value.location }}",
+ body=deepcopy(VALID_BODY),
+ task_id="id",
+ )
+ # Template rendering replaces the Jinja expression with the resolved
value before execute.
+ op.location = ""
+ with pytest.raises(AirflowException) as ctx:
+ op.execute(context=mock.MagicMock())
+ assert "The required parameter 'location' is missing" in str(ctx.value)
+ mock_hook.assert_not_called()
+
+
@mock.patch("airflow.providers.google.cloud.operators.functions.CloudFunctionsHook")
+ def test_templated_api_version_validates_after_rendering(self, mock_hook):
+ """A templated ``api_version`` must reach ``GcpBodyFieldValidator``
rendered.
+
+ ``sourceRepositoryUrl`` is gated on ``api_version == "v1beta2"`` in
+ ``CLOUD_FUNCTION_VALIDATION``. Building the validator in ``__init__``
pinned it to the
+ un-rendered ``"{{ ... }}"`` string, which matches no gated spec, so
every
+ version-specific field was silently skipped and an invalid value
passed validation.
+ """
+ body = deepcopy(VALID_BODY)
+ body.pop("sourceArchiveUrl", None)
+ body["sourceRepositoryUrl"] = ""
+ op = CloudFunctionDeployFunctionOperator(
+ project_id=GCP_PROJECT_ID,
+ location=GCP_LOCATION,
+ body=body,
+ api_version="{{ var.value.api_version }}",
+ task_id="id",
+ )
+ # Template rendering replaces the Jinja expression with the resolved
value before execute.
+ op.api_version = "v1beta2"
+ with pytest.raises(AirflowException, match="sourceRepositoryUrl"):
+ op.execute(context=mock.MagicMock())
+ mock_hook.return_value.create_new_function.assert_not_called()
@mock.patch("airflow.providers.google.cloud.operators.functions.CloudFunctionsHook")
def test_deploy_execute(self, mock_hook):
@@ -154,19 +238,21 @@ class TestGcfFunctionDeploy:
@mock.patch("airflow.providers.google.cloud.operators.functions.CloudFunctionsHook")
def test_empty_location(self, mock_hook):
+ op = CloudFunctionDeployFunctionOperator(
+ project_id="test_project_id", location="", body=None, task_id="id"
+ )
with pytest.raises(AirflowException) as ctx:
- CloudFunctionDeployFunctionOperator(
- project_id="test_project_id", location="", body=None,
task_id="id"
- )
+ op.execute(None)
err = ctx.value
assert "The required parameter 'location' is missing" in str(err)
@mock.patch("airflow.providers.google.cloud.operators.functions.CloudFunctionsHook")
def test_empty_body(self, mock_hook):
+ op = CloudFunctionDeployFunctionOperator(
+ project_id="test_project_id", location="test_region", body=None,
task_id="id"
+ )
with pytest.raises(AirflowException) as ctx:
- CloudFunctionDeployFunctionOperator(
- project_id="test_project_id", location="test_region",
body=None, task_id="id"
- )
+ op.execute(None)
err = ctx.value
assert "The required parameter 'body' is missing" in str(err)
@@ -381,20 +467,21 @@ class TestGcfFunctionDeploy:
),
],
)
- def test_invalid_source_code_union_field__init(self, source_code, message):
+ def test_invalid_source_code_union_field__preprocess(self, source_code,
message):
body = deepcopy(VALID_BODY)
body.pop("sourceUploadUrl", None)
body.pop("sourceArchiveUrl", None)
zip_path = source_code.pop("zip_path", None)
body.update(source_code)
+ op = CloudFunctionDeployFunctionOperator(
+ project_id="test_project_id",
+ location="test_region",
+ body=body,
+ task_id="id",
+ zip_path=zip_path,
+ )
with pytest.raises(AirflowException, match=message):
- CloudFunctionDeployFunctionOperator(
- project_id="test_project_id",
- location="test_region",
- body=body,
- task_id="id",
- zip_path=zip_path,
- )
+ op.execute(None)
@pytest.mark.parametrize(
("source_code", "project_id"),
diff --git a/scripts/ci/prek/validate_operators_init_exemptions.txt
b/scripts/ci/prek/validate_operators_init_exemptions.txt
index 675813b6031..afdc7deba3b 100644
--- a/scripts/ci/prek/validate_operators_init_exemptions.txt
+++ b/scripts/ci/prek/validate_operators_init_exemptions.txt
@@ -10,4 +10,3 @@
providers/amazon/src/airflow/providers/amazon/aws/operators/neptune.py::NeptuneS
providers/google/src/airflow/providers/google/cloud/operators/cloud_build.py::CloudBuildCreateBuildOperator
providers/google/src/airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py::CloudDataTransferServiceCreateJobOperator
providers/google/src/airflow/providers/google/cloud/operators/dataproc.py::DataprocCreateClusterOperator
-providers/google/src/airflow/providers/google/cloud/operators/functions.py::CloudFunctionDeployFunctionOperator
