Python-Driven Test Generation

What this does

A short Python script reads a spreadsheet and creates one Validatar standard test per row through the public API. Each row carries a test name, a SQL statement to run against the source, and a SQL statement to run against the target. Both statements return a single number. The generated test passes when the two numbers agree.

The tests it creates are ordinary standard tests. They are not linked to catalog metadata, they do not depend on a template, and each one can be opened and edited in Validatar like any test built by hand. The spreadsheet is simply where the definitions live between runs.

Choose this route when:

• The objects you compare are not in the Validatar catalog, or you do not want the tests tied to it.
• Someone maintains the list of comparisons in Excel already and wants to keep doing that.
• The comparison is always the same shape: one number on each side.

If the tests should follow catalog objects, carry per-row parameters into a template, and stay grouped under one parent, use a template test instead. Generating Child Tests from an Import File covers that approach and Template Test Examples: Import File has worked examples.

Two scripts follow. The first names the source and target data sources once at the top and applies them to every row. The second reads them from two extra columns, so one workbook can compare across several systems.

Prerequisites

• Python 3.9 or later with three packages: pip install requests pandas openpyxl
• A user token with the Authoring scope and access to the target project. See User Tokens. The script also reads the data source list through the Catalog endpoint; if that call returns 403, grant the token all scopes, or replace the name lookup with the numeric data source ids.
• Project ID and folder ID. Open the folder that should receive the tests in Validatar. The URL reads /projects/<project id>/folders/<folder id>.
• Data source names exactly as they appear in the Catalog. Case does not matter; spelling does.
• SQL that returns one row with one numeric column on each side. The column alias does not matter: the test matches by position.

The workbook

One sheet, headers in the first row. Extra columns are ignored, so annotate freely.

Fixed data sources (script 1):

Data sources per row (script 2):

Description is optional in both layouts. Test_Name must be unique within the folder; it is also how a re-run finds the test to update.

Script 1: fixed source and target data sources

Save as create_tests_from_excel.py. Fill in the block at the top, or set the environment variables named there.

"""Create one Validatar standard test per row of an Excel workbook.

Each row supplies a test name, a source SQL statement and a target SQL statement.
Both statements must return a single numeric value (a count, a sum, a max). The
script creates a standard test that runs the source SQL as the Test Data Set, the
target SQL as the Control Data Set, and passes when the two numbers match.

Every test in this script runs against the same two data sources, named once at
the top. For a workbook that names the data sources per row, use
create_tests_from_excel_dynamic_sources.py instead.

Workbook layout (first sheet, header row):

    Test_Name | Source_SQL | Target_SQL | Description (optional)

Requirements:  pip install requests pandas openpyxl
API token:     Validatar > your user name > Tokens > New Token, with the Authoring
               scope and access to the target project. Keep it out of source control.
"""
import os
import sys

import pandas as pd
import requests

# ---------------------------------------------------------------------------
# Placeholders. Edit these, or set the matching environment variable.
# ---------------------------------------------------------------------------
VALIDATAR_API_URL = os.environ.get("VALIDATAR_API_URL", "https://<your-instance>.cloud.validatar.com")  # no trailing slash
API_TOKEN = os.environ.get("VALIDATAR_API_TOKEN", "<paste your user token here>")

PROJECT_ID = int(os.environ.get("VALIDATAR_PROJECT_ID", "0"))          # from the project URL: /projects/<id>
TARGET_FOLDER_ID = int(os.environ.get("VALIDATAR_FOLDER_ID", "0"))     # test folder that receives the tests: /folders/<id>

SOURCE_DATA_SOURCE = os.environ.get("VALIDATAR_SOURCE_DS", "Sample Data Source - Snowflake Data Warehouse")
TARGET_DATA_SOURCE = os.environ.get("VALIDATAR_TARGET_DS", "Sample Data Source - Snowflake Data Warehouse")

EXCEL_FILE = os.environ.get("TESTS_EXCEL_FILE", "tests.xlsx")
SHEET_NAME = 0                     # first sheet; use a name to pick another

# Defaults applied to every test created.
SEVERITY = "High"                  # Critical, High, Medium, Low, Informational
QUALITY_DIMENSION = "Completeness" # Accuracy, Completeness, Consistency, Timeliness, Uniqueness, Validity
TOLERANCE = 0                      # allowed difference between the two numbers
UPDATE_EXISTING = True             # a test with the same name in the folder is updated, not duplicated

# ---------------------------------------------------------------------------
# API helpers
# ---------------------------------------------------------------------------
API = f"{VALIDATAR_API_URL.rstrip('/')}/core/api/v1"
HEADERS = {"x-val-api-token": API_TOKEN, "Accept": "application/json"}


def call(method, path, **kwargs):
    """One HTTP call. Raises with the response body on any error status."""
    response = requests.request(method, f"{API}{path}", headers=HEADERS, timeout=60, **kwargs)
    if response.status_code >= 400:
        raise SystemExit(f"{method} {path} -> HTTP {response.status_code}\n{response.text[:1000]}")
    return response.json() if response.content else {}


def data_source_ids_by_name():
    """Map data source name (lower-cased) -> id, from the catalog."""
    listing = call("GET", "/Catalog/data-sources")
    return {ds["dataSourceName"].strip().lower(): ds["dataSourceId"] for ds in listing.get("dataSources", [])}


def existing_tests_in_folder(project_id, folder_id):
    """Map test name -> test id for tests already in the folder."""
    folder = call("GET", f"/Authoring/projects/{project_id}/folders/{folder_id}")
    return {t["name"]: t["id"] for t in folder.get("tests", [])}


def single_value_script(data_source_id, sql):
    """A script data set that returns one numeric value."""
    return {
        "dataSourceId": data_source_id,
        "scriptText": sql,
        "isScalar": True,
        "dataSetProcessingType": "SingleValue",
        "columns": [{"name": "VALUE", "sequence": 1, "columnType": "Numeric", "roleId": "Value"}],
    }


def build_test(name, description, source_ds_id, source_sql, target_ds_id, target_sql):
    """The standard test definition: source count on the left, target count on the right."""
    return {
        "name": name,
        "description": description,
        "folderId": TARGET_FOLDER_ID,
        "severityLevel": SEVERITY,
        "qualityDimension": QUALITY_DIMENSION,
        "testMissingResultAction": "Fail",
        "controlMissingResultAction": "Fail",
        "valueSuccessConditionType": "Value",      # numeric difference within tolerance
        "valueSuccessTolerance": TOLERANCE,
        "overallSuccessConditionType": "Value",
        "overallSuccessTolerance": 0,
        "qualityScoreMethod": "AllOrNothing",
        "treatValuesAsType": "Numeric",
        "testDataSetColumnSelection": "Position",  # match by position, so the SQL alias does not matter
        "testDataSetColumnMappingMethod": "Automatic",
        "resultsAreOrdered": False,
        "onlyKeepFailures": False,
        "abortAfterFailures": False,
        "abortAfterRows": False,
        "purgeResults": False,
        "testDataSet": single_value_script(source_ds_id, source_sql),
        "controlDataSet": single_value_script(target_ds_id, target_sql),
    }


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main():
    rows = pd.read_excel(EXCEL_FILE, sheet_name=SHEET_NAME, dtype=str).fillna("")
    rows.columns = [c.strip() for c in rows.columns]
    required = ["Test_Name", "Source_SQL", "Target_SQL"]
    missing = [c for c in required if c not in rows.columns]
    if missing:
        raise SystemExit(f"{EXCEL_FILE} is missing column(s): {missing}. Found: {list(rows.columns)}")

    ds_ids = data_source_ids_by_name()
    try:
        source_ds_id = ds_ids[SOURCE_DATA_SOURCE.strip().lower()]
        target_ds_id = ds_ids[TARGET_DATA_SOURCE.strip().lower()]
    except KeyError as missing_name:
        raise SystemExit(f"Data source {missing_name} not found. Available: {sorted(ds_ids)}")

    existing = existing_tests_in_folder(PROJECT_ID, TARGET_FOLDER_ID) if UPDATE_EXISTING else {}

    created = updated = skipped = 0
    for i, row in rows.iterrows():
        name = row["Test_Name"].strip()
        source_sql = row["Source_SQL"].strip()
        target_sql = row["Target_SQL"].strip()
        if not (name and source_sql and target_sql):
            print(f"row {i + 2}: skipped (blank name or SQL)")
            skipped += 1
            continue
        description = row["Description"].strip() if "Description" in rows.columns else ""
        body = build_test(name, description, source_ds_id, source_sql, target_ds_id, target_sql)

        if name in existing:
            call("PUT", f"/Authoring/projects/{PROJECT_ID}/tests/{existing[name]}", json=body)
            print(f"row {i + 2}: updated  {name} (test {existing[name]})")
            updated += 1
        else:
            result = call("POST", f"/Authoring/projects/{PROJECT_ID}/tests", json=body)
            print(f"row {i + 2}: created  {name} (test {result.get('testId')})")
            created += 1

    print(f"\nDone. {created} created, {updated} updated, {skipped} skipped, in folder {TARGET_FOLDER_ID}.")


if __name__ == "__main__":
    if "<" in VALIDATAR_API_URL or "<" in API_TOKEN or not PROJECT_ID or not TARGET_FOLDER_ID:
        sys.exit("Set VALIDATAR_API_URL, API_TOKEN, PROJECT_ID and TARGET_FOLDER_ID at the top of the script first.")
    main()

Script 2: data sources named on each row

Save as create_tests_from_excel_dynamic_sources.py. The workbook gains Source_DataSource and Target_DataSource columns; the two data source placeholders disappear from the top of the script. Rows that name a data source Validatar does not have are skipped and reported, and the run finishes with the list of names it does know.

"""Create one Validatar standard test per row of an Excel workbook, with the
source and target data sources named on each row.

Same behaviour as create_tests_from_excel.py, except that every row carries the
names of the data sources its two SQL statements should run against. That lets
one workbook reconcile several systems: a row can compare an Oracle count with a
Snowflake count, the next row a SQL Server count with the same Snowflake table.

Workbook layout (first sheet, header row):

    Test_Name | Source_DataSource | Source_SQL | Target_DataSource | Target_SQL | Description (optional)

Data source names must match the names shown in Validatar's Catalog, case aside.

Requirements:  pip install requests pandas openpyxl
API token:     Validatar > your user name > Tokens > New Token, with the Authoring
               scope and access to the target project. Keep it out of source control.
"""
import os
import sys

import pandas as pd
import requests

# ---------------------------------------------------------------------------
# Placeholders. Edit these, or set the matching environment variable.
# ---------------------------------------------------------------------------
VALIDATAR_API_URL = os.environ.get("VALIDATAR_API_URL", "https://<your-instance>.cloud.validatar.com")  # no trailing slash
API_TOKEN = os.environ.get("VALIDATAR_API_TOKEN", "<paste your user token here>")

PROJECT_ID = int(os.environ.get("VALIDATAR_PROJECT_ID", "0"))          # from the project URL: /projects/<id>
TARGET_FOLDER_ID = int(os.environ.get("VALIDATAR_FOLDER_ID", "0"))     # test folder that receives the tests: /folders/<id>

EXCEL_FILE = os.environ.get("TESTS_EXCEL_FILE", "tests.xlsx")
SHEET_NAME = 0                     # first sheet; use a name to pick another

# Defaults applied to every test created.
SEVERITY = "High"                  # Critical, High, Medium, Low, Informational
QUALITY_DIMENSION = "Completeness" # Accuracy, Completeness, Consistency, Timeliness, Uniqueness, Validity
TOLERANCE = 0                      # allowed difference between the two numbers
UPDATE_EXISTING = True             # a test with the same name in the folder is updated, not duplicated

# ---------------------------------------------------------------------------
# API helpers
# ---------------------------------------------------------------------------
API = f"{VALIDATAR_API_URL.rstrip('/')}/core/api/v1"
HEADERS = {"x-val-api-token": API_TOKEN, "Accept": "application/json"}


def call(method, path, **kwargs):
    """One HTTP call. Raises with the response body on any error status."""
    response = requests.request(method, f"{API}{path}", headers=HEADERS, timeout=60, **kwargs)
    if response.status_code >= 400:
        raise SystemExit(f"{method} {path} -> HTTP {response.status_code}\n{response.text[:1000]}")
    return response.json() if response.content else {}


def data_source_ids_by_name():
    """Map data source name (lower-cased) -> id, from the catalog."""
    listing = call("GET", "/Catalog/data-sources")
    return {ds["dataSourceName"].strip().lower(): ds["dataSourceId"] for ds in listing.get("dataSources", [])}


def existing_tests_in_folder(project_id, folder_id):
    """Map test name -> test id for tests already in the folder."""
    folder = call("GET", f"/Authoring/projects/{project_id}/folders/{folder_id}")
    return {t["name"]: t["id"] for t in folder.get("tests", [])}


def single_value_script(data_source_id, sql):
    """A script data set that returns one numeric value."""
    return {
        "dataSourceId": data_source_id,
        "scriptText": sql,
        "isScalar": True,
        "dataSetProcessingType": "SingleValue",
        "columns": [{"name": "VALUE", "sequence": 1, "columnType": "Numeric", "roleId": "Value"}],
    }


def build_test(name, description, source_ds_id, source_sql, target_ds_id, target_sql):
    """The standard test definition: source count on the left, target count on the right."""
    return {
        "name": name,
        "description": description,
        "folderId": TARGET_FOLDER_ID,
        "severityLevel": SEVERITY,
        "qualityDimension": QUALITY_DIMENSION,
        "testMissingResultAction": "Fail",
        "controlMissingResultAction": "Fail",
        "valueSuccessConditionType": "Value",      # numeric difference within tolerance
        "valueSuccessTolerance": TOLERANCE,
        "overallSuccessConditionType": "Value",
        "overallSuccessTolerance": 0,
        "qualityScoreMethod": "AllOrNothing",
        "treatValuesAsType": "Numeric",
        "testDataSetColumnSelection": "Position",  # match by position, so the SQL alias does not matter
        "testDataSetColumnMappingMethod": "Automatic",
        "resultsAreOrdered": False,
        "onlyKeepFailures": False,
        "abortAfterFailures": False,
        "abortAfterRows": False,
        "purgeResults": False,
        "testDataSet": single_value_script(source_ds_id, source_sql),
        "controlDataSet": single_value_script(target_ds_id, target_sql),
    }


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main():
    rows = pd.read_excel(EXCEL_FILE, sheet_name=SHEET_NAME, dtype=str).fillna("")
    rows.columns = [c.strip() for c in rows.columns]
    required = ["Test_Name", "Source_DataSource", "Source_SQL", "Target_DataSource", "Target_SQL"]
    missing = [c for c in required if c not in rows.columns]
    if missing:
        raise SystemExit(f"{EXCEL_FILE} is missing column(s): {missing}. Found: {list(rows.columns)}")

    ds_ids = data_source_ids_by_name()
    existing = existing_tests_in_folder(PROJECT_ID, TARGET_FOLDER_ID) if UPDATE_EXISTING else {}

    created = updated = skipped = 0
    for i, row in rows.iterrows():
        name = row["Test_Name"].strip()
        source_sql = row["Source_SQL"].strip()
        target_sql = row["Target_SQL"].strip()
        source_ds = row["Source_DataSource"].strip().lower()
        target_ds = row["Target_DataSource"].strip().lower()
        if not (name and source_sql and target_sql and source_ds and target_ds):
            print(f"row {i + 2}: skipped (blank name, SQL or data source)")
            skipped += 1
            continue
        unknown = [n for n in (source_ds, target_ds) if n not in ds_ids]
        if unknown:
            print(f"row {i + 2}: skipped, data source not found: {unknown}")
            skipped += 1
            continue
        description = row["Description"].strip() if "Description" in rows.columns else ""
        body = build_test(name, description, ds_ids[source_ds], source_sql, ds_ids[target_ds], target_sql)

        if name in existing:
            call("PUT", f"/Authoring/projects/{PROJECT_ID}/tests/{existing[name]}", json=body)
            print(f"row {i + 2}: updated  {name} (test {existing[name]})")
            updated += 1
        else:
            result = call("POST", f"/Authoring/projects/{PROJECT_ID}/tests", json=body)
            print(f"row {i + 2}: created  {name} (test {result.get('testId')})")
            created += 1

    print(f"\nDone. {created} created, {updated} updated, {skipped} skipped, in folder {TARGET_FOLDER_ID}.")
    if skipped:
        print(f"Known data sources: {sorted(ds_ids)}")


if __name__ == "__main__":
    if "<" in VALIDATAR_API_URL or "<" in API_TOKEN or not PROJECT_ID or not TARGET_FOLDER_ID:
        sys.exit("Set VALIDATAR_API_URL, API_TOKEN, PROJECT_ID and TARGET_FOLDER_ID at the top of the script first.")
    main()

Run it

pip install requests pandas openpyxl
python create_tests_from_excel.py

Output from a six-row workbook against the Snowflake sample warehouse:

row 2: created  Customer row count: staging vs dimension (test 1434)
row 3: created  Product row count: staging vs dimension (test 1435)
row 4: created  Sales row count: staging vs fact (test 1436)
row 5: created  Sales amount: staging vs fact (test 1437)
row 6: created  Policy row count: raw vs dimension (test 1438)
row 7: created  Claim row count: raw vs dimension (test 1439)

Done. 6 created, 0 updated, 0 skipped, in folder 134.

Run it again without changing the workbook and every line reads updated instead of created. The tests keep their ids, their history, and their place in any job.

The scripts do not run the tests. Open the folder in Validatar and run it, add the folder to a job, or execute through the API as shown in Generic REST Orchestration. On the sample warehouse the first five rows pass and the claim row fails: RAW.CLAIM has 54,339 rows and STAR_SCHEMA.DIM_CLAIM has 53,313.

What each test looks like

Open any generated test in Validatar and you will find a plain standard test:

Anything you change in Validatar afterwards is overwritten the next time the script updates that test, because the workbook is the source of truth. Make durable changes in the workbook, or set UPDATE_EXISTING = False and manage the tests in Validatar from then on.

Adapting it

Per-row severity or tolerance. Add a Severity or Tolerance column and read it where build_test is called: severity = row.get("Severity", "").strip() or SEVERITY. Pass it into the body in place of the constant.

Coverage in trust scores. The tests carry no metadata links, so they do not count as coverage of the tables they compare. To link a test to a catalogued table, add to each data set:

"metaLinks": [{"key": "STAR_SCHEMA.DIM_CUSTOMER", "includeForCoverage": True}]

The key is schema.table (or schema.table.column) as it appears in the catalog of that data source. A key that does not resolve is reported when the test runs.

Other comparisons. A single number on each side covers counts, sums, min and max dates cast to numbers, and distinct counts. For row-by-row comparison of two result sets, change both data sets to "dataSetProcessingType": "Table", set "isScalar": False, and declare the key and value columns. Configuring a Test describes the column roles.

CSV instead of Excel. Replace pd.read_excel(...) with pd.read_csv(EXCEL_FILE, dtype=str). Nothing else changes.

Deleting tests. Removing a row does not remove its test. Delete it in Validatar, or add a pass that calls DELETE /Authoring/projects/{project}/tests/{id} for names present in the folder but absent from the workbook.

Troubleshooting

HTTP 401 or 403 on the first call. The token is wrong, expired, or lacks the Authoring scope or access to this project. Regenerate it under your user name > Tokens.

HTTP 403 on /Catalog/data-sources only. The token can author tests but not read the catalog. Give it all scopes, or hardcode the ids: replace data_source_ids_by_name() with a dictionary of lower-cased name to id.

Data source ... not found. The name does not match the Catalog. The message lists every name the token can see; copy from there.

HTTP 404 on the folder. TARGET_FOLDER_ID is not a test folder in PROJECT_ID. Check both numbers in the folder's URL. Job folders have ids too and will 404 here.

HTTP 400 on create. The response body names the field. The usual causes are a severityLevel or qualityDimension spelled differently from the values in your instance's settings.

Tests create fine but error when run. The API does not execute the SQL at creation time. A typo, a missing grant, or a statement that returns two columns surfaces as an execution error on the test's results page.

Related

• Generic REST Orchestration
• User Tokens
• Configuring a Test
• Generating Child Tests from an Import File
• Template Test Examples: Import File