dvp-test-setup-generator

name	dvp-test-setup-generator
description	Generate pytest files for source and migrated workloads based on entrypoints.json.

DVP Test Setup Generator

Overview

Generate a pytest project under dvp/03-tests/.

• Always generates source tests (dvp/03-tests/source/) to produce baseline CSV outputs.
• Generates exactly one migrated flavor per run:
  • dvp/03-tests/migrated/ (Snowpark API) or
  • dvp/03-tests/migrated_scos/ (SCOS / Snowpark Connect)

For detailed rationale and examples, see:

• skills/spark-migration/snowpark-api/dvp/docs/data-validator/dvp-test-setup-generator.md

Preconditions

• DVP workspace exists (dvp/01-source/, dvp/03-tests/, dvp/04-results/).
• Code has been adapted for testing by dvp-code-adapter.
• Exactly one migrated folder exists in the workspace:
  • dvp/02-migrated/ or dvp/02-migrated_scos/.

Inputs

Input	Required	Location
Entrypoints inventory	Yes	dvp/04-results/entrypoints.json
IO schema	Yes	dvp/04-results/data_io_schema.json
Synthetic data	Yes	dvp/04-results/synthetic_data/*.csv

Outputs

Output	Format	Location
pytest project	files	dvp/03-tests/
generated tests (source)	Python	dvp/03-tests/source/**/test_*.py
generated tests (migrated)	Python	dvp/03-tests/migrated/**/test_*.py (if selected)
generated tests (scos)	Python	dvp/03-tests/migrated_scos/**/test_*.py (if selected)

dvp/03-tests/data/expected_output/ is a runtime artifact created when running the source suite; it is not generated by this skill.

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: Locate required files...
[2026-03-24 14:05:45] Created test_job_customer_stats.py
[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 — if the orchestrator already initialized git, this is a no-op.

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

Step 1: Copy templates

Copy scaffolding from dvp-test-setup-generator/templates/ into <workload_path>/dvp/03-tests/:

IMPORTANT: .gitignore is a dotfile — shell globs like * do NOT match dotfiles by default. You MUST copy it explicitly: cp templates/.gitignore <workload_path>/.gitignore

• Copy these files into <workload_path>/dvp/03-tests/:
  • conftest.py, config.py, requirements.txt
  • source/conftest.py
• Copy DVP-TESTING.md to the workload root (<workload_path>/DVP-TESTING.md), NOT inside dvp/03-tests/.
• Copy .gitignore to the workload root (<workload_path>/.gitignore), NOT inside dvp/03-tests/:
  • If it already exists, merge the entries (append missing lines). Do not duplicate.
• Copy exactly one migrated flavor based on which migrated folder exists in the workspace:
  • If dvp/02-migrated/ exists: copy migrated/conftest.py and generate pytest.ini with testpaths = source migrated
  • If dvp/02-migrated_scos/ exists: copy migrated_scos/conftest.py and generate pytest.ini with testpaths = source migrated_scos
• Generate or merge .vscode/settings.json at the workload root (<workload_path>/.vscode/settings.json), NOT inside dvp/03-tests/:
  • If the file already exists, read it, add/update the pytest keys, and write it back (preserve existing settings).
  • If the file does not exist, create it:

  mkdir -p <workload_path>/.vscode
  

  {
      "python.testing.pytestEnabled": true,
      "python.testing.pytestArgs": [
          "dvp/03-tests",
          "-v",
          
      
       
          
          
          
      
  
  

Step 2: Read entrypoints

Load dvp/04-results/entrypoints.json and generate tests for entries where status == "detected".

For each entrypoint, determine the invocation target:

1. If adapted_source exists and contains :: → parse it: the last :: segment is the callable function name, preceding segments (if any) are scope (class/object). Example: workload.py:163::main → call main; App.scala:5::MyApp::main → call MyApp.main
2. If adapted_source exists but has no :: → the entrypoint is the whole file (execute as script)
3. If adapted_source does not exist → the code-adapter has not run; stop and report

The adapted_source field uses the hybrid format <path>:<lineno>(::segment)*. See entrypoints-source-spec.md.

Preserve subfolders: if an entrypoint lives under etl/daily_metrics.py, generate tests under:

• dvp/03-tests/source/etl/test_daily_metrics.py
• dvp/03-tests/<selected-suite>/etl/test_daily_metrics.py

Step 3: Generate test files

For each entrypoint:

• Parse adapted_source to extract the file path and function name.
• Load data_io_schema.json and filter entries whose source field starts with the same filename as the entrypoint (e.g., for entrypoint job_customer_stats with adapted_source: "job_customer_stats.py:9::main", filter entries where source starts with "job_customer_stats.py").
• Group the filtered entries into: INPUT_FILES (role=input, type=file), INPUT_TABLES (role=input, type=table), OUTPUT_FILES (role=output, type=file), OUTPUT_TABLES (role=output, type=table).
• Generate a source test file as a class that inherits from BaseSourceWorkloadTest.
• Generate a migrated or scos test file for the selected suite with the corresponding base class.

Required test method name: The execution test in every generated test_*.py file must be named test_validate_pipeline_runs. Do NOT invent alternative names like test_run_without_error, test_workload_completes, etc. This ensures consistent tracking across regenerations.

Required fixture parameter: test_validate_pipeline_runs must use self and workload_session as parameters (it's a class method). Do NOT invent fixture names like source_spark, spark, session, etc. The workload_session fixture is defined in each suite's conftest.py and returns the correct session type automatically:

• Source → PySpark SparkSession
• Migrated → Snowpark Session
• SCOS → SparkSession via Spark Connect

Why class-based: The base classes (BaseSourceWorkloadTest, BaseMigratedWorkloadTest, BaseScosWorkloadTest) inherit from BaseIOConfig, which provides three output validation tests automatically via inheritance:

• test_all_outputs_have_data — checks every output has at least one row
• test_all_outputs_match_baseline_row_count — compares row counts against baseline
• test_all_outputs_match_baseline — full data comparison (schema + values)

These tests iterate over OUTPUT_FILES + OUTPUT_TABLES. If both lists are empty, the tests run but have nothing to iterate over (no failures). The run_workload fixture in the base class also handles input setup and cleanup automatically.

Class naming: Convert the entrypoint name to PascalCase and prefix with Test. Example: job_customer_stats → TestJobCustomerStats.

Source test template (use this exact pattern):

"""Source test for <name> entrypoint."""
import importlib
import importlib.util
import sys
from pathlib import Path

# Load suite conftest to access BaseSourceWorkloadTest (importlib mode safe)
_conftest_path = Path(__file__).resolve().parent / "conftest.py"
_spec = importlib.util.spec_from_file_location("_source_conftest", _conftest_path)
_mod = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_mod)
BaseSourceWorkloadTest = _mod.BaseSourceWorkloadTest


class Test<ClassName>(BaseSourceWorkloadTest):
    """Test suite for the <name> source workload."""

    INPUT_FILES = <input_files_list>
    INPUT_TABLES = <input_tables_list>
    OUTPUT_FILES = <output_files_list>
    OUTPUT_TABLES = <output_tables_list>

    def _call_main(self, session):
        """Import and execute the source workload with the test session."""
        import os
        from pathlib import Path

        # --- env var isolation: always recompute OUTPUT_DATA_PATH ---
        input_path = os.environ.get("INPUT_DATA_PATH", ".")
        output_dir = str(Path(input_path).parent / "output")
        os.environ["OUTPUT_DATA_PATH"] = output_dir
        os.makedirs(output_dir, exist_ok=True)

        # --- module cache cleanup: prevent stale config/utils from prior test ---
        for mod_name in list(sys.modules.keys()):
            if mod_name.startswith(("config", "utils", "pipelines")):
                del sys.modules[mod_name]

        source_dir = Path(__file__).resolve().parent.parent.parent / 
        source_file = source_dir / 
        spec = importlib.util.spec_from_file_location(, source_file)
        mod = importlib.util.module_from_spec(spec)
        sys.modules[spec.name] = mod
        sys.path.insert(, (source_dir))
        :
            spec.loader.exec_module(mod)
            fn = (mod, , )
             fn   :
                result = fn(session)
                 result  result     
             
        :
            sys.path.pop()

     ():
        
          

Migrated test template (use this exact pattern):

"""Migrated test for <name> entrypoint (Snowpark API)."""
import importlib
import importlib.util
import sys
from pathlib import Path

# Load suite conftest to access BaseMigratedWorkloadTest (importlib mode safe)
_conftest_path = Path(__file__).resolve().parent / "conftest.py"
_spec = importlib.util.spec_from_file_location("_migrated_conftest", _conftest_path)
_mod = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_mod)
BaseMigratedWorkloadTest = _mod.BaseMigratedWorkloadTest


class Test<ClassName>(BaseMigratedWorkloadTest):
    """Test suite for the <name> migrated workload."""

    INPUT_FILES = <input_files_list>
    INPUT_TABLES = <input_tables_list>
    OUTPUT_FILES = <output_files_list>
    OUTPUT_TABLES = <output_tables_list>

    def _call_main(self, session):
        """Import and execute the migrated workload with the test session."""
        import os
        from pathlib import Path

        # --- module cache cleanup: prevent stale config/utils from prior test ---
        for mod_name in list(sys.modules.keys()):
            if mod_name.startswith(("config", "utils", "pipelines")):
                del sys.modules[mod_name]

        migrated_dir = Path(__file__).resolve().parent.parent.parent / "02-migrated"
        migrated_file = migrated_dir / "<filename>"
        spec = importlib.util.spec_from_file_location("<module_name>", migrated_file)
        mod = importlib.util.module_from_spec(spec)
        sys.modules[spec.name] = mod
        sys.path.insert(0, str(migrated_dir))
        try:
            spec.loader.exec_module(mod)
            fn = (mod, , )
             fn   :
                result = fn(session)
                 result  result     
             
        :
            sys.path.pop()

     ():
        
          

SCOS test template — same as migrated but:

• Load BaseScosWorkloadTest instead of BaseMigratedWorkloadTest (name the spec "_scos_conftest")
• Change "02-migrated" to "02-migrated_scos"
• Update docstrings accordingly

I/O attribute rules:

• Each INPUT_FILES, INPUT_TABLES, OUTPUT_FILES, OUTPUT_TABLES is a Python list of dicts, copied from matching data_io_schema.json entries.
• CRITICAL: data_io_schema.json is JSON — you MUST convert JSON null to Python None when writing dict literals in the test file. Writing null produces NameError: name 'null' is not defined.
• CRITICAL — output name mapping: The name field in OUTPUT_FILES entries must match the actual filesystem subdirectory name the workload writes to (e.g., churn_scores, daily_summary, fraud_flags), NOT the config variable name (e.g., OUTPUT_CHURN_PATH). Read 01-source/config/settings.py to find the actual path suffixes used by each output. The _try_read_output() and persist_baseline() functions in conftest resolve outputs via output/{name}.
• Include the full dict for each entry (name, full_name, type, format, role, columns, etc.) — the base class methods use fields like name, columns, key_columns, format.
• If no matching entries exist for a category, use an empty list [].
• Example: INPUT_FILES = [{"name": "orders", "type": "file", "format": "memory", "role": "input", "path": None, "columns": [...]}]

Step 4: Register tests in database (MANDATORY)

After generating the test files, run the registration script to populate sma_storage.sqlite3 for dashboard tracking.

python3 "<skills_path>/spark-migration/dvp/dvp-test-setup-generator/scripts/register_tests.py" \
  --workload-path "<workload_path>"

The script automatically:

1. Reads dvp/04-results/entrypoints.json for detected entrypoints
2. Scans dvp/03-tests/source/ and dvp/03-tests/migrated/ (or migrated_scos/) for test_*.py files
3. Matches test files to entrypoints by filename stem
4. Calls sma_api.register_tests() to insert them into the entrypoint_tests table

This enables the Test Tracker module in the SMA Dashboard.

Step 5: Remind user to verify Snowflake config

After generation, remind the user that dvp/03-tests/config.py auto-reads credentials from ~/.snowflake/connections.toml. If the user has a valid TOML config, no manual editing is needed. Environment variables (SNOWFLAKE_CONNECTION_NAME, SNOWFLAKE_TEST_DATABASE, etc.) override TOML values.

Step 6: Commit Changes to Git

After test project is generated, commit the changes:

result = sma_api.git_commit("<workload_path>", """DVP Test Setup: Generated test project with N test files

Test suites: source + migrated (or migrated_scos)
Entrypoints covered: N
Output: dvp/03-tests/""")

Verify branches:

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

Stopping points

• Missing entrypoints.json or data_io_schema.json stop and report which file is missing.
• Both migrated folders exist or neither exists stop and instruct the user to re-run the orchestrator (single migrated flavor per run).

Final Summary

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

Test Setup Complete

┌──────────────────┬──────────┬──────────────────────────────────────────────────────────┐
│ Step             │ Status   │ Details                                                  │
├──────────────────┼──────────┼──────────────────────────────────────────────────────────┤
│ Test Setup       │ Done     │ Generated test project with N test files                  │
│ DB Registration  │ Done     │ Registered N tests in sma_storage.sqlite3                 │
└──────────────────┴──────────┴──────────────────────────────────────────────────────────┘

Output location: <output>/

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

Next steps (always show verbatim after the table):

Next steps:
  1. Review DVP-TESTING.md for detailed testing instructions
  2. cd <workload_path>/dvp/03-tests && pip install -r requirements.txt
  3. Verify ~/.snowflake/connections.toml has your Snowflake connection
  4. pytest dvp/03-tests/source/ -v          (generate PySpark baselines)
  5. pytest dvp/03-tests/migrated/ -v        (validate Snowpark conversion)
  6. Open the SMA Dashboard to view results

Replace migrated/ with migrated_scos/ when the SCOS suite was generated.

Rules:

• Replace N with actual count of test files generated
• Replace <workload_path> with the actual workload path
• Status is Done, Skipped, or Failed
• If prerequisites were missing, show Failed with which file is missing
• The git branches section uses sma_api.git_verify_branches() to confirm both branches exist