refactor: deserializers to follow dry

src/azure_data_factory_testing_framework/deserializers/_deserializer_base.py

@@ -0,0 +1,21 @@
+from azure_data_factory_testing_framework.deserializers.shared._activity_deserializer import (
+    _get_activity_from_activity_data,
+)
+from azure_data_factory_testing_framework.deserializers.shared._data_factory_element_replacer import (
+    _find_and_replace_expressions_in_dict,
+)
+from azure_data_factory_testing_framework.models.pipeline import Pipeline
+
+
+def _parse_pipeline_from_json(name: str, json_data: dict) -> Pipeline:
+    properties = json_data.get("properties", {})
+    activities = properties.get("activities", [])
+
+    for activity_data in activities:
+        activities[activities.index(activity_data)] = _get_activity_from_activity_data(activity_data)
+
+    pipeline = Pipeline(name, **properties)
+
+    _find_and_replace_expressions_in_dict(pipeline)
+
+    return pipeline

src/azure_data_factory_testing_framework/deserializers/_deserializer_data_factory.py

@@ -1,26 +1,10 @@
 import json
 
-from azure_data_factory_testing_framework.deserializers.shared._activity_deserializer import (
-    _get_activity_from_activity_data,
-)
-from azure_data_factory_testing_framework.deserializers.shared._data_factory_element_replacer import (
-    _find_and_replace_expressions_in_dict,
-)
+from azure_data_factory_testing_framework.deserializers._deserializer_base import _parse_pipeline_from_json
 from azure_data_factory_testing_framework.models.pipeline import Pipeline
 
 
-def parse_pipeline_from_json(json_str: str) -> Pipeline:
-    json_data = json.loads(json_str)
-
+def parse_data_factory_pipeline_from_pipeline_json(pipeline_json: str) -> Pipeline:
+    json_data = json.loads(pipeline_json)
     name = json_data["name"]
-    properties = json_data.get("properties", {})
-    activities = properties.get("activities", [])
-
-    for activity_data in activities:
-        activities[activities.index(activity_data)] = _get_activity_from_activity_data(activity_data)
-
-    pipeline = Pipeline(name, **properties)
-
-    _find_and_replace_expressions_in_dict(pipeline)
-
-    return pipeline
+    return _parse_pipeline_from_json(name, json_data)

src/azure_data_factory_testing_framework/deserializers/_deserializer_fabric.py

@@ -1,26 +1,15 @@
 import json
 
-from azure_data_factory_testing_framework.deserializers.shared._activity_deserializer import (
-    _get_activity_from_activity_data,
-)
-from azure_data_factory_testing_framework.deserializers.shared._data_factory_element_replacer import (
-    _find_and_replace_expressions_in_dict,
-)
+from azure_data_factory_testing_framework.deserializers._deserializer_base import _parse_pipeline_from_json
 from azure_data_factory_testing_framework.models.pipeline import Pipeline
 
 
-def parse_pipeline_from_json(metadata_json: str, pipeline_json: str) -> Pipeline:
-    metadata_json = json.loads(metadata_json)
+def parse_fabric_pipeline_from_pipeline_json_files(metadata_json: str, pipeline_json: str) -> Pipeline:
+    pipeline_name = _get_pipeline_name_from_metadata_json(metadata_json)
     pipeline_json = json.loads(pipeline_json)
+    return _parse_pipeline_from_json(pipeline_name, pipeline_json)
 
-    properties = pipeline_json.get("properties", {})
-    activities = properties.get("activities", [])
-
-    for activity_data in activities:
-        activities[activities.index(activity_data)] = _get_activity_from_activity_data(activity_data)
-
-    pipeline_json = Pipeline(metadata_json["displayName"], **properties)
 
-    _find_and_replace_expressions_in_dict(pipeline_json)
-
-    return pipeline_json
+def _get_pipeline_name_from_metadata_json(metadata_json: str) -> str:
+    metadata_json = json.loads(metadata_json)
+    return metadata_json["displayName"]

src/azure_data_factory_testing_framework/repositories/data_factory_repository_factory.py

@@ -1,6 +1,8 @@
 import os
 
-from azure_data_factory_testing_framework.deserializers._deserializer_data_factory import parse_pipeline_from_json
+from azure_data_factory_testing_framework.deserializers._deserializer_data_factory import (
+    parse_data_factory_pipeline_from_pipeline_json,
+)
 from azure_data_factory_testing_framework.models.pipeline import Pipeline
 from azure_data_factory_testing_framework.repositories.base_repository_factory import BaseRepositoryFactory
 
@@ -14,6 +16,6 @@ def _get_data_factory_pipelines_by_folder_path(self, folder_path: str) -> list[P
             file_path = os.path.join(pipeline_path, file)
             if file.endswith(".json"):
                 with open(file_path, "r") as f:
-                    pipelines.append(parse_pipeline_from_json(f.read()))
+                    pipelines.append(parse_data_factory_pipeline_from_pipeline_json(f.read()))
 
         return pipelines

src/azure_data_factory_testing_framework/repositories/fabric_repository_factory.py

@@ -1,7 +1,9 @@
 import os
 from typing import List
 
-from azure_data_factory_testing_framework.deserializers._deserializer_fabric import parse_pipeline_from_json
+from azure_data_factory_testing_framework.deserializers._deserializer_fabric import (
+    parse_fabric_pipeline_from_pipeline_json_files,
+)
 from azure_data_factory_testing_framework.models.pipeline import Pipeline
 from azure_data_factory_testing_framework.repositories.base_repository_factory import BaseRepositoryFactory
 
@@ -14,7 +16,9 @@ def _get_data_factory_pipelines_by_folder_path(self, folder_path: str) -> list[P
             pipeline_file = os.path.join(pipeline_folder, "pipeline-content.json")
             pipeline_metadata_file = os.path.join(pipeline_folder, "item.metadata.json")
             with open(pipeline_file, "r") as pipeline_file, open(pipeline_metadata_file, "r") as pipeline_metadata_file:
-                pipelines.append(parse_pipeline_from_json(pipeline_metadata_file.read(), pipeline_file.read()))
+                pipelines.append(
+                    parse_fabric_pipeline_from_pipeline_json_files(pipeline_metadata_file.read(), pipeline_file.read())
+                )
 
         return pipelines