dvp-test-runner

name	dvp-test-runner
description	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.

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.

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/:

   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:

   <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:

   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:

   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):

     .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:

   java -version 2>&1
   

2. If Java is present, extract the major version and verify it is 17–21:

   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:

   # 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:

   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:

   <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):

   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:

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:

<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:

-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

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 parquet from synthetic data BEFORE the module executes. Pattern:

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:

   # 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

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 // statements in 's fixture with try/except blocks: . This allows the session to proceed with whatever defaults the connection provides.

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                        │
├──────────────────┼──────────────┼────────────────┼──────────────────────────────┤
│ process_orders   │ 42           │ 42             │ None ✅                       │
│ compute_totals   │ 18           │ 17             │ -1 row ⚠️                     │
│ flag_anomalies   │ 5            │ 0              │ All rows missing ❌           │
└──────────────────┴──────────────┴────────────────┴──────────────────────────────┘

Row counts for source come from the baseline CSV row counts. Row counts for migrated come from migrated test assertion output or sma_api.get_test_results().

Legend: ✅ Match · ⚠️ Minor drift (type difference or ≤ 5% row delta) · ❌ Significant drift

Present this table immediately after Step 7.6 and before Step 8.

Migrated test results (<flavor>):
  N collected, X passed, Y failed, Z errors

Failed tests (expected — migration validation differences):
  - test_workload::test_all_outputs_match_baseline
    AssertionError: Column mismatch in 'total_spend'

Runability issues fixed: <list of fixes applied, or "none">

Step 8: Export Test Results

Export all test results to a CSV for reporting:

result = sma_api.export_test_results("<workload_path>")

This writes a timestamped CSV to dvp/04-results/testing-results/.

Step 9: Commit Changes to Git

After tests are executed, commit results (baseline CSVs, test run records):

result = sma_api.git_commit("<workload_path>", """DVP Test Runner: Executed test suites

Source tests: N passed, M failed
Migrated tests: X passed, Y failed (or skipped)
Baselines: dvp/03-tests/data/expected_output/
Results: dvp/04-results/testing-results/""")

Verify branches:

result = sma_api.git_verify_branches("<workload_path>")

Parsing Pytest Output

When parsing pytest output to record individual test results, match lines like:

source/test_job_customer_stats.py::TestJobCustomerStats::test_validate_pipeline_runs PASSED
source/test_job_customer_stats.py::TestJobCustomerStats::test_all_outputs_have_data PASSED
source/test_job_customer_stats.py::TestJobCustomerStats::test_all_outputs_match_baseline FAILED

For each line:

• Extract the test file path (e.g., source/test_job_customer_stats.py)
• Extract the test class (e.g., TestJobCustomerStats)
• Extract the test method (e.g., test_validate_pipeline_runs)
• Extract the status (PASSED, FAILED, ERROR, SKIPPED)

Match test files to entrypoint_tests records by comparing the test_file column with the test file path from pytest output.

To extract duration, look for the summary line:

===== 4 passed in 12.34s =====

For failed tests, capture the traceback between the FAILURES section header and the next test or summary line.

Stopping Points

• No test files found → stop and report. Run dvp-test-setup-generator first.
• Python environment cannot be created → stop and report installation instructions.
• Java not available → warn but continue. Source tests will fail; migrated tests still run.
• Snowflake connection not configured → source tests still run; migrated tests are skipped.

These are NOT stopping points:

• Source test failures → fix and retry (up to 5 attempts per test), then proceed to migrated tests regardless.
• Migrated test assertion failures → expected and acceptable. Only runability errors trigger fixes.

Final Summary

MANDATORY: After completing all steps (whether running standalone or invoked from the orchestrator), ALWAYS present this summary table:

Test Execution Complete

┌──────────────────────────┬──────────┬──────────────────────────────────────────┐
│ Step                     │ Status   │ Details                                  │
├──────────────────────────┼──────────┼──────────────────────────────────────────┤
│ Environment Check        │ Done     │ Python, Java, packages verified          │
│ Source Tests             │ Done     │ N passed, 0 failed                       │
│ Migrated Tests           │ Done     │ X passed, Y failed                       │
│ Results Export           │ Done     │ Exported to dvp/04-results/testing-results/ │
│ Git Commit               │ Done     │ Committed baselines and results          │
└──────────────────────────┴──────────┴──────────────────────────────────────────┘

Output location: <output>/

Git branches:
• main — original code (unmodified)
• sma/migration-process — test execution results applied

Rules:

• Replace N, X, Y with actual counts from pytest output
• Status is Done, Skipped, or Failed
• If source tests failed, show Failed with count of failures
• If migrated tests were skipped (no Snowflake config), show Skipped with reason
• If environment check failed, show Failed with what is missing
• The git branches section uses sma_api.git_verify_branches() to confirm both branches exist

Next steps (always show verbatim after the table):

Next steps:
  1. Run dvp-migrated-test-fixer to make failing migrated tests pass
  2. Use dvp-ewi-fixer to resolve EWI issues in dvp/02-migrated/
  3. Re-run: dvp-test-runner to validate fixes
  4. Open the SMA Dashboard to view detailed test results per entrypoint

Replace dvp/02-migrated/ with dvp/02-migrated_scos/ when the SCOS suite was used.