diff --git a/.agents/rules/testing-python.md b/.agents/rules/testing-python.md index 71b02a1db3d..86b873c2187 100644 --- a/.agents/rules/testing-python.md +++ b/.agents/rules/testing-python.md @@ -63,6 +63,10 @@ Skip tests that only exercise library behavior: plain `Enum` value/round-trip ch If the logic needs only in-memory inputs (a `SchemaBranch`, a dataclass, a pure function), write a unit test without DB fixtures — don't default to a component test because a neighbor uses one. Use the database or containers only when behavior genuinely depends on them. +## Don't leak process-global state + +Every test in an xdist worker shares one interpreter. Change `logging` levels/handlers/filters, `structlog` config, module-level registries/singletons, `sys.path`/`sys.modules` or env vars only through a save/restore fixture (change it, `yield`, restore it), or `monkeypatch` where it applies. Never call an application startup routine such as `infrahub.log.configure_logging` from a test — it owns the whole process and undoes nothing, so it reconfigures every later test in the worker. Install only the piece under test and remove it after the `yield`. See `dev/guidelines/backend/testing.md` §"Leave process-global state as you found it". + ## Test file placement Test files mirror source structure: `infrahub/core/node.py` → `tests/unit/core/test_node.py` diff --git a/.agents/skills/analyzing-ci-flakiness/SKILL.md b/.agents/skills/analyzing-ci-flakiness/SKILL.md new file mode 100644 index 00000000000..6c504490af3 --- /dev/null +++ b/.agents/skills/analyzing-ci-flakiness/SKILL.md @@ -0,0 +1,128 @@ +--- +name: analyzing-ci-flakiness +description: >- + Analyzes recent CI failures on pull requests to identify flaky tests, using retry outcomes + (failed attempt → green re-run) and cross-PR recurrence as evidence, and maintains a local + longitudinal ledger so flakiness can be tracked over time. TRIGGER when: the user wants to find + flaky tests, correlate recent CI failures, check which tests fail across PRs or recover on + retry, or refresh the flakiness trend report. DO NOT TRIGGER when: babysitting a single PR's CI + until green → monitoring-pull-requests; diagnosing or fixing one specific failing test → the + bug-analysis skills. +argument-hint: "Optional base-branch glob(s) and window, e.g. `release-1.11 14` (default: all bases, last 7 days)" +allowed-tools: + - Bash(python3 .agents/skills/analyzing-ci-flakiness/scripts/collect.py:*) +compatibility: Requires the gh CLI authenticated against the repo. Python 3 (stdlib only). Writes a cache under ~/ci-cache. +metadata: + version: 0.1.0 + author: OpsMill +--- + +# CI Flakiness Analyzer + +## Introduction + +A test is *flaky* when its failure does not reproduce on the same code: the run was retried and +went green, or the same test fails on unrelated PRs. This skill mines both signals from GitHub +Actions history, downloads the failed job logs once into a local cache, and appends every +observation to a ledger (`~/ci-cache/-/ledger.jsonl`) so repeated invocations — +weekly, or ad hoc — accumulate trend data instead of starting from scratch. + +The mechanical part (fetching, caching, test-name extraction, known-signature classification) is +done by the bundled script. Your job is the judgment part: separating flakes from real +regressions, spotting new systemic signatures, and writing the report. + +## Step 1 — Parse arguments + +- Base-branch filter: any arguments that look like branch names or globs (`release-1.11`, + `release-*`, `stable`). Default: no filter (all PR bases), which is usually what "how flaky is + CI" means. Filter when the user names a branch. +- Window: a bare integer is a number of days (default 7). An ISO date means "since that date". + +## Step 2 — Collect + +Run the bundled collector (repo-root relative): + +```bash +python3 .agents/skills/analyzing-ci-flakiness/scripts/collect.py \ + [--base ...] [--days N | --since YYYY-MM-DD] [--repo owner/name] +``` + +It prints a JSON report to stdout and writes everything under +`~/ci-cache/-/windows/_/`: + +- `runs.jsonl` — every `pull_request` workflow run created in the window +- `failed_jobs_with_tests.json` — failed jobs of the interesting run-attempts, with extracted + failing tests, systemic-bucket tags, and a `recovered_same_run` flag +- `report-data.json` — headline numbers, ranked per-test table, per-bucket incident counts + (`bucket_incidents`: distinct jobs/runs/PRs per systemic bucket), and the ledger's weekly + history +- `joblogs/.log` — raw logs (ANSI intact; strip with `sed 's/\x1b\[[0-9;]*m//g'`) + +Notes the script already accounts for — don't re-derive them: + +- The runs API's `pull_requests` field is empty for many runs; the script joins runs to PRs + through every PR head commit SHA as well. Don't trust the field alone. +- "Interesting attempts" = every earlier attempt of a retried run (that's what the retry fixed) + plus final attempts that failed. Runs cancelled on attempt 1 are concurrency noise and skipped. +- Logs already on disk are never re-downloaded; the ledger is deduplicated by (job, test). Old + logs expire on GitHub's side (~90 days) — an empty `joblogs/*.log` means expired, not passing. + +## Step 3 — Investigate what the script could not name + +For failed jobs with an empty `tests` list and no bucket tag, read the log yourself (grep for +`##[error]`, `FAILED`, `Error:`, `Timeout`). Two outcomes: + +- It matches a *new* systemic signature (infra failure that cascades over many tests). Add a + regex for it to `BUCKETS` in `collect.py` and to the table below, so future runs classify it. +- It's a genuine test failure the extraction regexes missed — note the test manually and + consider extending `extract_tests`. + +### Known systemic signatures (as of 2026-08 — keep in sync with `BUCKETS` in collect.py) + +| Bucket | Signature | Meaning | +|---|---|---| +| `stack-readiness` | `ServerNotResponsiveError … /api/schema/load` | Seeded testcontainers stack not ready; the whole pytest-playwright shard errors. One incident, not N flaky tests. | +| `vitest-mock-corruption` | `TypeError: vi.mocked(...).mockX is not a function` | vitest browser-mode module-mocking race; hits a different test file each time. | +| `prefect-setup-triggers-timeout` | `Setup triggers` task `ReadTimeout` | Prefect hang at session setup; downstream tests hit their own timeouts. | +| `neo4j-deadlock` | `Neo.TransientError.Transaction.DeadlockDetected` | Concurrent-write deadlock, usually integration suites under xdist. | +| `compose-boot-failure` | `docker compose … up --wait` non-zero exit | Stack never booted; job-level infra failure. | +| `sqlite-locked` | `(sqlite3.OperationalError) database is locked` (also matches the raw `sqlite3.OperationalError:` form) | Prefect's sqlite under contention. | +| `runner-oom` | `Process completed with exit code 137` | Runner OOM/SIGKILL; the mass test failures in the same job are casualties, not flakes. | +| `docker-network-pool-exhausted` | `all predefined address pools have been fully subnetted` | Leaked compose networks exhausted the docker address pools on a self-hosted runner. | +| `actions-download-429` | `Failed to download action … 429` | GitHub rate-limited its own action download; pure platform flake. | +| `pytest-green-exit-1` | green pytest summary directly followed by exit 1 | Session-teardown/plugin abort after all tests passed (e.g. testcontainers result reporting). | + +## Step 4 — Judge: flake vs regression + +For each test in the ranked table, classify: + +- **Flaky (strong)** — fails on ≥2 unrelated PRs, or `recovered_on_retry > 0`. The more distinct + PRs, the stronger. +- **Flaky (weak)** — single occurrence with an infra-flavored error (locator timeout, transient + branch not found) and the PR later went green. List, but rank low. +- **Suspect regression, not a flake** — the same test fails on *every* attempt of the same + commit and the PR is still red, or the failures started only after a specific merge. Say so + explicitly; do not bury it in the flake list. Cross-check: does the test fail on any PR that + does not contain the suspect change? +- **Systemic bucket** — tests whose only failures carry a bucket tag are casualties, not causes. + Report the bucket (with the incident count from `bucket_incidents` in `report-data.json`), not + the individual tests. + +Different tests failing on successive attempts of the same run = two independent flakes, not a +regression. + +## Step 5 — Report + +Write `ANALYSIS.md` into the window directory, then give the user a summary. Lead with the +ranked flake candidates. Include: + +1. Headline numbers: PRs in scope, runs matched, retried runs, retried-and-recovered runs + (pure-flake evidence), hard failures. +2. Ranked flake candidates — test id, distinct PRs/runs, recovered-on-retry count, one-line + error cause. Group systemic buckets as single entries. +3. Suspected real regressions, clearly separated. +4. Trend — from `weekly_history` in `report-data.json`: which offenders are new this window, + which recur week over week, which disappeared (likely fixed). This section is the reason the + ledger exists; don't skip it once ≥2 windows of data exist. + +Do not propose fixes unless asked; the deliverable is the evidence-ranked candidate list. diff --git a/.agents/skills/analyzing-ci-flakiness/scripts/collect.py b/.agents/skills/analyzing-ci-flakiness/scripts/collect.py new file mode 100755 index 00000000000..47abca68088 --- /dev/null +++ b/.agents/skills/analyzing-ci-flakiness/scripts/collect.py @@ -0,0 +1,399 @@ +#!/usr/bin/env python3 +"""Collect CI failure data for flakiness analysis, incrementally, into a local cache. + +Fetches GitHub Actions runs for pull requests, matches them to PRs (via the +runs' ``pull_requests`` field *and* a head-SHA join, since the field is often +empty), identifies retried runs and failed attempts, downloads the failed job +logs, extracts failing test identifiers, classifies known systemic failure +signatures, and appends everything to a longitudinal ledger so successive +invocations build trend data. + +Only stdlib + the ``gh`` CLI (must be authenticated). Safe to re-run: the runs +listing is refreshed each time, but job logs already on disk are never +re-downloaded and the ledger is deduplicated. + +Usage: + collect.py [--repo OWNER/NAME] [--base GLOB ...] [--days N | --since YYYY-MM-DD] + [--cache DIR] + +Outputs (under /-/): + ledger.jsonl one record per (job, test) ever observed + windows/_/ this invocation's window + runs.jsonl all pull_request runs created in the window + failed_jobs_with_tests.json failed jobs of interesting attempts + tests + report-data.json ranked frequency table, per-bucket incident counts + headline numbers + joblogs/.log raw logs of failed jobs (ANSI intact; empty = expired) +""" + +from __future__ import annotations + +import argparse +import datetime as dt +import fnmatch +import json +import re +import subprocess # noqa: S404 +import sys +from collections import defaultdict +from pathlib import Path + +ANSI = re.compile(r"\x1b\[[0-9;]*m") + +# The Actions list-runs API silently returns at most this many results per query. +API_RESULT_CAP = 1000 + +# Playwright's breadcrumb separator (U+203A) as it appears in job logs. +PW_SEP = "\u203a" + +# Known systemic failure signatures. When one matches a job log, the job is +# tagged with the bucket. The tags feed the report's per-bucket incident counts +# (``bucket_incidents``) and the judgment step (SKILL.md Step 4), which reports +# a bucketed cascade as one incident rather than N flaky tests; the per-test +# table still lists every test, annotated with its buckets, so casualties can +# be discounted. Keep in sync with the table in SKILL.md. +BUCKETS: list[tuple[str, str]] = [ + ("stack-readiness", r"ServerNotResponsiveError: Unable to read from '[^']*/api/schema/load"), + ("vitest-mock-corruption", r"TypeError: (?:vi\.mocked\(\.\.\.\)|\w+)\.mock\w+ is not a function"), + ("prefect-setup-triggers-timeout", r"'Setup triggers'.*ReadTimeout|Task run encountered an exception ReadTimeout"), + ("neo4j-deadlock", r"Neo\.TransientError\.Transaction\.DeadlockDetected"), + ("compose-boot-failure", r"'docker', 'compose'.*'up', '--wait'.*non-zero exit status"), + ("sqlite-locked", r"sqlite3\.OperationalError[):] database is locked"), + ("runner-oom", r"Process completed with exit code 137|exit code: 137"), + ("docker-network-pool-exhausted", r"all predefined address pools have been fully subnetted"), + ("actions-download-429", r"Failed to download action .*429"), + # pytest summary is green (no "N failed") yet the process exits 1: a + # session-teardown/plugin abort, e.g. the testcontainers result reporting. + ( + "pytest-green-exit-1", + r"=+ \d+ passed(?:(?!\d+ failed)[^\n])*=+[^\n]*\n(?:[^\n]*\n){0,3}[^\n]*Process completed with exit code 1\.", + ), +] + +LEDGER_FIELDS = ( + "run", + "attempt", + "final_conclusion", + "recovered_same_run", + "run_created", + "workflow", + "job_id", + "job", + "prs", + "buckets", +) + + +def gh(args: list[str], *, check: bool = True) -> str: + res = subprocess.run(["gh", *args], capture_output=True, text=True, errors="replace", check=False) # noqa: S603, S607 + if res.returncode != 0 and check: + raise RuntimeError(f"gh {' '.join(args[:3])}... failed: {res.stderr.strip()[:300]}") + return res.stdout + + +def gh_json_lines(args: list[str]) -> list[dict]: + out = gh(args) + return [json.loads(line) for line in out.splitlines() if line.strip()] + + +def fetch_job_log(repo: str, job_id: int, log_path: Path) -> None: + """Download one job log, distinguishing gone from transiently unavailable. + + On success the log is written to ``log_path``. On HTTP 404/410 (the log + expired or was deleted on GitHub's side) an empty file is written as a + durable sentinel so the job is never re-fetched. On any other failure + (rate limit, network) — including a successful call with an empty body, + which a real job log never has — nothing is written, so the next + collection retries. + """ + res = subprocess.run( # noqa: S603 + ["gh", "api", f"repos/{repo}/actions/jobs/{job_id}/logs"], # noqa: S607 + capture_output=True, + text=True, + errors="replace", + check=False, + ) + if res.returncode == 0 and res.stdout: + log_path.write_text(res.stdout, encoding="utf-8") + elif res.returncode == 0: + print(f"[collect] WARN log {job_id}: empty response, leaving unfetched for retry", file=sys.stderr) + elif "HTTP 404" in res.stderr or "HTTP 410" in res.stderr: + log_path.write_text("", encoding="utf-8") + else: + print(f"[collect] WARN log {job_id}: {res.stderr.strip()[:300]}", file=sys.stderr) + + +def list_prs(repo: str, since: dt.date, base_globs: list[str]) -> list[dict]: + # Look back further than the run window: a re-run in the window can belong + # to a PR whose updatedAt predates it. + pr_since = since - dt.timedelta(days=21) + prs = json.loads( + gh( + [ + "pr", + "list", + "--repo", + repo, + "--state", + "all", + "--limit", + "500", + "--search", + f"updated:>={pr_since.isoformat()}", + "--json", + "number,title,state,baseRefName,headRefName,updatedAt", + ] + ) + ) + if base_globs: + prs = [p for p in prs if any(fnmatch.fnmatch(p["baseRefName"], g) for g in base_globs)] + return prs + + +def pr_head_shas(repo: str, numbers: list[int]) -> dict[str, set[int]]: + sha2pr: dict[str, set[int]] = defaultdict(set) + for n in numbers: + out = gh(["api", f"repos/{repo}/pulls/{n}/commits?per_page=100", "--paginate", "--jq", ".[].sha"], check=False) + for sha in out.split(): + sha2pr[sha].add(n) + return sha2pr + + +def _runs_query(repo: str, created: str) -> str: + return f"repos/{repo}/actions/runs?event=pull_request&created={created}&per_page=100" + + +def list_runs(repo: str, since: dt.date, until: dt.date) -> list[dict]: + """List runs in [since, until], splitting the date range to stay under the API's 1000-result cap.""" + jq = ( + ".workflow_runs[] | {id, name, head_branch, head_sha, run_attempt, " + "conclusion, status, created_at, prs: [.pull_requests[] | " + "{number, base: .base.ref}]}" + ) + created = f"{since.isoformat()}..{until.isoformat()}" + total = int(gh(["api", _runs_query(repo, created).replace("per_page=100", "per_page=1"), "--jq", ".total_count"])) + if total > API_RESULT_CAP and since < until: + mid = since + (until - since) // 2 + print(f"[collect] {total} runs in {created} exceeds the API result cap; splitting", file=sys.stderr) + return list_runs(repo, since, mid) + list_runs(repo, mid + dt.timedelta(days=1), until) + if total > API_RESULT_CAP: + print(f"[collect] WARN {total} runs on {since} alone; the API returns only the newest results", file=sys.stderr) + return gh_json_lines(["api", _runs_query(repo, created), "--paginate", "--jq", jq]) + + +def failed_jobs_for_attempt(repo: str, run_id: int, attempt: int) -> list[dict]: + return gh_json_lines( + [ + "api", + f"repos/{repo}/actions/runs/{run_id}/attempts/{attempt}/jobs?per_page=100", + "--paginate", + "--jq", + '.jobs[] | select(.conclusion=="failure") | {id, name}', + ] + ) + + +def extract_tests(job_name: str, text: str) -> list[str]: + """Pull failing test identifiers out of a cleaned (ANSI-stripped) job log.""" + fails: set[str] = set() + # pytest — backend suites and the pytest-playwright e2e suite + fails.update( + m.group(1).split(" - ")[0].rstrip(",") + for m in re.finditer(r"(?:FAILED|ERROR) ((?:backend/)?tests/\S+::\S+)", text) + ) + # legacy TS Playwright — numbered entries of the failure report + if "E2E-testing-playwright" in job_name: + for m in re.finditer(rf"\d+\)\s+\[[\w-]+\]\s+{PW_SEP}\s+(tests/e2e/[^\n{PW_SEP}]+){PW_SEP}([^\n]+)", text): + spec = m.group(1).strip().split(":")[0] + title = re.sub(r"\s+", " ", m.group(2)).strip()[:120] + fails.add(f"PW {spec} {PW_SEP} {title}") + # vitest browser mode + if job_name == "frontend-tests": + fails.update( + f"VITEST {m.group(1)}" for m in re.finditer(r"FAIL\s+\|?\s*\w*\s*\|?\s+(src/\S+\.test\.\w+)", text) + ) + return sorted(fails) + + +def classify(text: str) -> list[str]: + return [name for name, pat in BUCKETS if re.search(pat, text)] + + +def match_runs_to_prs(runs: list[dict], pr_by_num: dict[int, dict], sha2pr: dict[str, set[int]]) -> list[dict]: + matched = [] + for r in runs: + nums = {p["number"] for p in r["prs"] if p["number"] in pr_by_num} + nums |= {n for n in sha2pr.get(r["head_sha"], set()) if n in pr_by_num} + if nums: + r["pr_nums"] = sorted(nums) + matched.append(r) + return matched + + +def collect_failed_jobs( + repo: str, targets: list[tuple[dict, int]], pr_by_num: dict[int, dict], win_dir: Path +) -> list[dict]: + """Fetch failed jobs and their logs for each (run, attempt); build job entries.""" + jobs_out = [] + for r, attempt in targets: + try: + jobs = failed_jobs_for_attempt(repo, r["id"], attempt) + except RuntimeError as exc: + print(f"[collect] WARN jobs {r['id']}/{attempt}: {exc}", file=sys.stderr) + continue + for job in jobs: + log_path = win_dir / "joblogs" / f"{job['id']}.log" + if not log_path.exists(): + fetch_job_log(repo, job["id"], log_path) + text = ANSI.sub("", log_path.read_text(errors="replace")) if log_path.exists() else "" + jobs_out.append( + { + "run": r["id"], + "attempt": attempt, + "final_attempt": r["run_attempt"], + "final_conclusion": r["conclusion"], + "recovered_same_run": attempt < r["run_attempt"] and r["conclusion"] == "success", + "run_created": r["created_at"], + "workflow": r["name"], + "prs": [{"number": n, "base": pr_by_num[n]["baseRefName"]} for n in r["pr_nums"]], + "job_id": job["id"], + "job": job["name"], + "tests": extract_tests(job["name"], text), + "buckets": classify(text), + "log_ok": bool(text.strip()), + } + ) + return jobs_out + + +def append_ledger(ledger_path: Path, jobs_out: list[dict], repo: str, today: dt.date) -> int: + seen: set[str] = set() + if ledger_path.exists(): + seen = {json.loads(line)["dedup_key"] for line in ledger_path.open(encoding="utf-8")} + added = 0 + with ledger_path.open("a", encoding="utf-8") as fh: + for entry in jobs_out: + week = dt.datetime.fromisoformat(entry["run_created"]).strftime("%G-W%V") + for test in entry["tests"] or [""]: + key = f"{entry['job_id']}:{test}" + if key in seen: + continue + seen.add(key) + record = { + "dedup_key": key, + "fetched_at": today.isoformat(), + "week": week, + "repo": repo, + "test": test, + **{k: entry[k] for k in LEDGER_FIELDS}, + } + fh.write(json.dumps(record) + "\n") + added += 1 + return added + + +def ranked_tests(jobs_out: list[dict]) -> list[dict]: + freq: dict[str, list[dict]] = defaultdict(list) + for e in jobs_out: + for t in e["tests"]: + freq[t].append(e) + table = [ + { + "test": test, + "distinct_runs": len({e["run"] for e in entries}), + "distinct_prs": len({p["number"] for e in entries for p in e["prs"]}), + "attempts": len(entries), + "recovered_on_retry": sum(e["recovered_same_run"] for e in entries), + "buckets": sorted({b for e in entries for b in e["buckets"]}), + "prs": sorted({p["number"] for e in entries for p in e["prs"]}), + } + for test, entries in freq.items() + ] + table.sort(key=lambda x: (-x["distinct_prs"], -x["distinct_runs"], x["test"])) + return table + + +def bucket_incidents(jobs_out: list[dict]) -> dict[str, dict[str, int]]: + """Count distinct jobs/runs/PRs per systemic bucket, so a cascade reads as one incident.""" + jobs: dict[str, set[int]] = defaultdict(set) + runs: dict[str, set[int]] = defaultdict(set) + prs: dict[str, set[int]] = defaultdict(set) + for e in jobs_out: + for b in e["buckets"]: + jobs[b].add(e["job_id"]) + runs[b].add(e["run"]) + prs[b].update(p["number"] for p in e["prs"]) + return {b: {"jobs": len(jobs[b]), "runs": len(runs[b]), "prs": len(prs[b])} for b in sorted(jobs)} + + +def weekly_history(ledger_path: Path) -> dict[str, dict[str, int]]: + hist: dict[str, dict[str, int]] = {} + if ledger_path.exists(): + for line in ledger_path.open(encoding="utf-8"): + rec = json.loads(line) + if rec["test"]: + weeks = hist.setdefault(rec["test"], {}) + weeks[rec["week"]] = weeks.get(rec["week"], 0) + 1 + return {t: dict(sorted(w.items())) for t, w in sorted(hist.items())} + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--repo", default="opsmill/infrahub") + ap.add_argument( + "--base", + action="append", + default=[], + help="base-branch glob(s) to keep, e.g. release-1.11 or 'release-*' (default: all)", + ) + ap.add_argument("--days", type=int, default=7) + ap.add_argument("--since", type=dt.date.fromisoformat) + ap.add_argument("--cache", type=Path, default=Path.home() / "ci-cache") + args = ap.parse_args() + + today = dt.datetime.now(tz=dt.UTC).date() + since = args.since or today - dt.timedelta(days=args.days) + repo_dir = args.cache / args.repo.replace("/", "-") + win_dir = repo_dir / "windows" / f"{since.isoformat()}_{today.isoformat()}" + (win_dir / "joblogs").mkdir(parents=True, exist_ok=True) + + pr_by_num = {p["number"]: p for p in list_prs(args.repo, since, args.base)} + print(f"[collect] {len(pr_by_num)} PRs in scope (bases: {args.base or 'all'})", file=sys.stderr) + + runs = list_runs(args.repo, since, today) + (win_dir / "runs.jsonl").write_text("".join(json.dumps(r) + "\n" for r in runs)) + print(f"[collect] {len(runs)} pull_request runs since {since}", file=sys.stderr) + + matched = match_runs_to_prs(runs, pr_by_num, pr_head_shas(args.repo, list(pr_by_num))) + + # Attempts worth reading: every earlier attempt of a retried run (those + # failures are what the retry "fixed"), plus the final attempt when it + # failed outright. Runs cancelled on attempt 1 are concurrency noise. + targets = [(r, a) for r in matched for a in range(1, r["run_attempt"] + (r["conclusion"] == "failure"))] + print(f"[collect] {len(matched)} runs matched to PRs, {len(targets)} run-attempts to inspect", file=sys.stderr) + + jobs_out = collect_failed_jobs(args.repo, targets, pr_by_num, win_dir) + (win_dir / "failed_jobs_with_tests.json").write_text(json.dumps(jobs_out, indent=1)) + new_records = append_ledger(repo_dir / "ledger.jsonl", jobs_out, args.repo, today) + + report = { + "window": {"since": since.isoformat(), "until": today.isoformat()}, + "base_filter": args.base or "all", + "prs_in_scope": len(pr_by_num), + "runs_matched": len(matched), + "runs_retried": sum(1 for r in matched if r["run_attempt"] > 1), + "runs_recovered_on_retry": sum(1 for r in matched if r["run_attempt"] > 1 and r["conclusion"] == "success"), + "runs_failed_final": sum(1 for r in matched if r["conclusion"] == "failure"), + "failed_jobs": len(jobs_out), + "ranked_tests": ranked_tests(jobs_out), + "bucket_incidents": bucket_incidents(jobs_out), + "weekly_history": weekly_history(repo_dir / "ledger.jsonl"), + "new_ledger_records": new_records, + } + (win_dir / "report-data.json").write_text(json.dumps(report, indent=1)) + print(json.dumps(report, indent=1)) + print(f"[collect] window dir: {win_dir}", file=sys.stderr) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.github/workflows/ci-docker-image.yml b/.github/workflows/ci-docker-image.yml index 7d07688feb1..6d36a94e809 100644 --- a/.github/workflows/ci-docker-image.yml +++ b/.github/workflows/ci-docker-image.yml @@ -202,9 +202,33 @@ jobs: password: ${{ secrets.HARBOR_PASSWORD }} - name: Sign manifest + env: + IMAGE: ${{ vars.HARBOR_HOST }}/${{ github.repository }}@${{ needs.merge.outputs.digest }} run: | - cosign sign --yes --recursive --new-bundle-format=false --use-signing-config=false \ - "${{ vars.HARBOR_HOST }}/${{ github.repository }}@${{ needs.merge.outputs.digest }}" + # cosign gives up after two internal attempts when a Sigstore transparency + # log write fails, so a brief rekor.sigstore.dev blip is enough to fail a + # release. Retry the whole command instead; five attempts 60s apart cover a + # ~4-minute outage. The sbom job carries a verbatim copy of this function: + # the jobs run on separate runners, so sharing it would take a checkout or a + # third-party action in the signing path. Keep the two copies identical. + retry() { + local attempt + for attempt in 1 2 3 4 5; do + if "$@"; then + return 0 + fi + if [ "${attempt}" -lt 5 ]; then + echo "::warning::${1} ${2} failed (attempt ${attempt}/5), retrying in 60s" + sleep 60 + fi + done + echo "::error::${1} ${2} failed after 5 attempts" + return 1 + } + + retry cosign sign --yes --recursive \ + --new-bundle-format=false --use-signing-config=false \ + "${IMAGE}" sbom: needs: merge @@ -236,20 +260,16 @@ jobs: "${{ vars.HARBOR_HOST }}/${{ github.repository }}@${{ needs.merge.outputs.digest }}" \ --output cyclonedx-json=infrahub-sbom.cdx.json - - name: Attest SBOM (SPDX) - run: | - cosign attest --yes \ - --type spdxjson \ - --predicate infrahub-sbom.spdx.json \ - "${{ vars.HARBOR_HOST }}/${{ github.repository }}@${{ needs.merge.outputs.digest }}" - - - name: Attest SBOM (CycloneDX) - run: | - cosign attest --yes \ - --type cyclonedx \ - --predicate infrahub-sbom.cdx.json \ - "${{ vars.HARBOR_HOST }}/${{ github.repository }}@${{ needs.merge.outputs.digest }}" - + # Uploaded before the attestations so that a transparency-log outage cannot cost + # us the SBOMs themselves: v1.10.7 shipped without any because the attest step + # failed first and skipped this one. + # + # overwrite is required precisely because this now runs before a step that can + # fail. Artifacts are scoped to the run rather than the attempt, so on a re-run + # this name already exists from the earlier attempt and the default overwrite: + # false would fail the upload, breaking the recovery path this ordering exists to + # protect. Every caller passes a version unique to its invocation, so the only + # artifact this can replace is the same SBOM from a previous attempt. - name: Upload SBOM artifacts uses: actions/upload-artifact@v7 with: @@ -258,3 +278,37 @@ jobs: infrahub-sbom.spdx.json infrahub-sbom.cdx.json retention-days: 90 + overwrite: true + + - name: Attest SBOMs + env: + IMAGE: ${{ vars.HARBOR_HOST }}/${{ github.repository }}@${{ needs.merge.outputs.digest }} + run: | + # Same transparency-log flakiness the sign job guards against; this is a + # verbatim copy of that job's retry function (separate runners, so it cannot + # be shared without a checkout or a third-party action in the signing path). + # Keep the two copies identical. + retry() { + local attempt + for attempt in 1 2 3 4 5; do + if "$@"; then + return 0 + fi + if [ "${attempt}" -lt 5 ]; then + echo "::warning::${1} ${2} failed (attempt ${attempt}/5), retrying in 60s" + sleep 60 + fi + done + echo "::error::${1} ${2} failed after 5 attempts" + return 1 + } + + retry cosign attest --yes \ + --type spdxjson \ + --predicate infrahub-sbom.spdx.json \ + "${IMAGE}" + + retry cosign attest --yes \ + --type cyclonedx \ + --predicate infrahub-sbom.cdx.json \ + "${IMAGE}" diff --git a/CHANGELOG.md b/CHANGELOG.md index de01b74612f..a8c94ab4d7b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -584,6 +584,13 @@ docker compose restart - Restructured the node stages of `development/Dockerfile` (shared node base, frontend, docs) with BuildKit cache mounts, enabling parallel builds and much better layer-cache reuse; `pnpm install` only re-runs when a manifest changes. - Significantly reduced the size of the Infrahub container image: the build toolchain now lives in a dedicated build stage that is excluded from the runtime image, and the `numpy` and `pyarrow` dependencies are no longer installed by default (`pyarrow` remains available via the `object-transfer` extra for `infrahubctl object load`). +## [Infrahub - v1.10.9](https://github.com/opsmill/infrahub/tree/infrahub-v1.10.9) - 2026-08-19 + +### Fixed + +- Fixed a crash when loading the Tasks page where a task was tagged with a related node whose kind could no longer be resolved (for example a deleted definition or a stale tag). Such unresolvable related nodes are now omitted from the task instead of causing a GraphQL resolver error. ([#9662](https://github.com/opsmill/infrahub/issues/9662)) +- Renaming an attribute in a schema on a branch now correctly closes the old attribute instead of only opening a newer path to the new attribute. This issue would have been mostly invisible to the user unless an attribute was renamed on a user's branch and that branch was then rebased, in which case there could be duplicated paths to the new attribute which could result in unexpected behavior when updating its value. + ## [Infrahub - v1.10.8](https://github.com/opsmill/infrahub/tree/infrahub-v1.10.8) - 2026-08-14 ### Fixed diff --git a/backend/infrahub/graphql/mutations/branch.py b/backend/infrahub/graphql/mutations/branch.py index 9740a0ef117..04f345695a4 100644 --- a/backend/infrahub/graphql/mutations/branch.py +++ b/backend/infrahub/graphql/mutations/branch.py @@ -49,10 +49,19 @@ class BranchCreateInput(InputObjectType): id = String(required=False) name = String(required=True) description = String(required=False) - origin_branch = String(required=False) - branched_from = String(required=False) + origin_branch = InputField( + String(required=False), + deprecation_reason="Branches can only be created from the default branch. Will be removed after version 1.12.", + ) + branched_from = InputField( + String(required=False), + deprecation_reason="branched_from is set by the server and cannot be provided. Will be removed after version 1.12.", + ) sync_with_git = Boolean(required=False) - is_isolated = InputField(Boolean(required=False), deprecation_reason="Non isolated mode is not supported anymore") + is_isolated = InputField( + Boolean(required=False), + deprecation_reason="Non-isolated mode is not supported anymore. Will be removed after version 1.12.", + ) class BranchCreate(Mutation): @@ -77,13 +86,18 @@ async def mutate( background_execution: bool = False, wait_until_completion: bool = True, ) -> Self: - if data.origin_branch and data.origin_branch != registry.default_branch: - raise ValueError(f"origin_branch must be '{registry.default_branch}'") + origin_branch = data.get("origin_branch") + if origin_branch is not None and origin_branch != registry.default_branch: + raise ValidationError(f"origin_branch must be '{registry.default_branch}'") + if data.get("branched_from") is not None: + raise ValidationError( + "branched_from input is deprecated and cannot be set, it will be the create time of the branch." + ) graphql_context: GraphqlContext = info.context task: dict | None = None - model = BranchCreateModel(**data) + model = BranchCreateModel(**{key: value for key, value in data.items() if value is not None}) await apply_external_context(graphql_context=graphql_context, context_input=context) try: diff --git a/backend/infrahub/log.py b/backend/infrahub/log.py index e93a54053ab..563d7e1d8a9 100644 --- a/backend/infrahub/log.py +++ b/backend/infrahub/log.py @@ -59,6 +59,22 @@ def filter(self, record: logging.LogRecord) -> bool: return type(exception) not in self._suppressed_types +def install_traceback_suppression_filter() -> TracebackSuppressionFilter: + """Install the traceback suppression filter on the Prefect run loggers and return it. + + Prefect ships flow/task run logs to its API; drop tracebacks for failures that are reported as a + clean classified reason rather than a crash to debug. The filter reads the shared registry that + each expected-failure type opts into via suppress_traceback_in_logs. + + The installed filter is returned so a caller that must leave logging state as it found it can + remove it again from every logger in PREFECT_RUN_LOGGERS. + """ + traceback_filter = TracebackSuppressionFilter(_TRACEBACK_SUPPRESSED_TYPES) + for prefect_logger_name in PREFECT_RUN_LOGGERS: + logging.getLogger(prefect_logger_name).addFilter(traceback_filter) + return traceback_filter + + def clear_log_context() -> None: structlog.contextvars.clear_contextvars() @@ -86,13 +102,8 @@ def configure_logging(production: bool, log_level: str) -> None: # the infrahub logger importlib.import_module("prefect.main") - # Prefect ships flow/task run logs to its API; drop tracebacks for failures that - # are reported as a clean classified reason rather than a crash to debug. Installed after the - # prefect.main import above so it survives Prefect's logging reset; reads the shared registry that - # each expected-failure type opts into via suppress_traceback_in_logs. - traceback_filter = TracebackSuppressionFilter(_TRACEBACK_SUPPRESSED_TYPES) - for prefect_logger_name in PREFECT_RUN_LOGGERS: - logging.getLogger(prefect_logger_name).addFilter(traceback_filter) + # Installed after the prefect.main import above so it survives Prefect's logging reset. + install_traceback_suppression_filter() shared_processors: list[Processor] = [ structlog.contextvars.merge_contextvars, diff --git a/backend/tests/component/graphql/mutations/test_branch.py b/backend/tests/component/graphql/mutations/test_branch.py index 3132b397802..726213303bb 100644 --- a/backend/tests/component/graphql/mutations/test_branch.py +++ b/backend/tests/component/graphql/mutations/test_branch.py @@ -1,3 +1,4 @@ +from dataclasses import dataclass from typing import Any from unittest.mock import AsyncMock, patch @@ -19,6 +20,7 @@ from infrahub.core.schema.schema_branch import SchemaBranch from infrahub.core.timestamp import Timestamp from infrahub.database import InfrahubDatabase +from infrahub.exceptions import BranchNotFoundError from infrahub.graphql.initialization import prepare_graphql_params from infrahub.services import InfrahubServices from infrahub.services.adapters.workflow.local import WorkflowLocalExecution @@ -28,6 +30,158 @@ from tests.helpers.graphql import graphql, graphql_mutation from tests.helpers.test_app import TestInfrahubApp +BRANCH_CREATE = """ +mutation( + $name: String! + $description: String + $originBranch: String + $branchedFrom: String + $syncWithGit: Boolean +) { + BranchCreate( + data: { + name: $name + description: $description + origin_branch: $originBranch + branched_from: $branchedFrom + sync_with_git: $syncWithGit + } + ) { + ok + object { + id + name + description + origin_branch + branched_from + sync_with_git + } + } +} +""" + +BRANCHED_FROM_ERROR = "branched_from input is deprecated and cannot be set, it will be the create time of the branch." + + +@dataclass +class RejectedInputTestCase: + name: str + """Descriptive name for the test scenario.""" + + branch_name: str + """Name of the branch the mutation attempts to create.""" + + variables: dict[str, Any] + """Optional BranchCreate input variables sent alongside the branch name.""" + + expected_message: str + """The exact GraphQL error message the mutation must return.""" + + +REJECTED_INPUT_TEST_CASES: list[RejectedInputTestCase] = [ + RejectedInputTestCase( + name="branched_from_timestamp_rejected", + branch_name="own-branched-from", + variables={"branchedFrom": "2020-01-01T00:00:00.000Z"}, + expected_message=BRANCHED_FROM_ERROR, + ), + RejectedInputTestCase( + name="branched_from_empty_string_rejected", + branch_name="empty-branched-from", + variables={"branchedFrom": ""}, + expected_message=BRANCHED_FROM_ERROR, + ), + RejectedInputTestCase( + name="origin_branch_other_than_default_rejected", + branch_name="other-origin-branch", + variables={"originBranch": "not-the-default-branch"}, + expected_message="origin_branch must be 'main'", + ), + RejectedInputTestCase( + name="origin_branch_empty_string_rejected", + branch_name="empty-origin-branch", + variables={"originBranch": ""}, + expected_message="origin_branch must be 'main'", + ), +] + + +class TestBranchCreateInputValidation(TestInfrahubApp): + @pytest.mark.parametrize( + "test_case", + [pytest.param(tc, id=tc.name) for tc in REJECTED_INPUT_TEST_CASES], + ) + async def test_server_owned_input_is_rejected( + self, + db: InfrahubDatabase, + default_branch: Branch, + register_core_models_schema: SchemaBranch, + session_admin: AccountSession, + client: InfrahubClient, + service: InfrahubServices, + test_case: RejectedInputTestCase, + ) -> None: + """branched_from and origin_branch are decided by the server, so a client-supplied value is an input error.""" + result = await graphql_mutation( + query=BRANCH_CREATE, + db=db, + service=service, + branch=default_branch, + account_session=session_admin, + variables={"name": test_case.branch_name} | test_case.variables, + ) + + assert result.errors is not None + assert len(result.errors) == 1 + assert result.errors[0].message == test_case.expected_message + + with pytest.raises(BranchNotFoundError): + await Branch.get_by_name(db=db, name=test_case.branch_name) + + @pytest.mark.parametrize( + ("branch_name", "variables"), + [ + pytest.param("omitted-optional-input", {}, id="optional_input_omitted"), + pytest.param( + "null-optional-input", + {"description": None, "originBranch": None, "branchedFrom": None, "syncWithGit": None}, + id="optional_input_explicitly_null", + ), + ], + ) + async def test_unset_optional_input_falls_back_to_defaults( + self, + db: InfrahubDatabase, + default_branch: Branch, + register_core_models_schema: SchemaBranch, + session_admin: AccountSession, + client: InfrahubClient, + service: InfrahubServices, + branch_name: str, + variables: dict[str, Any], + ) -> None: + """An explicit null says no more than omitting the field, so both must land on the server defaults.""" + result = await graphql_mutation( + query=BRANCH_CREATE, + db=db, + service=service, + branch=default_branch, + account_session=session_admin, + variables={"name": branch_name} | variables, + ) + + assert result.errors is None + assert result.data + assert result.data["BranchCreate"]["ok"] is True + + branch = await Branch.get_by_name(db=db, name=branch_name) + assert isinstance(branch.description, str) + assert not branch.description + assert branch.origin_branch == default_branch.name + assert branch.sync_with_git is True + assert isinstance(branch.branched_from, str) + assert branch.branched_from + class TestBranchCreate(TestInfrahubApp): async def test_branch_create( diff --git a/backend/tests/component/webhook/test_traceback_suppression.py b/backend/tests/component/webhook/test_traceback_suppression.py index 701708b6590..06fcb850b04 100644 --- a/backend/tests/component/webhook/test_traceback_suppression.py +++ b/backend/tests/component/webhook/test_traceback_suppression.py @@ -1,12 +1,12 @@ from __future__ import annotations import logging +from typing import TYPE_CHECKING import httpx import pytest from prefect import flow, task -from infrahub.log import configure_logging from infrahub.webhook.classifier import ( EXPECTED_DELIVERY_ERRORS, ClassifiedFailure, @@ -14,6 +14,10 @@ WebhookDeliveryError, WebhookFailureClassifier, ) +from tests.helpers.log import traceback_suppression + +if TYPE_CHECKING: + from collections.abc import Generator CLASSIFIED_MESSAGE = "The target responded with HTTP 404." @@ -46,12 +50,14 @@ async def _send_classifying_in_task() -> None: @pytest.fixture -def configured_logging() -> None: - # Register the traceback filter on the Prefect run loggers, as production startup does. - configure_logging(production=False, log_level="DEBUG") +def traceback_suppression_installed() -> Generator[None, None, None]: + with traceback_suppression(): + yield -async def test_classified_failure_logs_no_traceback(configured_logging: None, caplog: pytest.LogCaptureFixture) -> None: +async def test_classified_failure_logs_no_traceback( + traceback_suppression_installed: None, caplog: pytest.LogCaptureFixture +) -> None: with ( caplog.at_level(logging.INFO, logger="prefect.flow_runs"), pytest.raises(WebhookDeliveryError, match=r"^The target responded with HTTP 404\.$"), @@ -66,7 +72,7 @@ async def test_classified_failure_logs_no_traceback(configured_logging: None, ca async def test_classified_failure_from_task_logs_no_traceback( - configured_logging: None, caplog: pytest.LogCaptureFixture + traceback_suppression_installed: None, caplog: pytest.LogCaptureFixture ) -> None: # The transport error is caught and classified inside the task, so the failure the engine records # for the task run is a delivery error whose traceback is dropped — not the raw transport stacktrace. @@ -84,7 +90,7 @@ async def test_classified_failure_from_task_logs_no_traceback( async def test_unclassified_failure_logs_a_traceback( - configured_logging: None, caplog: pytest.LogCaptureFixture + traceback_suppression_installed: None, caplog: pytest.LogCaptureFixture ) -> None: with ( caplog.at_level(logging.INFO, logger="prefect.flow_runs"), diff --git a/backend/tests/functional/webhook/conftest.py b/backend/tests/functional/webhook/conftest.py index c0940a99e24..584ec3e4416 100644 --- a/backend/tests/functional/webhook/conftest.py +++ b/backend/tests/functional/webhook/conftest.py @@ -12,6 +12,9 @@ from infrahub.core.node import Node from infrahub.events.models import EventBranchContext, EventContext from infrahub.task_manager.flow_run.prefect_client import PrefectClientAdapter +from infrahub.trigger.constants import NAME_SEPARATOR +from infrahub.trigger.models import TriggerType +from infrahub.trigger.setup import gather_all_automations from infrahub.webhook.tasks import process from infrahub.workflows.catalogue import ( WEBHOOK_CONFIGURE, @@ -101,6 +104,23 @@ async def prefect_client(prefect_test_fixture: None) -> AsyncGenerator[PrefectCl yield client +@pytest.fixture(scope="class", autouse=True) +async def delete_webhook_automations(prefect_client: PrefectClient) -> AsyncGenerator[None, None]: + """Delete the webhook automations a test class registered, once the class is done with them. + + The Prefect test server is session-scoped, so an automation outlives the class that created it + while the webhook node behind it is dropped with the class database. A surviving all-branches + automation then turns every event any later test emits in this session into a scheduled + webhook-process run that no worker ever executes, filling the server's database and pushing the + run count past the API's 200-row page size. + """ + yield + webhook_prefix = f"{TriggerType.WEBHOOK.value}{NAME_SEPARATOR}" + for automation in await gather_all_automations(client=prefect_client): + if automation.id and automation.name.startswith(webhook_prefix): + await prefect_client.delete_automation(automation_id=automation.id) + + @pytest.fixture(scope="class") def flow_run_querier(prefect_client: PrefectClient) -> FlowRunQuerying: """A read-only view of the Prefect client, exposing only flow-run querying to tests.""" diff --git a/backend/tests/functional/webhook/test_render.py b/backend/tests/functional/webhook/test_render.py index 9216e9e17d9..11627ee8cf2 100644 --- a/backend/tests/functional/webhook/test_render.py +++ b/backend/tests/functional/webhook/test_render.py @@ -6,6 +6,7 @@ from uuid import uuid4 from prefect.client.schemas.filters import DeploymentFilter, DeploymentFilterId +from prefect.client.schemas.sorting import FlowRunSort from prefect.events.schemas.events import Event, Resource from prefect.types import DateTime @@ -19,6 +20,7 @@ if TYPE_CHECKING: from infrahub_sdk import InfrahubClient from prefect.client.orchestration import PrefectClient + from prefect.client.schemas.objects import FlowRun from infrahub.database import InfrahubDatabase @@ -63,7 +65,22 @@ async def test_branchless_event_triggers_webhook_process( deployment = await prefect_client.read_deployment_by_name(f"{WEBHOOK_PROCESS.name}/{WEBHOOK_PROCESS.name}") deployment_filter = DeploymentFilter(id=DeploymentFilterId(any_=[deployment.id])) - runs_before = len(await prefect_client.read_flow_runs(deployment_filter=deployment_filter)) + + async def read_process_runs() -> list[FlowRun]: + # Earlier tests leave webhook-process runs behind and a read is capped at the server's + # 200-row page size, so run counts saturate and only a run's identity is a usable signal. + return await prefect_client.read_flow_runs( + deployment_filter=deployment_filter, sort=FlowRunSort.EXPECTED_START_TIME_DESC + ) + + runs_before = {run.id for run in await read_process_runs()} + + async def read_new_runs() -> list[FlowRun]: + return [ + run + for run in await read_process_runs() + if run.id not in runs_before and run.parameters.get("webhook_id") == webhook.id + ] # A branch-less event: the resource carries no infrahub.branch.name, the id is a UUID and the # occurred time a datetime -- all values the action parameters must render as plain strings. @@ -76,10 +93,20 @@ async def test_branchless_event_triggers_webhook_process( ) await prefect_client._client.post("/events", json=[event.model_dump(mode="json")]) - runs_after = runs_before + new_runs: list[FlowRun] = [] for _ in range(PREFECT_EVENT_WAIT_SECONDS): - runs_after = len(await prefect_client.read_flow_runs(deployment_filter=deployment_filter)) - if runs_after > runs_before: + new_runs = await read_new_runs() + if new_runs: break await asyncio.sleep(1) - assert runs_after > runs_before, "webhook-process deployment was not run; server-side parameter render failed" + assert new_runs, "webhook-process deployment was not run; server-side parameter render failed" + + # Every value the deployment receives has to be a plain string, an absent branch included. + parameters = new_runs[0].parameters + assert parameters["event_id"] == str(event.id) + assert parameters["event_type"] == "infrahub.node.created" + assert parameters["event_occured_at"] == "2026-01-01 00:00:00+00:00" + branch_name = parameters["branch_name"] + assert isinstance(branch_name, str) + assert not branch_name + assert parameters["event_payload"] == {"data": {"node_id": "abc"}, "context": {}} diff --git a/backend/tests/helpers/log.py b/backend/tests/helpers/log.py new file mode 100644 index 00000000000..448e894d30f --- /dev/null +++ b/backend/tests/helpers/log.py @@ -0,0 +1,25 @@ +"""Install the infrahub.log traceback suppression filter for a test, then remove it again.""" + +from __future__ import annotations + +import logging +from contextlib import contextmanager +from typing import TYPE_CHECKING + +from infrahub.log import PREFECT_RUN_LOGGERS, install_traceback_suppression_filter + +if TYPE_CHECKING: + from collections.abc import Iterator + + from infrahub.log import TracebackSuppressionFilter + + +@contextmanager +def traceback_suppression() -> Iterator[TracebackSuppressionFilter]: + """Register the traceback filter on the Prefect run loggers, as production startup does, then remove it.""" + traceback_filter = install_traceback_suppression_filter() + try: + yield traceback_filter + finally: + for prefect_logger_name in PREFECT_RUN_LOGGERS: + logging.getLogger(prefect_logger_name).removeFilter(traceback_filter) diff --git a/backend/tests/helpers/task_manager.py b/backend/tests/helpers/task_manager.py index 60ae083975a..13416d0a570 100644 --- a/backend/tests/helpers/task_manager.py +++ b/backend/tests/helpers/task_manager.py @@ -5,17 +5,46 @@ the calls are slow (several seconds of API round-trips), so fixtures should reuse a single setup per process instead of repeating it for every test or test class. +A failure is remembered the same way a success is. An unreachable Prefect test +server does not fail fast — the setup blocks until the pytest timeout fires — so +retrying it for every later test class costs that timeout each time and buries the +original cause under a wall of identical errors. + Tests that intentionally corrupt the shared task manager state must restore it themselves before yielding back, otherwise later tests will observe the corruption. """ +from collections.abc import Awaitable, Callable + from infrahub.workflows.initialization import setup_task_manager -_state = {"initialized": False} + +class TaskManagerSetup: + def __init__(self, setup: Callable[[], Awaitable[None]] = setup_task_manager) -> None: + self._setup = setup + self._initialized = False + self._failure: BaseException | None = None + + async def run_once(self) -> None: + if self._failure is not None: + raise RuntimeError("Prefect task manager setup already failed in this process") from self._failure + + if self._initialized: + return + + try: + await self._setup() + # The pytest timeout raises Failed, which derives from BaseException, and that is + # the failure worth remembering most. + except BaseException as exc: + self._failure = exc + raise + + self._initialized = True + + +_setup = TaskManagerSetup() async def setup_task_manager_once() -> None: - if _state["initialized"]: - return - await setup_task_manager() - _state["initialized"] = True + await _setup.run_once() diff --git a/backend/tests/helpers/test_app.py b/backend/tests/helpers/test_app.py index c550b891304..b51f7e46635 100644 --- a/backend/tests/helpers/test_app.py +++ b/backend/tests/helpers/test_app.py @@ -96,20 +96,28 @@ async def bus_simulator( # Creating another service object to get service correctly initialized is a hack. # We should either reuse `service` fixture (leading to circular fixture dependencies issue atm), # or ideally properly patch production code responsible for Bus instantiation instead + original = config.OVERRIDE.message_bus bus = BusSimulator() _ = await InfrahubServices.new(database=db, workflow=WorkflowLocalExecution(), message_bus=bus) config.OVERRIDE.message_bus = bus - with dependency_provider.scope(build_message_bus, lambda: bus): - yield bus + try: + with dependency_provider.scope(build_message_bus, lambda: bus): + yield bus + finally: + config.OVERRIDE.message_bus = original @pytest.fixture(scope="class") async def memory_cache( self, db: InfrahubDatabase, dependency_provider: Provider ) -> AsyncGenerator[MemoryCache, None]: + original = config.OVERRIDE.cache cache = MemoryCache() config.OVERRIDE.cache = cache - with dependency_provider.scope(build_cache, lambda: cache): - yield cache + try: + with dependency_provider.scope(build_cache, lambda: cache): + yield cache + finally: + config.OVERRIDE.cache = original @pytest.fixture(scope="class") async def register_internal_schema(self, db: InfrahubDatabase, default_branch: Branch) -> SchemaBranch: diff --git a/backend/tests/unit/helpers/test_task_manager.py b/backend/tests/unit/helpers/test_task_manager.py new file mode 100644 index 00000000000..2ad537a9331 --- /dev/null +++ b/backend/tests/unit/helpers/test_task_manager.py @@ -0,0 +1,70 @@ +import pytest + +from tests.helpers.task_manager import TaskManagerSetup + + +class TimeoutFailure(BaseException): + """Stands in for pytest's Failed, which derives from BaseException rather than Exception.""" + + +class RecordingSetup: + """Counts how many times the task manager setup was actually run.""" + + def __init__(self) -> None: + self.calls = 0 + + async def __call__(self) -> None: + self.calls += 1 + + +class FailingSetup(RecordingSetup): + """Stands in for a Prefect test server that accepts connections but never answers.""" + + def __init__(self, error: BaseException) -> None: + super().__init__() + self.error = error + + async def __call__(self) -> None: + await super().__call__() + raise self.error + + +async def test_setup_runs_once_across_repeated_calls() -> None: + setup = RecordingSetup() + once = TaskManagerSetup(setup=setup) + + await once.run_once() + await once.run_once() + await once.run_once() + + assert setup.calls == 1 + + +async def test_failed_setup_is_reported_without_being_rerun() -> None: + setup = FailingSetup(TimeoutError("prefect server is unreachable")) + once = TaskManagerSetup(setup=setup) + + with pytest.raises(TimeoutError, match=r"^prefect server is unreachable$"): + await once.run_once() + + for _ in range(3): + with pytest.raises( + RuntimeError, match=r"^Prefect task manager setup already failed in this process$" + ) as exc_info: + await once.run_once() + assert isinstance(exc_info.value.__cause__, TimeoutError) + + assert setup.calls == 1 + + +async def test_failure_that_bypasses_exception_is_remembered() -> None: + setup = FailingSetup(TimeoutFailure("Timeout >300.0s")) + once = TaskManagerSetup(setup=setup) + + with pytest.raises(TimeoutFailure, match=r"^Timeout >300\.0s$"): + await once.run_once() + + with pytest.raises(RuntimeError, match=r"^Prefect task manager setup already failed in this process$"): + await once.run_once() + + assert setup.calls == 1 diff --git a/backend/tests/unit/test_log.py b/backend/tests/unit/test_log.py index cbe2ad9663b..a01c74b052e 100644 --- a/backend/tests/unit/test_log.py +++ b/backend/tests/unit/test_log.py @@ -1,9 +1,19 @@ from __future__ import annotations import logging +from typing import TYPE_CHECKING -from infrahub.log import _TRACEBACK_SUPPRESSED_TYPES, TracebackSuppressionFilter, suppress_traceback_in_logs +from infrahub.log import ( + _TRACEBACK_SUPPRESSED_TYPES, + PREFECT_RUN_LOGGERS, + TracebackSuppressionFilter, + suppress_traceback_in_logs, +) from infrahub.webhook.classifier import ClassifiedFailure, StatusClass, WebhookDeliveryError +from tests.helpers.log import traceback_suppression + +if TYPE_CHECKING: + from collections.abc import Sequence def _record(exception: BaseException | None) -> logging.LogRecord: @@ -42,3 +52,33 @@ class _ExpectedFailureError(Exception): ... # The production filter is wired to this shared registry, so a decorated type is suppressed. assert TracebackSuppressionFilter(_TRACEBACK_SUPPRESSED_TYPES).filter(_record(_ExpectedFailureError())) is False + + +def test_startup_installs_the_filter_on_the_prefect_run_loggers() -> None: + """Importing infrahub.log configures logging for the process, which is what installs the filter.""" + installed_on = [ + name + for name in PREFECT_RUN_LOGGERS + if any(isinstance(log_filter, TracebackSuppressionFilter) for log_filter in logging.getLogger(name).filters) + ] + assert installed_on == list(PREFECT_RUN_LOGGERS) + + +def _run_logger_filters() -> dict[str, Sequence[object]]: + # Logger.filters is a union of filter forms; the identity of what is attached is all that matters here. + return {name: list(logging.getLogger(name).filters) for name in PREFECT_RUN_LOGGERS} + + +def test_traceback_suppression_leaves_logging_state_unchanged() -> None: + """The suppression context must hand logging back exactly as it found it.""" + root_logger = logging.getLogger() + level_before, filters_before = root_logger.level, _run_logger_filters() + + with traceback_suppression() as traceback_filter: + assert _run_logger_filters() == { + name: [*filters_before[name], traceback_filter] for name in PREFECT_RUN_LOGGERS + } + assert root_logger.level == level_before + + assert _run_logger_filters() == filters_before + assert root_logger.level == level_before diff --git a/changelog/+attribute-rename-close-branch-owned-edge.fixed.md b/changelog/+attribute-rename-close-branch-owned-edge.fixed.md deleted file mode 100644 index b371cb5e674..00000000000 --- a/changelog/+attribute-rename-close-branch-owned-edge.fixed.md +++ /dev/null @@ -1 +0,0 @@ -Renaming an attribute in a schema on a branch now correctly closes the old attribute instead of only opening a newer path to the new attribute. This issue would have been mostly invisible to the user unless an attribute was renamed on a user's branch and that branch was then rebased, in which case there could be duplicated paths to the new attribute which could result in unexpected behavior when updating its value. diff --git a/changelog/+block-user-branched-from.fixed.md b/changelog/+block-user-branched-from.fixed.md new file mode 100644 index 00000000000..80ad8511e77 --- /dev/null +++ b/changelog/+block-user-branched-from.fixed.md @@ -0,0 +1 @@ +The `BranchCreate` GraphQL mutation now rejects a client-supplied `branched_from` value with an error, and the field is marked deprecated. `branched_from` is an internal field managed by the application. The `origin_branch` field, which was already rejected for any value other than the default branch, is now also marked deprecated. An empty string is now rejected for either field instead of being silently accepted, and an explicit `null` on an optional field is treated the same as omitting it. diff --git a/changelog/9662.fixed.md b/changelog/9662.fixed.md deleted file mode 100644 index 4969f244638..00000000000 --- a/changelog/9662.fixed.md +++ /dev/null @@ -1 +0,0 @@ -Fixed a crash when loading the Tasks page where a task was tagged with a related node whose kind could no longer be resolved (for example a deleted definition or a stale tag). Such unresolvable related nodes are now omitted from the task instead of causing a GraphQL resolver error. diff --git a/dev/guidelines/backend/testing.md b/dev/guidelines/backend/testing.md index 74ce53aba82..a2e73271146 100644 --- a/dev/guidelines/backend/testing.md +++ b/dev/guidelines/backend/testing.md @@ -111,6 +111,33 @@ The module provides individual node/generic schemas (`CAR`, `DEVICE`, `TAG`, `PE `config.SETTINGS` is populated from `INFRAHUB_*` environment variables at process start, so values exported in the developer's shell leak into the test process. Any test whose behavior depends on a settings field must pin it in a save/restore fixture (set the value, `yield`, restore the original) — see `import_every_remote_branch` in `backend/tests/integration/git/conftest.py`. Never assume a field holds its default. +## Leave process-global state as you found it + +Under `pytest-xdist` every test in a worker shares one interpreter, so whatever a test changes outside +its own fixtures stays changed for every test that follows it there. Touch global state only through a +save/restore fixture (change it, `yield`, restore the original). Pinning a setting, above, is one case +of that rule; it also covers: + +- the `logging` module — root and per-logger levels, handlers, filters +- `structlog` configuration +- module-level registries, caches and singletons +- environment variables (prefer `monkeypatch.setenv`, which restores on teardown) +- `sys.path`, `sys.modules`, warning filters + +**Never call an application startup routine from a test.** `infrahub.log.configure_logging` is the +example to learn from: it runs once at process start and owns the process when it does — setting the +root log level, replacing the root handler and reconfiguring structlog — so, being startup code, it has +no counterpart that undoes any of that. Called from a fixture it silently reconfigures every later test +in the worker. Install only the piece the test needs, extracting it from the startup routine when it is +not already reusable, and undo it after the `yield` — see `traceback_suppression` in +`backend/tests/helpers/log.py`, which the webhook suppression tests use to install the traceback +suppression filter alone rather than calling `configure_logging`. + +Such a leak is invisible locally and expensive in CI. A root logger left at `DEBUG` overrides the +`WARNING` level `pytest_configure` pins, and the Neo4j driver then logs a line per Bolt message for +every test that follows in that worker: one job produced 185k lines of driver output and pushed three +unrelated tests past their 300s timeout. + ## Dataclass Test Case Pattern For parametrized tests with multiple scenarios, use dataclasses to define test cases. This pattern provides type safety, readable test IDs, and clear separation between test data and test logic. diff --git a/dev/knowledge/backend/testing.md b/dev/knowledge/backend/testing.md index b5d0bae4624..046ec5ebfd7 100644 --- a/dev/knowledge/backend/testing.md +++ b/dev/knowledge/backend/testing.md @@ -418,6 +418,20 @@ async def test_logs_warning(caplog: pytest.LogCaptureFixture) -> None: This matches the pattern used in `test_webhook_header.py` and `test_models.py`. +### Prefect Server State Outlives the Test Class + +The Prefect test server is session-scoped — one per xdist worker — while the database and the +fixtures that populate it are class-scoped, so whatever a class registers on that server survives +it. Two rules follow: + +- Delete the automations a class created at its teardown. A surviving all-branches webhook + automation turns every event any later test emits into a scheduled flow run — no worker runs in + the functional suite, so nothing executes them — filling the server's SQLite database. +- Never assert on a flow-run count. `read_flow_runs()` returns at most `PREFECT_API_DEFAULT_LIMIT` + (200) rows and the API rejects a larger `limit`, so once that page is full a before/after + comparison saturates and can never be true again. Read newest-first + (`FlowRunSort.EXPECTED_START_TIME_DESC`) and identify the run by its id or parameters instead. + ### Functional Tests with `TestInfrahubApp` `TestInfrahubApp` provides a `memory_cache` fixture (class-scoped) that injects a `MemoryCache` via `dependency_provider.scope(build_cache, ...)`. Use it in functional tests to pre-fill and assert on cache state: diff --git a/docker-compose.yml b/docker-compose.yml index 62d92738aff..534e664caf7 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -267,7 +267,7 @@ services: - 6362:6362 task-manager: - image: "${INFRAHUB_DOCKER_IMAGE:-registry.opsmill.io/opsmill/infrahub}:${VERSION:-1.10.8}" + image: "${INFRAHUB_DOCKER_IMAGE:-registry.opsmill.io/opsmill/infrahub}:${VERSION:-1.11.0}" command: uvicorn --host 0.0.0.0 --port 4200 --factory infrahub.prefect_server.app:create_infrahub_prefect restart: unless-stopped depends_on: @@ -300,7 +300,7 @@ services: retries: 5 infrahub-server: - image: "${INFRAHUB_DOCKER_IMAGE:-registry.opsmill.io/opsmill/infrahub}:${VERSION:-1.10.8}" + image: "${INFRAHUB_DOCKER_IMAGE:-registry.opsmill.io/opsmill/infrahub}:${VERSION:-1.11.0}" restart: unless-stopped command: > gunicorn --config backend/infrahub/serve/gunicorn_config.py @@ -346,7 +346,7 @@ services: deploy: mode: replicated replicas: 2 - image: "${INFRAHUB_DOCKER_IMAGE:-registry.opsmill.io/opsmill/infrahub}:${VERSION:-1.10.8}" + image: "${INFRAHUB_DOCKER_IMAGE:-registry.opsmill.io/opsmill/infrahub}:${VERSION:-1.11.0}" command: prefect worker start --type infrahubasync --pool infrahub-worker --with-healthcheck restart: unless-stopped depends_on: diff --git a/docs/docs/release-notes/infrahub/release-1_10_9.mdx b/docs/docs/release-notes/infrahub/release-1_10_9.mdx new file mode 100644 index 00000000000..b9d0d4f215b --- /dev/null +++ b/docs/docs/release-notes/infrahub/release-1_10_9.mdx @@ -0,0 +1,27 @@ +--- +title: Release 1.10.9 +release_date: 2026-08-19 +release_type: patch +description: "Fixes a crash on the Tasks page when a task references a related node whose kind can no longer be resolved, and an attribute rename on a branch leaving the old attribute open, which could duplicate paths to the renamed attribute after a rebase." +--- + + + + + + + + + + + + + + + +
Release Number1.10.9
Release DateAugust 19th, 2026
Tag[infrahub-v1.10.9](https://github.com/opsmill/infrahub/releases/tag/infrahub-v1.10.9)
+ +### Fixed + +- Fixed a crash when loading the Tasks page where a task was tagged with a related node whose kind could no longer be resolved (for example a deleted definition or a stale tag). Such unresolvable related nodes are now omitted from the task instead of causing a GraphQL resolver error. ([#9662](https://github.com/opsmill/infrahub/issues/9662)) +- Renaming an attribute in a schema on a branch now correctly closes the old attribute instead of only opening a newer path to the new attribute. This issue would have been mostly invisible to the user unless an attribute was renamed on a user's branch and that branch was then rebased, in which case there could be duplicated paths to the new attribute which could result in unexpected behavior when updating its value. diff --git a/frontend/app/src/shared/api/graphql/generated/types.ts b/frontend/app/src/shared/api/graphql/generated/types.ts index 13b61aed644..422fcb3dd92 100644 --- a/frontend/app/src/shared/api/graphql/generated/types.ts +++ b/frontend/app/src/shared/api/graphql/generated/types.ts @@ -296,12 +296,14 @@ export type BranchCreate = { }; export type BranchCreateInput = { + /** @deprecated branched_from is set by the server and cannot be provided */ branched_from?: InputMaybe; description?: InputMaybe; id?: InputMaybe; /** @deprecated Non isolated mode is not supported anymore */ is_isolated?: InputMaybe; name: Scalars['String']['input']; + /** @deprecated Branches can only be created from the default branch */ origin_branch?: InputMaybe; sync_with_git?: InputMaybe; }; diff --git a/pyproject.toml b/pyproject.toml index 5954f3e6607..1d63780d5ed 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -45,10 +45,10 @@ dependencies = [ "lunr>=0.7.0.post1,<0.8", "starlette-exporter>=0.23,<0.24", "prometheus-client>=0.25,<0.26", - "python-multipart==0.0.27", # Required by FastAPI to upload large files + "python-multipart==0.0.31", # Required by FastAPI to upload large files "asgi-correlation-id==4.2.0", # Middleware for FastAPI to generate ID per request "bcrypt>=4.1,<4.2", # Used to hash and validate password - "pyjwt==2.12.1", # Used to manage JWT tokens + "pyjwt==2.13.0", # Used to manage JWT tokens "uvicorn[standard]>=0.32,<0.33", "opentelemetry-instrumentation-aio-pika==0.65b0", "opentelemetry-instrumentation-fastapi==0.65b0", diff --git a/python_sdk b/python_sdk index f9e28cfd595..99a380ac145 160000 --- a/python_sdk +++ b/python_sdk @@ -1 +1 @@ -Subproject commit f9e28cfd5958946759f113fd9fe29422adc8fcea +Subproject commit 99a380ac145cb549687bc2b8030cf5edf2f5a492 diff --git a/python_testcontainers/infrahub_testcontainers/docker-compose-cluster.test.yml b/python_testcontainers/infrahub_testcontainers/docker-compose-cluster.test.yml index 9801b98d859..58f5ce12ecf 100644 --- a/python_testcontainers/infrahub_testcontainers/docker-compose-cluster.test.yml +++ b/python_testcontainers/infrahub_testcontainers/docker-compose-cluster.test.yml @@ -38,7 +38,10 @@ services: RABBITMQ_DEFAULT_USER: infrahub RABBITMQ_DEFAULT_PASS: infrahub healthcheck: - test: rabbitmq-diagnostics -q check_port_connectivity + # raw TCP probe instead of rabbitmq-diagnostics: each diagnostics call + # boots a full Erlang VM (~2s CPU), which at a 1s interval pegs two + # cores per broker for the life of the stack + test: ["CMD", "bash", "-c", "