SQLMesh Model Unit Test & Breaking Change Gate

id: sqlmesh-model-unit-test-and-breaking-change-gate
namespace: company.analytics

description: |
  Gates SQLMesh data transformation releases by executing the project's own model unit
  tests, detecting breaking schema mutations in the plan diff, and materializing an
  ephemeral virtual data preview environment before warehouse deployment.

inputs:
  - id: target_environment
    type: STRING
    displayName: Target Virtual Environment
    defaults: "pr_preview"
    description: Name of the ephemeral virtual data environment (e.g. pr_preview or
      staging) created for the pull request.

  - id: allow_breaking_changes
    type: BOOL
    displayName: Allow Breaking Model Changes
    defaults: false
    description: When false, any Direct or Indirect Breaking schema mutations
      (dropped columns, altered types, modified grain) halt the deployment gate.
      When true, breaking changes proceed with explicit warning.

  - id: project_bundle
    type: FILE
    displayName: SQLMesh Project Archive
    description: tar.gz or zip bundle of a SQLMesh project, including its models/
      and tests/ directories.
    required: true

  - id: slack_channel
    type: STRING
    displayName: Slack Notification Channel
    defaults: "#analytics-engineering-alerts"
    description: Destination Slack channel for unit test regressions and breaking
      change notifications.

triggers:
  - id: pull_request_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Webhook endpoint called by GitHub Actions or GitLab CI on pull
      request creation or update to evaluate SQLMesh models. Streams the
      uploaded project archive into Kestra internal storage and exposes it as
      trigger.uri.
    key: "{{ secret('SQLMESH_GATE_WEBHOOK_KEY') }}"
    fetchType: STORE
    disabled: true

tasks:
  - id: run_model_gate
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: Extract the uploaded SQLMesh project once so unit testing and plan
      evaluation operate on the exact same checkout.
    tasks:
      - id: extract_project_bundle
        type: io.kestra.plugin.scripts.shell.Commands
        description: Unpack the uploaded SQLMesh project archive into ./project.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: ghcr.io/kestra-io/sqlmesh
        commands:
          - |
            set -euo pipefail
            BUNDLE_PATH="{{ trigger.uri is defined ? trigger.uri : inputs.project_bundle }}"
            mkdir -p project

            if [[ "${BUNDLE_PATH}" == *.tar.gz ]] || [[ "${BUNDLE_PATH}" == *.tgz ]]; then
              tar -xzf "${BUNDLE_PATH}" -C project
            elif [[ "${BUNDLE_PATH}" == *.zip ]]; then
              unzip -q "${BUNDLE_PATH}" -d project
            else
              echo "Unsupported archive format: ${BUNDLE_PATH}" >&2
              exit 1
            fi

      - id: run_sqlmesh_unit_tests
        type: io.kestra.plugin.sqlmesh.cli.SQLMeshCLI
        description: Execute the project's own SQLMesh model unit tests against its own
          fixtures in tests/.
        containerImage: ghcr.io/kestra-io/sqlmesh
        allowFailure: true
        outputFiles:
          - "test_results.log"
          - "test_status.json"
        commands:
          - |
            set +e
            ORIGINAL_DIR="$(pwd)"
            cd project

            sqlmesh test > "${ORIGINAL_DIR}/test_results.log" 2>&1
            TEST_EXIT=$?

            cat "${ORIGINAL_DIR}/test_results.log"

            if [ $TEST_EXIT -eq 0 ]; then
              echo '{"test_status": "PASSED", "failures": []}' > "${ORIGINAL_DIR}/test_status.json"
            else
              echo '{"test_status": "FAILED", "failures": ["Unit tests failed. Check test_results.log for details."]}' > "${ORIGINAL_DIR}/test_status.json"
            fi

      - id: evaluate_plan_changes
        type: io.kestra.plugin.sqlmesh.cli.SQLMeshCLI
        description: Build the target virtual environment and classify model
          modifications into direct breaking, indirect breaking, and
          non-breaking categories from the plan diff.
        containerImage: ghcr.io/kestra-io/sqlmesh
        allowFailure: true
        outputFiles:
          - "plan_output.txt"
          - "plan_classification.json"
        commands:
          - |
            set +e
            ORIGINAL_DIR="$(pwd)"
            cd project

            sqlmesh plan "{{ inputs.target_environment }}" --auto-apply --no-prompts > "${ORIGINAL_DIR}/plan_output.txt" 2>&1
            PLAN_EXIT=$?

            cat "${ORIGINAL_DIR}/plan_output.txt"
            cd "${ORIGINAL_DIR}"

            python3 -c "
            import json, re

            with open('plan_output.txt', 'r') as f:
                content = f.read()

            direct_breaking = bool(re.search(r'Directly Modified:.*\(Breaking\)', content))
            indirect_breaking = '(Indirect Breaking)' in content

            breaking_models = []
            for line in content.splitlines():
                stripped = line.strip()
                if stripped.endswith('(Breaking)') or stripped.endswith('(Indirect Breaking)'):
                    breaking_models.append(stripped)

            classification = {
                'has_breaking': direct_breaking or indirect_breaking,
                'direct_breaking': direct_breaking,
                'indirect_breaking': indirect_breaking,
                'breaking_models': breaking_models,
                'plan_exit_code': $PLAN_EXIT
            }

            with open('plan_classification.json', 'w') as out:
                json.dump(classification, out, indent=2)
            "

  - id: parse_gate_verdict
    type: io.kestra.plugin.scripts.python.Script
    description: Synthesize unit test results and plan change classification into an
      actionable gate verdict.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: python:3.11-slim
    inputFiles:
      test_status.json: "{{
        read(outputs.run_sqlmesh_unit_tests.outputFiles['test_status.json']) }}"
      plan_classification.json: "{{
        read(outputs.evaluate_plan_changes.outputFiles['plan_classification.jso\
        n']) }}"
    outputFiles:
      - "gate_summary.json"
    script: |
      import json

      with open("test_status.json", "r") as f:
          test_data = json.load(f)

      with open("plan_classification.json", "r") as f:
          plan_data = json.load(f)

      test_status = test_data.get("test_status", "UNKNOWN")
      has_breaking = plan_data.get("has_breaking", False)
      allow_breaking = "{{ inputs.allow_breaking_changes }}".lower() == "true"

      if test_status != "PASSED":
          verdict = "TESTS_FAILED"
          reason = "One or more SQLMesh model unit tests failed assertion checks."
      elif has_breaking and not allow_breaking:
          verdict = "UNAPPROVED_BREAKING_CHANGE"
          reason = "Direct or Indirect Breaking model modifications detected without allow_breaking_changes approval."
      else:
          verdict = "GATE_PASSED"
          reason = "All unit tests passed and model plan classification is compliant."

      summary = {
          "verdict": verdict,
          "test_status": test_status,
          "has_breaking": has_breaking,
          "allow_breaking": allow_breaking,
          "direct_breaking": plan_data.get("direct_breaking", False),
          "indirect_breaking": plan_data.get("indirect_breaking", False),
          "breaking_models": plan_data.get("breaking_models", []),
          "reason": reason
      }

      with open("gate_summary.json", "w") as f:
          json.dump(summary, f, indent=2)

      print(f"SQLMesh Gate Evaluation: {verdict}")
      print(f"Reason: {reason}")

  - id: route_gate_verdict
    type: io.kestra.plugin.core.flow.Switch
    description: Route deployment gate actions based on unit testing and model plan
      classification results.
    value: "{{
      fromJson(read(outputs.parse_gate_verdict.outputFiles['gate_summary.json']\
      )).verdict }}"
    cases:
      "TESTS_FAILED":
        - id: alert_test_failure
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Notify analytics engineering channel when SQLMesh model unit tests
            fail.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": ":x: *SQLMesh Model Unit Test Failure*",
              "attachments": [
                {
                  "color": "#D32F2F",
                  "title": "Data Transformation Test Suite Regression",
                  "fields": [
                    {
                      "title": "Target Environment",
                      "value": "{{ inputs.target_environment }}",
                      "short": true
                    },
                    {
                      "title": "Test Status",
                      "value": "FAILED",
                      "short": true
                    },
                    {
                      "title": "Remediation",
                      "value": "Review test_results.log execution artifact. Verify model SQL calculations against fixture expectations in tests/.",
                      "short": false
                    }
                  ],
                  "footer": "Kestra SQLMesh Deployment Gate"
                }
              ]
            }

        - id: fail_on_test_regression
          type: io.kestra.plugin.core.execution.Fail
          description: Halt execution due to failing SQLMesh unit test assertions.
          errorMessage: "SQLMesh model unit tests failed. Deployment blocked to prevent
            faulty transformation logic."

      "UNAPPROVED_BREAKING_CHANGE":
        - id: alert_unapproved_breaking
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Alert data governance team when breaking model changes are
            attempted without explicit approval.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": ":warning: *SQLMesh Breaking Model Changes Blocked*",
              "attachments": [
                {
                  "color": "#FFA000",
                  "title": "Unapproved Direct or Indirect Breaking Changes Detected",
                  "fields": [
                    {
                      "title": "Target Environment",
                      "value": "{{ inputs.target_environment }}",
                      "short": true
                    },
                    {
                      "title": "Action",
                      "value": "Deployment halted. Re-run flow with allow_breaking_changes: true after downstream BI dependencies are validated.",
                      "short": false
                    }
                  ],
                  "footer": "Kestra SQLMesh Deployment Gate"
                }
              ]
            }

        - id: fail_on_breaking_schema
          type: io.kestra.plugin.core.execution.Fail
          description: Halt deployment due to unapproved breaking schema modifications.
          errorMessage: "SQLMesh plan detected Direct or Indirect Breaking model
            modifications. Set allow_breaking_changes=true to proceed."

      "GATE_PASSED":
        - id: log_preview_ready
          type: io.kestra.plugin.core.log.Log
          description: Log confirmation of successful unit testing and virtual preview
            environment deployment.
          message: "SQLMesh deployment gate PASSED. All unit tests succeeded and virtual
            environment '{{ inputs.target_environment }}' is verified and ready
            for consumer testing."