log rows returned. Tests Pass

Author: nhsdevws

.github/workflows/sonarcloud.yml

@@ -13,7 +13,7 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v4
         with:
           fetch-depth: 0
 
@@ -60,7 +60,7 @@ jobs:
         continue-on-error: true
         run: |
           poetry install
-          poetry run coverage run -m unittest discover -p "*batch*.py" || echo "recordforwarder tests failed" >> ../failed_tests.txt
+          poetry run coverage run -m unittest discover -s "./tests" -p "*batch*.py" || echo "recordforwarder tests failed" >> ../failed_tests.txt
           poetry run coverage xml -o ../recordforwarder-coverage.xml
 
       - name: Run unittest with coverage-ack-lambda

recordprocessor/poetry.lock

@@ -1,34 +1,34 @@
 [tool.poetry]
-name        = "processor"
-version     = "0.1.0"
+name = "filenameprocessor"
+version = "0.1.0"
 description = ""
-authors     = ["Your Name <[email protected]>"]
-readme      = "README.md"
-packages    = [{include = "src"}]
+readme = "README.md"
+authors = ["Your Name <[email protected]>"]
+packages = [{ include = "src" }]
 
 [tool.poetry.dependencies]
-python                  = "~3.11"
-"fhir.resources"        = "~7.0.2"
-boto3                   = "~1.38.42"
-boto3-stubs-lite        = {extras = ["dynamodb"], version = "~1.38.42"}
-aws-lambda-typing       = "~2.20.0"
-moto                    = "^4"
-requests                = "~2.32.4"
-responses               = "~0.25.7"
-pydantic                = "~1.10.13"
-pyjwt                   = "~2.10.1"
-cryptography            = "~42.0.4"
-cffi                    = "~1.17.1"
-jsonpath-ng             = "^1.6.0"
-simplejson              = "^3.20.1"
-structlog               = "^24.1.0"
-pandas                  = "^2.3.0"
-freezegun               = "^1.5.2"
-coverage                = "^7.9.1"
-redis = "^6.2.0"
-numpy = "~2.2.6"
+python = "~3.11"
+"fhir.resources" = "~7.0.2"
+boto3 = "~1.38.42"
+boto3-stubs-lite = { extras = ["dynamodb"], version = "~1.38.42" }
+aws-lambda-typing = "~2.20.0"
+requests = "~2.32.4"
+responses = "~0.25.7"
+pydantic = "~1.10.13"
+pyjwt = "~2.10.1"
+cryptography = "~42.0.4"
+cffi = "~1.17.1"
+jsonpath-ng = "^1.6.0"
+simplejson = "^3.20.1"
+structlog = "^24.1.0"
+redis = "^5.1.1"
+coverage = "^7.9.1"
+freezegun = "^1.5.2"
+fakeredis = "^2.30.1"
 
-[build-system]
-requires      = ["poetry-core ~= 1.5.0"]
+[tool.poetry.group.dev.dependencies]
+moto = {extras = ["s3"], version = "^5.1.12"}
 
-build-backend = "poetry.core.masonry.api"
+[build-system]
+requires = ["poetry-core >= 1.5.0"]
+build-backend = "poetry.core.masonry.api"

recordprocessor/pyproject.toml

@@ -13,10 +13,12 @@
 from typing import Optional
 
 
-def process_csv_to_fhir(incoming_message_body: dict) -> None:
+def process_csv_to_fhir(incoming_message_body: dict) -> int:
     """
     For each row of the csv, attempts to transform into FHIR format, sends a message to kinesis,
     and documents the outcome for each row in the ack file.
+    Returns the number of rows processed. While this is not used by the handler, the number of rows
+    processed must be correct and therefore is returned for logging and test purposes.
     """
     encoder = "utf-8"  # default encoding
     try:
@@ -50,10 +52,10 @@ def process_csv_to_fhir(incoming_message_body: dict) -> None:
     logger.info(f"process with encoder {encoder} from row {row_count+1}")
     row_count, err = process_rows(file_id, vaccine, supplier, file_key, allowed_operations,
                                   created_at_formatted_string, csv_reader, target_disease)
+
     if err:
-        logger.warning(f"Error processing: {err}.")
-        # check if it's a decode error
-        if err.reason == "invalid continuation byte":
+        logger.warning(f"Processing Error: {err}.")
+        if isinstance(err, InvalidEncoding):
             new_encoder = "cp1252"
             logger.info(f"Encode error at row {row_count} with {encoder}. Switch to {new_encoder}")
             encoder = new_encoder
@@ -63,10 +65,9 @@ def process_csv_to_fhir(incoming_message_body: dict) -> None:
             row_count, err = process_rows(file_id, vaccine, supplier, file_key, allowed_operations,
                                           created_at_formatted_string, csv_reader, target_disease, row_count)
         else:
-            logger.error(f"Non-decode error: {err}. Cannot retry. Call someone.")
+            logger.error(f"Row Processing error: {err}")
             raise err
 
-    logger.info("Total rows processed: %s", row_count)
     return row_count
 
 
@@ -108,7 +109,7 @@ def process_rows(file_id, vaccine, supplier, file_key, allowed_operations, creat
         # if error reason is 'invalid continuation byte', then it's a decode error
         logger.error("Error processing row %s: %s", row_count, error)
         if hasattr(error, 'reason') and error.reason == "invalid continuation byte":
-            return total_rows_processed_count, error
+            return total_rows_processed_count, InvalidEncoding("Invalid continuation byte")
         else:
             raise error
     return total_rows_processed_count, None
@@ -118,11 +119,13 @@ def main(event: str) -> None:
     """Process each row of the file"""
     logger.info("task started")
     start = time.time()
+    n_rows_processed = 0
     try:
-        process_csv_to_fhir(incoming_message_body=json.loads(event))
+        n_rows_processed = process_csv_to_fhir(incoming_message_body=json.loads(event))
     except Exception as error:  # pylint: disable=broad-exception-caught
         logger.error("Error processing message: %s", error)
     end = time.time()
+    logger.info("Total rows processed: %s", n_rows_processed)
     logger.info("Total time for completion: %ss", round(end - start, 5))