refactor utility function

Author: Akol125

lambdas/shared/src/common/validator/expression_checker.py

@@ -1,15 +1,7 @@
-import json
-
-
 class FHIRParser:
     def __init__(self):
         self.fhir_resources = {}
 
-    # This opens a file with FHIR resource data
-    def parse_fhir_file(self, fhir_file_name: str) -> None:
-        with open(fhir_file_name) as json_file:
-            self.fhir_resources = json.load(json_file)
-
     # This is used to parse FHIR data events passed in as a dictionary
     def parse_fhir_data(self, fhir_data: dict) -> None:
         self.fhir_resources = fhir_data

lambdas/shared/src/common/validator/parsers/fhir_parser.py

@@ -211,15 +211,15 @@ def run_validation(
         return self.error_records
 
     # Build the error Report
-    def build_error_report(self, event_id):
+    def build_error_report(self, event_id: str) -> dict:
         occurrence_date_time = self.data_parser.get_fhir_value("occurrenceDateTime")
         dq_reporter = DQReporter()
         dq_report = dq_reporter.generate_error_report(event_id, occurrence_date_time, self.error_records)
 
         return dq_report
 
     # Check all errors to see if we have a critical error that would fail the validation
-    def has_validation_failed(self):
+    def has_validation_failed(self) -> bool:
         for error_record in self.error_records:
             if error_record.error_level == ErrorLevels.CRITICAL_ERROR:
                 return True

lambdas/shared/src/common/validator/validator.py

@@ -1,10 +1,10 @@
 # Test application file
-import json
 import time
 import unittest
 from pathlib import Path
 
 from common.validator.validator import Validator
+from tests.test_common.validator.testing_utils.csv_fhir_utils import parse_test_file
 
 # TODO this needs success and fail cases
 
@@ -20,11 +20,9 @@ def test_validation(self):
         start = time.time()
 
         # get the JSON of the schema, changed to cope with elasticache
-        with open(self.schemaFilePath) as JSON:
-            SchemaFile = json.load(JSON)
+        SchemaFile = parse_test_file(self.schemaFilePath)
 
-        with open(self.FHIRFilePath) as json_file:
-            self.fhir_resources = json.load(json_file)
+        self.fhir_resources = parse_test_file(self.FHIRFilePath)
 
         validator = Validator(SchemaFile)  # FHIR File Path not needed
         print(f"FHIR Resources Loaded: {len(self.fhir_resources.get('entry', []))} entries")

lambdas/shared/tests/test_common/validator/test_application_fhir.py

@@ -8,7 +8,7 @@
 
 from common.validator.parsers.csv_line_parser import CSVLineParser
 from test_common.validator.testing_utils.constants import CSV_HEADER, CSV_VALUES
-from test_common.validator.testing_utils.csv_utils import build_row
+from tests.test_common.validator.testing_utils.csv_fhir_utils import build_row
 
 
 class TestCSVLineParser(unittest.TestCase):

lambdas/shared/tests/test_common/validator/test_csv_line_parser.py

@@ -193,7 +193,7 @@ def tearDown(self):
 
     def test_validate_datetime_valid(self):
         result = self.expression_checker.validate_expression(
-            "DATETIME", rule="", field_name="timestamp", field_value="2022-01-01T12:00:00", row={}
+            "DATETIME", expression_rule="", field_name="timestamp", field_value="2022-01-01T12:00:00", row={}
         )
         self.assertEqual(
             result.message,
@@ -204,13 +204,13 @@ def test_validate_datetime_valid(self):
 
     def test_validate_uuid_valid(self):
         result = self.expression_checker.validate_expression(
-            "UUID", rule="", field_name="id", field_value="550e8400-e29b-41d4-a716-446655440000", row={}
+            "UUID", expression_rule="", field_name="id", field_value="550e8400-e29b-41d4-a716-446655440000", row={}
         )
         self.assertTrue(result is None)
 
     def test_validate_integer_invalid(self):
         result = self.expression_checker.validate_expression(
-            "INT", rule="", field_name="age", field_value="hello world", row={}
+            "INT", expression_rule="", field_name="age", field_value="hello world", row={}
         )
         self.assertEqual(result.code, ExceptionLevels.UNEXPECTED_EXCEPTION)
         self.assertEqual(result.field, "age")
@@ -221,14 +221,14 @@ def test_validate_in_array(self):
         self.mock_data_parser.get_key_values.return_value = ["val1", "val2"]
 
         result = self.expression_checker.validate_expression(
-            "INARRAY", rule="", field_name="some_field", field_value="val2", row={}
+            "INARRAY", expression_rule="", field_name="some_field", field_value="val2", row={}
         )
         self.assertEqual(result.message, "Value not in array check failed")
         self.assertEqual(result.field, "some_field")
 
     def test_validate_expression_type_not_found(self):
         result = self.expression_checker.validate_expression(
-            "UNKNOWN", rule="", field_name="field", field_value="value", row={}
+            "UNKNOWN", expression_rule="", field_name="field", field_value="value", row={}
         )
         self.assertIn("Schema expression not found", result)
 

lambdas/shared/tests/test_common/validator/test_expression_checker.py

@@ -3,25 +3,24 @@
 from pathlib import Path
 
 from common.validator.parsers.fhir_parser import FHIRParser
+from tests.test_common.validator.testing_utils.csv_fhir_utils import parse_test_file
 
 
 class TestParse(unittest.TestCase):
     def setUp(self):
         self.fhir_data_folder = Path(__file__).parent / "sample_data"
-
-    def test_parse_fhir_key_exists(self):
         fhirFilePath = self.fhir_data_folder / "vaccination.json"
+        self.fhir_data = parse_test_file(fhirFilePath)
 
+    def test_parse_fhir_key_exists(self):
         fhir_parser = FHIRParser()
-        fhir_parser.parse_fhir_file(fhirFilePath)
+        fhir_parser.parse_fhir_data(self.fhir_data)
         my_value = fhir_parser.get_fhir_value_list("vaccineCode|coding|0|code")
         self.assertEqual(my_value, ["42223111000001107"])
 
     def test_parse_fhir_key_not_exists(self):
-        fhirFilePath = self.fhir_data_folder / "vaccination.json"
-
         fhir_parser = FHIRParser()
-        fhir_parser.parse_fhir_file(fhirFilePath)
+        fhir_parser.parse_fhir_data(self.fhir_data)
         my_value = fhir_parser.get_fhir_value_list("vaccineCode|coding|1")
         self.assertEqual(my_value, [""])
         my_value = fhir_parser.get_fhir_value_list("vaccineCode|coding|1|codes")

lambdas/shared/tests/test_common/validator/test_fhir_parser.py

@@ -1,8 +1,8 @@
-import json
 import unittest
 from pathlib import Path
 
 from common.validator.validator import Validator
+from tests.test_common.validator.testing_utils.csv_fhir_utils import parse_test_file
 
 
 class TestValidator(unittest.TestCase):
@@ -14,8 +14,7 @@ class TestValidator(unittest.TestCase):
     def setUp(self):
         self.parent_folder = Path(__file__).parent
         schema_file_path = self.parent_folder / "test_schemas/test_small_schema.json"
-        with open(schema_file_path) as file:
-            self.schema = json.load(file)
+        self.schema = parse_test_file(schema_file_path)
 
     def test_run_validation_csv_success(self):
         good_file_path = self.parent_folder / "sample_data/valid_csv_data.csv"

lambdas/shared/tests/test_common/validator/test_validation_csv_file.py

@@ -1,11 +1,10 @@
 # Test application file
-import json
 import unittest
 from pathlib import Path
 
 from common.validator.validator import Validator
 from test_common.validator.testing_utils.constants import CSV_HEADER, CSV_VALUES
-from test_common.validator.testing_utils.csv_utils import build_row
+from tests.test_common.validator.testing_utils.csv_fhir_utils import build_row, parse_test_file
 
 schema_data_folder = Path(__file__).parent / "test_schemas"
 schemaFilePath = schema_data_folder / "test_schema.json"
@@ -17,9 +16,7 @@ class TestValidator(unittest.TestCase):
     """
 
     def setUp(self):
-        with open(schemaFilePath, encoding="utf-8") as file:
-            schema_file = json.load(file)
-        self.validator = Validator(schema_file)
+        self.validator = Validator(parse_test_file(schemaFilePath))
 
     def test_run_validation_on_valid_csv_row(self):
         valid_rows = build_row(CSV_HEADER, CSV_VALUES)

lambdas/shared/tests/test_common/validator/test_validation_csv_row.py

@@ -1,10 +1,17 @@
-# Utility function to build CSV row strings
+import json
 
 
+# Utility function to build CSV row strings
 def build_row(header: str, csv_file: dict) -> str:
     """
     Construct a CSV row string from the provided csv_file.
     Any missing header columns get empty string values.
     """
     cols = header.split(",")
     return ",".join(str(csv_file.get(col, "")) for col in cols)
+
+
+# Utility function to parse test (FHIR or schema) files
+def parse_test_file(test_file_name: str) -> dict:
+    with open(test_file_name) as json_file:
+        return json.load(json_file)