gyxx-flow/tests/test_cli_workflows.py

from __future__ import annotations

import io
import json
import os
from datetime import date
from pathlib import Path
from types import SimpleNamespace

import gyxx_flow.cli as cli
from gyxx_flow.catalog import WorkflowCatalog
from gyxx_flow.cli import (
    EXIT_CONFIGURATION,
    EXIT_SUCCESS,
    EXIT_WORKFLOW_FAILED,
    build_default_registry,
    main,
)
from gyxx_flow.core.config import Settings
from gyxx_flow.core.context import RunContext
from gyxx_flow.ops import RunIndex
from gyxx_flow.workflow.model import StepDefinition, WorkflowDefinition
from gyxx_flow.workflow.registry import WorkflowRegistry
from gyxx_flow.workflow.steps import StepExecution


class RecordingStep:
    def __init__(self, *, fail_on: date | None = None) -> None:
        self.fail_on = fail_on
        self.calls: list[RunContext] = []

    def execute(
        self,
        *,
        context: RunContext,
        timeout_seconds: float | None,
        dry_run: bool,
    ) -> StepExecution:
        del timeout_seconds, dry_run
        self.calls.append(context)
        if context.business_date == self.fail_on:
            return StepExecution(exit_code=9, error="fixture failure")
        return StepExecution(exit_code=0)


def test_default_cli_registry_exposes_all_project_scheduled_workflows(
    tmp_path: Path,
) -> None:
    project_root = Path(__file__).parents[1]
    registry = build_default_registry(
        Settings(project_root=project_root, data_root=tmp_path)
    )

    scheduled_ids = {
        entry.workflow_id for entry in registry.catalog.scheduled_workflows()
    }
    assert len(scheduled_ids) == 39
    assert scheduled_ids == {
        schedule.workflow_id for schedule in registry.catalog.schedules
    }
    assert all(registry.is_registered(workflow_id) for workflow_id in scheduled_ids)
    assert registry.is_registered("product.video_upload")
    assert registry.is_registered("product.jd_video_upload")


def _catalog(tmp_path: Path) -> WorkflowCatalog:
    config = tmp_path / "config"
    config.mkdir()
    (config / "workflows.json").write_text(
        json.dumps(
            {
                "schema_version": 3,
                "workflows": [
                    {
                        "id": "shop.metrics.weekly",
                        "module": "shop_intelligence",
                        "trigger": "scheduled",
                        "execution": {"entry": "run.py"},
                        "provenance": {
                            "source_project": "shop",
                            "task_name": "shop-weekly",
                        },
                    },
                    {
                        "id": "shop.missing",
                        "module": "shop_intelligence",
                        "trigger": "unavailable",
                        "execution": {"entry": "missing.py"},
                        "provenance": {"source_project": "shop"},
                    },
                ],
            }
        ),
        encoding="utf-8",
    )
    (config / "schedules.json").write_text(
        json.dumps(
            {
                "schema_version": 1,
                "timezone": "Asia/Shanghai",
                "schedules": [
                    {
                        "workflow_id": "shop.metrics.weekly",
                        "kind": "weekly",
                        "at": "12:00",
                        "days": ["Monday"],
                    }
                ],
            }
        ),
        encoding="utf-8",
    )
    return WorkflowCatalog.load(config)


def _runtime(tmp_path: Path, workflow: WorkflowDefinition) -> tuple[WorkflowRegistry, Settings]:
    registry = WorkflowRegistry(_catalog(tmp_path))
    registry.register(workflow)
    settings = Settings(project_root=tmp_path, data_root=tmp_path / "runtime")
    return registry, settings


def _invoke(
    argv: list[str], registry: WorkflowRegistry, settings: Settings
) -> tuple[int, str, str]:
    stdout = io.StringIO()
    stderr = io.StringIO()
    code = main(
        argv,
        registry=registry,
        settings=settings,
        stdout=stdout,
        stderr=stderr,
    )
    return code, stdout.getvalue(), stderr.getvalue()


def test_list_reports_only_scheduled_catalog_entries_as_json(tmp_path: Path) -> None:
    action = RecordingStep()
    registry, settings = _runtime(
        tmp_path,
        WorkflowDefinition(
            "shop.metrics.weekly", (StepDefinition("collect", action),)
        ),
    )

    code, output, error = _invoke(["list", "--json"], registry, settings)

    assert code == EXIT_SUCCESS
    assert error == ""
    rows = json.loads(output)
    assert rows == [
        {
            "enabled": True,
            "id": "shop.metrics.weekly",
            "module": "shop_intelligence",
            "registered": True,
            "trigger": "scheduled",
        }
    ]


def test_run_is_dry_run_by_default_and_records_a_traceable_run(tmp_path: Path) -> None:
    action = RecordingStep()
    registry, settings = _runtime(
        tmp_path,
        WorkflowDefinition(
            "shop.metrics.weekly", (StepDefinition("collect", action),)
        ),
    )

    code, output, error = _invoke(
        ["run", "shop.metrics.weekly", "--date", "2026-07-27"],
        registry,
        settings,
    )

    assert code == EXIT_SUCCESS
    assert error == ""
    payload = json.loads(output)
    assert payload["dry_run"] is True
    assert payload["status"] == "success"
    assert payload["steps"]["collect"] == "skipped"
    assert payload["journal_path"].startswith("runs/")
    assert action.calls == []
    journals = list((settings.data_root / "runs").rglob("run.json"))
    assert len(journals) == 1
    assert json.loads(journals[0].read_text(encoding="utf-8"))["status"] == "success"
    assert RunIndex(settings.data_root).get(payload["run_id"]).status == "success"


def test_run_execute_returns_the_workflow_failure_exit_code(tmp_path: Path) -> None:
    action = RecordingStep(fail_on=date(2026, 7, 27))
    registry, settings = _runtime(
        tmp_path,
        WorkflowDefinition(
            "shop.metrics.weekly", (StepDefinition("collect", action),)
        ),
    )

    code, output, error = _invoke(
        [
            "run",
            "shop.metrics.weekly",
            "--date",
            "2026-07-27",
            "--execute",
        ],
        registry,
        settings,
    )

    assert code == EXIT_WORKFLOW_FAILED
    assert error == ""
    assert json.loads(output)["status"] == "failed"
    assert len(action.calls) == 1


def test_run_unknown_unavailable_or_unregistered_is_a_configuration_error(
    tmp_path: Path,
) -> None:
    registry = WorkflowRegistry(_catalog(tmp_path))
    settings = Settings(project_root=tmp_path, data_root=tmp_path / "runtime")

    for workflow_id in ("unknown.workflow", "shop.missing", "shop.metrics.weekly"):
        code, output, error = _invoke(
            ["run", workflow_id, "--date", "2026-07-27"], registry, settings
        )
        assert code == EXIT_CONFIGURATION
        assert output == ""
        assert workflow_id in error

    assert not settings.data_root.exists()


def test_backfill_runs_each_business_date_and_aggregates_final_exit_code(
    tmp_path: Path,
) -> None:
    action = RecordingStep(fail_on=date(2026, 7, 28))
    registry, settings = _runtime(
        tmp_path,
        WorkflowDefinition(
            "shop.metrics.weekly", (StepDefinition("collect", action),)
        ),
    )

    code, output, error = _invoke(
        [
            "backfill",
            "shop.metrics.weekly",
            "--from",
            "2026-07-27",
            "--to",
            "2026-07-29",
            "--execute",
        ],
        registry,
        settings,
    )

    assert code == EXIT_WORKFLOW_FAILED
    assert error == ""
    payload = json.loads(output)
    assert payload["status"] == "failed"
    assert [run["business_date"] for run in payload["runs"]] == [
        "2026-07-27",
        "2026-07-28",
        "2026-07-29",
    ]
    assert [context.business_date for context in action.calls] == [
        date(2026, 7, 27),
        date(2026, 7, 28),
        date(2026, 7, 29),
    ]


def test_jd_after_sales_backfill_executes_one_range_and_restores_end_env(
    tmp_path: Path, monkeypatch,
) -> None:
    settings = Settings(project_root=tmp_path, data_root=tmp_path / "runtime")
    workflow = SimpleNamespace(workflow_id=cli.JD_AFTER_SALES_ISSUES_WORKFLOW_ID)
    registry = SimpleNamespace(resolve=lambda _workflow_id: SimpleNamespace(definition=workflow))
    seen: list[tuple[date, str | None, bool]] = []

    def fake_execute_once(
        _workflow,
        *,
        business_date: date,
        shadow: bool,
        dry_run: bool,
        settings: Settings,
    ) -> tuple[int, dict[str, object]]:
        del shadow, settings
        seen.append((business_date, os.environ.get("GYXX_JD_AFTER_SALES_TO"), dry_run))
        return EXIT_SUCCESS, {"business_date": business_date.isoformat()}

    monkeypatch.delenv("GYXX_JD_AFTER_SALES_TO", raising=False)
    monkeypatch.setattr(cli, "_execute_once", fake_execute_once)
    arguments = SimpleNamespace(
        workflow_id=cli.JD_AFTER_SALES_ISSUES_WORKFLOW_ID,
        from_date="2026-06-01",
        to_date="2026-09-15",
        shadow=False,
        execute=True,
        rerun_step=None,
        resume_from=None,
    )
    output = io.StringIO()

    code = cli._backfill_command(arguments, registry, settings, output)

    assert code == EXIT_SUCCESS
    assert seen == [(date(2026, 6, 1), "2026-09-15", False)]
    assert "GYXX_JD_AFTER_SALES_TO" not in os.environ
    payload = json.loads(output.getvalue())
    assert payload["execution_mode"] == "single_range"
    assert payload["runs"][0]["business_date_end"] == "2026-09-15"


def test_step_rerun_and_resume_select_only_the_requested_execution_slice(
    tmp_path: Path,
) -> None:
    collect = RecordingStep()
    normalize = RecordingStep()
    publish = RecordingStep()
    registry, settings = _runtime(
        tmp_path,
        WorkflowDefinition(
            "shop.metrics.weekly",
            (
                StepDefinition("collect", collect),
                StepDefinition("normalize", normalize, depends_on=("collect",)),
                StepDefinition("publish", publish, depends_on=("normalize",)),
            ),
        ),
    )

    rerun_code, _, rerun_error = _invoke(
        [
            "run",
            "shop.metrics.weekly",
            "--date",
            "2026-07-27",
            "--execute",
            "--rerun-step",
            "normalize",
        ],
        registry,
        settings,
    )
    resume_code, _, resume_error = _invoke(
        [
            "run",
            "shop.metrics.weekly",
            "--date",
            "2026-07-28",
            "--execute",
            "--resume-from",
            "normalize",
        ],
        registry,
        settings,
    )

    assert (rerun_code, resume_code) == (EXIT_SUCCESS, EXIT_SUCCESS)
    assert rerun_error == resume_error == ""
    assert [context.business_date for context in collect.calls] == []
    assert [context.business_date for context in normalize.calls] == [
        date(2026, 7, 27),
        date(2026, 7, 28),
    ]
    assert [context.business_date for context in publish.calls] == [date(2026, 7, 28)]


def test_backfill_rejects_reverse_or_excessive_ranges_without_creating_runs(
    tmp_path: Path,
) -> None:
    action = RecordingStep()
    registry, settings = _runtime(
        tmp_path,
        WorkflowDefinition(
            "shop.metrics.weekly", (StepDefinition("collect", action),)
        ),
    )

    for start, end in (("2026-07-29", "2026-07-27"), ("2025-01-01", "2026-07-27")):
        code, output, error = _invoke(
            [
                "backfill",
                "shop.metrics.weekly",
                "--from",
                start,
                "--to",
                end,
                "--execute",
            ],
            registry,
            settings,
        )
        assert code == EXIT_CONFIGURATION
        assert output == ""
        assert error

    assert action.calls == []
    assert not settings.data_root.exists()