test: Add assemble stage coverage and refactor tests to use run-scoped paths

Author: BLMgithub

data_pipeline/stages/assemble_validated_events.py

@@ -20,7 +20,7 @@
 
 
 def init_report():
-    return {"status": "success", "error": [], "info": []}
+    return {"status": "success", "errors": [], "info": []}
 
 
 def log_info(message: str, report: Dict[str, List[str]]) -> None:
@@ -30,7 +30,7 @@ def log_info(message: str, report: Dict[str, List[str]]) -> None:
 
 def log_error(message: str, report: Dict[str, list[str]]) -> None:
     print(f"[ERROR] {message}")
-    report["error"].append(message)
+    report["errors"].append(message)
 
 
 # ------------------------------------------------------------
@@ -153,12 +153,22 @@ def freeze_schema(df: pd.DataFrame) -> pd.DataFrame:
 
     ENFORCED_DTYPES = {
         "order_id": "string",
-        "order_year": "int64",
+        "seller_id": "string",
+        "product_id": "string",
+        "order_status": "string",
+        "order_purchase_timestamp": "datetime64[ns]",
+        "order_approved_at": "datetime64[ns]",
+        "order_delivered_timestamp": "datetime64[ns]",
         "lead_time_days": "int64",
         "approval_lag_days": "int64",
         "delivery_delay_days": "int64",
+        "order_year": "int64",
     }
 
+    missing_cols = set(ENFORCED_SCHEMA) - set(df.columns)
+    if missing_cols:
+        raise RuntimeError(f"missing required columns: {sorted(missing_cols)}")
+
     df_contract = df[ENFORCED_SCHEMA].copy()
     df_contract = df_contract.astype(ENFORCED_DTYPES)
     df_contract = df_contract.sort_values("order_id").reset_index(drop=True)

data_pipeline/tests/stages/test_apply_raw_data_contract.py

@@ -4,7 +4,6 @@
 
 import pandas as pd
 import pytest
-from shutil import copytree
 
 from data_pipeline.shared.run_context import RunContext
 from data_pipeline.stages.apply_raw_data_contract import (
@@ -181,9 +180,6 @@ def test_remove_impossible_timestamps_drops_invalid_rows(invalid_temporal_order_
 
 def test_apply_contract_event_fact_success(tmp_path):
 
-    raw_dir = tmp_path / "raw"
-    raw_dir.mkdir()
-
     df = pd.DataFrame(
         {
             "order_id": [1, 2, 3, 4],
@@ -214,12 +210,10 @@ def test_apply_contract_event_fact_success(tmp_path):
         }
     )
 
-    df.to_csv(raw_dir / "df_orders_2026_01.csv", index=False)
-
     run_context = RunContext.create(base_path=tmp_path)
     run_context.initialize_directories()
 
-    copytree(raw_dir, run_context.raw_snapshot_path, dirs_exist_ok=True)
+    df.to_csv(run_context.raw_snapshot_path / "df_orders_2026_01.csv", index=False)
 
     report, _ = apply_contract(run_context, "df_orders")
 
@@ -246,9 +240,6 @@ def test_apply_contract_unknown_table(tmp_path):
 
 def test_apply_contract_duplicate_on_entity_reference(tmp_path):
 
-    raw_dir = tmp_path / "raw"
-    raw_dir.mkdir()
-
     df = pd.DataFrame(
         {
             "customer_id": [1, 1, 3],  # 1 exact duplicate
@@ -258,12 +249,10 @@ def test_apply_contract_duplicate_on_entity_reference(tmp_path):
         }
     )
 
-    df.to_csv(raw_dir / "df_customers_2026_01.csv", index=False)
-
     run_context = RunContext.create(base_path=tmp_path)
     run_context.initialize_directories()
 
-    copytree(raw_dir, run_context.raw_snapshot_path, dirs_exist_ok=True)
+    df.to_csv(run_context.raw_snapshot_path / "df_customers_2026_01.csv", index=False)
 
     report, _ = apply_contract(run_context, "df_customers")
 
@@ -274,9 +263,6 @@ def test_apply_contract_duplicate_on_entity_reference(tmp_path):
 
 def test_apply_contract_duplicate_on_transactional_detail(tmp_path):
 
-    raw_dir = tmp_path / "raw"
-    raw_dir.mkdir()
-
     df = pd.DataFrame(
         {
             "order_id": [
@@ -292,12 +278,10 @@ def test_apply_contract_duplicate_on_transactional_detail(tmp_path):
         }
     )
 
-    df.to_csv(raw_dir / "df_payments_2026_01.csv", index=False)
-
     run_context = RunContext.create(base_path=tmp_path)
     run_context.initialize_directories()
 
-    copytree(raw_dir, run_context.raw_snapshot_path, dirs_exist_ok=True)
+    df.to_csv(run_context.raw_snapshot_path / "df_payments_2026_01.csv", index=False)
 
     report, _ = apply_contract(run_context, "df_payments")
 
@@ -308,9 +292,6 @@ def test_apply_contract_duplicate_on_transactional_detail(tmp_path):
 
 def test_apply_contract_cascade_drop_with_order_id(tmp_path):
 
-    raw_dir = tmp_path / "raw"
-    raw_dir.mkdir()
-
     df_order = pd.DataFrame(
         {
             "order_id": ["o1", "o2", "o3"],
@@ -355,14 +336,21 @@ def test_apply_contract_cascade_drop_with_order_id(tmp_path):
         }
     )
 
-    df_order.to_csv(raw_dir / "df_orders_2026_01.csv", index=False)
-    df_payments.to_csv(raw_dir / "df_payments_2026_01.csv", index=False)
-    df_order_items.to_csv(raw_dir / "df_order_items_2026_01.csv", index=False)
-
     run_context = RunContext.create(base_path=tmp_path)
     run_context.initialize_directories()
 
-    copytree(raw_dir, run_context.raw_snapshot_path, dirs_exist_ok=True)
+    df_order.to_csv(
+        run_context.raw_snapshot_path / "df_orders_2026_01.csv",
+        index=False,
+    )
+    df_payments.to_csv(
+        run_context.raw_snapshot_path / "df_payments_2026_01.csv",
+        index=False,
+    )
+    df_order_items.to_csv(
+        run_context.raw_snapshot_path / "df_order_items_2026_01.csv",
+        index=False,
+    )
 
     invalid_ids = set()