# Dvp Test Setup Generator

> Generate pytest files for source and migrated workloads based on entrypoints.json.

- Skill: `snowflake-labs/dvp-test-setup-generator` (Agent Skill, multi-file: 10 files)
- Install (CLI): `npx skillmds@latest add snowflake-labs/dvp-test-setup-generator`
- Raw SKILL.md: https://api.skillmd.com/api/skills/snowflake-labs/dvp-test-setup-generator/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-setup-generator

---


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

```python
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:
  ```bash
  mkdir -p <workload_path>/.vscode
  ```
  ```json
  {
      "python.testing.pytestEnabled": true,
      "python.testing.pytestArgs": [
          "dvp/03-tests",
          "-v",
          "--import-mode=importlib"
      ],
      "python.analysis.extraPaths": [
          "dvp/03-tests",
          "dvp/01-source",
          "dvp/02-migrated"
      ]
  }
  ```

### 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](../docs/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):
```python
"""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 / "01-source"
        source_file = source_dir / "<filename>"
        spec = importlib.util.spec_from_file_location("<module_name>", source_file)
        mod = importlib.util.module_from_spec(spec)
        sys.modules[spec.name] = mod
        sys.path.insert(0, str(source_dir))
        try:
            spec.loader.exec_module(mod)
            fn = getattr(mod, "<function_name>", None)
            if fn is not None:
                result = fn(session)
                return result if result is not None else 0
            return 0
        finally:
            sys.path.pop(0)

    def test_validate_pipeline_runs(self):
        """Confirm the workload executed without error (run_workload fixture)."""
        pass  # run_workload (autouse) already executed _call_main and asserted success
```

**Migrated test template** (use this exact pattern):
```python
"""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 = getattr(mod, "<function_name>", None)
            if fn is not None:
                result = fn(session)
                return result if result is not None else 0
            return 0
        finally:
            sys.path.pop(0)

    def test_validate_pipeline_runs(self):
        """Confirm the workload executed without error (run_workload fixture)."""
        pass  # run_workload (autouse) already executed _call_main and asserted success
```

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

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

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