604 lines
21 KiB
Python
604 lines
21 KiB
Python
"""Restart-safe cron worker handoff and ownership contracts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def execution_ledger(tmp_path, monkeypatch):
|
|
import cron.executions as executions
|
|
|
|
monkeypatch.setattr(executions, "EXECUTIONS_FILE", tmp_path / "executions.db")
|
|
return executions
|
|
|
|
|
|
def test_execution_owner_moves_to_external_worker_before_running(
|
|
execution_ledger, monkeypatch
|
|
):
|
|
record = execution_ledger.create_execution("job-1", source="builtin")
|
|
assert execution_ledger.mark_execution_handoff_pending(record["id"]) is not None
|
|
monkeypatch.setattr(execution_ledger.os, "getpid", lambda: 4242)
|
|
monkeypatch.setattr(execution_ledger, "_process_start_time", lambda pid: 9876)
|
|
|
|
adopted = execution_ledger.adopt_claimed_execution(record["id"])
|
|
|
|
assert adopted is not None
|
|
assert adopted["pid"] == 4242
|
|
assert adopted["process_started_at"] == 9876
|
|
assert adopted["status"] == "running"
|
|
assert execution_ledger.adopt_claimed_execution(record["id"]) is None
|
|
assert execution_ledger.mark_execution_running(record["id"]) is None
|
|
|
|
|
|
def test_external_worker_cannot_adopt_execution_without_handoff_fence(
|
|
execution_ledger, monkeypatch
|
|
):
|
|
record = execution_ledger.create_execution("job-unfenced", source="builtin")
|
|
monkeypatch.setattr(execution_ledger.os, "getpid", lambda: 4242)
|
|
monkeypatch.setattr(execution_ledger, "_process_start_time", lambda _pid: 9876)
|
|
|
|
assert execution_ledger.adopt_claimed_execution(record["id"]) is None
|
|
assert execution_ledger.get_execution(record["id"])["status"] == "claimed"
|
|
|
|
|
|
def test_genuine_external_worker_crash_is_recovered_unknown(
|
|
execution_ledger, monkeypatch
|
|
):
|
|
record = execution_ledger.create_execution("job-crash", source="builtin")
|
|
assert execution_ledger.mark_execution_handoff_pending(record["id"]) is not None
|
|
script = (
|
|
"import os\n"
|
|
"from pathlib import Path\n"
|
|
"import cron.executions as executions\n"
|
|
f"executions.EXECUTIONS_FILE = Path({str(execution_ledger.EXECUTIONS_FILE)!r})\n"
|
|
f"assert executions.adopt_claimed_execution({record['id']!r}) is not None\n"
|
|
"os._exit(9)\n"
|
|
)
|
|
|
|
crashed = subprocess.run([sys.executable, "-c", script], check=False)
|
|
assert crashed.returncode == 9
|
|
|
|
monkeypatch.setattr(execution_ledger, "_PROCESS_ID", "replacement-scheduler")
|
|
assert execution_ledger.recover_interrupted_executions() == 1
|
|
recovered = execution_ledger.latest_execution("job-crash")
|
|
assert recovered["status"] == "unknown"
|
|
assert "whether side effects ran is unknown" in recovered["error"]
|
|
|
|
|
|
@pytest.mark.linux_only
|
|
def test_restart_safe_gateway_child_fails_closed_without_scope(monkeypatch):
|
|
import tools.process_registry as process_registry
|
|
|
|
monkeypatch.setattr(process_registry, "_is_supervised_gateway_process", lambda: True)
|
|
monkeypatch.setenv("INVOCATION_ID", "managed-service")
|
|
monkeypatch.setattr(process_registry, "_systemd_run_user_scope_available", lambda: False)
|
|
|
|
with pytest.raises(RuntimeError, match="systemd-run --user --scope is unavailable"):
|
|
process_registry.restart_safe_gateway_child_argv(
|
|
["python", "worker.py"], unit_suffix="cron-job-1"
|
|
)
|
|
|
|
|
|
def test_restart_safe_gateway_child_is_unchanged_outside_managed_gateway(monkeypatch):
|
|
import tools.process_registry as process_registry
|
|
|
|
command = ["python", "worker.py"]
|
|
monkeypatch.setattr(process_registry, "_is_supervised_gateway_process", lambda: False)
|
|
|
|
assert process_registry.restart_safe_gateway_child_argv(
|
|
command, unit_suffix="cron-job-1"
|
|
) is command
|
|
|
|
|
|
def test_restart_safe_gateway_child_never_probes_systemd_off_linux(monkeypatch):
|
|
import tools.process_registry as process_registry
|
|
|
|
command = ["python", "worker.py"]
|
|
probe = Mock(side_effect=AssertionError("systemd probe ran off Linux"))
|
|
monkeypatch.setattr(process_registry, "_IS_LINUX", False)
|
|
monkeypatch.setattr(process_registry, "_is_supervised_gateway_process", lambda: True)
|
|
monkeypatch.setattr(process_registry, "_systemd_run_user_scope_available", probe)
|
|
monkeypatch.setenv("INVOCATION_ID", "managed-service")
|
|
|
|
assert process_registry.restart_safe_gateway_child_argv(
|
|
command, unit_suffix="cron-job-1"
|
|
) is command
|
|
probe.assert_not_called()
|
|
|
|
|
|
def test_external_worker_adopts_execution_and_runs_payload_once(
|
|
tmp_path, monkeypatch
|
|
):
|
|
import cron.scheduler as scheduler
|
|
|
|
payload = tmp_path / "payload.json"
|
|
ack = tmp_path / "ready.json"
|
|
payload.write_text(
|
|
json.dumps({
|
|
"job": {"id": "job-1", "execution_id": "exec-1"},
|
|
"profile_home": str(tmp_path / "profile"),
|
|
}),
|
|
encoding="utf-8",
|
|
)
|
|
from hermes_constants import get_hermes_home
|
|
|
|
observed_homes = []
|
|
adopted = Mock(
|
|
side_effect=lambda execution_id: (
|
|
observed_homes.append(get_hermes_home().resolve())
|
|
or {"id": execution_id, "status": "running"}
|
|
)
|
|
)
|
|
run = Mock(
|
|
side_effect=lambda *_args, **_kwargs: (
|
|
observed_homes.append(get_hermes_home().resolve()) or True
|
|
)
|
|
)
|
|
monkeypatch.setattr("cron.executions.adopt_claimed_execution", adopted)
|
|
monkeypatch.setattr(scheduler, "run_one_job", run)
|
|
|
|
assert scheduler._run_external_worker_payload(payload, ack) is True
|
|
|
|
adopted.assert_called_once_with("exec-1")
|
|
run.assert_called_once()
|
|
assert run.call_args.args[0]["id"] == "job-1"
|
|
expected_home = (tmp_path / "profile").resolve()
|
|
assert observed_homes == [expected_home, expected_home]
|
|
assert ack.exists()
|
|
assert not payload.exists()
|
|
|
|
|
|
def test_external_worker_refuses_to_run_without_durable_ownership(
|
|
tmp_path, monkeypatch
|
|
):
|
|
import cron.scheduler as scheduler
|
|
|
|
payload = tmp_path / "payload.json"
|
|
ack = tmp_path / "ready.json"
|
|
payload.write_text(
|
|
json.dumps({
|
|
"job": {"id": "job-1", "execution_id": "exec-1"},
|
|
"profile_home": str(tmp_path / "profile"),
|
|
}),
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setattr("cron.executions.adopt_claimed_execution", lambda _id: None)
|
|
run = Mock()
|
|
monkeypatch.setattr(scheduler, "run_one_job", run)
|
|
|
|
assert scheduler._run_external_worker_payload(payload, ack) is False
|
|
|
|
run.assert_not_called()
|
|
assert not ack.exists()
|
|
|
|
|
|
def test_launch_external_worker_uses_restart_safe_scope_and_acknowledges(
|
|
tmp_path, monkeypatch
|
|
):
|
|
import cron.scheduler as scheduler
|
|
|
|
job = {"id": "job-1", "execution_id": "exec-1", "prompt": "work"}
|
|
monkeypatch.setattr(scheduler, "_get_hermes_home", lambda: tmp_path)
|
|
wrapped_commands = []
|
|
|
|
def wrap(command, *, unit_suffix):
|
|
wrapped_commands.append((command, unit_suffix))
|
|
return ["scope", "--", *command]
|
|
|
|
monkeypatch.setattr(
|
|
"tools.process_registry.restart_safe_gateway_child_argv", wrap
|
|
)
|
|
|
|
class FakeProcess:
|
|
returncode = None
|
|
|
|
def poll(self):
|
|
return self.returncode
|
|
|
|
def wait(self, timeout=None):
|
|
if self.returncode is None:
|
|
raise subprocess.TimeoutExpired(cmd="worker", timeout=timeout)
|
|
return self.returncode
|
|
|
|
spawned = []
|
|
|
|
payloads = []
|
|
|
|
def popen(command, **kwargs):
|
|
spawned.append((command, kwargs))
|
|
payload_index = command.index("--external-worker-file") + 1
|
|
payloads.append(json.loads(Path(command[payload_index]).read_text()))
|
|
ack_index = command.index("--ack-file") + 1
|
|
Path(command[ack_index]).write_text(
|
|
json.dumps({"pid": 4321, "execution_id": "exec-1"}),
|
|
encoding="utf-8",
|
|
)
|
|
return FakeProcess()
|
|
|
|
handoff = Mock(return_value={"id": "exec-1", "handoff_pending": 1})
|
|
monkeypatch.setattr(scheduler, "mark_execution_handoff_pending", handoff)
|
|
monkeypatch.setattr(scheduler.subprocess, "Popen", popen)
|
|
observed_statuses = iter(
|
|
[
|
|
{"id": "exec-1", "status": "running"},
|
|
{"id": "exec-1", "status": "completed"},
|
|
]
|
|
)
|
|
get = Mock(side_effect=lambda _execution_id: next(observed_statuses))
|
|
monkeypatch.setattr(scheduler, "get_execution", get)
|
|
monkeypatch.setenv("ANTHROPIC_API_KEY", "should-not-cross-profile")
|
|
from agent.secret_scope import set_multiplex_active
|
|
|
|
set_multiplex_active(True)
|
|
try:
|
|
assert scheduler._launch_external_cron_worker(job) is True
|
|
finally:
|
|
set_multiplex_active(False)
|
|
assert wrapped_commands[0][1] == "cron-job-1-exec-exec-1"
|
|
assert spawned[0][0][0:2] == ["scope", "--"]
|
|
assert spawned[0][1]["start_new_session"] is True
|
|
assert "ANTHROPIC_API_KEY" not in spawned[0][1]["env"]
|
|
handoff.assert_called_once_with("exec-1")
|
|
assert get.call_count == 2
|
|
assert payloads[0]["multiplex_active"] is True
|
|
# Once the attempt is terminal the parent reaps its own handoff artifacts.
|
|
assert not (tmp_path / "cron/external-workers/exec-1.json").exists()
|
|
|
|
|
|
def test_external_worker_exit_rechecks_exact_execution_before_failure(monkeypatch):
|
|
import cron.scheduler as scheduler
|
|
|
|
statuses = iter(
|
|
[
|
|
{"id": "exec-1", "status": "running"},
|
|
{"id": "exec-1", "status": "completed"},
|
|
]
|
|
)
|
|
get = Mock(side_effect=lambda _execution_id: next(statuses))
|
|
monkeypatch.setattr(scheduler, "get_execution", get, raising=False)
|
|
process = Mock()
|
|
process.poll.return_value = 0
|
|
process.wait.return_value = 0
|
|
|
|
assert scheduler._wait_for_external_cron_worker(
|
|
process, execution_id="exec-1"
|
|
) is True
|
|
assert get.call_count == 2
|
|
|
|
|
|
def test_external_worker_crash_recovers_uncertain_attempt(monkeypatch):
|
|
import cron.scheduler as scheduler
|
|
|
|
statuses = iter(
|
|
[
|
|
{"id": "exec-1", "status": "running"},
|
|
{"id": "exec-1", "status": "unknown"},
|
|
]
|
|
)
|
|
get = Mock(side_effect=lambda _execution_id: next(statuses))
|
|
recover = Mock(return_value=1)
|
|
monkeypatch.setattr(scheduler, "get_execution", get)
|
|
monkeypatch.setattr(
|
|
scheduler, "recover_interrupted_executions", recover, raising=False
|
|
)
|
|
process = Mock()
|
|
process.poll.return_value = 9
|
|
process.wait.return_value = 9
|
|
|
|
assert scheduler._wait_for_external_cron_worker(
|
|
process, execution_id="exec-1"
|
|
) is True
|
|
recover.assert_called_once_with()
|
|
assert get.call_count == 2
|
|
|
|
|
|
def test_launch_external_worker_stays_in_process_outside_managed_gateway(
|
|
monkeypatch,
|
|
):
|
|
import cron.scheduler as scheduler
|
|
|
|
command_calls = []
|
|
|
|
def unchanged(command, *, unit_suffix):
|
|
command_calls.append((command, unit_suffix))
|
|
return command
|
|
|
|
monkeypatch.setattr(
|
|
"tools.process_registry.restart_safe_gateway_child_argv", unchanged
|
|
)
|
|
popen = Mock()
|
|
monkeypatch.setattr(scheduler.subprocess, "Popen", popen)
|
|
|
|
assert scheduler._launch_external_cron_worker(
|
|
{"id": "job-1", "execution_id": "exec-1"}
|
|
) is False
|
|
assert command_calls
|
|
popen.assert_not_called()
|
|
|
|
|
|
def test_shared_run_path_hands_gateway_fire_to_external_worker(monkeypatch):
|
|
import cron.scheduler as scheduler
|
|
|
|
launch = Mock(return_value=True)
|
|
run = Mock(side_effect=AssertionError("agent ran inside gateway"))
|
|
monkeypatch.setattr(scheduler, "_launch_external_cron_worker", launch)
|
|
monkeypatch.setattr(scheduler, "run_job", run)
|
|
job = {"id": "job-1", "execution_id": "exec-1"}
|
|
|
|
assert scheduler.run_one_job(job, adapters={"discord": object()}) is True
|
|
|
|
launch.assert_called_once_with(job)
|
|
run.assert_not_called()
|
|
|
|
|
|
def test_shutdown_does_not_interrupt_restart_safe_waiter():
|
|
import cron.scheduler as scheduler
|
|
|
|
job_id = "external-waiter"
|
|
scheduler._running_job_ids.add(job_id)
|
|
scheduler._restart_safe_waiter_job_ids.add(job_id)
|
|
try:
|
|
assert scheduler.mark_running_jobs_interrupted("gateway restart") == []
|
|
assert job_id not in scheduler._interrupted_job_ids
|
|
finally:
|
|
scheduler._restart_safe_waiter_job_ids.discard(job_id)
|
|
scheduler._running_job_ids.discard(job_id)
|
|
scheduler._interrupted_job_ids.discard(job_id)
|
|
|
|
|
|
def test_worker_delivery_queue_is_keyed_by_the_delivering_jobs_own_execution(
|
|
monkeypatch, tmp_path
|
|
):
|
|
"""A nested in-process dispatch inside a worker (e.g. a script running
|
|
``hermes cron run <other>``) must not queue under the OUTER execution id."""
|
|
import cron.scheduler as scheduler
|
|
|
|
queued = []
|
|
monkeypatch.setattr(
|
|
"cron.delivery_queue.enqueue_and_wait",
|
|
lambda execution_id, job, content, **kw: (
|
|
queued.append(execution_id) or "queued-marker"
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
scheduler,
|
|
"_resolve_delivery_targets",
|
|
lambda job, for_failure=False: [{"platform": "telegram", "chat_id": "123"}],
|
|
)
|
|
|
|
def _standalone(*_args, **_kwargs):
|
|
raise RuntimeError("standalone path reached")
|
|
|
|
# First call the standalone (non-queue) path makes after the guard; the
|
|
# failure is reported as the delivery error string.
|
|
monkeypatch.setattr("gateway.config.load_gateway_config", _standalone)
|
|
monkeypatch.setenv("_HERMES_CRON_EXTERNAL_WORKER", "exec-outer")
|
|
|
|
# Own attempt: routed through the durable queue.
|
|
assert scheduler._deliver_result(
|
|
{"id": "job-1", "execution_id": "exec-outer", "deliver": "telegram:123"},
|
|
"done",
|
|
adapters=None,
|
|
loop=None,
|
|
) == "queued-marker"
|
|
assert queued == ["exec-outer"]
|
|
|
|
# A different job's attempt: must NOT be queued under exec-outer; it falls
|
|
# through to the standalone path.
|
|
error = scheduler._deliver_result(
|
|
{"id": "job-2", "execution_id": "exec-inner", "deliver": "telegram:123"},
|
|
"done",
|
|
adapters=None,
|
|
loop=None,
|
|
)
|
|
assert error == "failed to load gateway config: standalone path reached"
|
|
assert queued == ["exec-outer"]
|
|
|
|
|
|
def test_gateway_tool_run_without_adapter_objects_hands_off(monkeypatch):
|
|
import cron.scheduler as scheduler
|
|
|
|
created = Mock(return_value={"id": "exec-tool"})
|
|
launch = Mock(return_value=True)
|
|
run = Mock(side_effect=AssertionError("agent ran inside gateway"))
|
|
monkeypatch.setattr(scheduler, "create_execution", created)
|
|
monkeypatch.setattr(scheduler, "_launch_external_cron_worker", launch)
|
|
monkeypatch.setattr(scheduler, "run_job", run)
|
|
job = {"id": "tool-job"}
|
|
|
|
assert scheduler.run_one_job(job, adapters=None) is True
|
|
|
|
created.assert_called_once_with("tool-job", source="direct")
|
|
assert job["execution_id"] == "exec-tool"
|
|
launch.assert_called_once_with(job)
|
|
run.assert_not_called()
|
|
|
|
|
|
def test_shared_run_path_creates_execution_before_managed_handoff(monkeypatch):
|
|
import cron.scheduler as scheduler
|
|
|
|
created = Mock(return_value={"id": "exec-new"})
|
|
launch = Mock(return_value=True)
|
|
monkeypatch.setattr(scheduler, "create_execution", created)
|
|
monkeypatch.setattr(scheduler, "_launch_external_cron_worker", launch)
|
|
job = {"id": "manual-job"}
|
|
|
|
assert scheduler.run_one_job(job, adapters={"discord": object()}) is True
|
|
|
|
created.assert_called_once_with("manual-job", source="direct")
|
|
assert job["execution_id"] == "exec-new"
|
|
launch.assert_called_once_with(job)
|
|
|
|
|
|
def test_lost_execution_start_cas_prevents_side_effects(monkeypatch):
|
|
import cron.scheduler as scheduler
|
|
|
|
run = Mock(side_effect=AssertionError("side effect ran without ownership"))
|
|
monkeypatch.setattr(scheduler, "claim_dispatch", lambda _job_id: True)
|
|
monkeypatch.setattr(scheduler, "mark_execution_running", lambda _execution_id: None)
|
|
monkeypatch.setattr(scheduler, "run_job", run)
|
|
|
|
assert scheduler.run_one_job(
|
|
{"id": "job-1", "execution_id": "exec-1"}, adapters=None
|
|
) is True
|
|
run.assert_not_called()
|
|
|
|
|
|
@pytest.mark.linux_only
|
|
@pytest.mark.live_system_guard_bypass
|
|
def test_managed_gateway_restart_preserves_active_worker_and_single_side_effect(
|
|
tmp_path, monkeypatch
|
|
):
|
|
import cron.delivery_queue as delivery_queue
|
|
import cron.executions as executions
|
|
import cron.scheduler as scheduler
|
|
from cron.jobs import create_job, use_cron_store
|
|
from gateway.config import Platform, PlatformConfig
|
|
from gateway.status import _pid_exists
|
|
from tools import process_registry
|
|
|
|
if not process_registry._systemd_run_user_scope_available():
|
|
pytest.skip("systemd-run --user --scope is unavailable on this host")
|
|
|
|
home = tmp_path / "profile"
|
|
scripts_dir = home / "scripts"
|
|
scripts_dir.mkdir(parents=True)
|
|
started = tmp_path / "started"
|
|
release = tmp_path / "release"
|
|
side_effect = tmp_path / "side-effect"
|
|
probe = scripts_dir / "restart_probe.py"
|
|
probe.write_text(
|
|
"import pathlib, time\n"
|
|
f"started = pathlib.Path({str(started)!r})\n"
|
|
f"release = pathlib.Path({str(release)!r})\n"
|
|
f"side_effect = pathlib.Path({str(side_effect)!r})\n"
|
|
"started.write_text('started')\n"
|
|
"deadline = time.monotonic() + 15\n"
|
|
"while not release.exists() and time.monotonic() < deadline:\n"
|
|
" time.sleep(0.05)\n"
|
|
"if not release.exists():\n"
|
|
" raise SystemExit('release timeout')\n"
|
|
"with side_effect.open('a') as handle:\n"
|
|
" handle.write('once\\n')\n"
|
|
"print('completed')\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(home))
|
|
with use_cron_store(home):
|
|
job = create_job(
|
|
prompt=None,
|
|
schedule="every 1h",
|
|
name="restart probe",
|
|
script=probe.name,
|
|
no_agent=True,
|
|
deliver="telegram:123",
|
|
)
|
|
payload = tmp_path / "job.json"
|
|
launched = tmp_path / "launched.json"
|
|
payload.write_text(json.dumps(job), encoding="utf-8")
|
|
|
|
sent = []
|
|
adapter = Mock()
|
|
|
|
async def send(_chat_id, content, metadata=None):
|
|
sent.append((content, metadata))
|
|
return {"success": True, "message_id": "restart-delivery-1"}
|
|
|
|
adapter.send = send
|
|
gateway_config = Mock()
|
|
gateway_config.platforms = {
|
|
Platform.TELEGRAM: PlatformConfig(enabled=True),
|
|
}
|
|
gateway_config.get_home_channel = lambda _platform: None
|
|
monkeypatch.setattr(
|
|
"gateway.config.load_gateway_config", lambda: gateway_config
|
|
)
|
|
monkeypatch.setattr(
|
|
scheduler, "load_config", lambda: {"cron": {"wrap_response": False}}
|
|
)
|
|
replacement_loop = asyncio.new_event_loop()
|
|
replacement_thread = threading.Thread(
|
|
target=replacement_loop.run_forever,
|
|
daemon=True,
|
|
)
|
|
replacement_thread.start()
|
|
deadline = time.monotonic() + 2
|
|
while not replacement_loop.is_running() and time.monotonic() < deadline:
|
|
time.sleep(0.01)
|
|
assert replacement_loop.is_running()
|
|
|
|
harness = (
|
|
"import json, os, pathlib, time\n"
|
|
f"os.environ['HERMES_HOME'] = {str(home)!r}\n"
|
|
"os.environ['INVOCATION_ID'] = 'restart-fixture'\n"
|
|
"from cron import scheduler\n"
|
|
"from tools import process_registry\n"
|
|
"process_registry._is_supervised_gateway_process = lambda: True\n"
|
|
f"job = json.loads(pathlib.Path({str(payload)!r}).read_text())\n"
|
|
"if not scheduler.run_one_job(job, adapters=None, loop=None):\n"
|
|
" raise SystemExit('worker was not isolated')\n"
|
|
f"pathlib.Path({str(launched)!r}).write_text('returned')\n"
|
|
)
|
|
parent = subprocess.Popen([sys.executable, "-c", harness])
|
|
worker_pid = None
|
|
try:
|
|
deadline = time.monotonic() + 10
|
|
current = None
|
|
while time.monotonic() < deadline:
|
|
if parent.poll() is not None:
|
|
pytest.fail(f"gateway fixture exited early with {parent.returncode}")
|
|
current = executions.latest_execution(job["id"])
|
|
if started.exists() and current and current.get("pid") != os.getpid():
|
|
break
|
|
time.sleep(0.05)
|
|
assert started.exists()
|
|
assert current is not None
|
|
execution = current
|
|
worker_pid = int(current["pid"])
|
|
assert not launched.exists(), "handoff returned before execution completed"
|
|
|
|
# Replacing a managed gateway kills its old process tree. The active
|
|
# cron owner must remain in its transient scope and keep the same PID.
|
|
parent.terminate()
|
|
parent.wait(timeout=5)
|
|
assert _pid_exists(worker_pid)
|
|
|
|
release.write_text("go", encoding="utf-8")
|
|
deadline = time.monotonic() + 10
|
|
while time.monotonic() < deadline:
|
|
row = delivery_queue.get_status(execution["id"])
|
|
if row and row["status"] == "pending":
|
|
scheduler.drain_delivery_queue(
|
|
{Platform.TELEGRAM: adapter}, replacement_loop
|
|
)
|
|
current = executions.latest_execution(job["id"])
|
|
if current and current["status"] == "completed":
|
|
break
|
|
time.sleep(0.05)
|
|
assert executions.latest_execution(job["id"])["status"] == "completed"
|
|
assert side_effect.read_text(encoding="utf-8").splitlines() == ["once"]
|
|
assert delivery_queue.get_status(execution["id"])["status"] == "delivered"
|
|
assert len(sent) == 1
|
|
assert "completed" in sent[0][0]
|
|
finally:
|
|
replacement_loop.call_soon_threadsafe(replacement_loop.stop)
|
|
replacement_thread.join(timeout=2)
|
|
replacement_loop.close()
|
|
if parent.poll() is None:
|
|
parent.terminate()
|
|
parent.wait(timeout=5)
|
|
if worker_pid is not None and _pid_exists(worker_pid):
|
|
os.kill(worker_pid, signal.SIGKILL)
|