Validate Cloud Function deploy body after template rendering

providers/google/src/airflow/providers/google/cloud/operators/functions.py

@@ -171,20 +171,18 @@ def __init__(
         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 @@ def extra_links_params(self) -> dict[str, Any]:
         }
 
     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()

providers/google/tests/unit/google/cloud/operators/test_functions.py

@@ -80,10 +80,94 @@ def test_missing_fields(self, mock_hook, body, message):
             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 @@ def test_empty_project_id_is_ok(self, mock_hook):
 
     @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 @@ def test_invalid_source_code_union_field__execute(self, source_code, message):
             ),
         ],
     )
-    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"),

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