# Dvp Test Runner

> Execute generated pytest suites for source and migrated workloads. Fixes runability issues, ensures source tests pass, ensures migrated tests execute. Triggers: run tests, execute tests, validate migration, test runner.

- Skill: `snowflake-labs/dvp-test-runner` (Agent Skill)
- Install (CLI): `npx skillmds@latest add snowflake-labs/dvp-test-runner`
- Raw SKILL.md: https://api.skillmd.com/api/skills/snowflake-labs/dvp-test-runner/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Integrations & APIs
- Author: snowflake-labs (https://skillmd.com/u/snowflake-labs)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/snowflake-labs/dvp-test-runner

---


# DVP Test Runner

## Overview

Execute the pytest suites generated by `dvp-test-setup-generator`. This skill verifies that the test environment is ready, **creates the environment if needed**, runs source and migrated test suites, **actively fixes ALL issues** that prevent tests from running and passing, records results in `sma_storage.sqlite3`, and commits outcomes to git.

## Acceptance Criteria (SUCCESS DEFINITION)

**The skill is considered successful ONLY when ALL of the following are met:**

1. **ALL source tests MUST run AND pass (green).** Every single source test class must be collected by pytest and every test method within it must pass. Zero failures, zero errors, zero skips (unless a skip is explicitly justified). If any source test fails, the skill MUST diagnose and fix the root cause and re-run. The skill MUST NOT stop or declare success until all source tests pass. **Exception — truly unfixable tests:** If after exhausting all fix attempts (5 per-test attempts) a specific test CANNOT be fixed by the agent (e.g., the source code has a genuine bug, requires external services not available, or depends on infrastructure the agent cannot provision), the agent MUST: (a) add a `pytest.mark.skip(reason="...")` decorator with a detailed reason explaining WHY the test cannot pass, (b) add a code comment above the test class documenting the root cause, what was attempted, and why it's unfixable, (c) report the skipped test clearly in the final summary. This is a LAST RESORT — the agent must genuinely exhaust all alternatives before skipping.
2. **ALL migrated tests MUST run.** Every migrated test class must be collected and executed by pytest. Assertion failures (FAILED status) are expected and acceptable — they indicate migration differences. However, collection errors (ERROR during collection), setup errors (ERROR during setup/fixture), and import crashes are NOT acceptable — they must be fixed. The skill MUST NOT stop until all migrated tests at minimum execute (even if they fail assertions).
3. **The skill MUST iterate** over source tests and migrated tests, fixing issues in each cycle, until the acceptance criteria above are met. There is **no global cycle limit** — the skill keeps iterating as long as it is making progress (i.e., each cycle fixes at least one new issue or produces a different error). The limit is **per-test**: each individual test gets at most **5 fix attempts**. If a test has been fixed/retried 5 times and still fails, apply the unfixable-test policy (Step 6.5). If ALL failing tests have hit their per-test limit, mark as `Partial` but NEVER stop early — always attempt both suites.
4. **The skill MUST consider previous commits.** When applying fixes, the skill must check prior commits on `sma/migration-process` (from upstream DVP skills) to understand what has already been changed. If a fix modifies an artifact produced by an upstream skill (e.g., `data_io_schema.json`, `entrypoints.json`, synthetic data CSVs, or adapted source/migrated code), the skill must re-trigger the relevant downstream verification to ensure consistency. For example: if synthetic data columns are regenerated to fix a source test, the migrated test setup must be re-verified against the new schema. If adapted entrypoint signatures change, test files that call those entrypoints must be regenerated or updated. The skill treats each committed fix as a version — subsequent iterations validate against the latest committed state, not the original.

**The skill MUST NOT stop if:**
- Source tests don't initially run or pass — fix them and retry
- Migrated tests don't initially run — fix them and retry
- Individual fixes fail — try alternative approaches
- One suite has issues — the other suite must still be attempted

**Goals (in order of priority):**
1. **Source tests must all pass (green).** If they fail, diagnose and fix the root cause (missing stubs, synthetic data quality, schema mismatches, missing dependencies, path issues, env var pollution, module caching, pipeline dependencies) and re-run until all pass.
2. **Migrated tests must execute.** They will likely fail with assertion errors — that is expected and acceptable. The goal is that pytest **collects and runs** every migrated test without crashing on import errors, missing modules, or infrastructure setup failures.

**Execution order:**
1. Source tests run first — they produce baseline CSVs in `dvp/03-tests/data/expected_output/`
2. Migrated tests run second — they compare Snowflake results against those baselines
3. Both suites always run. Source failures trigger a fix-and-retry loop. Migrated failures are reported but do not block.

**Critical constraint:** Any changes made to synthetic data, schemas, or shared fixtures to fix source tests MUST also apply to migrated tests. Both suites use the same `dvp/04-results/synthetic_data/` and `dvp/04-results/data_io_schema.json`.

## Preconditions

- DVP workspace exists (`dvp/01-source/`, `dvp/03-tests/`, `dvp/04-results/`).
- `dvp-test-setup-generator` has run: test files exist under `dvp/03-tests/source/` and `dvp/03-tests/migrated/` (or `migrated_scos/`).
- `dvp/04-results/entrypoints.json` exists with detected entrypoints.

## Inputs

| Input | Required | Location |
|-------|----------|----------|
| Test project | Yes | `dvp/03-tests/` |
| Source test files | Yes | `dvp/03-tests/source/test_*.py` |
| Migrated test files | Yes | `dvp/03-tests/migrated/test_*.py` or `dvp/03-tests/migrated_scos/test_*.py` |
| Requirements | Yes | `dvp/03-tests/requirements.txt` |
| Config | Yes | `dvp/03-tests/config.py` |
| Entrypoints | Yes | `dvp/04-results/entrypoints.json` |

## Outputs

| Output | Format | Location |
|--------|--------|----------|
| Source test results | pytest stdout | Console + `sma_storage.sqlite3` |
| Migrated test results | pytest stdout | Console + `sma_storage.sqlite3` |
| Baseline CSVs | CSV files | `dvp/03-tests/data/expected_output/` (produced by source tests) |
| Test results export | CSV | `dvp/04-results/testing-results/` (via `sma_api.export_test_results`) |

## Output Format

Every time you begin a step, sub-step, or significant action, prefix the message with a timestamp in the format `[YYYY-MM-DD HH:MM:SS]`. Obtain the current time by running `date '+%Y-%m-%d %H:%M:%S'` in bash.

Example:
```
[2026-03-24 14:05:32] Starting Step 1: Verify test project exists...
[2026-03-24 14:05:45] Found 3 source test files, 3 migrated test files
[2026-03-24 14:05:46] Step 1 complete.
```

## Procedure

### Step 0: Initialize Git

Ensure the workload directory has a git repository on the `sma/migration-process` branch. This is idempotent.

```python
result = sma_api.git_ensure_ready("<workload_path>")
```

### Step 1: Verify Test Project Exists

Check that `dvp/03-tests/` contains the expected structure:

1. Verify `dvp/03-tests/conftest.py` exists
2. Verify `dvp/03-tests/config.py` exists
3. Verify `dvp/03-tests/requirements.txt` exists
4. Verify `dvp/03-tests/pytest.ini` exists
5. Count source test files: `dvp/03-tests/source/test_*.py` (must be >= 1)
6. Detect migrated flavor:
   - If `dvp/03-tests/migrated/` exists with `test_*.py` files → `migrated`
   - If `dvp/03-tests/migrated_scos/` exists with `test_*.py` files → `migrated_scos`
7. Verify `dvp/03-tests/source/conftest.py` and `dvp/03-tests/<migrated_flavor>/conftest.py` exist

**Stopping point:** If any required file is missing, stop and report:
```
Test project incomplete. Missing:
  - dvp/03-tests/conftest.py
  - dvp/03-tests/source/test_*.py (no test files found)

Run dvp-test-setup-generator first to generate the test project.
```

### Step 2: Create and Verify Python Environment

The skill MUST ensure a working Python environment exists. **Create it from scratch if needed** — do not assume it pre-exists.

1. **Check for existing venv** at `dvp/.venv/`:
   ```bash
   ls <workload_path>/dvp/.venv/bin/python 2>/dev/null
   ```

2. **If venv exists**, run a preflight check — also verify PySpark is 3.5.x:
   ```bash
   <workload_path>/dvp/.venv/bin/python -c "import pyspark; import pytest; import pytest_subtests; import snowflake.snowpark; assert pyspark.__version__.startswith('3.5'), f'PySpark 3.5.x required, got {pyspark.__version__}'; print('OK')"
   ```

3. **If venv does not exist or preflight fails**, CREATE IT:
   ```bash
   cd <workload_path>/dvp && uv venv .venv && uv pip install -r <workload_path>/dvp/03-tests/requirements.txt "pyspark~=3.5.0"
   ```
   The `pyspark~=3.5.0` override ensures PySpark 3.5.x even if `requirements.txt` is unpinned or pinned to 4.x.

4. **If `uv` is not available**, fall back to standard venv:
   ```bash
   cd <workload_path>/dvp && python3 -m venv .venv && .venv/bin/pip install -r <workload_path>/dvp/03-tests/requirements.txt "pyspark~=3.5.0"
   ```

5. **If pip install fails for specific packages** (e.g., `jpype1` build failure due to missing CMake/ANT):
   - Upgrade pip first: `.venv/bin/pip install --upgrade pip`
   - Install packages individually, skipping ones that fail to build (e.g., `snowpark-connect` depends on `jpype1`):
     ```bash
     .venv/bin/pip install "pyspark~=3.5.0" pytest pytest-subtests snowflake-snowpark-python
     ```
   - The goal is to have the core packages installed. Optional packages that fail can be skipped.

6. **Verify the final environment** with the preflight command. If it still fails, stop and report:
   ```
   Environment setup failed. Missing packages:
     - snowflake-snowpark-python (required for migrated tests)

   Install manually:
     cd <workload_path>/dvp && .venv/bin/pip install -r 03-tests/requirements.txt
   ```

Store the path to the Python binary for all subsequent commands:
```
PYTHON_BIN=<workload_path>/dvp/.venv/bin/python
```

### Step 3: Verify Java (Source Tests Requirement)

Source tests use PySpark, which requires a compatible Java runtime (17–21).

1. Check Java availability and version:
   ```bash
   java -version 2>&1
   ```

2. If Java is **present**, extract the major version and verify it is 17–21:
   ```bash
   java -version 2>&1 | head -1 | grep -oE '"[0-9]+' | tr -d '"'
   ```
   - Major version 17–21 → **Java OK**. Proceed.
   - Version < 17 or > 21 → treat as incompatible (see step 3 below).

3. If Java is **missing or incompatible**, attempt to set `JAVA_HOME` from an existing installation before asking the user to install:
   ```bash
   # macOS: probe java_home helper
   for v in 17 21 18 19 20; do
     candidate=$(/usr/libexec/java_home -v $v 2>/dev/null)
     if [ -n "$candidate" ]; then
       export JAVA_HOME="$candidate"
       echo "JAVA_HOME set to $JAVA_HOME"
       break
     fi
   done

   # Linux: probe SDKMAN / common paths
   for candidate in "$HOME/.sdkman/candidates/java/current" /usr/lib/jvm/java-17-openjdk-amd64 /usr/lib/jvm/temurin-17; do
     if [ -d "$candidate" ]; then
       export JAVA_HOME="$candidate"
       echo "JAVA_HOME set to $JAVA_HOME"
       break
     fi
   done
   ```
   Re-run `java -version` after setting `JAVA_HOME`. If Java is now found and compatible, proceed.

4. If Java is still not available after the auto-detection above, inform the user **and stop**:
   ```
   Source tests require a compatible Java runtime (17–21). Java is not found or JAVA_HOME is not set.

   Install a compatible JDK:
     macOS:   brew install --cask temurin@17
     Ubuntu:  sudo apt install openjdk-17-jdk
     Windows: Download from https://adoptium.net/

   Then set JAVA_HOME before running this skill:
     macOS:   export JAVA_HOME=$(/usr/libexec/java_home -v 17)
     Linux:   export JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64
   ```
   **Stop** — source tests cannot run without Java. Migrated tests may proceed independently if Snowflake is configured.

### Step 4: Verify Snowflake Configuration (Migrated Tests Requirement)

Migrated tests need a Snowflake connection. The skill reads connection config from **`~/.snowflake/config.toml`** (which may reference `connections.toml`) or from `~/.snowflake/connections.toml` directly. Environment variables override TOML values.

**IMPORTANT**: The Snowflake connection MUST be read from the user's config.toml inside the `.snowflake` directory in the user root folder (e.g., `/Users/<username>/.snowflake/config.toml`). This file may specify `default_connection_name` which points to a section in `connections.toml`.

1. **Check both TOML files exist:**
   ```bash
   ls ~/.snowflake/config.toml 2>/dev/null && echo "config.toml found" || echo "config.toml not found"
   ls ~/.snowflake/connections.toml 2>/dev/null && echo "connections.toml found" || echo "connections.toml not found"
   ```

2. **If TOML exists**, verify it has a usable connection:
   ```bash
   <PYTHON_BIN> -c "
   from pathlib import Path
   try:
       import tomllib
   except ModuleNotFoundError:
       import tomli as tomllib
   p = Path.home() / '.snowflake' / 'connections.toml'
   data = tomllib.load(open(p, 'rb'))
   conn_name = data.get('default_connection_name', 'default')
   conn = data.get(conn_name, data.get('default', {}))
   print(f'Connection: {conn_name}')
   print(f'Account: {conn.get(\"account\", \"<not set>\")}')
   print(f'Database: {conn.get(\"database\", \"<not set>\")}')
   print(f'Schema: {conn.get(\"schema\", \"<not set>\")}')
   print(f'Role: {conn.get(\"role\", \"<not set>\")}')
   "
   ```

3. **CRITICAL: Verify the connection points to the correct Snowflake account.** The `default_connection_name` in `connections.toml` may point to a different account (e.g., `SNOWHOUSE_AWS_US_WEST_2` pointing to an internal Snowflake account) while the `[default]` section has the correct target account. If `default_connection_name` points to the wrong account:
   - Set `SNOWFLAKE_CONNECTION_NAME=default` env var to override and use the `[default]` section
   - Or identify the correct connection section name and set `SNOWFLAKE_CONNECTION_NAME=<correct_section>`

4. **Also check environment variable overrides** (these take precedence over TOML):
   ```bash
   echo "SNOWFLAKE_CONNECTION_NAME=${SNOWFLAKE_CONNECTION_NAME:-<not set>}"
   echo "SNOWFLAKE_TEST_DATABASE=${SNOWFLAKE_TEST_DATABASE:-<not set>}"
   echo "SNOWFLAKE_TEST_SCHEMA=${SNOWFLAKE_TEST_SCHEMA:-<not set>}"
   echo "SNOWFLAKE_TEST_ROLE=${SNOWFLAKE_TEST_ROLE:-<not set>}"
   ```

5. **Evaluate readiness:**
   - If `~/.snowflake/connections.toml` exists with a valid default connection → Snowflake config is ready
   - If `SNOWFLAKE_CONNECTION_NAME` env var is set → Snowflake config is ready
   - Otherwise → Snowflake config is **not configured**

6. **If not configured**, inform the user but do NOT stop:
   ```
   Snowflake connection not configured. Migrated tests will be skipped.

   To enable migrated tests, create ~/.snowflake/connections.toml:

     [default]
     account = "your_account"
     user = "your_user"
     password = "your_password"
     role = "your_role"
     warehouse = "your_warehouse"
     database = "your_database"
     schema = "your_schema"

   Or set environment variables:
     export SNOWFLAKE_CONNECTION_NAME=my_connection
     export SNOWFLAKE_TEST_DATABASE=MY_DB
     export SNOWFLAKE_TEST_SCHEMA=MY_SCHEMA
   ```

   Track this as `snowflake_configured = true/false` for use in Step 7.

### Step 4.5: Select Validation Scope

Ask the user which entrypoints to validate. Present numeric options — **full workload is the default**:

```
Validation scope:
  1. Full workload — run all entrypoints (default)
  2. Select specific entrypoints — choose a subset

Press Enter or type 1 to run the full workload.
```

If the user chooses **1 (or presses Enter)**: set `SCOPE=full`. All entrypoints will be tested.

If the user chooses **2**: list detected entrypoints from `dvp/04-results/entrypoints.json`:
```bash
<PYTHON_BIN> -c "
import json
eps = json.load(open('<workload_path>/dvp/04-results/entrypoints.json'))
for i, ep in enumerate(eps, 1):
    print(f'  {i}. {ep[\"name\"]}')
"
```
Ask the user to type the numbers of entrypoints to include (e.g. `1 3`). Store the selection as `SCOPE=partial` with the chosen entrypoint names.

When `SCOPE=partial`, filter test collection in Steps 6 and 7 to only the selected entrypoints using pytest's `-k` flag:
```bash
-k "<name1> or <name2>"
```

### Step 5: Environment Readiness Summary

Present a readiness check table before running tests:

```
Environment Readiness Check

┌──────────────────────────┬──────────┬──────────────────────────────────────────┐
│ Check                    │ Status   │ Details                                  │
├──────────────────────────┼──────────┼──────────────────────────────────────────┤
│ Test project             │ Ready    │ N source + M migrated test files         │
│ Python environment       │ Ready    │ dvp/.venv with all packages              │
│ Java 17+                 │ Ready    │ openjdk 17.0.x                           │
│ Snowflake connection     │ Missing  │ config.py has placeholder values         │
└──────────────────────────┴──────────┴──────────────────────────────────────────┘
```

**Decision logic:**
- If `Test project` is not Ready → **STOP**. Cannot proceed.
- If `Python environment` is not Ready → **STOP**. Cannot proceed.
- If `Java 17+` is not Ready → **WARN** but proceed. Source tests will fail with a clear Java error.
- If `Snowflake connection` is Missing → **WARN** and skip migrated tests in Step 7. Source tests still run.

### Step 6: Run Source Tests (Fix-and-Retry Loop)

Source tests execute the original PySpark code locally and produce baseline CSV outputs. **ALL source tests MUST pass.** If any fail, diagnose and fix the root cause, then re-run. The skill keeps iterating as long as progress is being made — there is **no global cycle limit**. Each individual test gets at most **5 fix attempts** before being marked as unfixable (see Step 6.5). The skill MUST NOT give up — it must do whatever it takes to make source tests pass.

#### 6.1 Run pytest

```bash
cd <workload_path>/dvp/03-tests && <workload_path>/dvp/.venv/bin/python -m pytest source/ -v --tb=long 2>&1
```

Use `--tb=long` (not `--tb=short`) so you have full tracebacks for diagnosis.

#### 6.2 Parse results

- Count passed, failed, errored, skipped tests
- For each test file, record individual test results

#### 6.3 If ALL pass → done

- Verify baseline CSVs were generated: `ls <workload_path>/dvp/03-tests/data/expected_output/`
- Record results in database using `sma_api.insert_test_run()`
- Proceed to Step 7

#### 6.4 If any fail → diagnose and fix

Read the full traceback for each failure and apply the appropriate fix from the table below. After each fix, re-run pytest (go back to 6.1). **Do NOT stop after one fix — fix ALL errors found in a single run before retrying.**

| Error Pattern | Diagnosis | Fix |
|---|---|---|
| `ModuleNotFoundError: No module named 'xxx'` | Missing dependency or stub module | If it's a third-party package: install it via pip. If it's a project-internal module (e.g. `pyspark_wrapper`): create a stub file `dvp/01-source/xxx.py` with a single comment line `# xxx stub — placeholder for SMA migration compatibility`. |
| `JAVA_GATEWAY_EXITED` or `UnsupportedClassVersionError` | Java not installed or wrong version | Report to user — cannot auto-fix. Stop source tests. |
| `FileNotFoundError` or `PATH_NOT_FOUND` for input files | Synthetic data CSVs not where PySpark expects them | The conftest `monkeypatch.chdir()` only changes Python's CWD, not Spark's JVM CWD. PySpark resolves file paths via the JVM, which keeps its own CWD. Use absolute paths via `INPUT_DATA_PATH` and `OUTPUT_DATA_PATH` env vars. Copy synthetic data: `cp dvp/04-results/synthetic_data/*.csv dvp/03-tests/input/` (create `input/` dir if needed). |
| `AttributeError: 'NoneType'` on session/spark | Session not injected correctly | Check `_call_main()` in test file — ensure it passes the session argument to the entrypoint function. |
| Output table is empty (0 rows) | Synthetic data lacks join-key overlap between tables OR pipeline dependencies not satisfied (see below) | Read the source code to identify the JOIN condition and which columns must have matching values across tables. Edit the synthetic CSV files in `dvp/04-results/synthetic_data/` so that at least 2-3 rows share matching join keys. For pipeline dependencies, see the "Pipeline Dependency Chain" fix below. |
| **Pipeline Dependency Chain**: Job reads from another job's OUTPUT (e.g., `customer_churn` reads `OUTPUT_SALES_PATH` which is produced by `sales_ingestion`) | The test runs a single job in isolation — intermediate outputs from other jobs don't exist | In `_call_main()`, before executing the module, **pre-create the intermediate output as parquet** from synthetic data. Example: if job reads `OUTPUT_SALES_PATH`, create `output/sales/` parquet from `output_sales_path.csv` synthetic data BEFORE the module executes. Pattern: `if not os.path.exists(sales_out): tmp_df = session.read.option("header","true").option("inferSchema","true").csv(str(synth_csv)); tmp_df.write.mode("overwrite").parquet(sales_out)` |
| **OUTPUT_DATA_PATH env var pollution**: Second test class reads wrong output directory | When multiple test classes share a SparkSession, `os.environ["OUTPUT_DATA_PATH"]` set by one class persists to the next, causing outputs to be written to the wrong temp directory | **Always recompute** OUTPUT_DATA_PATH from INPUT_DATA_PATH in `_call_main()`: `output_dir = str(Path(os.environ.get("INPUT_DATA_PATH", ".")).parent / "output"); os.environ["OUTPUT_DATA_PATH"] = output_dir; os.makedirs(output_dir, exist_ok=True)` |
| **Module caching**: `config.settings` imported at module level freezes path values from first import | Python caches modules in `sys.modules`. If `config.settings` is imported at module level, its path variables are frozen with values from the first test class's temp directory | Clear `sys.modules` before re-importing: `for mod_name in list(sys.modules.keys()): if mod_name.startswith("config"): del sys.modules[mod_name]`. Also clear `utils` and `pipelines` modules if present. |
| **Output name mapping mismatch**: `_try_read_output("OUTPUT_CHURN_PATH")` fails because it looks for `output/OUTPUT_CHURN_PATH/` | `data_io_schema.json` uses config variable names (e.g., `OUTPUT_CHURN_PATH`) but `persist_baseline` and `_try_read_output` resolve via `output/{name}` — must use actual subdirectory names | In OUTPUT_FILES, set `name` to the actual filesystem subdirectory name (e.g., `churn_scores`, `daily_summary`, `fraud_flags`, `product_affinity`, `sales`, `inventory`) NOT the config variable name (e.g., `OUTPUT_CHURN_PATH`). Read `01-source/config/settings.py` to find the actual path suffixes. |
| Baseline comparison type mismatch (e.g. `datetime.date` vs string) | Spark outputs typed values but baseline CSV stores strings | This is typically a `test_all_outputs_match_baseline` issue. Read the test method in `source/conftest.py` to understand how baselines are compared. If the comparison is strict type-based, the fix may require casting in the comparison or regenerating baselines. Delete the old baseline directory (`dvp/03-tests/data/expected_output/`) and re-run so fresh baselines are generated from the current run. |
| **Boolean comparison mismatch** (e.g., `expected 'false', got False`) | CSV baseline stores `"false"` (lowercase string), PySpark returns Python `False` → `str(False)` = `"False"` (capital F) | Fix `_values_equal` in `conftest.py` to use case-insensitive string comparison: `exp_s = str(expected).strip().lower(); act_s = str(actual).strip().lower(); if exp_s == act_s: return True` |
| **Datetime comparison mismatch** (e.g., `expected '2025-11-12T09:40:17.000-06:00', got datetime.datetime(...)`) | CSV stores timezone-aware ISO string, PySpark returns naive `datetime.datetime` object | Fix `_values_equal` in `conftest.py` to handle datetime objects: detect `isinstance(actual, (datetime.datetime, datetime.date))`, convert to ISO string, and compare with multiple format parsing (with/without timezone, with/without microseconds). |
| **Synthetic data wrong columns**: Source code expects column `event_ts` but CSV has `event_date` | SMA SyntheticDataGenerator produces generic/minimal columns that may not match the actual source code schema | Read the source code to identify exact column names expected, then **regenerate the synthetic CSV** with the correct column headers and appropriate data. For complex jobs like `fraud_detection`, engineer the data to produce expected outputs (see Synthetic Data Engineering section below). |
| **Synthetic data produces empty output**: All rows filtered out by business logic | Synthetic data doesn't meet business logic thresholds (e.g., fraud score < 0.85, no co-purchase pairs) | Read the source code to understand the filtering/scoring logic, then **engineer synthetic data** that produces non-empty output. See Synthetic Data Engineering section below. |
| `enableHiveSupport` or Hive metastore errors | SparkSession not configured for Hive | Ensure `dvp/01-source/` workload's `create_spark_session()` includes `.enableHiveSupport()`. |
| Baseline comparison wrong row order (`expected '', got datetime.date(...)` or similar off-by-one) | Sort key in `compare_dataframes()` treats `None` as `str(None)` = `"None"` which sorts differently from `""` (empty string from CSV) | Fix the sort lambda in `conftest.py`'s `compare_dataframes()`: change `str(row.get(k, ""))` to `"" if row.get(k) is None else str(row.get(k, ""))`. This normalizes `None` to `""` so the sort order matches CSV empty fields. |
| Any other `ERROR` during collection | Syntax error or import-time crash in source code | Read the traceback, fix the source file in `dvp/01-source/`. |

#### Synthetic Data Engineering

When source tests produce empty outputs because synthetic data doesn't meet business logic thresholds, you MUST read the source code, understand the scoring/filtering logic, and engineer data that produces output.

**Example — fraud detection scoring:**
If fraud detection uses a composite score from signals (e.g., velocity_spike=0.30, amount_anomaly=0.25, geo_inconsistency=0.20, first_order_spike=0.15, repeated_decline=0.10) with threshold=0.85:
- Create a customer (e.g., C006) with 12+ events in a 1-hour window (triggers velocity_spike: 0.30)
- Include one event with amount $10000+ (triggers amount_anomaly: 0.25)
- Events from 4+ different regions (triggers geo_inconsistency: 0.20)
- Multiple DECLINED events for same product (triggers repeated_decline: 0.10)
- Total score: 0.30+0.25+0.20+0.10 = 0.85 ≥ threshold → flagged

**Example — product affinity (co-purchase pairs):**
If product affinity joins products within the same order, each order MUST have 2+ distinct products. Single-product orders produce no co-purchase pairs → empty output.

**IMPORTANT — shared data constraint:** If you modify synthetic data CSVs or `data_io_schema.json` to fix source tests, those same files are used by migrated tests. Do NOT create source-only data copies. Both suites read from `dvp/04-results/synthetic_data/`.

#### 6.5 Unfixable Source Tests (LAST RESORT)

If a specific source test has been attempted **5 times** and still genuinely CANNOT be fixed (or the same error recurs with no new fix strategy available):

1. **Verify it is truly unfixable** — not just hard. Ask: Is this a source code bug? An external dependency? A missing service? If it's a data issue, schema issue, or path issue, it IS fixable — keep trying. A test is unfixable when: (a) the same error persists after 5 different fix strategies, OR (b) the error type is inherently unresolvable by the agent (e.g., requires external service, hardware, or user credentials not available).
2. **Add `pytest.mark.skip` with a detailed reason:**
   ```python
   # DVP-SKIP: This test cannot pass because <detailed reason>.
   # Attempted fixes: <list what was tried>.
   # Root cause: <explain the fundamental issue>.
   @pytest.mark.skip(reason="DVP: <source code bug | external dependency | ...> — <brief explanation>")
   class TestXxx(BaseSourceWorkloadTest):
   ```
3. **Report the skip clearly** in the final summary with the reason.
4. **Still proceed to Step 7** (migrated tests) — do not block the entire pipeline.
5. Mark source tests as `Partial` in the final summary.

#### 6.6 Cross-Check: Apply Source Fixes to Migrated Code

**After source tests pass (or are marked as partial), review ALL fixes applied during Step 6 and determine if any also apply to migrated code.** Many issues affect both suites:

| Source Fix | Migrated Equivalent |
|---|---|
| Regenerated synthetic CSV with correct columns | Same CSV is used — no action needed (shared `synthetic_data/`) |
| Fixed `_values_equal` in root conftest.py (boolean, datetime) | Same conftest — no action needed (shared) |
| Added `sys.modules` cleanup in source `_call_main()` | MUST also add to ALL migrated test `_call_main()` methods |
| Fixed output name mapping (actual subdir names in OUTPUT_FILES) | MUST also fix in migrated test OUTPUT_FILES |
| Created intermediate parquet for pipeline dependencies | Migrated tests use stage paths — may need equivalent stage data setup |
| Created `__init__.py` in `01-source/config/` | MUST also create `02-migrated/config/__init__.py` if migrated has `config/` package |
| Rewrote `config/settings.py` for env var support | Must verify `02-migrated/config/settings.py` also supports `INPUT_DATA_STAGE`/`OUTPUT_DATA_STAGE` |

**Rule:** After every source fix cycle, scan the fix list and proactively apply equivalent fixes to `02-migrated/` before running migrated tests in Step 7. This prevents re-discovering the same issues.

### Step 7: Run Migrated Tests (Runability Fix Loop)

Migrated tests **always run** — they are not gated on source test results. The only skip condition is a missing Snowflake connection (`snowflake_configured == false` from Step 4).

**IMPORTANT — pre-apply source fixes:** Before running migrated tests, review ALL fixes applied in Step 6 (source tests) and proactively apply equivalent fixes to migrated code/tests where applicable (see Step 6.6 cross-check table). This prevents wasting retry cycles on issues already solved for source.

**Goal:** Every migrated test must be **collected and executed** by pytest. Assertion failures (FAILED) are expected and acceptable. Collection errors, import crashes, and infrastructure setup failures are NOT acceptable — fix them.

#### 7.1 Skip check

If `snowflake_configured == false`, skip with:
```
Migrated tests skipped — Snowflake connection not configured.
Configure connection and re-run dvp-test-runner to validate migrated code.
```

#### 7.2 Detect migrated flavor

- If `dvp/03-tests/migrated/` exists → `migrated`
- If `dvp/03-tests/migrated_scos/` exists → `migrated_scos`

#### 7.3 Run pytest

```bash
cd <workload_path>/dvp/03-tests && <workload_path>/dvp/.venv/bin/python -m pytest <migrated_flavor>/ -v --tb=long 2>&1
```

#### 7.4 Evaluate results — distinguish runability errors from test failures

Parse the pytest output and classify each result:

- **PASSED / FAILED** → The test was collected and executed. This is acceptable. Record the result.
- **ERROR during collection** → The test could not even be imported/collected. This is a runability problem — must fix.
- **ERROR during setup** (fixture failure) → Infrastructure issue (session creation, stage setup, file upload). Must fix.
- **ERROR in `_call_main`** with `ModuleNotFoundError` or `ImportError` → Missing stub or dependency in `dvp/02-migrated/`. Must fix.

#### 7.5 Fix runability issues

If there are collection errors, setup errors, or import crashes, apply fixes and re-run. There is **no global cycle limit** — keep iterating as long as each cycle fixes at least one new issue. Each individual test gets at most **5 fix attempts** before being reported as a persistent runability issue. **The skill MUST iterate until all migrated tests execute (even if they fail assertions). Do NOT stop after fixing one issue — fix ALL errors found in a single run before retrying.**

| Error Pattern | Diagnosis | Fix |
|---|---|---|
| `ModuleNotFoundError: No module named 'xxx'` in migrated code | Missing stub or dependency | If project-internal module (e.g. `pyspark_wrapper`): create stub at `dvp/02-migrated/xxx.py` with `# xxx stub — placeholder for SMA migration compatibility`. If third-party: install via pip. |
| `ImportError` from migrated workload | Bad import in SMA-generated code | Read the import line. If it imports a PySpark module that doesn't exist in Snowpark, add a compatibility stub or comment out the import. |
| **`'config' is not a package`** or `ModuleNotFoundError: No module named 'config.settings'` | Python finds `03-tests/config.py` (a module) instead of `02-migrated/config/` (a package directory) because `config/__init__.py` is missing | Create `dvp/02-migrated/config/__init__.py` (empty file) to make `config/` a proper Python package. Also add `sys.modules` cleanup in migrated test `_call_main()`: `for mod_name in list(sys.modules.keys()): if mod_name.startswith("config") or mod_name.startswith("utils") or mod_name.startswith("pipelines"): del sys.modules[mod_name]` |
| **Module caching in migrated tests** | Same as source: `config.settings` cached in `sys.modules` from previous test class | Add `sys.modules` cleanup in ALL migrated test files' `_call_main()` methods: clear `config`, `utils`, `pipelines` prefixes |
| `snowflake.snowpark.exceptions.SnowparkSessionException` | Session creation failed | Check config.py TOML reading. Verify `~/.snowflake/connections.toml` has valid credentials. |
| **`Object does not exist` on `USE ROLE`/`USE DATABASE`/`USE SCHEMA`** | Snowpark session fixture crashes because the role/database/schema doesn't exist or the user lacks privileges | Wrap `USE ROLE`/`USE DATABASE`/`USE SCHEMA` statements in `conftest.py`'s `snowpark_session` fixture with try/except blocks: `try: session.sql(f"USE ROLE {TEST_ROLE}").collect() except Exception: pass`. This allows the session to proceed with whatever defaults the connection provides. |
| `Cannot perform CREATE STAGE. This session does not have a current database` or `Object does not exist` (002043) on `USE DATABASE` | The TOML default connection points to a different Snowflake **account** than where the test database lives (e.g. `SNOWHOUSE` account vs the account hosting the DVP database) | Identify which Snowflake account has the test database. Set `SNOWFLAKE_CONNECTION_NAME` env var to a connection that targets the correct account (e.g. `default` or another connection in `connections.toml`). Also set `SNOWFLAKE_TEST_DATABASE` and `SNOWFLAKE_TEST_SCHEMA` env vars. |
| `config.py` resolves `CONNECTION_NAME` to `"default"` instead of the actual TOML connection name | `_toml.get("connection_name")` fails because the connection name is the TOML **section header**, not a key inside the section | Fix `_load_toml_connection()` to return a `(conn_name, conn_dict)` tuple: read `data.get("default_connection_name", "default")` for the name, then `data.get(conn_name, {})` for the dict. Use `conn_name` directly as `CONNECTION_NAME`. |
| `stage does not exist` or `PUT` failure | Test stage not created | Check that `test_stage` fixture runs `CREATE STAGE IF NOT EXISTS`. |
| `FileNotFoundError` for synthetic data during upload | Synthetic data not where conftest expects | Verify `dvp/04-results/synthetic_data/` contains the expected CSV files. |
| **`is an invalid Snowflake stage location`** or migrated config resolves to local filesystem paths | Migrated code uses `config.settings` which resolves to local paths, but Snowpark requires `@STAGE/file.csv` stage locations | Rewrite `dvp/02-migrated/config/settings.py` to support stage paths via `INPUT_DATA_STAGE`/`OUTPUT_DATA_STAGE` env vars. When these env vars are set, ALL path variables should resolve to `f"{_STAGE}/filename.csv"` for inputs and `f"{_OUT_STAGE}/subdir"` for outputs. When not set, fall back to local filesystem paths (for local development). This dual-mode config allows the same migrated code to work both locally and against Snowflake stages. |
| `SnowparkSQLAmbiguousJoinException` (ambiguous column after join) | PySpark join syntax `on="col"`, `how="left"` creates duplicate columns in Snowpark | Change all joins to Snowpark syntax: `using_columns=["COL"]` and `join_type="left"`. Also uppercase all column references in `col("...")`, `groupBy("...")`, etc. since Snowpark normalizes identifiers to uppercase. |
| **`TypeError: Unexpected item type: <class 'snowflake.snowpark.column.Column'>`** or PySpark `StructType`/`StructField` used in Snowpark | SMA migration left PySpark-specific API patterns in migrated code (e.g., `StructType([StructField(...)])` for schema definition) | This is a **migration quality issue** in `02-migrated/` code, not test infrastructure. Report it as requiring `dvp-ewi-fixer` resolution. The test should still be collected and execute (it will fail with this error during `_call_main`). Mark as `ERROR` but do not consider it a runability infrastructure failure. |
| DataFrameReader returns wrong columns (e.g. `returns_data` shows `raw_transactions` columns) | **Snowpark DataFrameReader caches schema** when a single reader object is reused for multiple CSVs — all subsequent files inherit the first file's schema | Create a **new reader per file**: use a helper like `def _read_csv(path): return spark.read.option("header", True).option("inferSchema", True).csv(path)`. Never reuse a reader object across different CSV files. |
| CSV column names are quoted lowercase (`"col_name"`) instead of `COL_NAME` | Snowpark CSV reader preserves original case and wraps names in double quotes | Add a normalization step: `df.to_df(*[c.strip('"').upper() for c in df.columns])` after every `read.csv()` call. |
| `SQL compilation error` during table creation | DDL mismatch with `data_io_schema.json` | Read the error, fix column types in `data_io_schema.json`. Remember: this file is shared with source tests. |
| Any other collection/setup `ERROR` | Read traceback | Fix the specific issue in the migrated code, test file, or conftest. |

**IMPORTANT:** Fixes to migrated code go in `dvp/02-migrated/`. Never modify `dvp/01-source/` to fix migrated test issues (except for shared data files in `dvp/04-results/`).

#### 7.6 Record and report results

After the final run (all tests collected and executed, or max retries reached):

1. Record results in database using `sma_api.insert_test_run()` for each registered test where `test_type = '<migrated_flavor>'`.

2. Report results — assertion failures are expected:

#### 7.7 Side-by-side PySpark vs SCOS comparison

After both suites complete, present a side-by-side diff to expose behavioral and data drift between local PySpark execution and SCOS/Snowflake execution.

**Schema diff — compare column names and types per entrypoint:**

```
Schema Comparison
┌──────────────────┬─────────────────────────────┬─────────────────────────────┐
│ Entrypoint       │ PySpark (source)             │ SCOS / Snowflake (migrated) │
├──────────────────┼─────────────────────────────┼─────────────────────────────┤
│ process_orders   │ order_id: long               │ ORDER_ID: long        ✅    │
│                  │ total: double                │ TOTAL: double         ✅    │
│                  │ created_at: timestamp        │ CREATED_AT: string    ⚠️    │
└──────────────────┴─────────────────────────────┴─────────────────────────────┘
```

Build this table by reading the baseline CSVs (`dvp/03-tests/data/expected_output/`) for source columns and the migrated test assertion errors for SCOS columns.

**Data diff — row-count and sample mismatch per entrypoint:**

```
Data Comparison
┌──────────────────┬──────────────┬────────────────┬──────────────────────────────┐
│ Entrypoint       │ PySpark rows │ SCOS rows      │ Drift                        │
├─────────

…(truncated)