777 lines
25 KiB
Python
777 lines
25 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from collections import deque
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import psutil
|
|
import pytest
|
|
|
|
import gyxx_flow.workflow.steps as steps_module
|
|
from gyxx_flow.core.context import RunContext
|
|
from gyxx_flow.core.layout import DataLayout
|
|
from gyxx_flow.core.locks import LockManager, ResourceBusyError
|
|
from gyxx_flow.core.records import RunJournal
|
|
from gyxx_flow.ops.effects import EffectLedger
|
|
from gyxx_flow.workflow.engine import WorkflowEngine
|
|
from gyxx_flow.workflow.model import (
|
|
StepDefinition,
|
|
WorkflowDefinition,
|
|
WorkflowValidationError,
|
|
)
|
|
from gyxx_flow.workflow.steps import CommandStep, StepExecution
|
|
|
|
|
|
def _context(*, shadow: bool = False) -> RunContext:
|
|
return RunContext.create(
|
|
"shop.weekly",
|
|
"2026-07-27",
|
|
shadow=shadow,
|
|
now=datetime(2026, 7, 27, 4, 0, tzinfo=timezone.utc),
|
|
random_suffix="abc123",
|
|
)
|
|
|
|
|
|
class FakeStep:
|
|
def __init__(self, *outcomes: StepExecution) -> None:
|
|
self.outcomes = deque(outcomes or (StepExecution(exit_code=0),))
|
|
self.calls: list[dict[str, Any]] = []
|
|
|
|
def execute(
|
|
self,
|
|
*,
|
|
context: RunContext,
|
|
timeout_seconds: float | None,
|
|
dry_run: bool,
|
|
) -> StepExecution:
|
|
self.calls.append(
|
|
{
|
|
"context": context,
|
|
"timeout_seconds": timeout_seconds,
|
|
"dry_run": dry_run,
|
|
}
|
|
)
|
|
return self.outcomes.popleft()
|
|
|
|
|
|
def test_workflow_validates_relations_and_uses_stable_topological_order() -> None:
|
|
action = FakeStep()
|
|
workflow = WorkflowDefinition(
|
|
workflow_id="shop.weekly",
|
|
steps=(
|
|
StepDefinition("publish", action, depends_on=("collect",)),
|
|
StepDefinition("audit", action),
|
|
StepDefinition("collect", action),
|
|
),
|
|
)
|
|
|
|
assert [step.step_id for step in workflow.ordered_steps()] == [
|
|
"audit",
|
|
"collect",
|
|
"publish",
|
|
]
|
|
|
|
with pytest.raises(WorkflowValidationError, match="duplicate step id"):
|
|
WorkflowDefinition(
|
|
"shop.weekly",
|
|
(StepDefinition("collect", action), StepDefinition("collect", action)),
|
|
)
|
|
|
|
with pytest.raises(WorkflowValidationError, match="unknown dependency"):
|
|
WorkflowDefinition(
|
|
"shop.weekly",
|
|
(StepDefinition("publish", action, depends_on=("missing",)),),
|
|
)
|
|
|
|
with pytest.raises(WorkflowValidationError, match="cycle"):
|
|
WorkflowDefinition(
|
|
"shop.weekly",
|
|
(
|
|
StepDefinition("first", action, depends_on=("second",)),
|
|
StepDefinition("second", action, depends_on=("first",)),
|
|
),
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("kwargs", "message"),
|
|
[
|
|
({"max_attempts": 0}, "max_attempts"),
|
|
({"timeout_seconds": 0}, "timeout_seconds"),
|
|
({"timeout_seconds": True}, "timeout_seconds"),
|
|
({"timeout_seconds": "60"}, "timeout_seconds"),
|
|
({"retry_delay_seconds": -1}, "retry_delay_seconds"),
|
|
({"depends_on": ("one", "one")}, "duplicate dependency"),
|
|
],
|
|
)
|
|
def test_step_definition_validates_retry_timeout_and_dependencies(
|
|
kwargs: dict[str, object], message: str
|
|
) -> None:
|
|
with pytest.raises(WorkflowValidationError, match=message):
|
|
StepDefinition("collect", FakeStep(), **kwargs)
|
|
|
|
|
|
def test_command_step_uses_argv_explicit_cwd_env_and_timeout(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
class FakeProcess:
|
|
pid = 424242
|
|
|
|
def wait(self, timeout: float | None = None) -> int:
|
|
captured["timeout"] = timeout
|
|
return 0
|
|
|
|
def fake_start(argv: list[str], **kwargs: object) -> tuple[FakeProcess, None]:
|
|
captured["argv"] = argv
|
|
captured.update(kwargs)
|
|
kwargs["stdout"].write("ok") # type: ignore[union-attr]
|
|
return FakeProcess(), None
|
|
|
|
monkeypatch.setattr(steps_module, "_start_process", fake_start)
|
|
step = CommandStep(
|
|
argv=("C:/Python312/python.exe", "worker.py", "--date", "2026-07-27"),
|
|
cwd=tmp_path,
|
|
env={"MODE": "test"},
|
|
)
|
|
|
|
outcome = step.execute(context=_context(), timeout_seconds=9.5, dry_run=False)
|
|
|
|
assert outcome.exit_code == 0
|
|
assert outcome.stdout == "ok"
|
|
assert captured["argv"] == [
|
|
"C:/Python312/python.exe",
|
|
"worker.py",
|
|
"--date",
|
|
"2026-07-27",
|
|
]
|
|
assert captured["cwd"] == tmp_path
|
|
assert captured["env"] == {"MODE": "test"}
|
|
assert captured["timeout"] == 9.5
|
|
assert captured["stdout"].closed is True # type: ignore[union-attr]
|
|
assert captured["stderr"].closed is True # type: ignore[union-attr]
|
|
|
|
|
|
def test_command_step_preserves_chinese_child_diagnostics(tmp_path: Path) -> None:
|
|
child_env = dict(os.environ)
|
|
child_env.pop("PYTHONIOENCODING", None)
|
|
expected = "天猫百亿补贴导入失败"
|
|
step = CommandStep(
|
|
argv=(sys.executable, "-c", f"print({expected!r})"),
|
|
cwd=tmp_path,
|
|
env=child_env,
|
|
)
|
|
|
|
outcome = step.execute(context=_context(), timeout_seconds=10, dry_run=False)
|
|
|
|
assert outcome.exit_code == 0
|
|
assert outcome.stdout.strip() == expected
|
|
|
|
|
|
def test_command_step_dry_run_never_starts_a_process(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
def fail_if_called(*args: object, **kwargs: object) -> None:
|
|
raise AssertionError("subprocess must not be started")
|
|
|
|
monkeypatch.setattr(steps_module, "_start_process", fail_if_called)
|
|
step = CommandStep(argv=("tool.exe", "--write"), cwd=tmp_path, env={})
|
|
|
|
outcome = step.execute(context=_context(), timeout_seconds=None, dry_run=True)
|
|
|
|
assert outcome.skipped is True
|
|
assert outcome.reason == "dry-run"
|
|
assert outcome.exit_code == 0
|
|
|
|
|
|
def test_command_step_uses_stdout_as_failure_detail_when_stderr_is_empty(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
class FakeProcess:
|
|
pid = 424243
|
|
|
|
def wait(self, timeout: float | None = None) -> int:
|
|
del timeout
|
|
return 1
|
|
|
|
def fake_start(argv: list[str], **kwargs: object) -> tuple[FakeProcess, None]:
|
|
del argv
|
|
kwargs["stdout"].write("business workflow failed") # type: ignore[union-attr]
|
|
return FakeProcess(), None
|
|
|
|
monkeypatch.setattr(steps_module, "_start_process", fake_start)
|
|
step = CommandStep(argv=("worker.exe",), cwd=tmp_path, env={})
|
|
|
|
outcome = step.execute(context=_context(), timeout_seconds=None, dry_run=False)
|
|
|
|
assert outcome.exit_code == 1
|
|
assert outcome.error == "business workflow failed"
|
|
|
|
|
|
def test_command_step_failure_detail_preserves_log_head_and_root_cause_tail(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
class FakeProcess:
|
|
pid = 424244
|
|
|
|
def wait(self, timeout: float | None = None) -> int:
|
|
del timeout
|
|
return 1
|
|
|
|
def fake_start(argv: list[str], **kwargs: object) -> tuple[FakeProcess, None]:
|
|
del argv
|
|
kwargs["stdout"].write( # type: ignore[union-attr]
|
|
"collector startup context\n" + "progress line\n" * 2_000
|
|
)
|
|
kwargs["stderr"].write( # type: ignore[union-attr]
|
|
"Traceback: ROOT CAUSE AT LOG TAIL"
|
|
)
|
|
return FakeProcess(), None
|
|
|
|
monkeypatch.setattr(steps_module, "_start_process", fake_start)
|
|
step = CommandStep(argv=("worker.exe",), cwd=tmp_path, env={})
|
|
|
|
outcome = step.execute(context=_context(), timeout_seconds=None, dry_run=False)
|
|
|
|
assert outcome.exit_code == 1
|
|
assert outcome.error is not None
|
|
assert outcome.error.startswith("collector startup context")
|
|
assert "[middle output omitted]" in outcome.error
|
|
assert outcome.error.endswith("Traceback: ROOT CAUSE AT LOG TAIL")
|
|
assert len(outcome.error) <= 16_000
|
|
|
|
|
|
|
|
def test_command_step_reports_timeout_as_a_failed_exit(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
terminated: list[object] = []
|
|
|
|
class TimeoutProcess:
|
|
pid = 424242
|
|
|
|
def wait(self, timeout: float | None = None) -> int:
|
|
raise subprocess.TimeoutExpired(cmd=["tool.exe"], timeout=timeout)
|
|
|
|
def fake_start(*_args: object, **_kwargs: object) -> tuple[TimeoutProcess, None]:
|
|
return TimeoutProcess(), None
|
|
|
|
monkeypatch.setattr(steps_module, "_start_process", fake_start)
|
|
monkeypatch.setattr(
|
|
steps_module,
|
|
"_terminate_process_tree",
|
|
lambda process, scope: terminated.append((process, scope)),
|
|
)
|
|
step = CommandStep(argv=("tool.exe",), cwd=tmp_path, env={})
|
|
|
|
outcome = step.execute(context=_context(), timeout_seconds=2, dry_run=False)
|
|
|
|
assert outcome.exit_code == 124
|
|
assert "timed out" in (outcome.error or "")
|
|
assert len(terminated) == 1
|
|
|
|
|
|
def test_command_step_timeout_terminates_grandchild_before_return(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
identity_file = tmp_path / "grandchild.json"
|
|
marker_file = tmp_path / "grandchild-finished.txt"
|
|
grandchild_code = (
|
|
"import json,os,pathlib,psutil,time; "
|
|
f"pathlib.Path({str(identity_file)!r}).write_text("
|
|
"json.dumps({'pid': os.getpid(), "
|
|
"'created': psutil.Process().create_time()}), encoding='utf-8'); "
|
|
"time.sleep(1.2); "
|
|
f"pathlib.Path({str(marker_file)!r}).write_text('alive', encoding='utf-8')"
|
|
)
|
|
parent_code = (
|
|
"import pathlib,subprocess,sys,time; "
|
|
f"subprocess.Popen([sys.executable, '-c', {grandchild_code!r}], "
|
|
"stdout=sys.stdout, stderr=sys.stderr, close_fds=False); "
|
|
f"identity = pathlib.Path({str(identity_file)!r}); "
|
|
"deadline = time.monotonic() + 2; "
|
|
"\nwhile not identity.exists() and time.monotonic() < deadline: "
|
|
"time.sleep(0.01)"
|
|
"\ntime.sleep(10)"
|
|
)
|
|
step = CommandStep(
|
|
argv=(sys.executable, "-c", parent_code),
|
|
cwd=tmp_path,
|
|
env=dict(os.environ),
|
|
)
|
|
|
|
outcome = step.execute(context=_context(), timeout_seconds=0.8, dry_run=False)
|
|
|
|
assert outcome.exit_code == 124
|
|
assert identity_file.is_file()
|
|
identity = json.loads(identity_file.read_text(encoding="utf-8"))
|
|
try:
|
|
descendant = psutil.Process(identity["pid"])
|
|
same_process_is_live = (
|
|
descendant.create_time() == pytest.approx(identity["created"], abs=0.01)
|
|
and descendant.is_running()
|
|
and descendant.status() != psutil.STATUS_ZOMBIE
|
|
)
|
|
except psutil.Error:
|
|
same_process_is_live = False
|
|
assert same_process_is_live is False
|
|
time.sleep(1.3)
|
|
assert marker_file.exists() is False
|
|
|
|
|
|
def test_command_step_does_not_wait_for_grandchild_output_handle(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
child_code = (
|
|
"import subprocess,sys; "
|
|
"subprocess.Popen([sys.executable, '-c', "
|
|
"'import time; time.sleep(3)'], stdout=sys.stdout, stderr=sys.stderr, "
|
|
"close_fds=False); "
|
|
"print('parent done')"
|
|
)
|
|
step = CommandStep(
|
|
argv=(sys.executable, "-c", child_code),
|
|
cwd=tmp_path,
|
|
env=dict(os.environ),
|
|
)
|
|
|
|
outcome = step.execute(context=_context(), timeout_seconds=1, dry_run=False)
|
|
|
|
assert outcome.exit_code == 0
|
|
assert "parent done" in outcome.stdout
|
|
|
|
|
|
def test_command_step_maps_reserved_cookie_exit_to_a_skip(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
class SkippedProcess:
|
|
pid = 424242
|
|
|
|
def wait(self, timeout: float | None = None) -> int:
|
|
del timeout
|
|
return 75
|
|
|
|
def skipped(*_args: object, **kwargs: object) -> tuple[SkippedProcess, None]:
|
|
kwargs["stdout"].write( # type: ignore[union-attr]
|
|
"[SKIPPED_COOKIE] state missing"
|
|
)
|
|
return SkippedProcess(), None
|
|
|
|
monkeypatch.setattr(steps_module, "_start_process", skipped)
|
|
step = CommandStep(argv=("collector.exe",), cwd=tmp_path, env={})
|
|
|
|
outcome = step.execute(context=_context(), timeout_seconds=2, dry_run=False)
|
|
|
|
assert outcome.exit_code == 75
|
|
assert outcome.skipped is True
|
|
assert outcome.reason == "cookie-preflight"
|
|
|
|
|
|
def test_command_step_maps_idempotent_replay_exit_to_a_skip(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
class SkippedProcess:
|
|
pid = 424243
|
|
|
|
def wait(self, timeout: float | None = None) -> int:
|
|
del timeout
|
|
return 76
|
|
|
|
monkeypatch.setattr(
|
|
steps_module,
|
|
"_start_process",
|
|
lambda *_args, **_kwargs: (SkippedProcess(), None),
|
|
)
|
|
step = CommandStep(argv=("collector.exe",), cwd=tmp_path, env={})
|
|
|
|
outcome = step.execute(context=_context(), timeout_seconds=2, dry_run=False)
|
|
|
|
assert outcome.exit_code == 76
|
|
assert outcome.skipped is True
|
|
assert outcome.reason == "idempotent-replay"
|
|
|
|
|
|
def test_engine_retries_records_every_attempt_and_aggregates_failures(tmp_path: Path) -> None:
|
|
flaky = FakeStep(
|
|
StepExecution(exit_code=7, error="temporary"),
|
|
StepExecution(exit_code=0),
|
|
)
|
|
optional = FakeStep(StepExecution(exit_code=5, error="optional failed"))
|
|
required = FakeStep(StepExecution(exit_code=9, error="required failed"))
|
|
workflow = WorkflowDefinition(
|
|
"shop.weekly",
|
|
(
|
|
StepDefinition(
|
|
"flaky",
|
|
flaky,
|
|
timeout_seconds=4,
|
|
max_attempts=2,
|
|
retry_delay_seconds=0,
|
|
resources=("browser:shop",),
|
|
),
|
|
StepDefinition("optional", optional, critical=False),
|
|
StepDefinition("required", required),
|
|
),
|
|
)
|
|
context = _context()
|
|
layout = DataLayout(tmp_path)
|
|
journal = RunJournal.create(layout, context)
|
|
|
|
result = WorkflowEngine(LockManager(tmp_path / "locks")).execute(
|
|
workflow,
|
|
context=context,
|
|
journal=journal,
|
|
)
|
|
|
|
assert result.status == "failed"
|
|
assert result.exit_code == 1
|
|
assert result.steps["flaky"].status == "success"
|
|
assert result.steps["flaky"].final_exit_code == 0
|
|
assert [attempt.exit_code for attempt in result.steps["flaky"].attempts] == [7, 0]
|
|
assert result.steps["optional"].status == "failed"
|
|
assert result.steps["required"].status == "failed"
|
|
assert len(flaky.calls) == 2
|
|
assert all(call["timeout_seconds"] == 4 for call in flaky.calls)
|
|
|
|
payload = json.loads(journal.path.read_text(encoding="utf-8"))
|
|
assert payload["status"] == "failed"
|
|
assert payload["steps"]["flaky.attempt-1"]["exit_code"] == 7
|
|
assert payload["steps"]["flaky.attempt-2"]["exit_code"] == 0
|
|
assert payload["steps"]["required.attempt-1"]["exit_code"] == 9
|
|
assert payload["trace"]["inputs"] == [
|
|
"step:flaky",
|
|
"step:optional",
|
|
"step:required",
|
|
]
|
|
assert "step:flaky:attempt:1:failed:exit:7" in payload["trace"]["outputs"]
|
|
assert "step:flaky:attempt:2:success:exit:0" in payload["trace"]["outputs"]
|
|
assert list((tmp_path / "locks").glob("*.lock")) == []
|
|
|
|
|
|
def test_noncritical_failure_does_not_fail_the_workflow(tmp_path: Path) -> None:
|
|
workflow = WorkflowDefinition(
|
|
"shop.weekly",
|
|
(
|
|
StepDefinition(
|
|
"optional",
|
|
FakeStep(StepExecution(exit_code=2, error="not available")),
|
|
critical=False,
|
|
),
|
|
StepDefinition("required", FakeStep()),
|
|
),
|
|
)
|
|
context = _context()
|
|
journal = RunJournal.create(DataLayout(tmp_path), context)
|
|
|
|
result = WorkflowEngine(LockManager(tmp_path / "locks")).execute(
|
|
workflow, context=context, journal=journal
|
|
)
|
|
|
|
assert result.status == "success"
|
|
assert result.exit_code == 0
|
|
assert result.warnings == ("optional",)
|
|
|
|
|
|
def test_engine_shadow_mode_skips_production_sinks_and_official_notifications(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
safe = FakeStep()
|
|
sink = FakeStep()
|
|
notify = FakeStep()
|
|
workflow = WorkflowDefinition(
|
|
"shop.weekly",
|
|
(
|
|
StepDefinition("safe", safe),
|
|
StepDefinition("publish", sink, production_sink=True, depends_on=("safe",)),
|
|
StepDefinition(
|
|
"notify",
|
|
notify,
|
|
official_notification=True,
|
|
depends_on=("publish",),
|
|
),
|
|
),
|
|
)
|
|
context = _context(shadow=True)
|
|
journal = RunJournal.create(DataLayout(tmp_path), context)
|
|
|
|
result = WorkflowEngine(LockManager(tmp_path / "locks")).execute(
|
|
workflow, context=context, journal=journal
|
|
)
|
|
|
|
assert result.status == "success"
|
|
assert result.steps["safe"].status == "success"
|
|
assert result.steps["publish"].status == "skipped"
|
|
assert result.steps["publish"].reason == "shadow-policy"
|
|
assert result.steps["notify"].status == "skipped"
|
|
assert sink.calls == []
|
|
assert notify.calls == []
|
|
|
|
payload = json.loads(journal.path.read_text(encoding="utf-8"))
|
|
assert payload["steps"]["publish.attempt-1"]["status"] == "skipped"
|
|
assert payload["steps"]["notify.attempt-1"]["status"] == "skipped"
|
|
assert payload["trace"]["external_writes"] == []
|
|
|
|
|
|
def test_successful_sink_records_declared_external_write_in_trace(tmp_path: Path) -> None:
|
|
workflow = WorkflowDefinition(
|
|
"shop.weekly",
|
|
(StepDefinition("publish", FakeStep(), production_sink=True),),
|
|
)
|
|
context = _context()
|
|
journal = RunJournal.create(DataLayout(tmp_path), context)
|
|
|
|
result = WorkflowEngine(LockManager(tmp_path / "locks")).execute(
|
|
workflow, context=context, journal=journal
|
|
)
|
|
|
|
assert result.status == "success"
|
|
payload = json.loads(journal.path.read_text(encoding="utf-8"))
|
|
assert payload["trace"]["external_writes"] == ["step:publish:production_sink"]
|
|
|
|
|
|
def test_effect_ledger_skips_a_successful_production_sink_replay(tmp_path: Path) -> None:
|
|
action = FakeStep()
|
|
workflow = WorkflowDefinition(
|
|
"shop.weekly",
|
|
(StepDefinition("publish", action, production_sink=True),),
|
|
)
|
|
ledger = EffectLedger(tmp_path)
|
|
engine = WorkflowEngine(
|
|
LockManager(tmp_path / "locks"), effect_ledger=ledger
|
|
)
|
|
first_context = _context()
|
|
first = engine.execute(
|
|
workflow,
|
|
context=first_context,
|
|
journal=RunJournal.create(DataLayout(tmp_path), first_context),
|
|
)
|
|
second_context = RunContext.create(
|
|
"shop.weekly", "2026-07-27", random_suffix="replay2"
|
|
)
|
|
second = engine.execute(
|
|
workflow,
|
|
context=second_context,
|
|
journal=RunJournal.create(DataLayout(tmp_path), second_context),
|
|
)
|
|
|
|
assert first.status == "success"
|
|
assert second.status == "success"
|
|
assert second.steps["publish"].reason == "idempotency-replay"
|
|
assert len(action.calls) == 1
|
|
|
|
|
|
def test_repeatable_sink_runs_again_without_business_date_effect_receipt(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
action = FakeStep(StepExecution(exit_code=0), StepExecution(exit_code=0))
|
|
workflow = WorkflowDefinition(
|
|
"product.tmall_baibu_apply",
|
|
(
|
|
StepDefinition(
|
|
"old_all_bag",
|
|
action,
|
|
production_sink=True,
|
|
replay_policy="repeatable",
|
|
),
|
|
),
|
|
)
|
|
ledger = EffectLedger(tmp_path)
|
|
engine = WorkflowEngine(
|
|
LockManager(tmp_path / "locks"), effect_ledger=ledger
|
|
)
|
|
first_context = RunContext.create(
|
|
"product.tmall_baibu_apply", "2026-07-27", random_suffix="repeatable1"
|
|
)
|
|
first = engine.execute(
|
|
workflow,
|
|
context=first_context,
|
|
journal=RunJournal.create(DataLayout(tmp_path), first_context),
|
|
)
|
|
second_context = RunContext.create(
|
|
"product.tmall_baibu_apply", "2026-07-27", random_suffix="repeatable2"
|
|
)
|
|
second = engine.execute(
|
|
workflow,
|
|
context=second_context,
|
|
journal=RunJournal.create(DataLayout(tmp_path), second_context),
|
|
)
|
|
|
|
assert first.status == "success"
|
|
assert second.status == "success"
|
|
assert second.steps["old_all_bag"].reason is None
|
|
assert len(action.calls) == 2
|
|
assert list((tmp_path / "state" / "ops" / "effects").glob("*.json")) == []
|
|
|
|
|
|
def test_idempotent_sink_runs_again_for_same_business_date(tmp_path: Path) -> None:
|
|
action = FakeStep(StepExecution(exit_code=0), StepExecution(exit_code=0))
|
|
workflow = WorkflowDefinition(
|
|
"shop.weekly",
|
|
(
|
|
StepDefinition(
|
|
"publish",
|
|
action,
|
|
production_sink=True,
|
|
replay_policy="idempotent",
|
|
),
|
|
),
|
|
)
|
|
ledger = EffectLedger(tmp_path)
|
|
first_context = _context()
|
|
second_context = RunContext.create(
|
|
"shop.weekly", "2026-07-27", random_suffix="replay3"
|
|
)
|
|
engine = WorkflowEngine(
|
|
LockManager(tmp_path / "locks"), effect_ledger=ledger
|
|
)
|
|
|
|
first = engine.execute(
|
|
workflow,
|
|
context=first_context,
|
|
journal=RunJournal.create(DataLayout(tmp_path), first_context),
|
|
)
|
|
second = engine.execute(
|
|
workflow,
|
|
context=second_context,
|
|
journal=RunJournal.create(DataLayout(tmp_path), second_context),
|
|
)
|
|
|
|
assert first.status == "success"
|
|
assert second.status == "success"
|
|
assert second.steps["publish"].reason is None
|
|
assert len(action.calls) == 2
|
|
assert list((tmp_path / "state" / "ops" / "effects").glob("*.json")) == []
|
|
|
|
|
|
def test_skipped_production_sink_cancels_effect_claim_for_future_runs(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
action = FakeStep(StepExecution(exit_code=75, skipped=True, reason="cookie-preflight"))
|
|
workflow = WorkflowDefinition(
|
|
"shop.weekly",
|
|
(StepDefinition("collect", action, production_sink=True),),
|
|
)
|
|
ledger = EffectLedger(tmp_path)
|
|
context = _context()
|
|
|
|
result = WorkflowEngine(
|
|
LockManager(tmp_path / "locks"),
|
|
effect_ledger=ledger,
|
|
).execute(
|
|
workflow,
|
|
context=context,
|
|
journal=RunJournal.create(DataLayout(tmp_path), context),
|
|
)
|
|
|
|
assert result.status == "success"
|
|
assert result.steps["collect"].status == "skipped"
|
|
assert list((tmp_path / "state" / "ops" / "effects").glob("*.json")) == []
|
|
|
|
|
|
def test_resource_busy_before_action_cancels_effect_claim(tmp_path: Path) -> None:
|
|
class BusyStepLockManager:
|
|
def acquire(self, resource, *, owner):
|
|
del owner
|
|
if resource == "module:product_commerce:product.daily":
|
|
raise ResourceBusyError(f"resource is busy: {resource}")
|
|
return LockManager(tmp_path / "locks").acquire(
|
|
resource,
|
|
owner="workflow-lock",
|
|
)
|
|
|
|
action = FakeStep()
|
|
workflow = WorkflowDefinition(
|
|
"product.daily",
|
|
(
|
|
StepDefinition(
|
|
"collect",
|
|
action,
|
|
resources=("module:product_commerce:product.daily",),
|
|
production_sink=True,
|
|
),
|
|
),
|
|
)
|
|
context = RunContext.create(
|
|
"product.daily",
|
|
"2026-08-11",
|
|
random_suffix="busy",
|
|
)
|
|
result = WorkflowEngine(
|
|
BusyStepLockManager(), # type: ignore[arg-type]
|
|
effect_ledger=EffectLedger(tmp_path),
|
|
).execute(
|
|
workflow,
|
|
context=context,
|
|
journal=RunJournal.create(DataLayout(tmp_path), context),
|
|
)
|
|
|
|
assert result.status == "failed"
|
|
assert result.steps["collect"].reason == "resource-busy-before-execution"
|
|
assert action.calls == []
|
|
assert list((tmp_path / "state" / "ops" / "effects").glob("*.json")) == []
|
|
|
|
|
|
def test_failed_dependency_skips_downstream_critical_step(tmp_path: Path) -> None:
|
|
downstream = FakeStep()
|
|
workflow = WorkflowDefinition(
|
|
"shop.weekly",
|
|
(
|
|
StepDefinition("collect", FakeStep(StepExecution(exit_code=3, error="bad input"))),
|
|
StepDefinition("publish", downstream, depends_on=("collect",)),
|
|
),
|
|
)
|
|
context = _context()
|
|
journal = RunJournal.create(DataLayout(tmp_path), context)
|
|
|
|
result = WorkflowEngine(LockManager(tmp_path / "locks")).execute(
|
|
workflow, context=context, journal=journal
|
|
)
|
|
|
|
assert result.status == "failed"
|
|
assert result.steps["publish"].status == "skipped"
|
|
assert result.steps["publish"].reason == "dependency-failed"
|
|
assert downstream.calls == []
|
|
|
|
|
|
def test_run_after_failure_executes_downstream_but_preserves_workflow_failure(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
product = FakeStep()
|
|
workflow = WorkflowDefinition(
|
|
"shop.weekly",
|
|
(
|
|
StepDefinition(
|
|
"brand",
|
|
FakeStep(StepExecution(exit_code=3, error="brand failed")),
|
|
),
|
|
StepDefinition(
|
|
"product",
|
|
product,
|
|
depends_on=("brand",),
|
|
run_after_failure=True,
|
|
),
|
|
),
|
|
)
|
|
context = _context()
|
|
|
|
result = WorkflowEngine(LockManager(tmp_path / "locks")).execute(
|
|
workflow,
|
|
context=context,
|
|
journal=RunJournal.create(DataLayout(tmp_path), context),
|
|
)
|
|
|
|
assert result.status == "failed"
|
|
assert result.exit_code == 1
|
|
assert result.steps["brand"].status == "failed"
|
|
assert result.steps["product"].status == "success"
|
|
assert len(product.calls) == 1
|