fix: fixed failing test for foundry pipeline introduced by previous commit

Author: stevenhsd

src/dve/pipeline/foundry_ddb_pipeline.py

@@ -20,17 +20,17 @@ class FoundryDDBPipeline(DDBDVEPipeline):
     def persist_audit_records(self, submission_info: SubmissionInfo) -> URI:
         """Write out key audit relations to parquet for persisting to datasets"""
         write_to = fh.joinuri(self.processed_files_path, submission_info.submission_id, "audit/")
-        if isinstance(_get_implementation(output_uri), LocalFilesystemImplementation):
-            output_uri = fh.file_uri_to_local_path(output_uri)
-            output_uri.parent.mkdir(parents=True, exist_ok=True)
-            output_uri = output_uri.as_posix()
+        if isinstance(_get_implementation(write_to), LocalFilesystemImplementation):
+            write_to = fh.file_uri_to_local_path(write_to)
+            write_to.parent.mkdir(parents=True, exist_ok=True)
+            write_to = write_to.as_posix()
         self.write_parquet(
             self._audit_tables._processing_status.get_relation(),
-            write_to + "processing_status.parquet",
+            fh.joinuri(write_to, "processing_status.parquet"),
         )
         self.write_parquet(
             self._audit_tables._submission_statistics.get_relation(),
-            write_to + "submission_statistics.parquet",
+            fh.joinuri(write_to, "submission_statistics.parquet"),
         )
         return write_to
 

tests/test_pipeline/test_foundry_ddb_pipeline.py

@@ -49,7 +49,7 @@ def test_foundry_runner_validation_fail(planet_test_files, temp_ddb_conn):
         output_loc, report_uri, audit_files = dve_pipeline.run_pipeline(sub_info)
         assert fh.get_resource_exists(report_uri)
         assert not output_loc
-        assert len(list(fh.iter_prefix(audit_files))) == 2
+        assert len(list(fh.iter_prefix(audit_files))) == 3
 
 
 def test_foundry_runner_validation_success(movies_test_files, temp_ddb_conn):
@@ -86,7 +86,7 @@ def test_foundry_runner_validation_success(movies_test_files, temp_ddb_conn):
         output_loc, report_uri, audit_files = dve_pipeline.run_pipeline(sub_info)
         assert fh.get_resource_exists(report_uri)
         assert len(list(fh.iter_prefix(output_loc))) == 2
-        assert len(list(fh.iter_prefix(audit_files))) == 2
+        assert len(list(fh.iter_prefix(audit_files))) == 3
 
 def test_foundry_runner_error(planet_test_files, temp_ddb_conn):
     # using spark reader config - should error in file transformation - check gracefully handled