360 lines
10 KiB
Python
360 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
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) == 27
|
|
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_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()
|