test: add foundry ddb pipeline multithreading test

Author: stevenhsd

tests/test_pipeline/test_foundry_ddb_pipeline.py

@@ -2,13 +2,17 @@
 # pylint: disable=missing-function-docstring
 # pylint: disable=protected-access
 
+from concurrent.futures import Future, ThreadPoolExecutor, as_completed
 from datetime import datetime
 from pathlib import Path
 import shutil
 import tempfile
+from typing import Any
 from uuid import uuid4
 
+from duckdb import DuckDBPyConnection, connect
 import polars as pl
+import pytest
 
 from dve.core_engine.backends.implementations.duckdb.auditing import DDBAuditingManager
 from dve.core_engine.backends.implementations.duckdb.reference_data import DuckDBRefDataLoader
@@ -24,6 +28,28 @@
     movies_test_files
 )
 
+@pytest.fixture(scope="function")
+def prep_multithreading_test():
+    sub_details: dict[str, tuple[DuckDBPyConnection, str, DDBAuditingManager]] = {}
+    for idx in range(1, 10):
+        db = f"dve_{uuid4().hex}"
+        tmp_dir = tempfile.mkdtemp(prefix="ddb_foundry_testing")
+        db_file = Path(tmp_dir, db + ".duckdb")
+        conn = connect(database=db_file, read_only=False)
+        ref_db_file = Path(tmp_dir, "movies_refdata.duckdb").as_posix()
+        conn.sql(f"ATTACH '{ref_db_file}' AS movies_refdata")
+        conn.read_parquet(
+            get_test_file_path("movies/refdata/movies_sequels.parquet").as_posix()
+        ).to_table("movies_refdata.sequels")
+        sub_details[f"submission_{idx}"] = (conn, tmp_dir, DDBAuditingManager(None, None, conn))
+    
+    yield sub_details
+    for con, db_dir, aud in sub_details.values():
+       con.close()
+       shutil.rmtree(db_dir)
+       aud.__exit__(None, None, None)       
+
+
 def test_foundry_runner_validation_fail(planet_test_files, temp_ddb_conn):
     db_file, conn = temp_ddb_conn
     processing_folder = planet_test_files
@@ -205,3 +231,39 @@ def test_foundry_runner_error_at_bi_rules(movies_test_files, temp_ddb_conn):
         assert len(list(fh.iter_prefix(audit_files))) == 2
         assert audit_manager.get_submission_status(sub_id).processing_failed
         assert audit_manager.get_latest_processing_records().select("submission_result").pl().to_dicts()[0]["submission_result"] == "processing_failed"
+
+
+def test_foundry_runner_multithreaded(prep_multithreading_test):
+    # get test files in submitted_files location - movies
+    sub_utils = prep_multithreading_test
+    sub_infos: dict[str, SubmissionInfo] = {}
+    with tempfile.TemporaryDirectory(prefix="ddb_mt_sub") as sub_dir, tempfile.TemporaryDirectory(prefix="ddb_mt_process") as proc_dir:
+        # get test files in submitted_files location
+        movie_file = get_test_file_path("movies").joinpath("good_movies.json")
+        for idx in range(1, 10):
+            dest = Path(sub_dir, f"good_movies_{idx}.json")
+            shutil.copyfile(movie_file, dest)
+            sub_infos[f"submission_{idx}"] = SubmissionInfo(submission_id=f"submission_{idx}",
+                                                            dataset_id="movies",
+                                                            file_name=f"good_movies_{idx}",
+                                                            file_extension="json",
+                                                            submitting_org="TEST",
+                                                            datetime_received=datetime(2025,11,5))
+        with ThreadPoolExecutor() as pool:
+            futures: dict[Future, str] = {}
+            for sub in sub_infos:
+                conn, _, aud = sub_utils.get(sub)
+                target = FoundryDDBPipeline(processed_files_path=proc_dir,
+                                            audit_tables=aud,
+                                            connection=conn,
+                                            rules_path=get_test_file_path("movies/movies_ddb.dischema.json").as_posix(),
+                                            submitted_files_path=sub_dir,).run_pipeline
+                futures[pool.submit(target, (sub_infos.get(sub)))] = sub
+            
+            for future in as_completed(futures):
+                sub = futures[future]
+                output_loc, report_uri, audit_files = future.result()
+                
+                assert audit_files
+                assert output_loc
+                assert report_uri