#!/usr/bin/env python3

"""Run the Spark→Feldera tests end to end, in a single pass.

For each test this script does everything in one go — no separate compile and

run phases:

1. translate the Spark SQL to Feldera SQL (felderize),

2. compile + start a Feldera pipeline from it,

3. insert the test data and query the final view,

4. compare the rows against the recorded ``.result``.

It is meant to be run by external users, so it first checks that everything it

needs is in place: the Anthropic API key (translation), the Feldera SQL

compiler, Docker, and a reachable Feldera instance.

Usage:

python3 run_tests.py # run every test under this folder

python3 run_tests.py --dir cast # only one subdirectory

python3 run_tests.py --file cast_001 # only one test

python3 run_tests.py --url http://localhost:8080

python3 run_tests.py --profile dev # fastest Feldera compile (default)

python3 run_tests.py --results-dir out/ # write per-test JSON + summary.json

python3 run_tests.py --check-only # only verify prerequisites

See README.md in this folder for setup details.

"""

from __future__ import annotations

import argparse

import ast

import contextlib

import io

import json

import logging

import re

import shutil

import subprocess

import sys

import urllib.error

import urllib.request

from pathlib import Path

# This file lives at <felderize>/tests/e2e/run_tests.py; fixtures are in ./fixtures.

_THIS_DIR = Path(__file__).resolve().parent

_FIXTURES_DIR = _THIS_DIR / "fixtures"

_FELDERIZE_ROOT = _THIS_DIR.parents[1] # python/felderize

_MONOREPO_PYTHON = _THIS_DIR.parents[2] # python/ (holds the in-repo feldera SDK)

_PIPELINE_PREFIX = "felderize-e2e-"

_DEFAULT_URL = "http://localhost:8080"

# Robustness fixtures (malformed inputs / unsupported features) have no `.result`

# and are exercised by run_behavior.py, not this result-comparison runner.

_BEHAVIOR_DIRS = {"malformed", "multiple_views_unsupported"}

# ---------------------------------------------------------------------------

# Prerequisite checks

# ---------------------------------------------------------------------------

def _check_api_key(config) -> list[str]:

if not config.api_key:

return [

"ANTHROPIC_API_KEY is not set. Translation needs it — put it in a .env "

"file or export it in your shell."

]

return []

def _check_compiler(config) -> list[str]:

compiler = config.feldera_compiler

if not compiler:

return [

"Feldera SQL compiler not configured. Set FELDERA_COMPILER in .env, or "

"run `felderize download-compiler` and use the path it prints."

]

path = Path(compiler).expanduser()

if not path.is_file():

return [

f"Feldera compiler not found at {path}. "

"Run `felderize download-compiler` to fetch it."

]

if path.suffix == ".jar" and not shutil.which("java"):

return ["Java not found — install Java 19–21 to run the compiler JAR."]

# Reject a compiler older than felderize's minimum (when the version is

# parseable from the filename) — older releases lack SQL features felderize

# relies on, so validation against them is unreliable.

try:

from felderize.install_feldera_sql_compiler import (

MINIMUM_COMPILER_VERSION,

is_supported_version,

jar_version,

)

tag = jar_version(str(path))

if tag and not is_supported_version(tag):

return [

f"Feldera compiler {tag} is older than the minimum supported "

f"{MINIMUM_COMPILER_VERSION}. Run `felderize download-compiler` to update."

]

except ModuleNotFoundError:

pass # felderize not importable here; the SDK prerequisite will flag the env

return []

def _check_docker() -> list[str]:

if not shutil.which("docker"):

return [

"Docker not found. Install Docker — Feldera runs as a Docker container."

]

try:

result = subprocess.run(

["docker", "info"], capture_output=True, timeout=20, text=True

)

except Exception as e: # noqa: BLE001 - report any failure to the user

return [f"Could not run `docker info`: {e}"]

if result.returncode != 0:

return [

"Docker is installed but its daemon is not running. Start Docker Desktop / the docker service."

]

return []

def _check_feldera(url: str) -> list[str]:

"""A reachable server (any HTTP response) is good enough; refused/timeout is not."""

try:

urllib.request.urlopen(url, timeout=5)

return []

except urllib.error.HTTPError:

return [] # server answered (some status) — it is up

except Exception: # noqa: BLE001 - connection refused, DNS, timeout, ...

return [

f"No Feldera instance reachable at {url}. Start one with Docker, e.g.:\n"

" docker run --rm -it -p 8080:8080 ghcr.io/feldera/pipeline-manager:latest\n"

" then re-run (see https://docs.feldera.com/get-started/)."

]

def _check_feldera_sdk() -> list[str]:

"""Import the full SDK client chain the runner uses at run time.

Done here so a missing dependency fails fast with one clear message, instead

of erroring identically on every test (a bare system python that lacks the

SDK's deps — pyarrow, requests, … — is the usual cause).

"""

try:

_feldera_imports()

return []

except Exception as e: # noqa: BLE001 - any import failure is actionable

return [

f"Feldera Python SDK is not fully importable ({type(e).__name__}: {e}).\n"

" Run inside the project venv — `.venv/bin/python tests/e2e/run_tests.py`\n"

" or `source .venv/bin/activate` first — or `pip install feldera`\n"

" into the interpreter you are using. A bare system python is the usual cause."

]

def check_prerequisites(config, url: str) -> bool:

"""Print a checklist; return True only if every prerequisite is satisfied."""

checks = [

("Anthropic API key", _check_api_key(config)),

("Feldera SQL compiler", _check_compiler(config)),

("Feldera Python SDK", _check_feldera_sdk()),

("Docker", _check_docker()),

(f"Feldera instance ({url})", _check_feldera(url)),

]

ok = True

print("Checking prerequisites:")

for label, problems in checks:

if problems:

ok = False

print(f" ✗ {label}")

for p in problems:

print(f" {p}")

else:

print(f" ✓ {label}")

print()

return ok

# ---------------------------------------------------------------------------

# SQL + result helpers

# ---------------------------------------------------------------------------

# CREATE VIEW with any modifier Feldera/Spark may carry. Group 1 captures the

# modifier (LOCAL / MATERIALIZED / TEMP[ORARY]) when present, group 2 the name.

# A LOCAL view is deliberately not observable (e.g. an INTERVAL output column,

# which a connector-exposed view cannot have), so it is detected, not rewritten.

_CREATE_VIEW_RE = re.compile(

r"CREATE\s+(?:OR\s+REPLACE\s+)?"

r"((?:LOCAL|MATERIALIZED|TEMP(?:ORARY)?)\s+)?VIEW\s+([\w`\".-]+)",

re.IGNORECASE,

)

def _output_view(query_sql: str) -> tuple[str | None, bool]:

"""Return (name, is_local) for the LAST CREATE VIEW — the final output view.

`is_local` is True when that view is a LOCAL view, which cannot be queried.

Returns (None, False) when no CREATE VIEW is present.

"""

code = "\n".join(

ln for ln in query_sql.splitlines() if not ln.lstrip().startswith("--")

)

matches = _CREATE_VIEW_RE.findall(code)

if not matches:

return None, False

modifier, name = matches[-1]

return name.strip('`"'), "LOCAL" in modifier.upper()

def _extract_view_name(query_sql: str) -> str | None:

"""Return the name of the final output view (see `_output_view`)."""

return _output_view(query_sql)[0]

# Make views materialized so they can be queried — but leave LOCAL views alone:

# they are intentionally non-observable and forcing MATERIALIZED would break

# compilation (e.g. INTERVAL columns are illegal in a connector-exposed view).

_MATERIALIZE_RE = re.compile(

r"CREATE\s+(?:OR\s+REPLACE\s+)?(?:(?:MATERIALIZED|TEMP(?:ORARY)?)\s+)?VIEW",

re.IGNORECASE,

)

def _normalize_feldera_sql(feldera_schema: str, feldera_query: str) -> str:

"""Combine schema + query, making non-local views materialized so they can be queried."""

query = _MATERIALIZE_RE.sub("CREATE MATERIALIZED VIEW", feldera_query.strip())

return feldera_schema.strip() + "\n\n" + query

def _parse_spark_result(result_text: str) -> list[list[str]]:

rows = []

for line in result_text.strip().splitlines():

if line.strip():

rows.append(line.split("\t"))

return rows

def _rows_match(

feldera_rows: list[dict], spark_rows: list[list[str]]

) -> tuple[bool, str]:

"""Compare Feldera output against the recorded result, order-independent."""

if not spark_rows and not feldera_rows:

return True, "both empty"

def _norm(v: str) -> str:

v = v.strip()

# Booleans: Feldera (Python) renders True/False; Spark renders true/false.

if v.lower() in ("true", "false"):

return v.lower()

row_m = re.match(r"Row\((.+)\)$", v)

dict_m = re.match(r"\{(.+)\}$", v)

if row_m or dict_m:

inner = row_m.group(1) if row_m else dict_m.group(1)

try:

pairs = re.findall(r"[\'\"]?(\w+)[\'\"]?\s*[=:]\s*([^,]+)", inner)

return str(sorted((k.strip(), _norm(vv.strip())) for k, vv in pairs))

except Exception: # noqa: BLE001

pass

if v.startswith("[") and v.endswith("]"):

try:

v_clean = re.sub(r"Decimal\('([^']+)'\)", r"\1", v)

lst = ast.literal_eval(v_clean)

if isinstance(lst, list):

return str(sorted(_norm(str(x)) for x in lst))

except Exception: # noqa: BLE001

pass

v_ts = v.replace("T", " ")

if v_ts != v and re.match(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}", v_ts):

return v_ts

try:

f = float(v)

return str(int(f)) if f == int(f) else f"{f:.10g}"

except (ValueError, OverflowError):

pass

return v

def row_to_key(row: dict) -> tuple:

return tuple(_norm(str(v) if v is not None else "None") for v in row.values())

feldera_keys = sorted(row_to_key(r) for r in feldera_rows)

spark_keys = sorted(tuple(_norm(c) for c in r) for r in spark_rows)

if len(feldera_keys) != len(spark_keys):

return (

False,

f"row count mismatch: feldera={len(feldera_keys)} expected={len(spark_keys)}",

)

mismatches = [(f, s) for f, s in zip(feldera_keys, spark_keys) if f != s]

if mismatches:

f, s = mismatches[0]

return False, f"{len(mismatches)} row(s) differ, e.g. got {f} vs expected {s}"

return True, f"{len(feldera_keys)} rows match"

# ---------------------------------------------------------------------------

# Feldera execution

# ---------------------------------------------------------------------------

def _feldera_imports():

"""Import the Feldera Python SDK (installed, or from the monorepo as fallback)."""

try:

from feldera.rest.feldera_client import FelderaClient

from feldera.pipeline_builder import PipelineBuilder

from feldera.enums import CompilationProfile, PipelineStatus

except ModuleNotFoundError:

sys.path.insert(0, str(_MONOREPO_PYTHON))

from feldera.rest.feldera_client import FelderaClient

from feldera.pipeline_builder import PipelineBuilder

from feldera.enums import CompilationProfile, PipelineStatus

return FelderaClient, PipelineBuilder, CompilationProfile, PipelineStatus

def _to_feldera_dml(stmt: str) -> str:

"""Convert Spark collection literals in a data statement to Feldera syntax.

Feldera's ad-hoc INSERT (DataFusion) rejects Spark's `array(...)`/`map(...)`

literals; sqlglot transpiles them to `ARRAY[...]` / `MAP(...)`. Only touched

when such a literal is present; falls back to the original on any failure.

"""

if not re.search(r"\b(?:array|map)\s*\(", stmt, re.IGNORECASE):

return stmt

try:

import sqlglot

return sqlglot.transpile(stmt, read="spark", write="postgres")[0]

except Exception: # noqa: BLE001 - keep the original statement on any failure

return stmt

def run_on_feldera(

test_id: str,

feldera_sql: str,

view_name: str,

data_sql: str,

url: str,

profile: str = "dev",

) -> list[dict]:

"""Compile + start a pipeline, insert data, query the view, then stop it.

profile is the Feldera compilation profile ("dev" is fastest to compile).

"""

import sqlparse

FelderaClient, PipelineBuilder, CompilationProfile, PipelineStatus = (

_feldera_imports()

)

pipeline_name = _PIPELINE_PREFIX + test_id.replace("_", "-")

client = FelderaClient(url)

pipeline = None

try:

pipeline = PipelineBuilder(

client,

name=pipeline_name,

sql=feldera_sql,

compilation_profile=CompilationProfile(profile),

).create_or_replace(wait=True)

pipeline.start()

pipeline.wait_for_status(PipelineStatus.RUNNING, timeout=120)

for stmt in sqlparse.split(data_sql):

stmt = stmt.strip().rstrip(";")

if not stmt:

continue

try:

pipeline.execute(_to_feldera_dml(stmt))

except Exception as e: # noqa: BLE001

# Tolerate inserts into tables trimmed from the schema (they are

# not referenced by the query, so their data is irrelevant).

print(

f" insert skipped: {str(e).splitlines()[0][:90]}",

file=sys.stderr,

)

pipeline.pause()

pipeline.wait_for_status(PipelineStatus.PAUSED, timeout=60)

return list(pipeline.query(f'SELECT * FROM "{view_name}"'))

finally:

# Always remove the pipeline so they do not accumulate across tests —

# even when compilation or startup failed above.

_cleanup_pipeline(client, pipeline, pipeline_name)

def _cleanup_pipeline(client, pipeline, name) -> None:

"""Best-effort: stop, clear storage, and delete the pipeline; never raises.

Feldera refuses to delete a pipeline until it is stopped and its storage is

cleared, so we use pipeline.delete(clear_storage=True). When the pipeline

object is unavailable (e.g. compilation failed), fetch a handle by name.

"""

if pipeline is None:

try:

from feldera.pipeline import Pipeline

pipeline = Pipeline.get(name, client)

except Exception: # noqa: BLE001 - nothing to clean up

return

if pipeline is None:

return

try:

pipeline.stop(force=True)

except Exception: # noqa: BLE001 - cleanup must not mask the real result

pass

try:

pipeline.delete(clear_storage=True)

except Exception: # noqa: BLE001

pass

# ---------------------------------------------------------------------------

# Test collection + single-pass execution

# ---------------------------------------------------------------------------

def _collect_tests(

subdir_filter: str | None,

file_filter: str | None,

):

tests = []

subdirs = sorted(

d

for d in _FIXTURES_DIR.iterdir()

if d.is_dir() and d.name not in _BEHAVIOR_DIRS

)

if subdir_filter:

subdirs = [d for d in subdirs if d.name == subdir_filter]

for subdir in subdirs:

schema_files = sorted(subdir.glob("*.schema.sql")) + sorted(

subdir.glob("*_schema.sql")

)

for schema_path in schema_files:

if schema_path.name.endswith(".schema.sql"):

name = schema_path.name[: -len(".schema.sql")]

query_path = subdir / f"{name}.query.sql"

data_path = subdir / f"{name}.data.sql"

result_path = subdir / f"{name}.result"

else:

name = schema_path.name[: -len("_schema.sql")]

query_path = subdir / f"{name}_query.sql"

data_path = subdir / f"{name}_data.sql"

result_path = subdir / f"{name}_result"

if file_filter and name != file_filter:

continue

if not query_path.exists():

continue

tests.append(

(subdir.name, name, schema_path, query_path, data_path, result_path)

)

return tests

def run_one_test(

subdir,

test_id,

schema_path,

query_path,

data_path,

result_path,

config,

url,

profile,

) -> dict:

"""Translate, compile, run, and compare one test — the whole pipeline.

Returns a JSON-serializable record describing the outcome and the artifacts.

"""

from felderize.translator import translate_spark_to_feldera

record: dict = {

"subdir": subdir,

"test_id": test_id,

"outcome": "error",

"detail": "",

}

schema_sql = schema_path.read_text() if schema_path.exists() else ""

query_sql = query_path.read_text()

try:

result = translate_spark_to_feldera(

schema_sql, query_sql, config, validate=True

)

except Exception as e: # noqa: BLE001 - a transient LLM/network error must not abort the run

record["detail"] = f"translation error: {e}"

return record

record["translation_status"] = result.status.value

record["feldera_schema"] = result.feldera_schema

record["feldera_query"] = result.feldera_query

if result.unsupported:

record["unsupported"] = result.unsupported

if not result.feldera_query.strip():

record["outcome"] = "skipped"

record["detail"] = f"no query translated ({result.status.value})"

return record

view, is_local = _output_view(result.feldera_query)

if not view:

record["detail"] = "could not find the output view name"

return record

if is_local:

# A LOCAL view is not observable, so its result cannot be compared. This

# is a genuine Feldera limitation (e.g. an INTERVAL output column), not a

# translation failure — record it as skipped rather than an error.

record["outcome"] = "skipped"

record["detail"] = (

"final output is a LOCAL VIEW (not observable, e.g. INTERVAL output)"

)

record["view"] = view

return record

record["view"] = view

feldera_sql = _normalize_feldera_sql(result.feldera_schema, result.feldera_query)

data_sql = data_path.read_text() if data_path and data_path.exists() else ""

expected = _parse_spark_result(

result_path.read_text() if result_path.exists() else ""

)

record["expected_row_count"] = len(expected)

try:

rows = run_on_feldera(test_id, feldera_sql, view, data_sql, url, profile)

except Exception as e: # noqa: BLE001 - surface any Feldera error per test

record["detail"] = str(e)

return record

record["feldera_row_count"] = len(rows)

match, detail = _rows_match(rows, expected)

record["outcome"] = "pass" if match else "fail"

record["detail"] = detail

return record

# ---------------------------------------------------------------------------

# Summary

# ---------------------------------------------------------------------------

_SUMMARY_FIELDS = ("subdir", "test_id", "outcome", "detail", "translation_status")

_OUTCOMES = ("pass", "fail", "error", "skipped")

# ---------------------------------------------------------------------------

# Console output

# ---------------------------------------------------------------------------

_LABELS = {"pass": "PASS", "fail": "FAIL", "error": "ERROR", "skipped": "SKIP"}

_COLORS = {"pass": "32", "fail": "31", "error": "31", "skipped": "33"} # ANSI

def _color(text: str, outcome: str, enabled: bool) -> str:

code = _COLORS.get(outcome)

return f"\033[{code}m{text}\033[0m" if (enabled and code) else text

def _short_detail(detail: str | None, width: int = 72) -> str:

"""Collapse a possibly multi-line detail to one trimmed line."""

one = " ".join((detail or "").split())

return one if len(one) <= width else one[: width - 1] + "…"

def _build_summary(records: list[dict], profile: str, url: str) -> dict:

"""Assemble a summary document from per-test records."""

counts = {o: 0 for o in _OUTCOMES}

for r in records:

outcome = r.get("outcome")

if outcome in counts:

counts[outcome] += 1

return {

"total": len(records),

"counts": counts,

"profile": profile,

"url": url,

"results": [{k: r.get(k) for k in _SUMMARY_FIELDS} for r in records],

}

def _merge_summaries(results_dir: Path, profile: str, url: str) -> dict:

"""Rebuild the canonical summary by scanning every per-test JSON on disk.

The per-test JSON files are the source of truth, so this reconstructs a

complete summary regardless of how many shards produced them. Records are

sorted by (subdir, test_id) for a stable, diff-friendly ordering.

"""

records: list[dict] = []

for path in results_dir.glob("*/*.json"):

try:

records.append(json.loads(path.read_text()))

except (OSError, json.JSONDecodeError):

continue # skip unreadable/partial files rather than abort the merge

records.sort(key=lambda r: (r.get("subdir", ""), r.get("test_id", "")))

return _build_summary(records, profile, url)

# ---------------------------------------------------------------------------

# Main

# ---------------------------------------------------------------------------

def main() -> None:

parser = argparse.ArgumentParser(

description="Run Spark→Feldera tests end to end (translate + compile + run + compare)."

)

parser.add_argument(

"--dir", metavar="SUBDIR", help="Only this subdirectory (e.g. cast)"

)

parser.add_argument("--file", metavar="NAME", help="Only this test (e.g. cast_001)")

parser.add_argument(

"--url", default=_DEFAULT_URL, help=f"Feldera API URL (default: {_DEFAULT_URL})"

)

parser.add_argument(

"--profile",

default="dev",

choices=["dev", "unoptimized", "optimized"],

help="Feldera compilation profile (default: dev — fastest to compile)",

)

parser.add_argument(

"--results-dir",

metavar="DIR",

default=None,

help="Write a JSON record per test (under DIR/<subdir>/) plus a summary "

"(summary.json, or summary.shard-K-of-M.json under --shard)",

)

parser.add_argument(

"--resume",

action="store_true",

help="Skip tests already recorded under --results-dir (continue a prior run)",

)

parser.add_argument(

"--shard",

metavar="K/M",

help="Run only shard K of M (0-based); for parallel runs across processes",

)

parser.add_argument(

"--merge-summaries",

action="store_true",

help="Rebuild summary.json from the per-test JSONs under --results-dir, "

"then exit (use after sharded runs complete)",

)

parser.add_argument(

"--check-only", action="store_true", help="Only verify prerequisites, then exit"

)

parser.add_argument(

"--verbose",

action="store_true",

help="Show the translator/LLM/compiler chatter (suppressed by default)",

)

args = parser.parse_args()

# Quiet the SDK's client/server version-mismatch warning (noise per pipeline).

logging.getLogger("feldera").setLevel(logging.ERROR)

# Merging only reads files already on disk — no Feldera, no prereq checks.

if args.merge_summaries:

if not args.results_dir:

print("--merge-summaries requires --results-dir")

sys.exit(2)

results_dir = Path(args.results_dir).expanduser().resolve()

summary = _merge_summaries(results_dir, args.profile, args.url)

(results_dir / "summary.json").write_text(json.dumps(summary, indent=2))

c = summary["counts"]

print(f"Merged {summary['total']} test(s) into {results_dir}/summary.json")

print(

f"PASS: {c['pass']} FAIL: {c['fail']} "

f"ERROR: {c['error']} SKIPPED: {c['skipped']}"

)

return

# felderize is importable when installed; fall back to the in-repo source.

try:

from felderize.config import Config

except ModuleNotFoundError:

sys.path.insert(0, str(_FELDERIZE_ROOT / "felderize"))

from felderize.config import Config

config = Config.from_env()

if not check_prerequisites(config, args.url):

print("Prerequisites not met — fix the items marked ✗ above and re-run.")

sys.exit(2)

if args.check_only:

print("All prerequisites satisfied.")

return

tests = _collect_tests(args.dir, args.file)

if not tests:

print("No tests found.")

sys.exit(1)

# Sharding for parallel runs: each process takes a disjoint stride of tests.

if args.shard:

k, m = (int(x) for x in args.shard.split("/"))

tests = tests[k::m]

print(f"[shard {k}/{m}] handling {len(tests)} tests")

results_dir = (

Path(args.results_dir).expanduser().resolve() if args.results_dir else None

)

total = len(tests)

counts = {"pass": 0, "fail": 0, "error": 0, "skipped": 0}

records: list[dict] = []

use_color = sys.stdout.isatty()

idx_w = len(str(total))

print(

f"Running {total} test(s) [profile={args.profile}"

+ (f", results-dir={results_dir}" if results_dir else "")

+ "]...\n"

)

for i, (

subdir,

test_id,

schema_path,

query_path,

data_path,

result_path,

) in enumerate(tests, 1):

prefix = f"[{i:>{idx_w}}/{total}] {subdir}/{test_id}"

cached = results_dir / subdir / f"{test_id}.json" if results_dir else None

if args.resume and cached and cached.is_file():

try:

res = json.loads(cached.read_text())

counts[res["outcome"]] += 1

records.append(res)

badge = _color(_LABELS[res["outcome"]], res["outcome"], use_color)

print(f"{prefix:<58} {badge} (resumed)")

continue

except Exception: # noqa: BLE001 - fall through and re-run if the cache is unreadable

pass

print(f"{prefix:<58} ", end="", flush=True)

try:

# The translator/LLM/compiler write progress to stderr; hide it unless

# --verbose so the per-test result stays one clean line.

stderr_sink = io.StringIO() if not args.verbose else sys.stderr

with contextlib.redirect_stderr(stderr_sink):

res = run_one_test(

subdir,

test_id,

schema_path,

query_path,

data_path,

result_path,

config,

args.url,

args.profile,

)

except Exception as e: # noqa: BLE001 - backstop: never let one test abort the whole run

res = {

"subdir": subdir,

"test_id": test_id,

"outcome": "error",

"detail": f"unexpected error: {e}",

}

counts[res["outcome"]] += 1

records.append(res)

if results_dir:

out = results_dir / subdir

out.mkdir(parents=True, exist_ok=True)

(out / f"{test_id}.json").write_text(json.dumps(res, indent=2))

badge = _color(f"{_LABELS[res['outcome']]:<5}", res["outcome"], use_color)

detail = _short_detail(res.get("detail"))

print(f"{badge} {detail}" if detail else badge)

print(f"\n{'=' * 60}")

parts = [

_color(f"PASS {counts['pass']}", "pass", use_color),

_color(f"FAIL {counts['fail']}", "fail", use_color),

_color(f"ERROR {counts['error']}", "error", use_color),

_color(f"SKIP {counts['skipped']}", "skipped", use_color),

f"TOTAL {total}",

]

print(" ".join(parts))

if results_dir:

summary = _build_summary(records, args.profile, args.url)

# A sharded run covers only its own stride, so each shard writes a

# distinct file rather than clobbering the others' summary.json. Rebuild

# the canonical summary.json with --merge-summaries once all shards finish.

if args.shard:

k, m = (int(x) for x in args.shard.split("/"))

name = f"summary.shard-{k}-of-{m}.json"

else:

name = "summary.json"

(results_dir / name).write_text(json.dumps(summary, indent=2))

print(f"Results written to {results_dir}/ ({name})")

if args.shard:

print("Run with --merge-summaries to rebuild the full summary.json.")

if counts["fail"] or counts["error"]:

sys.exit(1)

if __name__ == "__main__":

main()