45 lines
1.4 KiB
Python
45 lines
1.4 KiB
Python
import importlib.util
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
spec = importlib.util.spec_from_file_location(
|
|
"migration_worker_gate",
|
|
Path(__file__).resolve().parents[2] / "scripts/migration_worker_gate.py",
|
|
)
|
|
gate = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(gate)
|
|
|
|
|
|
@pytest.mark.parametrize("role", ["worker", "scheduler"])
|
|
def test_hold_prevents_outbound_process_until_release(role):
|
|
states = iter([True, True, False])
|
|
events = []
|
|
gate.start(
|
|
role,
|
|
check=lambda: next(states),
|
|
sleep=lambda seconds: events.append("wait"),
|
|
execute=lambda executable, args: events.append(args),
|
|
)
|
|
assert events == ["wait", "wait", gate.COMMANDS[role]]
|
|
|
|
|
|
def test_normal_startup_and_missing_hold(tmp_path):
|
|
assert not gate.hold_active(tmp_path / "absent")
|
|
events = []
|
|
gate.start("worker", check=lambda: False, execute=lambda executable, args: events.append(args))
|
|
assert events == [gate.COMMANDS["worker"]]
|
|
|
|
|
|
def test_hold_exists_and_unsupported_marker_fails_closed(tmp_path):
|
|
marker = tmp_path / "hold"
|
|
marker.write_text("fixture-attempt")
|
|
assert gate.hold_active(marker)
|
|
with pytest.raises(RuntimeError):
|
|
gate.hold_active(tmp_path)
|
|
|
|
|
|
def test_unknown_role_never_executes():
|
|
with pytest.raises(ValueError):
|
|
gate.start("shell", execute=lambda *args: pytest.fail("unreviewed command"))
|