"""Regression suite and report — run the tests, publish the result, alert on red.

Runs a project's test suite on a schedule (weekday mornings, 01:00 UTC), parses
the JUnit XML it produces, publishes a one-line summary, and notifies a webhook
only when something failed. A green run stays silent so the alert keeps meaning
something.

Written for **Airflow 3.x**, which exposes the TaskFlow decorators from
``airflow.sdk``. On Airflow 2.4-2.x, change the import to::

    from airflow.decorators import dag, task

Dependencies: ``requests`` and the standard library — the JUnit XML is parsed
with ``xml.etree``, so there is no provider package or test-framework plugin to
install on the worker beyond whatever runs your suite.

Before you enable it
--------------------
1. Set three Airflow Variables (Admin -> Variables): ``regression_working_dir``
   (the checkout the command runs in), ``regression_command`` (defaults to a
   pytest invocation writing JUnit XML), and ``regression_alert_webhook``
   (a Slack or Teams incoming webhook URL).
2. The worker must be able to run that command — the checkout, the interpreter
   and the dependencies all have to exist on the worker, not just the scheduler.
3. Trigger once manually and read the task log before unpausing.

If your suite does not emit JUnit XML, point ``regression_command`` at whatever
does and adjust ``parse_results``; everything downstream only reads the counts.

MIT licensed. Review it before you wire it to an on-call channel.
"""

from __future__ import annotations

import subprocess
import xml.etree.ElementTree as ElementTree
from datetime import timedelta
from pathlib import Path

import pendulum
import requests
from airflow.sdk import Variable, dag, task

REPORT_FILENAME = "junit-report.xml"
DEFAULT_COMMAND = f"pytest --junitxml={REPORT_FILENAME}"
SUITE_TIMEOUT_SECONDS = 60 * 45
WEBHOOK_TIMEOUT_SECONDS = 15


@dag(
    dag_id="regression_suite_and_report",
    description="Run the regression suite, publish a summary, alert only on failure.",
    schedule="0 1 * * 1-5",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    max_active_runs=1,
    tags=["codebrewerz", "quality-assurance"],
    default_args={
        "retries": 1,
        "retry_delay": timedelta(minutes=10),
    },
)
def regression_suite_and_report():
    @task
    def run_suite() -> str:
        """Run the suite and return the path to its JUnit XML.

        A non-zero exit code is expected when tests fail, so it is not raised
        here — the report is what the rest of the DAG reads. A missing report
        *is* fatal: it means the command never got as far as running tests.
        """
        working_dir = Path(Variable.get("regression_working_dir"))
        command = Variable.get("regression_command", default=DEFAULT_COMMAND)

        completed = subprocess.run(
            command,
            shell=True,
            cwd=working_dir,
            capture_output=True,
            text=True,
            timeout=SUITE_TIMEOUT_SECONDS,
            check=False,
        )
        print(completed.stdout[-8000:])
        if completed.stderr:
            print(completed.stderr[-4000:])
        print(f"exit code: {completed.returncode}")

        report = working_dir / REPORT_FILENAME
        if not report.is_file():
            raise FileNotFoundError(
                f"{report} was not written — the suite did not run. "
                f"Exit code {completed.returncode}."
            )
        return str(report)

    @task
    def parse_results(report_path: str) -> dict:
        """Sum the counters across every <testsuite> in the report."""
        root = ElementTree.parse(report_path).getroot()
        suites = (
            root.iter("testsuite") if root.tag == "testsuites" else [root]
        )

        totals = {"tests": 0, "failures": 0, "errors": 0, "skipped": 0}
        duration = 0.0
        for suite in suites:
            for key in totals:
                totals[key] += int(suite.get(key, 0))
            duration += float(suite.get("time", 0))

        totals["duration_seconds"] = round(duration, 1)
        totals["broken"] = totals["failures"] + totals["errors"]
        totals["passed"] = totals["tests"] - totals["broken"] - totals["skipped"]
        print(totals)
        return totals

    @task
    def publish_summary(totals: dict, dag_run=None) -> str:
        """One line in the task log, and the run's return value."""
        run_id = getattr(dag_run, "run_id", "manual")
        summary = (
            f"{totals['passed']}/{totals['tests']} passed, "
            f"{totals['broken']} broken, {totals['skipped']} skipped "
            f"in {totals['duration_seconds']}s ({run_id})"
        )
        print(summary)
        return summary

    @task
    def alert_if_broken(totals: dict, summary: str) -> str:
        """Post to the webhook only when the suite is red."""
        if totals["broken"] == 0:
            return "green — no alert sent"

        response = requests.post(
            Variable.get("regression_alert_webhook"),
            json={"text": f":rotating_light: Regression suite red — {summary}"},
            timeout=WEBHOOK_TIMEOUT_SECONDS,
        )
        response.raise_for_status()
        return f"alert sent — {summary}"

    results = parse_results(run_suite())
    alert_if_broken(results, publish_summary(results))


regression_suite_and_report()