> ## Documentation Index > Fetch the complete documentation index at: https://docs.paradime.io/llms.txt > Use this file to discover all available pages before exploring further. # dbt™ Test Maintainer > Write missing dbt™ tests for changed models, validate them with `dbt test`, and commit the result back to the PR branch automatically from GitHub Actions or a parent agent. A specialist DinoAI agent that writes missing dbt™ tests for any model added or changed in a pull request, runs them to confirm they pass, and commits the result directly to the PR branch. Triggered automatically by a GitHub Actions workflow whenever a `.sql` file is touched — so test coverage gaps never reach `main`. The agent can also be invoked as a sub-agent by the PR reviewer, allowing the reviewer to delegate test-writing as part of a broader code review rather than just flagging the gap in its verdict. **Before You Start** **Paradime** * Your Paradime API endpoint, API key, and API secret — generate these under Workspace Settings → API. Make sure to enable `DinoAI agent API` capabilities. Requires Admin access. **GitHub** * Write access to the repository you want to run the workflow on * Ability to add repository secrets and create GitHub Actions workflows **Recommended reading** Before proceeding, read the Programmable Agents section under **Products → DinoAI**: * Quick Start * YAML Configuration * Tools Reference * Agent-to-Agent Delegation — required reading if you plan to invoke this agent from the PR reviewer **Integrations** The following must already be connected in Paradime: * **Slack** — the agent posts a test summary to `#data-quality` via `post_slack_message` ## What You'll Build By the end of this guide you'll have: * A `test-maintainer` DinoAI agent YAML that reads changed models, infers appropriate tests, writes them to the schema YAML, runs `dbt test`, iterates until green, and commits the result * A Python driver script that collects changed `.sql` files from the PR and triggers one agent session per model in parallel * A GitHub Actions workflow that runs automatically on every PR that adds or modifies a dbt™ model ### What the Agent Does Per Model Once triggered for a model, the agent works through five steps without stopping: ``` STEP 1 — Read the model's .sql file and its schema YAML STEP 2 — Inspect upstream sources to infer test candidates (primary keys, not_null columns, accepted values, FK relationships) STEP 3 — Write the missing tests into the model's schema YAML STEP 4 — Run dbt test --select and iterate until all tests pass STEP 5 — Commit the updated YAML to the PR branch and post a summary to #data-quality and back to the PR as a comment ``` The agent always runs `dbt test` before reporting completion — it never commits tests it hasn't verified pass. If a test fails after being written, the agent diagnoses the failure, adjusts the test definition or SQL, and retries before giving up. The absence of `invoke_agent` from the tool allowlist means this agent cannot delegate further. When used as a child of the PR reviewer, the delegation graph stays exactly two levels deep — the reviewer delegates, the test maintainer executes. ### Tests the Agent Writes The agent infers test candidates by reading the model SQL and upstream source definitions. It writes the following categories of tests: | Test Type | When Written | | --------------------- | ---------------------------------------------------------------------------------------------- | | `unique` + `not_null` | Any column that appears to be a primary key based on name (`_id`, `_key`, `_sk`) or usage | | `not_null` | Columns used in JOIN conditions or GROUP BY clauses in downstream models | | `accepted_values` | Columns with a small, finite set of values inferred from CASE WHEN or WHERE clauses | | `relationships` | Foreign key columns that reference a known `ref()` or `source()` model | | Custom generic tests | Where dbt™ built-in tests are insufficient and a simple SQL assertion would be more expressive | The agent never writes a test it cannot justify from the SQL. When a test candidate is ambiguous it adds a `# TODO: confirm test logic with owner` comment in the YAML rather than guessing. ### Architecture Overview ```mermaid theme={"system"} flowchart TD A([Pull Request opened / updated\n.sql file changed]) --> B[GitHub Actions\ntest-maintainer workflow] B --> C[driver script\ntest_maintainer.py] C --> D[Collect changed .sql files\nfrom git diff] C --> E[Trigger one DinoAI session\nper changed model\nin parallel] E --> F1([test-maintainer\nsession — model A]) E --> F2([test-maintainer\nsession — model B]) F1 & F2 --> G[read_file · ripgrep_search\nread model SQL and schema YAML] F1 & F2 --> H[run_sql_query\ninspect upstream sources] F1 & F2 --> I[write_file\nwrite tests to schema YAML] F1 & F2 --> J[run_terminal_command\ndbt test --select model] F1 & F2 --> K[run_terminal_command\ngit commit · push to PR branch] F1 & F2 --> L[post_slack_message\n→ #data-quality] K & L --> M[driver script\nposts completion comment to PR] ``` ## How It Works When a PR is opened or updated, GitHub Actions runs `test_maintainer.py`, which collects all changed `.sql` files from the diff and fires one `test-maintainer` agent session per model in parallel. Each session reads the model, infers and writes tests, runs `dbt test`, iterates until green, and commits to the PR branch. The driver script polls all sessions concurrently and posts a single completion comment to the PR once every session has finished, summarising what was written per model. Create the following file in your repository at `.dinoai/agents/test-maintainer.yml`. ```yaml title=".dinoai/agents/test-maintainer.yml" lineNumbers theme={"system"} name: test-maintainer version: 1 role: > dbt™ Test Specialist responsible for writing and updating tests across analytics models. goal: > Ensure the target model has appropriate, passing tests committed to the repo. Always write meaningful tests — not just for coverage. ── STEP 1: READ THE MODEL ──────────────────────────────────────────────── Read the target model's .sql file in full using read_file. Then find its schema YAML (e.g. _.yml or schema.yml in the same directory) and read the existing column and test definitions. ── STEP 2: INFER TEST CANDIDATES ──────────────────────────────────────── Inspect the model SQL and upstream sources to identify what tests are appropriate. For each column, consider: - Does the name suggest a primary key? (_id, _key, _sk, _pk) → unique + not_null - Is the column used in a JOIN or GROUP BY in a downstream model? → not_null - Does a CASE WHEN or WHERE clause imply a finite value set? → accepted_values - Does the column reference a known ref() or source()? → relationships Use run_sql_query to inspect the actual data distribution for ambiguous columns before deciding. Never write a test you cannot justify from the SQL or the data. ── STEP 3: WRITE THE TESTS ────────────────────────────────────────────── Write the inferred tests into the model's schema YAML using write_file. Preserve all existing formatting and ordering. Only add — never remove or reorder existing test entries. If a test candidate is ambiguous, add a comment `# TODO: confirm test logic with owner` on the line above the test entry rather than omitting it entirely. ── STEP 4: RUN AND ITERATE ────────────────────────────────────────────── Run the tests via run_terminal_command: dbt test --select If any tests fail: 1. Read the failure output carefully. 2. Diagnose whether the failure is a bad test definition or a genuine data quality issue. 3. If a bad definition: fix the test in the YAML and re-run. 4. If a genuine data quality issue: add a severity: warn to the test config so it does not block the pipeline, and add a comment `# DATA ISSUE: — investigate before promoting to error` above the test entry. 5. Iterate until all tests are either passing or set to severity: warn with an explanation comment. NEVER commit tests that are failing at severity: error. ── STEP 5: COMMIT AND NOTIFY ──────────────────────────────────────────── Stage and commit all changed YAML files to the PR branch: git add git commit -m "test: add missing tests for [DinoAI]" git push origin HEAD: Always use `git push origin HEAD:` — never plain `git push`. Then post a summary to #data-quality via post_slack_message: ✅ *Test Maintainer — * *Tests added:* () *Tests passing:* / *Data quality warnings:* (severity: warn — investigate separately) *Committed to branch:* `` If invoked as a sub-agent, call notify_parent_session with the same summary so the parent agent (e.g. pr-reviewer) can include the result in its own report. ── GUARDRAILS ──────────────────────────────────────────────────────────── - NEVER commit tests that are failing at severity: error. - NEVER remove or modify existing passing tests. - NEVER guess at accepted_values without checking the actual data via run_sql_query first. - NEVER reformat or reorder unchanged YAML sections. - If dbt test output is ambiguous, re-run with --debug before concluding. - If the schema YAML does not exist yet, create it following the dbt™ schema YAML convention for the project (check an existing file first). backstory: > You are precise and methodical. You read the model and its upstream sources before writing tests. You always run dbt test before reporting completion. You write tests that mean something — not just not_null on every column for the sake of coverage. A test that fires on real bad data is worth ten cosmetic tests. tools: mode: allowlist list: - read_file - write_file - run_sql_query - run_terminal_command - search_files_and_directories - ripgrep_search - post_slack_message - notify_parent_session slack: channel: "#data-quality" ``` `notify_parent_session` is included in the allowlist so this agent can be invoked as a sub-agent by the PR reviewer. When triggered directly from the GitHub Actions workflow, the tool is present but never called — it only activates when a parent session ID is passed via `invoke_agent`. Create `scripts/test_maintainer.py`. This script runs inside GitHub Actions and is responsible for: * Collecting all changed `.sql` files from the PR diff * Firing one `test-maintainer` agent session per changed model in parallel * Posting a "started" comment to the PR immediately * Polling all sessions concurrently until each completes * Posting a single completion comment to the PR with a per-model summary ```python title="scripts/test:maintainer.py" lineNumbers theme={"system"} """ test_maintainer.py ------------------ GitHub Actions driver for the `test-maintainer` DinoAI agent. Fires one agent session per changed dbt™ model in parallel, polls until all sessions complete, and posts a summary back to the PR. Environment variables (required): PARADIME_API_ENDPOINT – Paradime API endpoint PARADIME_API_KEY – Paradime API key PARADIME_API_SECRET – Paradime API secret GITHUB_EVENT_PATH – Set automatically by GitHub Actions GITHUB_TOKEN – Set automatically by GitHub Actions GITHUB_REPOSITORY – Set automatically by GitHub Actions """ import json import os import sys import subprocess import time import urllib.request from concurrent.futures import ThreadPoolExecutor, as_completed from paradime import Paradime from paradime.apis.dinoai_agents.types import DinoaiAgentRunStatus AGENT_NAME = "test-maintainer" TIMEOUT = 1800 # 30 minutes per session POLL_INTERVAL = 20 def _build_client() -> Paradime: required = { "PARADIME_API_ENDPOINT": os.environ.get("PARADIME_API_ENDPOINT"), "PARADIME_API_KEY": os.environ.get("PARADIME_API_KEY"), "PARADIME_API_SECRET": os.environ.get("PARADIME_API_SECRET"), } missing = [k for k, v in required.items() if not v] if missing: print(f"ERROR: Missing env vars: {', '.join(missing)}", file=sys.stderr) raise SystemExit(1) return Paradime( api_endpoint=required["PARADIME_API_ENDPOINT"], api_key=required["PARADIME_API_KEY"], api_secret=required["PARADIME_API_SECRET"], ) def _post_pr_comment(repo: str, pr_number: int, body: str, gh_token: str) -> None: req = urllib.request.Request( f"https://api.github.com/repos/{repo}/issues/{pr_number}/comments", data=json.dumps({"body": body}).encode(), headers={ "Authorization": f"Bearer {gh_token}", "Accept": "application/vnd.github+json", }, method="POST", ) urllib.request.urlopen(req).read() def _model_name_from_path(path: str) -> str: """Extract the dbt™ model name from a file path, e.g. models/marts/fct_orders.sql → fct_orders.""" return os.path.splitext(os.path.basename(path))[0] def _run_single_model( paradime: Paradime, model_name: str, sql_path: str, branch_name: str, ) -> dict: """ Trigger a test-maintainer session for one model, poll until completion, and return a result dict with model name, status, and final output. """ message = ( f"Add missing dbt™ tests for model `{model_name}` (file: `{sql_path}`).\n\n" f"Branch: {branch_name}\n\n" f"Read the model SQL and its schema YAML, infer appropriate tests " f"(unique, not_null, accepted_values, relationships), write them, " f"run `dbt test --select {model_name}` and iterate until all tests " f"pass or are set to severity: warn with an explanation. " f"Commit the changes to branch `{branch_name}` using " f"`git push origin HEAD:{branch_name}` and post a summary to #data-quality." ) trigger = paradime.dinoai_agents.trigger_run(agent=AGENT_NAME, message=message) session_id = trigger.agent_session_id print(f"[test-maintainer] 🚀 {model_name} → session {session_id}") start = time.time() while True: run = paradime.dinoai_agents.get_run(agent_session_id=session_id) elapsed = int(time.time() - start) if run.status == DinoaiAgentRunStatus.COMPLETED: print(f"[test-maintainer] ✅ {model_name} completed ({elapsed}s)") return { "model": model_name, "status": "completed", "output": run.messages[-1].content if run.messages else "(no output)", "session": session_id, } if run.status == DinoaiAgentRunStatus.FAILED: last = run.messages[-1].content if run.messages else "no output" print(f"[test-maintainer] ❌ {model_name} failed ({elapsed}s)") return { "model": model_name, "status": "failed", "output": last, "session": session_id, } if elapsed > TIMEOUT: print(f"[test-maintainer] ⏱ {model_name} timed out") return { "model": model_name, "status": "timeout", "output": f"Timed out after {TIMEOUT}s", "session": session_id, } print(f"[test-maintainer] {model_name} — {run.status.value} ({elapsed}s) … {POLL_INTERVAL}s") time.sleep(POLL_INTERVAL) def main() -> None: paradime = _build_client() gh_token = os.environ.get("GITHUB_TOKEN", "") repo = os.environ.get("GITHUB_REPOSITORY", "") # 1. Read PR context with open(os.environ["GITHUB_EVENT_PATH"]) as f: event = json.load(f) pr = event["pull_request"] pr_number = pr["number"] pr_title = pr["title"] branch_name = pr["head"]["ref"] base_sha = pr["base"]["sha"] head_sha = pr["head"]["sha"] # 2. Find changed .sql files changed = subprocess.check_output( ["git", "diff", "--name-only", f"{base_sha}...{head_sha}"] ).decode().splitlines() sql_files = [f for f in changed if f.endswith(".sql") and "models/" in f] if not sql_files: print("[test-maintainer] No dbt™ model SQL files changed. Skipping.") raise SystemExit(0) models = [(_model_name_from_path(p), p) for p in sql_files] print(f"[test-maintainer] Found {len(models)} changed model(s):") for name, path in models: print(f" - {name} ({path})") # 3. Post "started" comment to PR if gh_token: model_list = "\n".join(f"- `{name}` (`{path}`)" for name, path in models) _post_pr_comment( repo, pr_number, f"🧪 **DinoAI test maintainer started** — PR #{pr_number}: {pr_title}\n\n" f"Writing and running missing tests for {len(models)} model(s):\n{model_list}\n\n" f"Changes will be committed to `{branch_name}`. Check `#data-quality` for live updates.", gh_token, ) # 4. Trigger all sessions in parallel results = [] with ThreadPoolExecutor(max_workers=len(models)) as executor: futures = { executor.submit(_run_single_model, paradime, name, path, branch_name): name for name, path in models } for future in as_completed(futures): results.append(future.result()) # 5. Post completion comment to PR if gh_token: completed = [r for r in results if r["status"] == "completed"] failed = [r for r in results if r["status"] != "completed"] sections = [] for r in results: icon = "✅" if r["status"] == "completed" else "❌" sections.append( f"### {icon} `{r['model']}`\n\n" f"{r['output']}\n\n" f"*Session: `{r['session']}`*" ) footer = ( f"**{len(completed)}/{len(results)} models** had tests written and committed to `{branch_name}`." ) if failed: footer += f"\n\n⚠️ {len(failed)} session(s) did not complete — check `#data-quality` for details." _post_pr_comment( repo, pr_number, f"🧪 **DinoAI test maintainer — complete**\n\n" + "\n\n---\n\n".join(sections) + f"\n\n---\n\n{footer}", gh_token, ) # Exit with error if any session failed so the workflow is marked red if any(r["status"] != "completed" for r in results): raise SystemExit(1) if __name__ == "__main__": main() ``` Sessions run in parallel via `ThreadPoolExecutor` — if a PR changes three models, all three agent sessions start at the same time. Total wall-clock time is bounded by the slowest single model, not the sum of all three. The script exits with code `1` if any session fails or times out, which marks the GitHub Actions job as failed. This makes test-writing failures visible in the PR checks panel alongside lint and CI results. The driver script authenticates with Paradime using three values. Add the following as GitHub Actions secrets in your repository under **Settings → Secrets and variables → Actions**: * `PARADIME_API_KEY` * `PARADIME_API_SECRET` * `PARADIME_API_ENDPOINT` `GITHUB_TOKEN` and `GITHUB_REPOSITORY` are provided automatically by GitHub Actions — you do not need to add them as secrets. Create `.github/workflows/test-maintainer.yml`. This triggers the driver script automatically whenever a PR adds or modifies a file inside `models/`. ```yaml title=".github/workflows/test-maintainer.yml" lineNumbers theme={"system"} name: DinoAI dbt™ test maintainer on: pull_request: types: [opened, synchronize, ready_for_review] paths: - "models/**/*.sql" permissions: pull-requests: write contents: write # required so the agent can push test commits to the PR branch jobs: write-tests: runs-on: ubuntu-latest timeout-minutes: 35 steps: - uses: actions/checkout@v4 with: fetch-depth: 0 ref: ${{ github.head_ref }} # check out the PR branch so the agent can push to it - uses: actions/setup-python@v5 with: python-version: "3.11" - name: Install Paradime SDK run: pip install paradime-io - name: Configure git for agent commits run: | git config user.name "DinoAI Test Maintainer" git config user.email "dinoai-tests@noreply.github.com" - name: Run test maintainer env: PARADIME_API_ENDPOINT: ${{ secrets.PARADIME_API_ENDPOINT }} PARADIME_API_KEY: ${{ secrets.PARADIME_API_KEY }} PARADIME_API_SECRET: ${{ secrets.PARADIME_API_SECRET }} GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} run: python scripts/test_maintainer.py ``` The `paths:` filter restricts the workflow to `.sql` files inside `models/` only. PRs that touch only YAML, Python, or documentation files will not trigger the agent. `contents: write` is required so the agent can commit and push the updated schema YAML files back to the PR branch. Without it, the `git push` inside the agent session will fail with a permissions error. ## What the PR Experience Looks Like Once the workflow is set up, the experience for a PR author is: 1. They open a PR adding or modifying a dbt™ model 2. A comment appears on the PR immediately: > 🧪 **DinoAI test maintainer started** — PR #42: feat/add-revenue-mart > > Writing and running missing tests for 2 model(s): > > * `fct_orders` (`models/marts/fct_orders.sql`) > * `stg_payments` (`models/staging/stg_payments.sql`) > > Changes will be committed to `feat/add-revenue-mart`. Check `#data-quality` for live updates. 3. A new commit appears on their branch: `test: add missing tests for fct_orders [DinoAI]` 4. A completion comment appears on the PR with a per-model breakdown: > 🧪 **DinoAI test maintainer — complete** > > #### ✅ `fct_orders` > > Tests added: `unique` + `not_null` on `order_id`, `not_null` on `customer_id`, `accepted_values` on `status` (values: pending, shipped, delivered, returned) All 4 tests passing. Committed to branch: `feat/add-revenue-mart` > > #### ✅ `stg_payments` > > Tests added: `unique` + `not_null` on `payment_id`, `relationships` on `order_id` → `stg_orders` All 3 tests passing. 1 data quality warning on `amount` (severity: warn — negative values found, investigate before promoting to error) > > **2/2 models** had tests written and committed to `feat/add-revenue-mart`. 5. The same summary is posted to `#data-quality` on Slack. ## Using as a Sub-Agent of the PR Reviewer If you have the PR reviewer set up, you can add `test-maintainer` to its `agents_squad` so the reviewer delegates test-writing rather than just flagging the gap in its verdict. Add the following to your `pr-reviewer-e2e.yml`: ```yaml theme={"system"} agents_squad: - test-maintainer tools: mode: allowlist list: - read_file - search_files_and_directories - ripgrep_search - get_linear_issue - run_sql_query - post_slack_message - invoke_agent # ← add this to allow delegation ``` The PR reviewer can then call `invoke_agent("test-maintainer", ...)` when it detects missing test coverage, and the test maintainer will call `notify_parent_session` with its findings once complete so the reviewer can include the result in its final verdict. When used as a sub-agent, the test maintainer receives the branch name and model name via the `invoke_agent` message rather than from the GitHub Actions context. Make sure the PR reviewer includes both in its delegation message. ## File Structure Your repository should look like this after completing the setup: ``` your-repo/ ├── dbt_project.yml ├── .dinoai/ │ └── agents/ │ └── test-maintainer.yml ├── .github/ │ └── workflows/ │ └── test-maintainer.yml └── scripts/ └── test_maintainer.py ``` ## Related Docs * [**PR Reviewer tutorial** — the parent agent that can delegate to this one](/guides/programmable-agents/end-to-end-pr-reviewer) * [**Agent-to-Agent Delegation** — how to invoke this agent from the PR reviewer](/products/dino-ai/programmable-agents/agent-to-agent-delegation) * [**Programmable Agents — Quick Start** — getting started with DinoAI agents](/products/dino-ai/programmable-agents/quick-start) * [**Programmable Agents — YAML Configuration** — full reference for agent config options](/products/dino-ai/programmable-agents/yaml-configuration) * [**Programmable Agents — Tools Reference** — all available tools including `run_terminal_command` and `notify_parent_session`](/products/dino-ai/programmable-agents/tools-reference) * [**Slack Integration** — connecting Slack to Paradime](/integrations/slack) * [**Paradime API & Credentials** — where to find your API endpoint, key, and secret](/developers/generate-api-keys-legacy) ## Related topics - [dbt™ assets](/products/data-catalog/data-assets/dbt-assets/index.md) - [Run and Test all your dbt™ Models](/products/bolt/creating-schedules/templates/dbt-tm-templates/run-and-test-all-your-dbt-tm-models.md) - [dbt™ Commands](/integrations/pre-commit/dbt-tm-checkpoint-hooks/dbt-tm-commands.md) - [Testing Data Quality](/guides/dbt-fundamentals/getting-started-with-dbt/testing-data-quality.md) - [Unit Testing](/guides/dbt-fundamentals/configuring-your-dbt-project/unit-testing.md)