re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", re.I), # UUID

re.compile(r"^[a-z0-9]([a-z0-9\-\.]*[a-z0-9])?:[^:]+$"), # tag (env:prod, host:web-01)

re.compile(r"^\d+$"), # pure numeric id

re.compile(r"^[a-z][a-z0-9_\-\.]+\.[a-z][a-z0-9_\-\.]+"), # dotted.metric.name

re.compile(r"^[a-z][a-z0-9_]*(-[a-z0-9_]+)+$"), # hyphenated service names (apm-service, account-config)

re.compile(r".*\.tags$"),

re.compile(r".*\.meta$"),

re.compile(r".*\.attributes$"),

re.compile(r".*\.attributes\.\*$"), # inner log/event attribute dicts (CN, action, client_ip …)

re.compile(r".*\.metrics$"),

# Relationships dicts (on cases, incidents, etc.) have resource-specific optional

# keys (assignee, project, created_by, …) that vary per item. Collapse to {*: …}.

re.compile(r".*\.relationships$"),

re.compile(r".*\.usage$"),

# Infrastructure hosts: tags_by_source keys are cloud provider names ("Amazon Web

# Services", "Google Cloud Platform", …) that vary depending on which hosts

# the API returns on a given run — collapse to {"*": …} so snapshots are stable.

re.compile(r".*\.tags_by_source$"),

# Any path already two levels deep in dynamic-key territory (.*.*) will vary

# too much across API calls to snapshot reliably — collapse it.

re.compile(r".*\.\*\.\*.*"),

(re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", re.I), "<uuid>"),

(re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}"), "<timestamp>"),

(re.compile(r"^\d{10,13}$"), "<unix_ts>"),

(re.compile(r"^https?://"), "<url>"),

(re.compile(r"^[0-9a-f]{32,64}$", re.I), "<hash>"),

"""Return True when every key in a dict looks like a dynamic identifier."""

if len(keys) < DYNAMIC_KEY_MIN:

return False

return all(

any(p.fullmatch(k) for p in DYNAMIC_KEY_PATTERNS)

for k in keys

for pattern, placeholder in VALUE_NORMALIZERS:

if pattern.search(val):

return placeholder

"""

if max_depth == 0:

return "<...>"

if isinstance(data, dict):

keys = list(data.keys())

if not keys:

return {}

if _path_forces_dynamic(path) or _all_keys_dynamic(keys):

child_path = f"{path}.*"

# Short-circuit: if the child path would also be forced dynamic, don't

# recurse further — the values are too variable to snapshot reliably.

if _path_forces_dynamic(child_path):

return {"*": "<...>"}

# Merge ALL value schemas into a union so the result is stable across

# API calls that return different subsets of optional fields.

merged: Any = None

for v in data.values():

child = extract_json_schema(v, max_depth - 1, child_path)

if merged is None:

merged = child

elif isinstance(merged, dict) and isinstance(child, dict):

for k, cv in child.items():

if k not in merged:

merged[k] = cv

return {"*": merged if merged is not None else "<empty>"}

return {k: extract_json_schema(v, max_depth - 1, f"{path}.{k}") for k, v in sorted(data.items())}

if isinstance(data, list):

if not data:

return ["<empty_list>"]

# Merge schemas from ALL elements so the snapshot is a stable union of

# optional fields rather than an unstable sample of whichever element

# happens to appear first on this particular API call.

child_path = f"{path}[]"

merged = extract_json_schema(data[0], max_depth - 1, child_path)

for item in data[1:]:

child = extract_json_schema(item, max_depth - 1, child_path)

if isinstance(merged, dict) and isinstance(child, dict):

for k, cv in child.items():

if k not in merged:

merged[k] = cv

return ["<list>", merged]

if isinstance(data, str):

normalized = normalize_value(data)

if normalized != data:

return normalized

return "str"

if isinstance(data, bool):

return "bool"

p = _last_output_path(label, mode)

if p.exists():

try:

return p.read_text()

except Exception:

return None

p = snapshot_path(label, mode)

if p.exists():

try:

return json.loads(p.read_text())

except Exception:

return None

"""

regressions: list[str] = []

additions: list[str] = []

if type(old) != type(new):

regressions.append(f"{path or 'root'}: type changed {type(old).__name__} → {type(new).__name__}")

return regressions, additions

if isinstance(old, dict):

old_keys = set(old)

new_keys = set(new)

for k in sorted(old_keys - new_keys):

full_path = f"{path}.{k}"

if full_path in _OPTIONAL_SCHEMA_PATHS:

continue # known optional field — its absence is not a regression

regressions.append(f"{full_path}: key removed")

for k in sorted(new_keys - old_keys):

additions.append(f"{path}.{k}: key added")

for k in sorted(old_keys & new_keys):

r, a = diff_schemas(old[k], new[k], f"{path}.{k}")

regressions.extend(r)

additions.extend(a)

elif isinstance(old, list):

old_inner = old[1] if len(old) > 1 else None

new_inner = new[1] if len(new) > 1 else None

if old[0] != new[0]:

# Treat any transition between empty and populated list as additive.

# "<empty_list>" → "<list>": baseline had no data; API now returns items.

# "<list>" → "<empty_list>": baseline had items; this run happens to be

# empty (no outages, no results, etc.) — keep the richer baseline schema.

# Both directions are non-regressing: list emptiness is a runtime condition,

# not a structural change in the API or pup's output format.

empty_to_list = (old[0] == "<empty_list>" and new[0] == "<list>")

list_to_empty = (old[0] == "<list>" and new[0] == "<empty_list>")

if empty_to_list or list_to_empty:

additions.append(f"{path}: list emptiness changed '{old[0]}' → '{new[0]}'")

else:

regressions.append(f"{path}: list structure changed '{old[0]}' → '{new[0]}'")

elif old_inner is not None and new_inner is not None:

r, a = diff_schemas(old_inner, new_inner, f"{path}[]")

regressions.extend(r)

additions.extend(a)

elif old != new:

# If both values are normalized placeholders they represent the same

# thing ("some string") — the content merely changed category, which

# is not a structural regression worth failing on.

if old in _NORMALIZED_PLACEHOLDERS and new in _NORMALIZED_PLACEHOLDERS:

additions.append(f"{path or 'root'}: placeholder changed '{old}' → '{new}'")

else:

regressions.append(f"{path or 'root'}: value changed '{old}' → '{new}'")

"""

if depth <= 0:

return "<...>"

if isinstance(schema, dict):

return {k: _truncate_schema(v, depth - 1) for k, v in schema.items()}

if isinstance(schema, list):

if not schema:

return schema

result: list = [schema[0]]

if len(schema) > 1:

result.append(_truncate_schema(schema[1], depth - 1))

return result

"""Merge new schema into base, adding any keys present in new but absent from base."""

if isinstance(base, dict) and isinstance(new, dict):

merged = dict(base)

for k, v in new.items():

if k in merged:

merged[k] = _merge_schemas(merged[k], v)

else:

merged[k] = v

return merged

if isinstance(base, list) and isinstance(new, list):

# Upgrade ["<empty_list>"] → ["<list>", schema] when the API finally

# returns data for a list that was previously empty.

if base and base[0] == "<empty_list>" and new and new[0] == "<list>":

return new

# Merge list element schemas (keep the richer union)

if base and new and base[0] == new[0] == "<list>":

merged_inner = _merge_schemas(base[1], new[1]) if len(base) > 1 and len(new) > 1 else (base[1] if len(base) > 1 else new[1] if len(new) > 1 else None)

return ["<list>", merged_inner] if merged_inner is not None else base

result: "TestResult",

update_snapshots: bool,

max_regression_depth: int = 0,

) -> tuple[str | None, bool]:

"""

if result.exit_code != 0 or not result.stdout.strip():

return None, False

try:

data = json.loads(result.stdout.strip())

except (json.JSONDecodeError, ValueError):

return None, False

current_schema = extract_json_schema(data)

existing = load_snapshot(result.label, result.mode)

if existing is None:

save_snapshot(result.label, current_schema, result.mode)

return None, True

if update_snapshots:

save_snapshot(result.label, current_schema, result.mode)

return None, False

# Apply depth truncation when requested. Both schemas are truncated

# identically so structural changes above the threshold still register.

if max_regression_depth > 0:

cmp_existing = _truncate_schema(existing, max_regression_depth)

cmp_current = _truncate_schema(current_schema, max_regression_depth)

else:

cmp_existing = existing

cmp_current = current_schema

regressions, additions = diff_schemas(cmp_existing, cmp_current)

# Auto-merge new keys into the snapshot so they're included in future baselines.

if additions:

merged = _merge_schemas(existing, current_schema)

save_snapshot(result.label, merged, result.mode)

if regressions:

return "\n".join(regressions), bool(additions)

re.compile(p, re.IGNORECASE) for p in [

r"401",

r"403",

r"authentication",

r"not authenticated",

r"no credentials",

r"run `pup auth login`",

r"invalid api key",

r"forbidden",

r"unauthorized",

r"DD_API_KEY",

r"DD_APP_KEY",

]

(re.compile(r"\b400\b"), "400 Bad Request"),

(re.compile(r"\b401\b"), "401 Unauthorized"),

(re.compile(r"\b403\b"), "403 Forbidden"),

(re.compile(r"\b404\b"), "404 Not Found"),

(re.compile(r"\b429\b"), "429 Too Many Requests"),

(re.compile(r"\b5\d\d\b"), "5xx Server Error"),

"""

combined = stderr + "\n" + stdout

for pat, label in _HTTP_STATUS_REASONS:

if pat.search(combined):

return label

# Fallback: generic auth description from pup's error text

if re.search(r"no credentials|authentication required|set DD_API_KEY", combined, re.I):

return "no credentials configured"

if re.search(r"forbidden", combined, re.I):

return "403 Forbidden"

if re.search(r"unauthorized", combined, re.I):

return "401 Unauthorized"

result: "TestResult",

expect_json: bool,

expect_exit: int | None = None,

) -> list[str]:

defects = []

# Rust runtime signals (panics, backtraces, internal errors) go to stderr.

# Checking stdout for these patterns produces false positives when API

# responses contain phrases like "internal error:" in their payload text

# (e.g. events list describing a monitored system error).

stderr_only = result.stderr

combined = result.stdout + "\n" + result.stderr

if PANIC_PATTERN.search(stderr_only):

defects.append("PANIC detected in output")

if UNWRAP_PATTERN.search(stderr_only):

defects.append("unwrap() on None detected")

if BACKTRACE_PATTERN.search(stderr_only) and STACK_TRACE_PATTERN.search(combined):

defects.append("Rust stack trace in output")

if INTERNAL_ERROR_PATTERN.search(stderr_only):

defects.append("Internal error message in output")

if expect_exit is not None and result.exit_code != expect_exit:

defects.append(f"Unexpected exit code: got {result.exit_code}, expected {expect_exit}")

if expect_json and result.exit_code == 0 and result.stdout.strip():

try:

parsed = json.loads(result.stdout.strip())

if parsed is None:

defects.append("JSON output is null on success")

except json.JSONDecodeError as exc:

if not is_auth_failure(result.stdout) and not is_auth_failure(result.stderr):

defects.append(f"Invalid JSON output: {exc}")

if (expect_exit is None or expect_exit == 0) and \

result.exit_code == 0 and \

not result.stdout.strip() and not result.stderr.strip():

defects.append("Empty output on successful exit")

label: str

args: list[str]

category: str

exit_code: int

stdout: str

stderr: str

duration_ms: float

expect_exit: int | None = None

note: str = ""

defects: list[str] = field(default_factory=list)

regression: str | None = None

snapshot_created: bool = False

skipped: bool = False

skip_reason: str = ""

# Override computed status (used for synthetic untested/write rows)

status_override: str | None = None

# "human" | "agent" | "" (synthetic rows that were not run)

mode: str = "human"

# HTTP status classification when auth_fail (e.g. "401 Unauthorized")

auth_reason: str = ""

# Per-row diff: current stdout vs previous run's stdout (same mode)

row_diff: tuple[str, str] = field(default_factory=lambda: ("", ""))

@property

def test_id(self) -> str:

"""Stable slug derived from the label and mode, usable as an HTML anchor."""

base = re.sub(r"[^a-z0-9]+", "-", self.label.lower()).strip("-")

return f"{base}-{self.mode}" if self.mode else base

@property

def status(self) -> str:

if self.status_override:

return self.status_override

if self.skipped:

return "skipped"

if self.defects or self.regression:

return "fail"

if self.expect_exit is not None:

return "pass" if self.exit_code == self.expect_exit else "fail"

if self.exit_code != 0:

return "auth_fail"

# ── no-auth commands ──────────────────────────────────────────────────

{

"label": "version",

"args": ["version"],

"category": "no_auth",

"expect_json": False,

},

{

"label": "test",

"args": ["test"],

"category": "no_auth",

"expect_json": False,

"note": "Diagnostic command: shows configured site, API host, key presence, and output format.",

},

{

"label": "agent schema",

"args": ["agent", "schema"],

"category": "no_auth",

"expect_json": True,

},

{

"label": "agent schema --compact",

"args": ["agent", "schema", "--compact"],

"category": "no_auth",

"expect_json": True,

},

{

"label": "agent guide",

"args": ["agent", "guide"],

"category": "no_auth",

"expect_json": False,

},

{

"label": "misc ip-ranges",

"args": ["misc", "ip-ranges"],

"category": "no_auth",

"expect_json": True,

},

{

"label": "misc status",

"args": ["misc", "status"],

"category": "no_auth",

"expect_json": True,

},

{

"label": "completions bash",

"args": ["completions", "bash"],

"category": "no_auth",

"expect_json": False,

"note": "BUG (debug build only): panics with clap debug_assert — "

"'type_id' referenced in conflicts_with* does not exist. "

"Release builds suppress debug_asserts and generate completions normally.",

},

{

"label": "completions zsh",

"args": ["completions", "zsh"],

"category": "no_auth",

"expect_json": False,

"note": "BUG (debug build only): same clap debug_assert panic as 'completions bash'.",

},

# ── --help smoke tests ────────────────────────────────────────────────

{

"label": "help (root)",

"args": ["--help"],

"category": "no_auth",

"expect_json": False,

"expect_exit": 0,

},

{

"label": "monitors --help",

"args": ["monitors", "--help"],

"category": "no_auth",

"expect_json": False,

"expect_exit": 0,

},

{

"label": "logs --help",

"args": ["logs", "--help"],

"category": "no_auth",

"expect_json": False,

"expect_exit": 0,

},

# ── auth status ───────────────────────────────────────────────────────

{

"label": "auth status",

"args": ["auth", "status"],

"category": "auth_status",

"expect_json": False,

},

# ── api-keys ──────────────────────────────────────────────────────────

{

"label": "api-keys list",

"args": ["api-keys", "list"],

"category": "auth_required",

"expect_json": True,

},

# ── app-keys ──────────────────────────────────────────────────────────

{

"label": "app-keys list",

"args": ["app-keys", "list"],

"category": "auth_required",

"expect_json": True,

},

# ── apm ───────────────────────────────────────────────────────────────

{

"label": "apm services list",

"args": ["apm", "services", "list", "--env=prod", "--from=1h"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "apm services stats",

"args": ["apm", "services", "stats", "--env=prod", "--from=1h"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "apm entities list",

"args": ["apm", "entities", "list", "--env=prod", "--from=1h"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "apm dependencies list",

"args": ["apm", "dependencies", "list", "--env=prod", "--from=1h"],

"category": "auth_required",

"expect_json": True,

"skip_regression": True,

# The service dependency map has 6000+ service names as top-level dict keys.

# They are a mix of hyphenated, underscore_separated, dotted, and single-word

# names — DYNAMIC_KEY_PATTERNS cannot collapse them all because _all_keys_dynamic

# requires EVERY key to match at least one pattern, and single-word names like

# "abacus" don't match any. Service names appear/disappear between API calls,

# making the snapshot inherently volatile. skip_regression is required.

},

{

"label": "apm flow-map",

"args": ["apm", "flow-map", "--env=prod", "--query=*", "--from=1h"],

"category": "auth_required",

"expect_json": True,

"note": "pup bug: --from is not propagated into the API query body; API returns 400 'missing parameter from in query'.",

},

# ── audit-logs ────────────────────────────────────────────────────────

{

"label": "audit-logs list",

"args": ["audit-logs", "list", "--from=1h"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "audit-logs search",

"args": ["audit-logs", "search", "--from=1h", "--query=*"],

"category": "auth_required",

"expect_json": True,

},

# ── cases ─────────────────────────────────────────────────────────────

{

"label": "cases projects list",

"args": ["cases", "projects", "list"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "cases search",

"args": ["cases", "search", "--query=*"],

"category": "auth_required",

"expect_json": True,

},

# ── cicd ──────────────────────────────────────────────────────────────

{

"label": "cicd pipelines list",

"args": ["cicd", "pipelines", "list", "--from=1h"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "cicd events search",

"args": ["cicd", "events", "search", "--from=1h", "--query=*"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "cicd tests list",

"args": ["cicd", "tests", "list", "--from=1h"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "cicd tests search",

"args": ["cicd", "tests", "search", "--from=1h", "--query=*"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "cicd flaky-tests search",

"args": ["cicd", "flaky-tests", "search", "--query=*"],

"category": "auth_required",

"expect_json": True,

},

# ── cloud ─────────────────────────────────────────────────────────────

{

"label": "cloud aws list",

"args": ["cloud", "aws", "list"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "cloud azure list",

"args": ["cloud", "azure", "list"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "cloud gcp list",

"args": ["cloud", "gcp", "list"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "cloud oci tenancies list",

"args": ["cloud", "oci", "tenancies", "list"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "cloud oci products list",

"args": ["cloud", "oci", "products", "list"],

"category": "auth_required",

"expect_json": True,

},

# ── dashboards ────────────────────────────────────────────────────────

{

"label": "dashboards list",

"args": ["dashboards", "list"],

"category": "auth_required",

"expect_json": True,

},

# ── data-governance ───────────────────────────────────────────────────

{

"label": "data-governance scanner rules list",

"args": ["data-governance", "scanner", "rules"],

"category": "auth_required",

"expect_json": True,

},

# ── downtime ──────────────────────────────────────────────────────────

{

"label": "downtime list",

"args": ["downtime", "list"],

"category": "auth_required",

"expect_json": True,

},

# ── error-tracking ────────────────────────────────────────────────────

{

"label": "error-tracking issues search",

"args": ["error-tracking", "issues", "search", "--from=1h", "--query=*"],

"category": "auth_required",

"expect_json": True,

},

# ── events ────────────────────────────────────────────────────────────

{

"label": "events list",

"args": ["events", "list", "--from=1h"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "events search",

"args": ["events", "search", "--from=1h", "--query=*"],

"category": "auth_required",

"expect_json": True,

},

# ── fleet ─────────────────────────────────────────────────────────────

{

"label": "fleet agents list",

"args": ["fleet", "agents", "list"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "fleet agents versions",

"args": ["fleet", "agents", "versions"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "fleet deployments list",

"args": ["fleet", "deployments", "list"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "fleet schedules list",

"args": ["fleet", "schedules", "list"],

"category": "auth_required",

"expect_json": True,

},

# ── incidents ─────────────────────────────────────────────────────────

{

"label": "incidents list",

"args": ["incidents", "list"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "incidents handles list",

"args": ["incidents", "handles", "list"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "incidents postmortem-templates list",

"args": ["incidents", "postmortem-templates", "list"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "incidents settings get",

"args": ["incidents", "settings", "get"],

"category": "auth_required",

"expect_json": True,

"note": "pup bug: serde deserialization error — missing field `analytics_dashboard_id` in API response model.",

},

# ── infrastructure ────────────────────────────────────────────────────

{

"label": "infrastructure hosts list",

"args": ["infrastructure", "hosts", "list"],

"category": "auth_required",

"expect_json": True,

"max_regression_depth": 2,

"note": "Per-host fields like aws_id/aws_name are optional (only on AWS hosts) and vary per run; depth-2 checks root envelope and that host_list is a populated list without inspecting per-host fields.",

},

# ── integrations ──────────────────────────────────────────────────────

{

"label": "integrations jira accounts",

"args": ["integrations", "jira", "accounts"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "integrations jira accounts list",

"args": ["integrations", "jira", "accounts", "list"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "integrations jira templates list",

"args": ["integrations", "jira", "templates", "list"],

"category": "auth_required",

"expect_json": True,

"note": "pup bug: serde deserialization error — missing field `attributes` in Jira template response model.",

},

{

"label": "integrations pagerduty list",

"args": ["integrations", "pagerduty", "list"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "integrations servicenow assignment-groups list",

"args": ["integrations", "servicenow", "assignment-groups", "list"],

"category": "auth_required",

"expect_json": True,

"note": "pup bug: validates INSTANCE_NAME positional arg as UUID; real instance names like 'dev186409' are rejected.",

},

{

"label": "integrations servicenow business-services list",

"args": ["integrations", "servicenow", "business-services", "list"],

"category": "auth_required",

"expect_json": True,

"note": "pup bug: same UUID validation bug as assignment-groups list.",

},

{

"label": "integrations servicenow instances list",

"args": ["integrations", "servicenow", "instances", "list"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "integrations servicenow templates list",

"args": ["integrations", "servicenow", "templates", "list"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "integrations servicenow users list",

"args": ["integrations", "servicenow", "users", "list"],

"category": "auth_required",

"expect_json": True,

"note": "pup bug: same UUID validation bug as assignment-groups list.",

},

{

"label": "integrations slack list",

"args": ["integrations", "slack", "list"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "integrations webhooks list",

"args": ["integrations", "webhooks", "list"],

"category": "auth_required",

"expect_json": True,

},

# ── hamr ──────────────────────────────────────────────────────────────

{

"label": "hamr connections get",

"args": ["hamr", "connections", "get"],

"category": "auth_required",

"expect_json": True,

"note": "pup bug: serde error 'invalid type: null, expected a mapping' — HAMR API may return null for connection config.",

},

# ── investigations ────────────────────────────────────────────────────

{

"label": "investigations list",

"args": ["investigations", "list"],

"category": "auth_required",

"expect_json": True,

},

# ── logs ──────────────────────────────────────────────────────────────

{

"label": "logs archives list",

"args": ["logs", "archives", "list"],

"category": "auth_required",

"expect_json": True,

},

{

"label": "logs custom-destinations list",