fixed unit tests

Author: ravi-databricks

tests/test_dataflow_pipeline.py

@@ -4,8 +4,6 @@
 import sys
 import tempfile
 import copy
-
-
 from pyspark.sql.functions import lit, expr
 import pyspark.sql.types as T
 from pyspark.sql import DataFrame
@@ -20,11 +18,6 @@
 dlt = MagicMock()
 dlt.expect_all_or_drop = MagicMock()
 dlt.apply_changes_from_snapshot = MagicMock()
-
-
-
-
-
 raw_delta_table_stream = MagicMock()
 
 
@@ -198,12 +191,6 @@ def test_invoke_dlt_pipeline_silver_positive(self, run_dlt):
             f"{database}.{silver_dataflow_table}",
         )
         self.spark.sql("CREATE DATABASE IF NOT EXISTS bronze")
-
-
-
-
-
-
         options = {"rescuedDataColumn": "_rescued_data", "inferColumnTypes": "true", "multiline": True}
         customers_parquet_df = self.spark.read.options(**options).json("tests/resources/data/customers")
         (customers_parquet_df.withColumn("_rescued_data", lit("Test")).write.format("delta")
@@ -229,7 +216,6 @@ def test_run_dlt_pipeline_silver_positive(self, read):
         silver_spec_map.update(source_details)
         silver_dataflow_spec = SilverDataflowSpec(**silver_spec_map)
         self.spark.sql("CREATE DATABASE IF NOT EXISTS bronze")
-
         options = {"rescuedDataColumn": "_rescued_data", "inferColumnTypes": "true", "multiline": True}
         customers_parquet_df = self.spark.read.options(**options).json("tests/resources/data/customers")
         (customers_parquet_df.withColumn("_rescued_data", lit("Test")).write.format("delta")
@@ -291,9 +277,6 @@ def test_get_silver_schema_positive(self):
         silver_spec_map.update(source_details)
         silver_dataflow_spec = SilverDataflowSpec(**silver_spec_map)
         self.spark.sql("CREATE DATABASE IF NOT EXISTS bronze")
-
-
-
         options = {"rescuedDataColumn": "_rescued_data", "inferColumnTypes": "true", "multiline": True}
         customers_parquet_df = self.spark.read.options(**options).json("tests/resources/data/customers")
         (customers_parquet_df.withColumn("_rescued_data", lit("Test")).write.format("delta")
@@ -318,9 +301,6 @@ def test_get_silver_schema_where_clause(self):
         silver_dataflow_spec = SilverDataflowSpec(**silver_spec_map)
 
         self.spark.sql("CREATE DATABASE IF NOT EXISTS bronze")
-
-
-
         options = {"rescuedDataColumn": "_rescued_data", "inferColumnTypes": "true", "multiline": True}
         customers_parquet_df = self.spark.read.options(**options).json("tests/resources/data/customers")
         (customers_parquet_df.withColumn("_rescued_data", lit("Test")).write.format("delta")
@@ -354,9 +334,6 @@ def test_read_silver_positive(self):
         }
         silver_spec_map.update(source_details)
         self.spark.sql("CREATE DATABASE IF NOT EXISTS bronze")
-
-
-
         options = {"rescuedDataColumn": "_rescued_data", "inferColumnTypes": "true", "multiline": True}
         customers_parquet_df = self.spark.read.options(**options).json("tests/resources/data/customers")
         (customers_parquet_df.withColumn("_rescued_data", lit("Test")).write.format("delta")
@@ -400,9 +377,6 @@ def test_read_silver_with_where(self, get_silver_schema):
         }
         silver_spec_map.update(source_details)
         self.spark.sql("CREATE DATABASE IF NOT EXISTS bronze")
-
-
-
         options = {"rescuedDataColumn": "_rescued_data", "inferColumnTypes": "true", "multiline": True}
         customers_parquet_df = self.spark.read.options(**options).json("tests/resources/data/customers")
         (customers_parquet_df.withColumn("_rescued_data", lit("Test")).write.format("delta")
@@ -464,7 +438,6 @@ def test_cdc_apply_changes_scd_type2(self, cdc_apply_changes):
         silver_dataflow_spec = SilverDataflowSpec(**silver_spec_map)
         silver_dataflow_spec.cdcApplyChanges = json.dumps(self.silver_cdc_apply_changes_scd2)
         self.spark.sql("CREATE DATABASE IF NOT EXISTS bronze")
-
         options = {"rescuedDataColumn": "_rescued_data", "inferColumnTypes": "true", "multiline": True}
         customers_parquet_df = self.spark.read.options(**options).json("tests/resources/data/customers")
         (customers_parquet_df.withColumn("_rescued_data", lit("Test")).write.format("delta")
@@ -1338,4 +1311,3 @@ def test_get_silver_schema_uc_disabled(self, mock_read_stream):
         #     format="delta"
         # )
         # mock_read_stream.load.return_value.selectExpr.assert_called_once_with(*silver_dataflow_spec.selectExp)
-