Added unit tests for silver layer fanout example

src/onboard_dataflowspec.py

@@ -733,7 +733,7 @@ def __get_silver_dataflow_spec_dataframe(self, onboarding_df, env):
         data = []
 
         onboarding_rows = onboarding_df.collect()
-        mandatory_fields = ["data_flow_id", "data_flow_group", "source_details", f"silver_database_{env}",
+        mandatory_fields = ["data_flow_id", "data_flow_group", f"silver_database_{env}",
                             "silver_table", f"silver_transformation_json_{env}"]  # f"silver_table_path_{env}",
 
         for onboarding_row in onboarding_rows:

tests/resources/onboarding_silverfanout.json

@@ -0,0 +1,36 @@
+[
+    {
+        "data_flow_id": "104",
+        "data_flow_group": "A1",
+        "bronze_database_dev": "bronze",
+        "bronze_database_staging": "bronze",
+        "bronze_database_prd": "bronze",
+        "bronze_table": "customers",
+        "bronze_table_path_dev": "tests/resources/delta/customers",        
+        "silver_database_dev": "silver",
+        "silver_database_staging": "silver",
+        "silver_database_prd": "silver",
+        "silver_table": "customers_clean",
+        "silver_cdc_apply_changes": {
+            "keys": [
+                "id"
+            ],
+            "sequence_by": "operation_date",
+            "scd_type": "1",
+            "apply_as_deletes": "operation = 'DELETE'",
+            "except_column_list": [
+                "operation",
+                "operation_date",
+                "_rescued_data"
+            ]
+        },
+        "silver_table_path_dev": "tests/resources/data/silver/customers",
+        "silver_table_properties": {
+            "pipelines.autoOptimize.managed": "false",
+            "pipelines.reset.allowed": "false",
+            "pipelines.autoOptimize.zOrderCols": "id,email"
+        },
+        "silver_transformation_json_dev": "tests/resources/silver_transformations_fanout.json",
+        "silver_data_quality_expectations_json_dev": "tests/resources/dqe/customers/silver_data_quality_expectations.json"
+    }
+]

tests/resources/silver_transformations_fanout.json

@@ -0,0 +1,19 @@
+[
+  {
+    "target_table": "customers_clean",
+    "select_exp": [
+      "address",
+      "email",
+      "firstname",
+      "id",
+      "lastname",
+      "operation_date",
+      "operation",
+      "_rescued_data"
+    ],
+    "where_clause": [
+      "id IS NOT NULL",
+      "email is not NULL"
+    ]
+  }
+]

tests/test_onboard_dataflowspec.py

@@ -310,3 +310,20 @@ def test_bronze_dataflow_spec_append_flow(self):
         silver_dataflowSpec_df.show(truncate=False)
         self.assertEqual(bronze_dataflowSpec_df.count(), 3)
         self.assertEqual(silver_dataflowSpec_df.count(), 3)
+
+    def test_silver_fanout_dataflow_spec_dataframe(self):
+        """Test for onboardDataflowspec with fanout scenario."""
+        local_params = copy.deepcopy(self.onboarding_bronze_silver_params_map)
+        onboardDataFlowSpecs = OnboardDataflowspec(self.spark, local_params)
+        onboardDataFlowSpecs.onboard_dataflow_specs()
+        local_params["onboarding_file_path"] = self.onboarding_silver_fanout_json_file
+        del local_params["bronze_dataflowspec_table"]
+        del local_params["bronze_dataflowspec_path"]
+        local_params["overwrite"] = "False"
+        onboardDataFlowSpecs = OnboardDataflowspec(self.spark, local_params)
+        onboardDataFlowSpecs.onboard_silver_dataflow_spec()
+        silver_dataflowSpec_df = self.read_dataflowspec(
+            self.onboarding_bronze_silver_params_map['database'],
+            self.onboarding_bronze_silver_params_map['silver_dataflowspec_table'])
+        silver_dataflowSpec_df.show(truncate=False)
+        self.assertEqual(silver_dataflowSpec_df.count(), 4)

tests/utils.py

@@ -39,6 +39,7 @@ def setUp(self):
         self.onboarding_type2_json_file = "tests/resources/onboarding_ac_type2.json"
         self.onboarding_bronze_type2_json_file = "tests/resources/onboarding_ac_bronze_type2.json"
         self.onboarding_append_flow_json_file = "tests/resources/onboarding_append_flow.json"
+        self.onboarding_silver_fanout_json_file = "tests/resources/onboarding_silverfanout.json"
         self.deltaPipelinesMetaStoreOps.drop_database("ravi_dlt_demo")
         self.deltaPipelinesMetaStoreOps.create_database("ravi_dlt_demo", "Unittest")
         self.onboarding_bronze_silver_params_map = {