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 988578f3bab5e..afe4a29b705a1 100644 --- a/providers/google/src/airflow/providers/google/cloud/operators/functions.py +++ b/providers/google/src/airflow/providers/google/cloud/operators/functions.py @@ -171,12 +171,9 @@ 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._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: @@ -227,6 +224,12 @@ def extra_links_params(self) -> dict[str, Any]: } def execute(self, context: Context): + self.zip_path_preprocessor = ZipPathPreprocessor(self.body, self.zip_path) + if self.validate_body: + self._field_validator = GcpBodyFieldValidator( + CLOUD_FUNCTION_VALIDATION, api_version=self.api_version + ) + self._validate_inputs() hook = CloudFunctionsHook( gcp_conn_id=self.gcp_conn_id, api_version=self.api_version, 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 9a08d695c68b5..51ae2a9769c92 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,25 @@ 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_deploys_after_rendering(self, mock_hook): + 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.create_new_function.return_value = True + op = CloudFunctionDeployFunctionOperator( + project_id=GCP_PROJECT_ID, location=GCP_LOCATION, body="{{ var.value.body }}", task_id="id" + ) + # Template rendering replaces the Jinja expression with the resolved value before execute. + op.body = deepcopy(VALID_BODY) + op.execute(context=mock.MagicMock()) + mock_hook.return_value.create_new_function.assert_called_once() @mock.patch("airflow.providers.google.cloud.operators.functions.CloudFunctionsHook") def test_deploy_execute(self, mock_hook): @@ -154,19 +169,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 +398,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"), diff --git a/scripts/ci/prek/validate_operators_init_exemptions.txt b/scripts/ci/prek/validate_operators_init_exemptions.txt index e6869670a5bea..e949aa4ff9a5c 100644 --- a/scripts/ci/prek/validate_operators_init_exemptions.txt +++ b/scripts/ci/prek/validate_operators_init_exemptions.txt @@ -14,7 +14,6 @@ providers/google/src/airflow/providers/google/cloud/operators/cloud_build.py::Cl 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/dataproc.py::DataprocSubmitJobOperator -providers/google/src/airflow/providers/google/cloud/operators/functions.py::CloudFunctionDeployFunctionOperator providers/google/src/airflow/providers/google/cloud/operators/gcs.py::GCSFileTransformOperator providers/google/src/airflow/providers/google/cloud/sensors/bigquery_dts.py::BigQueryDataTransferServiceTransferRunSensor providers/google/src/airflow/providers/google/cloud/sensors/cloud_composer.py::CloudComposerExternalTaskSensor