1#!/usr/bin/env python3
2"""Create Testmo runs from reviewable JSON filter specs.
3
4Reproduces the manual "filter cases -> create run" workflow via the API so it is repeatable.
5Pass --spec once per run; a batch fetches the folder tree and case list once, refuses --create
6unless every run is ready, and links every run to one milestone when --milestone-id or
7--milestone-name is given. Read-only unless --create is passed.
8
9Example:
10  python3 testmo_create_runs.py --spec a.json --spec b.json --milestone-name "2026.8.0 Manual Regression"
11  python3 testmo_create_runs.py --spec a.json --spec b.json --milestone-name "2026.8.0 Manual Regression" --create
12
13The API key is read from the environment and never printed,
14and never placed on a command line: argv is world-readable via `ps`. Requests go through urllib

testmo_create_runs.py on GitHub

in-process; on a network with TLS interception, where OpenSSL cannot verify the chain, they fall

back to curl with the header passed on stdin.

Filter model (all keys under "filters" are optional; a case must match every key present):

tags list of Testmo tag names or ids. Applied server-side via the /cases API

(?tags=...), then combined with any other filters below. e.g. ["oldnew"].

folder_paths list of folder paths, e.g. "Web > Password Manager". Matched against the

project folder tree; each expands to that folder AND all descendants

(unless include_subfolders is false). Paths are OR'd together.

folder_ids list of folder ids, same subtree expansion as folder_paths. OR'd with them.

include_subfolders default true. When false, folder_paths/folder_ids match that exact folder only.

test_type_ids case matches if its Test Type set intersects this list (multiselect field).

team_ids case matches if its Team set intersects this list (multiselect field).

case_state_ids case state_id must be in this list (e.g. [4] = Active).

automation_type_ids case automation type must be in this list. Use null in the list to include

cases with no automation type set.

exclude_automation_type_ids case is dropped if its automation type is in this list

(e.g. [10, 23, 24] = Automated / Automated-Android / Automated-iOS).

has_automation bool; case has_automation flag must equal this.

"""

import argparse, json, os, shutil, ssl, subprocess, sys, tempfile

from pathlib import Path

import http.client, urllib.error, urllib.parse, urllib.request

BASE = "https://bitwarden.testmo.net/api/v1"

KEY = os.environ.get("TESTMO_API_KEY")

TIMEOUT = 60

# Paging guard: project 1 holds ~13.7k cases, ~137 pages at the 100/page the API allows.

MAX_PAGES = 1000

# Set once a TLS verification failure proves this Python cannot validate the chain (see _curl_call).

_USE_CURL = False

class TestmoAPIError(Exception):

"""A Testmo API request did not return a usable JSON response.

Raised rather than exiting so callers can decide: a read failure aborts the batch, while a

failed write is recorded and the create loop carries on to report a full tally.

Messages are built from the request line and the response body only, never the API key.

"""

class _TLSVerificationError(Exception):

"""Internal: the certificate chain did not verify. Triggers the curl fallback."""

def _redact(text):

"""Strip the API key from anything before it is printed or raised."""

return text.replace(KEY, "<redacted>") if KEY else text

def _decode(method, path, raw):

try:

return json.loads(raw)

except json.JSONDecodeError as e:

# A proxy or login-wall HTML page can arrive with a 200; don't let it become a traceback.

raise TestmoAPIError(

f"{method} {path} returned a non-JSON body ({e}). First 200 characters:\n{raw[:200]!r}"

)

def _auth_hint(code):

return " (is TESTMO_API_KEY current?)" if code in (401, 403) else ""

def _unknown_outcome_hint(method):

"""A write that timed out or lost its connection after sending may still have been applied."""

if method == "GET":

return ""

return (" The request was sent, so the server may still have applied it — check the milestone "

"in Testmo before re-running.")

# curl exit codes for a failure after the request was sent: 18 partial transfer, 28 timeout,

# 52 empty reply, 56 receive error. A connect failure (6, 7) means nothing reached the server.

_CURL_AFTER_SEND_FAILURES = (18, 28, 52, 56)

class _NoRedirect(urllib.request.HTTPRedirectHandler):

"""Refuse every redirect, so a 3xx surfaces as an HTTPError instead of being followed.

urllib's default handler copies the request headers — Authorization included — onto the

redirected request, even when it points at another host. The Testmo API has no reason to

redirect, so a 3xx here means a proxy or misconfiguration, and following it would hand the key

to whatever host it names. curl, the fallback transport, does not follow redirects without -L.

"""

def redirect_request(self, req, fp, code, msg, headers, newurl):

return None

_OPENER = urllib.request.build_opener(_NoRedirect)

def _urllib_call(method, path, body):

"""Preferred transport: the Authorization header never leaves this process."""

data = json.dumps(body).encode() if body is not None else None

req = urllib.request.Request(BASE + path, data=data, method=method)

req.add_header("Authorization", f"Bearer {KEY}")

req.add_header("Accept", "application/json")

if data is not None:

req.add_header("Content-Type", "application/json")

try:

with _OPENER.open(req, timeout=TIMEOUT) as resp:

raw = resp.read().decode(errors="replace")

except urllib.error.HTTPError as e:

# e.read() is the API's error body; it echoes the request but not the auth header.

detail = e.read().decode(errors="replace").strip()

raise TestmoAPIError(

f"{method} {path} failed: HTTP {e.code} {e.reason}{_auth_hint(e.code)}\n{detail[:500]}"

)

except urllib.error.URLError as e:

if isinstance(e.reason, ssl.SSLCertVerificationError):

raise _TLSVerificationError(str(e.reason))

raise TestmoAPIError(f"{method} {path} failed: {e.reason}")

except (OSError, http.client.HTTPException) as e:

# urllib wraps only connect/send errors in URLError. A read timeout or a reset raised by

# getresponse() or resp.read() arrives bare (TimeoutError, RemoteDisconnected,

# IncompleteRead), and must still become TestmoAPIError so the create loop records

# the failure and carries on to its tally instead of dying with a traceback.

raise TestmoAPIError(

f"{method} {path} failed: {type(e).__name__}: {e}.{_unknown_outcome_hint(method)}"

)

return _decode(method, path, raw)

def _curl_call(method, path, body):

"""Fallback transport for networks running TLS interception.

A corporate proxy (Zscaler and friends) re-signs traffic with a root that is installed in the

OS trust store but is absent from certifi, and whose Basic Constraints are not marked

critical — which Python 3.13+ rejects outright, since it enables VERIFY_X509_STRICT by

default. curl uses the platform trust store and is not strict, so it succeeds where urllib

cannot.

The key is still kept out of argv: the Authorization header is fed to `curl --config -` on

stdin. Any request body goes to a 0600 temp file, referenced by path — the body is not

secret, and this keeps stdin free for the config.

*Only* the Authorization header goes through the config; every other directive is an argv

element. curl's config format is one directive per line, so a value carrying a `"` and a

newline would inject further directives onto the same config that holds the key — a second

`url =` would send the authenticated request to an attacker-chosen host. On argv each value is

one token no matter what it contains, so nothing spec-derived can reach the config parser. The

remaining interpolation is the key itself, which `_check_key` rejects if it holds a quote or a

newline.

"""

marker = "__TESTMO_HTTP_STATUS__"

argv = [

"curl", "--silent", "--show-error", "--max-time", str(TIMEOUT),

"--request", method,

"--header", "Accept: application/json",

"--write-out", f"\n{marker}%{{http_code}}",

"--url", BASE + path,

]

config = [f'header = "Authorization: Bearer {KEY}"']

body_file = None

try:

if body is not None:

fd, body_file = tempfile.mkstemp(prefix="testmo-body-", suffix=".json")

with os.fdopen(fd, "w") as fh:

json.dump(body, fh)

argv += ["--header", "Content-Type: application/json",

"--data-binary", f"@{body_file}"]

proc = subprocess.run(argv + ["--config", "-"], input="\n".join(config) + "\n",

capture_output=True, text=True)

finally:

if body_file:

os.unlink(body_file)

if proc.returncode != 0:

# Redact: a config parse error can echo the offending line back on stderr.

hint = _unknown_outcome_hint(method) if proc.returncode in _CURL_AFTER_SEND_FAILURES else ""

raise TestmoAPIError(

f"{method} {path} failed: curl exited {proc.returncode}.{hint}\n"

f"{_redact(proc.stderr.strip())}"

)

raw, found, status = proc.stdout.rpartition(marker)

if not found:

raise TestmoAPIError(f"{method} {path} failed: curl returned no status code")

try:

code = int(status.strip())

except ValueError:

raise TestmoAPIError(f"{method} {path} failed: unreadable status {status.strip()!r}")

if not 200 <= code < 300:

raise TestmoAPIError(

f"{method} {path} failed: HTTP {code}{_auth_hint(code)}\n{raw.strip()[:500]}"

)

return _decode(method, path, raw)

def _check_key():

"""Exit unless TESTMO_API_KEY is set and safe to interpolate into a curl config line."""

if not KEY:

sys.exit("TESTMO_API_KEY not set")

# A backslash cannot inject a directive (that needs a raw newline), but curl processes escapes

# inside a quoted config value, so it would silently alter the header that is sent.

if any(c in KEY for c in '"\\\r\n'):

sys.exit("TESTMO_API_KEY contains a quote, a backslash, or a newline, which cannot be carried "

"safely in an HTTP header. Re-copy the key from Testmo (no surrounding quotes).")

def call(method, path, body=None):

"""Issue a Testmo API request and return the decoded JSON body.

Prefers urllib, so the Authorization header stays in this process — passing it as a curl

argument would expose the key to any local user for the life of the request (`ps auxww`,

/proc/<pid>/cmdline). Falls back to curl only when the TLS chain will not verify, which on a

corporate network is not a fixable condition; that path keeps the key off argv too.

Any non-2xx status raises, so an expired key (401) or a rejected write (403) surfaces as an

error instead of an empty case list or a run "created" with id None.

"""

global _USE_CURL

_check_key()

if _USE_CURL:

return _curl_call(method, path, body)

try:

return _urllib_call(method, path, body)

except _TLSVerificationError as e:

if not shutil.which("curl"):

raise TestmoAPIError(

f"{method} {path} failed: TLS certificate verification failed ({e}), and curl is "

f"not on PATH to fall back to.\nThis usually means a TLS-inspecting proxy whose "

f"root CA this Python does not trust. Try the system python3, or install curl."

)

print(f"NOTE: TLS verification failed ({e}).\n"

f" A TLS-inspecting proxy is likely in the path. Falling back to curl, which "

f"uses the OS trust store, for this and all later requests.", file=sys.stderr)

_USE_CURL = True

return _curl_call(method, path, body)

def project_path_id(value):

"""Return `value` as an int, for interpolation into a request path.

Every path is built as f"/projects/{project}/…" from the spec's `project_id`, so a non-integer

value would let a spec steer the URL itself (`1/../..`, a whole `https://…`) rather than just

name a project. Reject anything that is not an integer id — including `True` and `1.9`, which

int() would otherwise quietly turn into project 1.

"""

if isinstance(value, bool):

raise TestmoAPIError(f'"project_id" must be an integer project id, got {value!r}')

if isinstance(value, int):

return value

if isinstance(value, str) and value.strip().isdigit():

return int(value)

raise TestmoAPIError(f'"project_id" must be an integer project id, got {value!r}')

def fetch_all(project, resource, query=""):

project_id = project_path_id(project)

out, page = [], 1

sep = "&" if query else ""

while True:

d = call("GET", f"/projects/{project_id}/{resource}?{query}{sep}page={page:d}")

out += d.get("result", [])

if not d.get("next_page"):

return out

page += 1

if page > MAX_PAGES:

raise TestmoAPIError(

f"GET /projects/{project_id}/{resource} kept reporting next_page past {MAX_PAGES} "

f"pages ({len(out)} records so far). Aborting rather than paging forever — check "

f"the query above for an unencoded character that truncated it."

)

# A tag name no case can carry. verify_tag_filter() asks /cases for it and expects nothing back.

_SENTINEL_TAG = "testmo-tools-probe-no-such-tag"

_TAG_FILTER_VERIFIED = set()

def verify_tag_filter(project):

"""Prove /cases actually applies `?tags=` before any tag filter is trusted.

This API ignores query parameters it does not recognize — 0.6.1 found `?milestone=240` returning

every run in the project. A tag-only spec has no other constraint, so were `?tags=` renamed or

dropped, /cases would return all ~13.7k project-1 cases, every one would pass matches(), and

--create would post a 13k-case run. Asking for a tag that cannot exist separates the two:

an honored parameter yields zero cases, an ignored one yields the whole repository.

"""

project = project_path_id(project)

if project in _TAG_FILTER_VERIFIED:

return

probe = f"/projects/{project}/cases?tags={_SENTINEL_TAG}&page=1"

try:

result = call("GET", probe).get("result") or []

except TestmoAPIError:

# The API rejected the value outright, which still shows it parses `?tags=`.

_TAG_FILTER_VERIFIED.add(project)

return

if result:

raise TestmoAPIError(

f"Refusing to trust filters.tags: GET {probe} returned {len(result)} case(s) for a tag "

f"no case can carry, so /cases is not applying `?tags=`. A tag filter would therefore "

f"select every case in project {project}. Confirm the current tag-filter parameter name "

f"against the Testmo API before creating any tag-driven run."

)

_TAG_FILTER_VERIFIED.add(project)

def fetch_cases(project, filters):

"""Fetch repository cases, applying any tag filter server-side."""

tags = filters.get("tags")

if not tags:

return fetch_all(project, "cases")

verify_tag_filter(project)

# Percent-encode each tag but keep the commas that separate them: Testmo tag names are free

# text, and a raw space raises http.client.InvalidURL, a non-ASCII character raises

# UnicodeEncodeError, `&` appends a parameter, and `#` truncates the URL at the fragment —

# dropping `page=` so fetch_all would re-request page 1 forever.

query = "tags=" + ",".join(urllib.parse.quote(str(t), safe="") for t in tags)

return fetch_all(project, "cases", query)

def build_folder_index(folders):

by_id = {f["id"]: f for f in folders}

children = {}

for f in folders:

children.setdefault(f.get("parent_id"), []).append(f["id"])

def path(fid):

parts, f = [], by_id.get(fid)

while f:

parts.append(f["name"])

f = by_id.get(f.get("parent_id"))

return " > ".join(reversed(parts))

paths = {fid: path(fid) for fid in by_id}

return by_id, children, paths

def subtree(root, children):

seen, stack = set(), [root]

while stack:

n = stack.pop()

seen.add(n)

stack += children.get(n, [])

return seen

def resolve_folders(spec_filters, project, folders=None):

"""Return (folder_id_set, notes). Empty set means 'no folder filter'.

Pass a prefetched `folders` list to avoid refetching (e.g. when resolving many specs)."""

want_paths = spec_filters.get("folder_paths") or []

want_ids = spec_filters.get("folder_ids") or []

if not want_paths and not want_ids:

return None, []

include_sub = spec_filters.get("include_subfolders", True)

if folders is None:

folders = fetch_all(project, "folders")

by_id, children, paths = build_folder_index(folders)

norm = {p.strip().lower(): fid for fid, p in paths.items()}

roots, notes = [], []

for p in want_paths:

fid = norm.get(p.strip().lower())

if fid is None:

notes.append(f"UNMATCHED folder_path: {p!r}")

else:

roots.append(fid)

notes.append(f"ok {p} -> id={fid}")

for fid in want_ids:

if fid in by_id:

roots.append(fid)

notes.append(f"ok id={fid} -> {paths[fid]}")

else:

notes.append(f"UNKNOWN folder_id: {fid}")

result = set()

for r in roots:

result |= subtree(r, children) if include_sub else {r}

return result, notes

def matches(case, spec, folder_ids):

if folder_ids is not None and case.get("folder_id") not in folder_ids:

return False

if spec.get("test_type_ids"):

ids = {t["id"] for t in (case.get("custom_test_type") or [])}

if not ids & set(spec["test_type_ids"]):

return False

if spec.get("team_ids"):

ids = {t["id"] for t in (case.get("custom_team") or [])}

if not ids & set(spec["team_ids"]):

return False

# custom_automation_type is a single {"id", "name"} object, or null when unset.

# Compare on the id so specs can list plain ints (and null for "no type set").

auto = case.get("custom_automation_type")

auto_id = auto["id"] if auto else None

if "automation_type_ids" in spec and auto_id not in spec["automation_type_ids"]:

return False

if spec.get("exclude_automation_type_ids") and auto_id in spec["exclude_automation_type_ids"]:

return False

if spec.get("case_state_ids") and case.get("state_id") not in spec["case_state_ids"]:

return False

if "has_automation" in spec and bool(case.get("has_automation")) != bool(spec["has_automation"]):

return False

return True

# Keys main() indexes directly. Checked when a spec is loaded, before any API call: a KeyError

# raised later — mid-way through the create loop — fires after earlier runs have already been

# POSTed, producing a half-populated milestone.

REQUIRED_SPEC_KEYS = ("project_id", "run_name")

class SpecError(Exception):

"""A spec file is missing, is not valid JSON, or lacks a required key."""

def load_spec(path):

"""Load and validate one spec file. Raises SpecError naming the file and the problem."""

try:

with open(path) as fh:

spec = json.load(fh)

except FileNotFoundError:

raise SpecError(f"Spec file not found: {path}")

except json.JSONDecodeError as e:

raise SpecError(f"{path} is not valid JSON: {e}")

if not isinstance(spec, dict):

raise SpecError(f"{path} must contain a JSON object, got {type(spec).__name__}")

absent = [k for k in REQUIRED_SPEC_KEYS if spec.get(k) is None]

if absent:

raise SpecError(f"{path} is missing {', '.join(repr(k) for k in absent)} "

f"(every spec needs {', '.join(REQUIRED_SPEC_KEYS)})")

return spec

def existing_runs_for_milestone(project, milestone_id):

"""Find runs already linked to this milestone, for the duplicate-run guards.

Returns (runs, checked). `checked` is False when the runs payload carries no milestone_id

field at all: the guard then cannot tell "no duplicates" from "cannot see the link", and the

caller must say so rather than report a clean result — a guardrail that quietly no-ops is

worse than none.

"""

# ?milestone_id= is honoured server-side (verified 2026-09-02: project 1 pages from 20 down to 4

# for a busy milestone). The client-side filter below is kept regardless, because the API ignores

# query parameters it does not recognize — were the name ever to change, that would silently return

# every run in the project, and this guard must not widen into a false positive.

runs = fetch_all(project, "runs", f"milestone_id={milestone_id:d}")

if runs and not any("milestone_id" in r for r in runs):

return [], False

return [r for r in runs if r.get("milestone_id") == milestone_id], True

def same_run(existing, spec):

"""True when an existing run is (or cannot be told apart from) the run this spec creates.

Multi-configuration runs share a run name by design, so the name alone is not enough: a run

with a different config_id is a sibling, not a duplicate. When the run object exposes no

config_id, the two cannot be told apart, so it counts — the guard errs toward refusing.

"""

if existing.get("name") != spec["run_name"]:

return False

if spec.get("config_id") is None or "config_id" not in existing:

return True

return existing.get("config_id") == spec["config_id"]

def load_specs(paths):

"""Load every spec up front, so a bad file fails before any API call is made.

Problems are collected rather than raised one at a time, so several bad specs are reported in a

single pass. Each spec is labelled by its file name, since multi-configuration runs share a

run_name by design and the name alone would not say which spec a row belongs to.

"""

specs, invalid = [], []

for path in paths:

try:

specs.append((Path(path).stem, load_spec(path)))

except SpecError as e:

invalid.append(str(e))

if invalid:

detail = "\n".join(f" {line}" for line in invalid)

sys.exit(f"{len(invalid)} spec(s) are invalid:\n{detail}")

return specs

def resolve_project(specs):

"""Return the single project id every spec targets.

The spec is the only source of truth for this — there is deliberately no --project override,

since a flag that outranked each spec's project_id could create a spec into a project it was

never written for. One invocation fetches one project's cases, so the specs must agree.

"""

by_project = {}

for name, spec in specs:

by_project.setdefault(spec["project_id"], []).append(name)

if len(by_project) > 1:

lines = "\n".join(f" {pid}: {', '.join(names)}" for pid, names in by_project.items())

sys.exit(f"Specs disagree on project_id:\n{lines}\n"

f"Every spec in one invocation must target the same project.")

try:

# Request paths interpolate this, so it has to be an integer id and nothing else.

return project_path_id(next(iter(by_project)))

except TestmoAPIError as e:

sys.exit(str(e))

def resolve_milestone(project, name):

"""Return the id of the one milestone with this exact name. The API cannot create milestones."""

hits = [m for m in fetch_all(project, "milestones") if m.get("name") == name]

if not hits:

sys.exit(f"No milestone named {name!r} in project {project}. "

f"Create it in the Testmo UI first (the API cannot create milestones).")

if len(hits) > 1:

ids = ", ".join(str(m["id"]) for m in hits)

sys.exit(f"{len(hits)} milestones named {name!r} (ids: {ids}). Rename or use a unique name.")

return hits[0]["id"]

def needs_manual_step_2(spec):

"""True when this run cannot be finished by the API alone.

A `config_id` marks a run that is one of several same-named variants distinguished only by

Testmo Configuration. /cases cannot filter by configuration and exposes no per-case config

assignment, so the script only ever reproduces step 1 — removing the other configurations'

cases is a manual UI pass. Each spec's `_comment` spells out its own step 2, and `_comment` is

never displayed, so the reminder has to come from here.

"""

return spec.get("config_id") is not None

def print_step_2_reminder(labels, created):

"""Name every run that still needs its manual Testmo UI pass."""

if not labels:

return

verb = "need" if created else "will need"

print(f"\nMANUAL STEP 2 REQUIRED — {len(labels)} run(s) {verb} a Testmo UI pass:")

for label in labels:

print(f" {label}")

print(" These are multi-configuration runs: the API reproduced step 1 only. In the Testmo UI, "

"remove\n the cases belonging to the other configurations. Each spec's \"_comment\" gives "

"its own step 2.\n Until that pass is done these runs hold the wrong case set.")

def build_payload(spec, ids):

payload = {"name": spec["run_name"], "state_id": spec.get("run_state_id", 7),

"include_all": False, "cases": ids}

for k in ("milestone_id", "config_id", "tags", "note"):

if k in spec and spec[k] is not None:

payload[k] = spec[k]

return payload

def print_details(notes, cases, selected, payload):

"""Single-spec dry-run detail: folder resolution, a sample of the matched cases, the payload."""

if notes:

print("--- folder resolution ---")

for n in notes:

print(f" {n}")

print()

print(f"{len(cases)} cases fetched -> {len(selected)} match filter")

for c in selected[:10]:

tt = ",".join(t["name"] for t in (c.get("custom_test_type") or [])) or "-"

print(f' id={c["id"]} folder={c.get("folder_id")} type={tt} '

f'auto={c.get("custom_automation_type")} state={c.get("state_id")}')

if len(selected) > 10:

print(f" ... +{len(selected) - 10} more")

print("\n--- run payload (cases truncated) ---")

print(json.dumps({**payload, "cases": f"[{len(payload['cases'])} ids]"}, indent=2))

print()

# Why each non-ready status blocks --create, keyed by the status text before any " — " detail.

_BLOCKED_HINTS = {

"NEEDS CONFIG_ID": "these runs are distinguished from their same-named siblings only by "

"Configuration — look the ids up via GET /projects/{project}/configs and set "

"them in the specs",

"EMPTY": "0 cases is almost always a renamed tag or folder — check the spec's filters against "

"the live repository",

"DUPLICATE": "the run already exists in the milestone, or another spec in this batch creates the "

"same run — drop that spec, or pass --allow-existing if the duplicate is intended",

}

def main():

ap = argparse.ArgumentParser(

description="Create Testmo runs from JSON filter specs. Dry-run unless --create is passed.")

ap.add_argument("--spec", action="append", required=True, metavar="PATH",

help="path to a JSON filter spec. Repeat to create several runs in one pass; the "

"folder tree and case list are fetched once for all of them.")

milestone = ap.add_mutually_exclusive_group()

milestone.add_argument("--milestone-id", type=int, default=None,

help="link every run to this existing milestone id (overrides the specs)")

milestone.add_argument("--milestone-name", default=None,

help="link every run to the existing milestone with this exact name. "

"Milestones must be created in the Testmo UI first.")

ap.add_argument("--create", action="store_true", help="actually POST the runs (omit for dry-run)")

ap.add_argument("--allow-existing", action="store_true",

help="create runs even when a run with the same name (and config) is already "

"linked to the milestone. Without this, --create refuses.")

args = ap.parse_args()

specs = load_specs(args.spec)

project = resolve_project(specs)

override = args.milestone_id

if args.milestone_name is not None:

override = resolve_milestone(project, args.milestone_name)

for name, spec in specs:

if override is not None:

spec["milestone_id"] = override

mid = spec.get("milestone_id")

# bool is a subclass of int, so `true` would otherwise pass as milestone 1.

if mid is not None and (not isinstance(mid, int) or isinstance(mid, bool)):

sys.exit(f'{name}: "milestone_id" must be an integer milestone id, got {mid!r}')

print(f"Specs : {len(specs)}")

print(f"Project : {project} (from the specs)")

if args.milestone_name is not None:

print(f"Milestone : {args.milestone_name!r} -> id {override}")

elif override is not None:

print(f"Milestone : id {override}")

print(f"Mode : {'CREATE' if args.create else 'DRY RUN'}")

# Idempotency is checked per spec, against the runs already linked to its milestone. It needs a

# milestone to scope the lookup — run names like "Mobile" recur every cycle, so a project-wide

# name match means nothing.

linked = {}

for mid in dict.fromkeys(s.get("milestone_id") for _, s in specs):

if mid is None:

print("Existing : NOT CHECKED for specs with no milestone — pass --milestone-id or "

"--milestone-name to check for duplicates")

continue

runs, checked = existing_runs_for_milestone(project, mid)

linked[mid] = (runs, checked)

if not checked:

print(f"Existing : CANNOT CHECK milestone {mid} — the runs API returned no "

f"milestone_id field, so duplicates cannot be ruled out")

else:

print(f"Existing : {len(runs)} run(s) already linked to milestone {mid}")

print()

# Fetch shared data once so N specs don't trigger N full downloads. Tag-driven specs each issue

# their own server-side filtered fetch instead of using the full case list.

all_filters = [s.get("filters") or {} for _, s in specs]

folders = (fetch_all(project, "folders")

if any(f.get("folder_paths") or f.get("folder_ids") for f in all_filters) else None)

all_cases = fetch_all(project, "cases") if any(not f.get("tags") for f in all_filters) else None

rows, results, batch = [], [], {}

for i, (name, spec) in enumerate(specs):

filters = spec.get("filters") or {}

label = f"{spec['run_name']} [{name}]"

folder_ids, notes = resolve_folders(filters, project, folders=folders)

bad = [n for n in notes if n.startswith(("UNMATCHED", "UNKNOWN"))]

if bad:

rows.append((label, "FOLDER ERROR", "; ".join(bad)))

continue

cases = fetch_cases(project, filters) if filters.get("tags") else all_cases

selected = [c for c in cases if matches(c, filters, folder_ids)]

ids = [c["id"] for c in selected]

mid = spec.get("milestone_id")

existing = [r for r in linked.get(mid, ([], None))[0] if same_run(r, spec)]

# Index by position, not file name: the same spec passed twice is still a duplicate.

twin = batch.setdefault((mid, spec["run_name"], spec.get("config_id")), i)

# "config_id": null is an explicit placeholder — the run needs a Testmo Configuration that

# has not been looked up yet, and creating it would produce indistinguishable runs.

if "config_id" in spec and spec["config_id"] is None:

status = "NEEDS CONFIG_ID"

elif not ids:

status = "EMPTY — 0 cases"

elif existing and not args.allow_existing:

status = f"DUPLICATE — run {', '.join(str(r.get('id')) for r in existing)} exists"

elif twin != i and not args.allow_existing:

status = f"DUPLICATE — same run as [{specs[twin][0]}]"

else:

status = "ready"

rows.append((label, str(len(ids)), status))

results.append((name, spec, ids))

if len(specs) == 1:

print_details(notes, cases, selected, build_payload(spec, ids))

print(f"{'RUN NAME [SPEC]':<60}{'CASES':>7} STATUS")

print("-" * 86)

for label, count, status in rows:

print(f"{label[:59]:<60}{count:>7} {status}")

ready = sum(1 for r in rows if r[2] == "ready")

print("-" * 86)

print(f"{'TOTAL':<60}{sum(len(ids) for _, _, ids in results):>7} "

f"{ready}/{len(specs)} runs ready")

errors = [r for r in rows if r[1] == "FOLDER ERROR"]

if errors:

sys.exit(f"\n{len(errors)} spec(s) reference folders that did not resolve — fix them before "

f"continuing.")

blocked = [r for r in rows if r[2] != "ready"]

step_2 = [f"{spec['run_name']} [{name}]" for name, spec, _ in results if needs_manual_step_2(spec)]

if not args.create:

print("\nDRY RUN — no runs created. Re-run with --create to POST the above.")

print_step_2_reminder(step_2, created=False)

if blocked:

print(f"NOTE: {len(blocked)} run(s) are not ready (see STATUS above); --create would "

f"refuse until they are fixed or dropped.")

return

if blocked:

# Refuse the whole batch rather than create the ready runs and skip the rest: a skipped run

# leaves the milestone incomplete, and the dry-run is the place to have noticed.

detail = "\n".join(f" {label} — {status}" for label, _, status in blocked)

kinds = dict.fromkeys(status.split(" — ")[0] for _, _, status in blocked)

reasons = "\n".join(f" {k}: {_BLOCKED_HINTS[k].format(project=project)}" for k in kinds)

sys.exit(f"\nRefusing to create: {len(blocked)} run(s) are not ready:\n{detail}\n{reasons}")

if any(checked is False for _, checked in linked.values()):

print("WARNING: could not verify whether these runs already exist — the runs API returned no "

"milestone_id field. Check the Testmo UI before continuing.")

# Create every run we can rather than aborting on the first failure: stopping midway through a

# batch leaves a half-populated milestone and no record of which runs made it.

print()

created, failed, step_2 = [], [], []

for name, spec, ids in results:

label = f"{spec['run_name']} [{name}]"

try:

res = call("POST", f"/projects/{project}/runs", build_payload(spec, ids))

except TestmoAPIError as e:

reason = str(e).splitlines()[0]

print(f" FAILED: {label}\n {reason}")

failed.append((label, reason))

continue

rid = (res.get("result") or {}).get("id")

if rid is None:

print(f" FAILED: {label}\n POST succeeded but the response carried no run id")

failed.append((label, "response carried no run id"))

continue

created.append(rid)

if needs_manual_step_2(spec):

step_2.append(f"{label} -> https://bitwarden.testmo.net/run/{rid}")

print(f" CREATED run {rid} ({len(ids)} cases) {label} https://bitwarden.testmo.net/run/{rid}")

print(f"\n{len(created)} created, {len(failed)} failed — of {len(results)} run(s) attempted.")

print_step_2_reminder(step_2, created=True)

if failed:

print("\nNOT created:")

for label, reason in failed:

print(f" {label} — {reason}")

sys.exit(f"\n{len(failed)} run(s) were not created. Fix the cause and re-run with only those "

f"specs — the {len(created)} run(s) above already exist.")

if __name__ == "__main__":

try:

main()

except TestmoAPIError as e:

sys.exit(str(e))