MERGE FIX

Author: bubriks

Diff for: python/tests/engine/test_spark.py

@@ -969,137 +969,6 @@ def test_save_stream_dataframe(self, mocker, backend_fixtures):
             == 0
         )
 
-    def test_save_stream_dataframe_transformations(self, mocker, backend_fixtures):
-        # Arrange
-        mock_common_client_get_instance = mocker.patch(
-            "hopsworks_common.client.get_instance"
-        )
-        mocker.patch("hopsworks_common.client._is_external", return_value=False)
-        mock_spark_engine_serialize_to_avro = mocker.patch(
-            "hsfs.engine.spark.Engine._serialize_to_avro"
-        )
-
-        mock_engine_get_instance = mocker.patch("hsfs.engine.get_instance")
-        mock_engine_get_instance.return_value.add_file.return_value = (
-            "result_from_add_file"
-        )
-
-        mock_storage_connector_api = mocker.patch(
-            "hsfs.core.storage_connector_api.StorageConnectorApi"
-        )
-
-        mock_spark_engine_apply_transformations = mocker.patch(
-            "hsfs.engine.spark.Engine._apply_transformation_function"
-        )
-
-        json = backend_fixtures["storage_connector"]["get_kafka_external"]["response"]
-        sc = storage_connector.StorageConnector.from_response_json(json)
-        mock_storage_connector_api.return_value.get_kafka_connector.return_value = sc
-
-        spark_engine = spark.Engine()
-
-        @udf(int)
-        def test(feature):
-            return feature + 1
-
-        fg = feature_group.FeatureGroup(
-            name="test",
-            version=1,
-            featurestore_id=99,
-            primary_key=[],
-            partition_key=[],
-            id=10,
-            online_topic_name="test_online_topic_name",
-            transformation_functions=[test],
-        )
-        fg.feature_store = mocker.Mock()
-        project_id = 1
-        fg.feature_store.project_id = project_id
-
-        mock_common_client_get_instance.return_value._project_name = "test_project_name"
-
-        # Act
-        spark_engine.save_stream_dataframe(
-            feature_group=fg,
-            dataframe=None,
-            query_name=None,
-            output_mode="test_mode",
-            await_termination=None,
-            timeout=None,
-            checkpoint_dir=None,
-            write_options={"test_name": "test_value"},
-        )
-
-        # Assert
-        assert (
-            mock_spark_engine_serialize_to_avro.return_value.withColumn.call_args[0][0]
-            == "headers"
-        )
-        assert (
-            mock_spark_engine_serialize_to_avro.return_value.withColumn.return_value.writeStream.outputMode.call_args[
-                0
-            ][0]
-            == "test_mode"
-        )
-        assert (
-            mock_spark_engine_serialize_to_avro.return_value.withColumn.return_value.writeStream.outputMode.return_value.format.call_args[
-                0
-            ][0]
-            == "kafka"
-        )
-        assert (
-            mock_spark_engine_serialize_to_avro.return_value.withColumn.return_value.writeStream.outputMode.return_value.format.return_value.option.call_args[
-                0
-            ][0]
-            == "checkpointLocation"
-        )
-        assert (
-            mock_spark_engine_serialize_to_avro.return_value.withColumn.return_value.writeStream.outputMode.return_value.format.return_value.option.call_args[
-                0
-            ][1]
-            == f"/Projects/test_project_name/Resources/{self._get_spark_query_name(project_id, fg)}-checkpoint"
-        )
-        assert (
-            mock_spark_engine_serialize_to_avro.return_value.withColumn.return_value.writeStream.outputMode.return_value.format.return_value.option.return_value.options.call_args[
-                1
-            ]
-            == {
-                "kafka.bootstrap.servers": "test_bootstrap_servers",
-                "kafka.security.protocol": "test_security_protocol",
-                "kafka.ssl.endpoint.identification.algorithm": "test_ssl_endpoint_identification_algorithm",
-                "kafka.ssl.key.password": "test_ssl_key_password",
-                "kafka.ssl.keystore.location": "result_from_add_file",
-                "kafka.ssl.keystore.password": "test_ssl_keystore_password",
-                "kafka.ssl.truststore.location": "result_from_add_file",
-                "kafka.ssl.truststore.password": "test_ssl_truststore_password",
-                "kafka.test_option_name": "test_option_value",
-                "test_name": "test_value",
-            }
-        )
-        assert (
-            mock_spark_engine_serialize_to_avro.return_value.withColumn.return_value.writeStream.outputMode.return_value.format.return_value.option.return_value.options.return_value.option.call_args[
-                0
-            ][0]
-            == "topic"
-        )
-        assert (
-            mock_spark_engine_serialize_to_avro.return_value.withColumn.return_value.writeStream.outputMode.return_value.format.return_value.option.return_value.options.return_value.option.call_args[
-                0
-            ][1]
-            == "test_online_topic_name"
-        )
-        assert (
-            mock_spark_engine_serialize_to_avro.return_value.withColumn.return_value.writeStream.outputMode.return_value.format.return_value.option.return_value.options.return_value.option.return_value.queryName.call_args[
-                0
-            ][0]
-            == self._get_spark_query_name(project_id, fg)
-        )
-        assert (
-            mock_spark_engine_serialize_to_avro.return_value.withColumn.return_value.writeStream.outputMode.return_value.format.return_value.option.return_value.options.return_value.option.return_value.queryName.return_value.start.return_value.awaitTermination.call_count
-            == 0
-        )
-        assert mock_spark_engine_apply_transformations.call_count == 1
-
     def test_save_stream_dataframe_query_name(self, mocker, backend_fixtures):
         # Arrange
         mock_common_client_get_instance = mocker.patch(