diff --git a/CHANGELOG.md b/CHANGELOG.md
index 233bb63e..ea28ef4b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## Unreleased
+### Added
+- Added `ci` command for CI/CD-optimized test runs: multi-file support, GitHub Actions annotations and step summary, Azure DevOps annotations, `--fail-on` flag, `--json` output
+
### Fixed
- Fix SQL export generating multiple PRIMARY KEY constraints for composite keys (#1026)
- Preserve parametrized physicalTypes for SQL export (#1086)
diff --git a/README.md b/README.md
index 70704e99..c18fda95 100644
--- a/README.md
+++ b/README.md
@@ -261,6 +261,7 @@ Commands
- [init](#init)
- [lint](#lint)
- [test](#test)
+- [ci](#ci)
- [export](#export)
- [import](#import)
- [catalog](#catalog)
@@ -374,6 +375,8 @@ Data Contract CLI connects to a data source and runs schema and quality tests to
$ datacontract test --server production datacontract.yaml
```
+For CI/CD pipelines, see [`ci`](#ci).
+
To connect to the databases the `server` block in the datacontract.yaml is used to set up the connection.
In addition, credentials, such as username and passwords, may be defined with environment variables.
@@ -1066,6 +1069,148 @@ models:
```
+### ci
+```
+
+ Usage: datacontract ci [OPTIONS] [LOCATIONS]...
+
+ Run tests for CI/CD pipelines. Emits GitHub Actions annotations and step
+ summary.
+
+╭─ Arguments ──────────────────────────────────────────────────────────────────╮
+│ locations [LOCATIONS]... The location(s) (url or path) of the data │
+│ contract yaml file(s). │
+╰──────────────────────────────────────────────────────────────────────────────╯
+╭─ Options ────────────────────────────────────────────────────────────────────╮
+│ --schema TEXT The location (url │
+│ or path) of the │
+│ ODCS JSON Schema │
+│ --server TEXT The server │
+│ configuration to │
+│ run the schema and │
+│ quality tests. Use │
+│ the key of the │
+│ server object in │
+│ the data contract │
+│ yaml file to refer │
+│ to a server, e.g., │
+│ `production`, or │
+│ `all` for all │
+│ servers (default). │
+│ [default: all] │
+│ --publish TEXT The url to publish │
+│ the results after │
+│ the test. │
+│ --output PATH Specify the file │
+│ path where the test │
+│ results should be │
+│ written to (e.g., │
+│ './test-results/TE… │
+│ --output-format [json|junit] The target format │
+│ for the test │
+│ results. │
+│ --logs --no-logs Print logs │
+│ [default: no-logs] │
+│ --json --no-json Print test results │
+│ as JSON to stdout. │
+│ [default: no-json] │
+│ --fail-on TEXT Minimum severity │
+│ that causes a │
+│ non-zero exit code: │
+│ 'warning', 'error', │
+│ or 'never'. │
+│ [default: error] │
+│ --ssl-verification --no-ssl-verific… SSL verification │
+│ when publishing the │
+│ data contract. │
+│ [default: │
+│ ssl-verification] │
+│ --debug --no-debug Enable debug │
+│ logging │
+│ --help Show this message │
+│ and exit. │
+╰──────────────────────────────────────────────────────────────────────────────╯
+
+```
+
+The `ci` command wraps [`test`](#test) with CI/CD-specific features:
+
+- **Multiple contracts**: `datacontract ci contracts/*.yaml`
+- **CI annotations:** Inline annotations for failed checks (GitHub Actions and Azure DevOps)
+- **Markdown summary** of the test results (GitHub Actions)
+- **`--json`**: Print test results as JSON to stdout for machine-readable output
+- **`--fail-on`**: Control the minimum severity that causes a non-zero exit code. Default is `error`; set to `warning` to also fail on warnings, or `never` to always exit 0.
+
+See the [test command](#test) for supported server types and their configuration.
+
+```bash
+# Single contract
+$ datacontract ci datacontract.yaml
+
+# Multiple contracts
+$ datacontract ci contracts/*.yaml
+
+# Fail on warnings too
+$ datacontract ci --fail-on warning datacontract.yaml
+
+# JSON output for scripting
+$ datacontract ci --json datacontract.yaml
+```
+
+
+GitHub Actions workflow example
+
+```yaml
+# .github/workflows/datacontract.yml
+name: Data Contract CI
+
+on:
+ push:
+ branches: [main]
+ pull_request:
+
+jobs:
+ datacontract-ci:
+ runs-on: ubuntu-latest
+ steps:
+ - uses: actions/checkout@v4
+ - uses: actions/setup-python@v5
+ with:
+ python-version: "3.11"
+ - run: pip install datacontract-cli
+ # Test one or more data contracts (supports globs, e.g. contracts/*.yaml)
+ - run: datacontract ci datacontract.yaml
+```
+
+
+
+
+Azure DevOps pipeline example
+
+```yaml
+# azure-pipelines.yml
+trigger:
+ branches:
+ include:
+ - main
+
+pool:
+ vmImage: "ubuntu-latest"
+
+steps:
+ - task: UsePythonVersion@0
+ inputs:
+ versionSpec: "3.11"
+ - script: pip install datacontract-cli
+ displayName: "Install datacontract-cli"
+ # Test one or more data contracts (supports globs, e.g. contracts/*.yaml)
+ - script: datacontract ci datacontract.yaml
+ displayName: "Run data contract tests"
+```
+
+
+
+
### export
```
@@ -1881,10 +2026,11 @@ Create a data contract based on the actual data. This is the fastest way to get
$ datacontract lint
```
-4. Set up a CI pipeline that executes daily for continuous quality checks. You can also report the
- test results to tools like [Data Mesh Manager](https://datamesh-manager.com)
+4. Set up a CI pipeline that executes daily for continuous quality checks. Use the [`ci`](#ci) command for
+ CI-optimized output (GitHub Actions annotations and step summary, Azure DevOps annotations).
+ You can also report the test results to tools like [Data Mesh Manager](https://datamesh-manager.com).
```bash
- $ datacontract test --publish https://api.datamesh-manager.com/api/test-results
+ $ datacontract ci --publish https://api.datamesh-manager.com/api/test-results
```
### Contract-First
diff --git a/datacontract/cli.py b/datacontract/cli.py
index 2eb13997..fea5f851 100644
--- a/datacontract/cli.py
+++ b/datacontract/cli.py
@@ -5,6 +5,7 @@
from pathlib import Path
from typing import Iterable, List, Optional
+import click
import typer
from click import Context
from rich.console import Console
@@ -20,6 +21,7 @@
)
from datacontract.lint.resolve import resolve_data_contract, resolve_data_contract_dict
from datacontract.model.exceptions import DataContractException
+from datacontract.output.ci_output import write_ci_output, write_ci_summary, write_json_results
from datacontract.output.output_format import OutputFormat
from datacontract.output.test_results_writer import write_test_result
@@ -187,6 +189,102 @@ def test(
write_test_result(run, console, output_format, output, data_contract)
+@app.command(name="ci")
+def ci(
+ locations: Annotated[
+ Optional[list[str]],
+ typer.Argument(help="The location(s) (url or path) of the data contract yaml file(s)."),
+ ] = None,
+ schema: Annotated[
+ str,
+ typer.Option(help="The location (url or path) of the ODCS JSON Schema"),
+ ] = None,
+ server: Annotated[
+ str,
+ typer.Option(
+ help="The server configuration to run the schema and quality tests. "
+ "Use the key of the server object in the data contract yaml file "
+ "to refer to a server, e.g., `production`, or `all` for all "
+ "servers (default)."
+ ),
+ ] = "all",
+ publish: Annotated[str, typer.Option(help="The url to publish the results after the test.")] = None,
+ output: Annotated[
+ Path,
+ typer.Option(
+ help="Specify the file path where the test results should be written to (e.g., './test-results/TEST-datacontract.xml')."
+ ),
+ ] = None,
+ output_format: Annotated[OutputFormat, typer.Option(help="The target format for the test results.")] = None,
+ logs: Annotated[bool, typer.Option(help="Print logs")] = False,
+ json_output: Annotated[bool, typer.Option("--json", help="Print test results as JSON to stdout.")] = False,
+ fail_on: Annotated[
+ str,
+ typer.Option(
+ click_type=click.Choice(["warning", "error", "never"], case_sensitive=False),
+ help="Minimum severity that causes a non-zero exit code.",
+ ),
+ ] = "error",
+ ssl_verification: Annotated[
+ bool,
+ typer.Option(help="SSL verification when publishing the data contract."),
+ ] = True,
+ debug: debug_option = None,
+):
+ """
+ Run tests for CI/CD pipelines. Emits GitHub Actions annotations and step summary.
+ """
+ enable_debug_logging(debug)
+
+ if not locations:
+ locations = ["datacontract.yaml"]
+
+ if output and len(locations) > 1:
+ console.print("Error: --output cannot be used with multiple contracts (results would overwrite each other).")
+ raise typer.Exit(code=1)
+
+ if server == "all":
+ server = None
+
+ # Plain text output for CI logs; --json sends human output to stderr.
+ out = Console(stderr=True, no_color=True) if json_output else Console(no_color=True)
+
+ results = []
+ fail_results = {
+ "warning": {"warning", "failed", "error"},
+ "error": {"failed", "error"},
+ "never": set(),
+ }
+ should_fail = False
+
+ for location in locations:
+ out.print(f"Testing {location}")
+ run = DataContract(
+ data_contract_file=location,
+ schema_location=schema,
+ publish_url=publish,
+ server=server,
+ ssl_verification=ssl_verification,
+ ).test()
+ if logs:
+ _print_logs(run, out)
+ results.append((location, run))
+ write_ci_output(run, location, json_mode=json_output)
+ try:
+ write_test_result(run, out, output_format, output)
+ except typer.Exit:
+ pass
+ if run.result in fail_results[fail_on]:
+ should_fail = True
+
+ write_ci_summary(results)
+ if json_output:
+ write_json_results(results)
+
+ if should_fail:
+ raise typer.Exit(code=1)
+
+
@app.command(name="export")
def export(
format: Annotated[ExportFormat, typer.Option(help="The export format.")],
@@ -508,10 +606,12 @@ def api(
uvicorn.run(**uvicorn_args)
-def _print_logs(run):
- console.print("\nLogs:")
+def _print_logs(run, out=None):
+ if out is None:
+ out = console
+ out.print("\nLogs:")
for log in run.logs:
- console.print(log.timestamp.strftime("%y-%m-%d %H:%M:%S"), log.level.ljust(5), log.message)
+ out.print(log.timestamp.strftime("%y-%m-%d %H:%M:%S"), log.level.ljust(5), log.message)
if __name__ == "__main__":
diff --git a/datacontract/output/ci_output.py b/datacontract/output/ci_output.py
new file mode 100644
index 00000000..4df73065
--- /dev/null
+++ b/datacontract/output/ci_output.py
@@ -0,0 +1,151 @@
+import json
+import os
+import sys
+from typing import List, Tuple
+
+from datacontract.model.run import Run
+from datacontract.output.test_results_writer import to_field
+
+
+def _sanitize_md_cell(text: str) -> str:
+ """Escape pipe characters and collapse newlines for use in markdown table cells."""
+ return text.replace("|", "\\|").replace("\r\n", " ").replace("\r", " ").replace("\n", " ").strip()
+
+
+def write_ci_output(run: Run, data_contract_file: str, json_mode: bool = False):
+ """Write CI-specific output for a single contract: annotations only."""
+ out = sys.stderr if json_mode else sys.stdout
+ if os.environ.get("GITHUB_ACTIONS") == "true":
+ _write_github_annotations(run, data_contract_file, out)
+ elif os.environ.get("TF_BUILD") == "True":
+ _write_azure_annotations(run, data_contract_file, out)
+
+
+def write_ci_summary(results: List[Tuple[str, Run]]):
+ """Write aggregated CI step summary for all contracts."""
+ summary_path = os.environ.get("GITHUB_STEP_SUMMARY")
+ if not summary_path:
+ return
+ _write_github_step_summary(results, summary_path)
+
+
+def _sanitize_annotation(text: str | None) -> str:
+ """Collapse newlines and trim text for use in CI annotations."""
+ if not text:
+ return ""
+ return text.replace("%", "%25").replace("\r\n", " ").replace("\r", " ").replace("\n", " ").strip()
+
+
+def _write_github_annotations(run: Run, data_contract_file: str, out=sys.stdout):
+ for check in run.checks:
+ name = _sanitize_annotation(check.name)
+ reason = _sanitize_annotation(check.reason)
+ if check.result in ("failed", "error"):
+ print(f"::error file={data_contract_file}::{name}: {reason}", file=out)
+ elif check.result == "warning":
+ print(f"::warning file={data_contract_file}::{name}: {reason}", file=out)
+
+
+def _write_azure_annotations(run: Run, data_contract_file: str, out=sys.stdout):
+ for check in run.checks:
+ name = _sanitize_annotation(check.name)
+ reason = _sanitize_annotation(check.reason)
+ if check.result in ("failed", "error"):
+ print(f"##vso[task.logissue type=error;sourcepath={data_contract_file}]{name}: {reason}", file=out)
+ elif check.result == "warning":
+ print(f"##vso[task.logissue type=warning;sourcepath={data_contract_file}]{name}: {reason}", file=out)
+
+
+RESULT_EMOJI = {
+ "passed": "🟢 passed",
+ "warning": "🟠 warning",
+ "failed": "🔴 failed",
+ "error": "🔴 error",
+}
+
+
+def _write_github_step_summary(results: List[Tuple[str, Run]], summary_path: str):
+ lines = []
+
+ # Aggregate header (only when multiple contracts)
+ if len(results) > 1:
+ n_total = len(results)
+ result_values = [run.result for _, run in results]
+ has_failures = any(r in ("failed", "error") for r in result_values)
+ has_warnings = any(r == "warning" for r in result_values)
+ if has_failures:
+ overall = "🔴 failed"
+ elif has_warnings:
+ overall = "🟠 warning"
+ else:
+ overall = "🟢 passed"
+ lines.append("## Data Contract CI")
+ lines.append("")
+ n_passed = sum(1 for r in result_values if r == "passed")
+ lines.append(f"**{overall}** — {n_passed}/{n_total} contracts passed")
+ lines.append("")
+ lines.append("| Result | Contract |")
+ lines.append("|--------|----------|")
+ for data_contract_file, run in results:
+ result = RESULT_EMOJI.get(run.result, run.result.value if hasattr(run.result, "value") else str(run.result))
+ lines.append(f"| {result} | {data_contract_file} |")
+ lines.append("")
+
+ # Per-contract detail sections
+ for data_contract_file, run in results:
+ result_display = RESULT_EMOJI.get(run.result, run.result.value if hasattr(run.result, "value") else str(run.result))
+
+ n_total = len(run.checks) if run.checks else 0
+ n_passed = sum(1 for c in run.checks if c.result == "passed") if run.checks else 0
+ n_failed = sum(1 for c in run.checks if c.result == "failed") if run.checks else 0
+ n_warnings = sum(1 for c in run.checks if c.result == "warning") if run.checks else 0
+ n_errors = sum(1 for c in run.checks if c.result == "error") if run.checks else 0
+
+ duration = (
+ (run.timestampEnd - run.timestampStart).total_seconds() if run.timestampStart and run.timestampEnd else 0
+ )
+
+ heading_level = "###" if len(results) > 1 else "##"
+ lines.append(f"{heading_level} Data Contract CI: {data_contract_file}")
+ lines.append("")
+ lines.append(
+ f"**Result: {result_display}** | {n_total} checks | {n_passed} passed | {n_failed} failed | {n_warnings} warnings | {n_errors} errors | {duration:.1f}s"
+ )
+ lines.append("")
+
+ if run.checks:
+ lines.append("| Result | Check | Field | Details |")
+ lines.append("|--------|-------|-------|---------|")
+ for check in sorted(
+ run.checks,
+ key=lambda c: (
+ c.result.value if hasattr(c.result, "value") else str(c.result or ""),
+ c.model or "",
+ c.field or "",
+ ),
+ ):
+ field = _sanitize_md_cell(to_field(run, check) or "")
+ reason = _sanitize_md_cell(check.reason or "")
+ name = _sanitize_md_cell(check.name or "")
+ result = check.result.value if hasattr(check.result, "value") else str(check.result)
+ lines.append(f"| {result} | {name} | {field} | {reason} |")
+ lines.append("")
+
+ with open(summary_path, "a", encoding="utf-8") as f:
+ f.write("\n".join(lines) + "\n")
+
+
+def write_json_results(results: List[Tuple[str, Run]]):
+ """Print test results as JSON to stdout."""
+ if len(results) == 1:
+ location, run = results[0]
+ obj = json.loads(run.model_dump_json(exclude_none=True))
+ obj["location"] = location
+ print(json.dumps(obj, indent=2))
+ else:
+ output = []
+ for location, run in results:
+ obj = json.loads(run.model_dump_json(exclude_none=True))
+ obj["location"] = location
+ output.append(obj)
+ print(json.dumps(output, indent=2))
diff --git a/tests/test_ci_output.py b/tests/test_ci_output.py
new file mode 100644
index 00000000..411c2752
--- /dev/null
+++ b/tests/test_ci_output.py
@@ -0,0 +1,342 @@
+import json
+import os
+import tempfile
+from unittest.mock import patch
+
+from typer.testing import CliRunner
+
+from datacontract.cli import app
+from datacontract.model.run import Check, ResultEnum, Run
+from datacontract.output.ci_output import (
+ _sanitize_annotation,
+ _sanitize_md_cell,
+ write_ci_output,
+ write_ci_summary,
+ write_json_results,
+)
+
+runner = CliRunner()
+
+
+def _make_run(checks):
+ run = Run.create_run()
+ run.checks = checks
+ run.finish()
+ return run
+
+
+# --- Annotation tests ---
+
+
+def test_github_annotations_emitted(capsys):
+ run = _make_run(
+ [
+ Check(type="schema", name="Check col types", result=ResultEnum.failed, reason="type mismatch"),
+ Check(type="schema", name="Check nullability", result=ResultEnum.warning, reason="nullable changed"),
+ Check(type="schema", name="Check row count", result=ResultEnum.passed, reason=None),
+ ]
+ )
+ with patch.dict(os.environ, {"GITHUB_ACTIONS": "true"}):
+ write_ci_output(run, "datacontract.yaml")
+
+ captured = capsys.readouterr()
+ assert "::error file=datacontract.yaml::Check col types: type mismatch" in captured.out
+ assert "::warning file=datacontract.yaml::Check nullability: nullable changed" in captured.out
+ assert "Check row count" not in captured.out
+
+
+def test_no_annotations_outside_ci(capsys):
+ run = _make_run(
+ [
+ Check(type="schema", name="Check col types", result=ResultEnum.failed, reason="type mismatch"),
+ ]
+ )
+ env = {k: v for k, v in os.environ.items() if k not in ("GITHUB_ACTIONS", "TF_BUILD")}
+ with patch.dict(os.environ, env, clear=True):
+ write_ci_output(run, "datacontract.yaml")
+
+ captured = capsys.readouterr()
+ assert "::error" not in captured.out
+ assert "##vso" not in captured.out
+
+
+def test_annotation_format_for_errors(capsys):
+ run = _make_run(
+ [
+ Check(type="quality", name="freshness", result=ResultEnum.error, reason="connection timeout"),
+ ]
+ )
+ with patch.dict(os.environ, {"GITHUB_ACTIONS": "true"}):
+ write_ci_output(run, "my/contract.yaml")
+
+ captured = capsys.readouterr()
+ assert captured.out.strip() == "::error file=my/contract.yaml::freshness: connection timeout"
+
+
+def test_azure_annotations_emitted(capsys):
+ run = _make_run(
+ [
+ Check(type="schema", name="Check col types", result=ResultEnum.failed, reason="type mismatch"),
+ Check(type="schema", name="Check nullability", result=ResultEnum.warning, reason="nullable changed"),
+ Check(type="schema", name="Check row count", result=ResultEnum.passed, reason=None),
+ ]
+ )
+ env = {k: v for k, v in os.environ.items() if k != "GITHUB_ACTIONS"}
+ env["TF_BUILD"] = "True"
+ with patch.dict(os.environ, env, clear=True):
+ write_ci_output(run, "datacontract.yaml")
+
+ captured = capsys.readouterr()
+ assert "##vso[task.logissue type=error;sourcepath=datacontract.yaml]Check col types: type mismatch" in captured.out
+ assert (
+ "##vso[task.logissue type=warning;sourcepath=datacontract.yaml]Check nullability: nullable changed"
+ in captured.out
+ )
+ assert "Check row count" not in captured.out
+
+
+def test_azure_annotation_format_for_errors(capsys):
+ run = _make_run(
+ [
+ Check(type="quality", name="freshness", result=ResultEnum.error, reason="connection timeout"),
+ ]
+ )
+ env = {k: v for k, v in os.environ.items() if k != "GITHUB_ACTIONS"}
+ env["TF_BUILD"] = "True"
+ with patch.dict(os.environ, env, clear=True):
+ write_ci_output(run, "my/contract.yaml")
+
+ captured = capsys.readouterr()
+ assert (
+ captured.out.strip()
+ == "##vso[task.logissue type=error;sourcepath=my/contract.yaml]freshness: connection timeout"
+ )
+
+
+# --- Step summary tests ---
+
+
+def test_step_summary_single_contract():
+ run = _make_run(
+ [
+ Check(type="schema", name="Check types", result=ResultEnum.passed, reason=None),
+ Check(type="schema", name="Check nulls", result=ResultEnum.failed, reason="not nullable"),
+ ]
+ )
+ with tempfile.NamedTemporaryFile(mode="w", suffix=".md", delete=False) as f:
+ summary_path = f.name
+
+ try:
+ env = {k: v for k, v in os.environ.items() if k != "GITHUB_ACTIONS"}
+ env["GITHUB_STEP_SUMMARY"] = summary_path
+ with patch.dict(os.environ, env, clear=True):
+ write_ci_summary([("datacontract.yaml", run)])
+
+ with open(summary_path) as f:
+ content = f.read()
+ assert "## Data Contract CI: datacontract.yaml" in content
+ assert "| Result | Check | Field | Details |" in content
+ assert "Check types" in content
+ assert "Check nulls" in content
+ # Single contract should not have aggregate header
+ assert "| Result | Contract |" not in content
+ finally:
+ os.unlink(summary_path)
+
+
+def test_no_summary_without_env():
+ run = _make_run(
+ [
+ Check(type="schema", name="Check types", result=ResultEnum.passed, reason=None),
+ ]
+ )
+ with tempfile.NamedTemporaryFile(mode="w", suffix=".md", delete=False) as f:
+ summary_path = f.name
+
+ try:
+ env = {k: v for k, v in os.environ.items() if k not in ("GITHUB_ACTIONS", "GITHUB_STEP_SUMMARY")}
+ with patch.dict(os.environ, env, clear=True):
+ write_ci_summary([("datacontract.yaml", run)])
+
+ with open(summary_path) as f:
+ content = f.read()
+ assert content == ""
+ finally:
+ os.unlink(summary_path)
+
+
+def test_step_summary_markdown_structure():
+ run = _make_run(
+ [
+ Check(type="schema", name="Check types", model="orders", field="id", result=ResultEnum.passed, reason=None),
+ Check(type="quality", name="Row count", model="orders", result=ResultEnum.failed, reason="0 rows"),
+ ]
+ )
+ with tempfile.NamedTemporaryFile(mode="w", suffix=".md", delete=False) as f:
+ summary_path = f.name
+
+ try:
+ env = {k: v for k, v in os.environ.items() if k != "GITHUB_ACTIONS"}
+ env["GITHUB_STEP_SUMMARY"] = summary_path
+ with patch.dict(os.environ, env, clear=True):
+ write_ci_summary([("datacontract.yaml", run)])
+
+ with open(summary_path) as f:
+ content = f.read()
+ assert "**Result: 🔴 failed**" in content
+ assert "2 checks" in content
+ assert "1 passed" in content
+ assert "1 failed" in content
+ finally:
+ os.unlink(summary_path)
+
+
+def test_sanitize_md_cell():
+ assert _sanitize_md_cell("foo | bar\nbaz") == "foo \\| bar baz"
+ assert _sanitize_md_cell("line1\r\nline2\rline3") == "line1 line2 line3"
+
+
+def test_sanitize_annotation():
+ assert _sanitize_annotation("error\non line 2\r\nand line 3") == "error on line 2 and line 3"
+ assert _sanitize_annotation(None) == ""
+ assert _sanitize_annotation("50% done") == "50%25 done"
+
+
+def test_step_summary_multi_contract():
+ run_passed = _make_run([Check(type="schema", name="Check types", result=ResultEnum.passed, reason=None)])
+ run_failed = _make_run([Check(type="schema", name="Check nulls", result=ResultEnum.failed, reason="not nullable")])
+ with tempfile.NamedTemporaryFile(mode="w", suffix=".md", delete=False) as f:
+ summary_path = f.name
+
+ try:
+ env = {k: v for k, v in os.environ.items() if k != "GITHUB_ACTIONS"}
+ env["GITHUB_STEP_SUMMARY"] = summary_path
+ with patch.dict(os.environ, env, clear=True):
+ write_ci_summary([("orders.yaml", run_passed), ("customers.yaml", run_failed)])
+
+ with open(summary_path) as f:
+ content = f.read()
+ # Aggregate header
+ assert "## Data Contract CI" in content
+ assert "1/2 contracts passed" in content
+ assert "| Result | Contract |" in content
+ assert "orders.yaml" in content
+ assert "customers.yaml" in content
+ # Per-contract detail sections use ### when multiple
+ assert "### Data Contract CI: orders.yaml" in content
+ assert "### Data Contract CI: customers.yaml" in content
+ finally:
+ os.unlink(summary_path)
+
+
+# --- CLI integration tests ---
+
+
+def test_ci_help():
+ result = runner.invoke(app, ["ci", "--help"])
+ assert result.exit_code == 0
+ assert "CI/CD" in result.stdout
+
+
+def test_ci_with_valid_contract():
+ result = runner.invoke(app, ["ci", "fixtures/lint/valid_datacontract.yaml"])
+ assert result.exit_code == 0
+
+
+def test_ci_with_missing_file():
+ result = runner.invoke(app, ["ci", "nonexistent.yaml"])
+ assert result.exit_code == 1
+
+
+def test_ci_multiple_files():
+ result = runner.invoke(
+ app, ["ci", "fixtures/lint/valid_datacontract.yaml", "fixtures/lint/valid_datacontract.yaml"]
+ )
+ assert result.exit_code == 0
+
+
+def test_ci_multiple_files_with_failure():
+ result = runner.invoke(app, ["ci", "fixtures/lint/valid_datacontract.yaml", "nonexistent.yaml"])
+ assert result.exit_code == 1
+
+
+def test_ci_fail_on_never():
+ result = runner.invoke(app, ["ci", "--fail-on", "never", "nonexistent.yaml"])
+ assert result.exit_code == 0
+
+
+def test_ci_fail_on_warning():
+ # valid_datacontract.yaml produces a warning ("Schema block is missing")
+ result = runner.invoke(app, ["ci", "--fail-on", "warning", "fixtures/lint/valid_datacontract.yaml"])
+ assert result.exit_code == 1
+
+
+def test_ci_fail_on_error_is_default():
+ # valid_datacontract.yaml produces warnings but not errors/failures — should pass
+ result = runner.invoke(app, ["ci", "fixtures/lint/valid_datacontract.yaml"])
+ assert result.exit_code == 0
+
+
+def test_ci_continues_after_failure():
+ """CI should test all contracts even if one fails."""
+ result = runner.invoke(app, ["ci", "nonexistent.yaml", "fixtures/lint/valid_datacontract.yaml"])
+ # Should still report on the second file
+ assert "fixtures/lint/valid_datacontract.yaml" in result.stdout
+ # But exit 1 because the first failed
+ assert result.exit_code == 1
+
+
+def test_ci_output_rejects_multi_with_output():
+ result = runner.invoke(
+ app,
+ [
+ "ci",
+ "--output",
+ "results.xml",
+ "--output-format",
+ "junit",
+ "fixtures/lint/valid_datacontract.yaml",
+ "fixtures/lint/valid_datacontract.yaml",
+ ],
+ )
+ assert result.exit_code == 1
+ assert "cannot be used with multiple contracts" in result.stdout
+
+
+# --- JSON output tests ---
+
+
+def test_json_output_single(capsys):
+ run = _make_run([Check(type="schema", name="Check types", result=ResultEnum.passed, reason=None)])
+ write_json_results([("datacontract.yaml", run)])
+ captured = capsys.readouterr()
+ data = json.loads(captured.out)
+ assert data["result"] == "passed"
+ assert data["location"] == "datacontract.yaml"
+ assert len(data["checks"]) == 1
+
+
+def test_json_output_multi(capsys):
+ run1 = _make_run([Check(type="schema", name="Check types", result=ResultEnum.passed, reason=None)])
+ run2 = _make_run([Check(type="schema", name="Check nulls", result=ResultEnum.failed, reason="not nullable")])
+ write_json_results([("orders.yaml", run1), ("customers.yaml", run2)])
+ captured = capsys.readouterr()
+ data = json.loads(captured.out)
+ assert isinstance(data, list)
+ assert len(data) == 2
+ assert data[0]["result"] == "passed"
+ assert data[0]["location"] == "orders.yaml"
+ assert data[1]["result"] == "failed"
+ assert data[1]["location"] == "customers.yaml"
+
+
+def test_ci_json_flag():
+ env = {k: v for k, v in os.environ.items() if k not in ("GITHUB_ACTIONS", "TF_BUILD")}
+ with patch.dict(os.environ, env, clear=True):
+ result = runner.invoke(app, ["ci", "--json", "fixtures/lint/valid_datacontract.yaml"])
+ assert result.exit_code == 0
+ # With --json, stdout should be clean JSON (human output goes to stderr)
+ data = json.loads(result.stdout)
+ assert "result" in data
+ assert "location" in data
+ assert "checks" in data