tegwick
84b4089f8a
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02b6f-7db1-7222-918b-e813a6bda38d
78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import time
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from rein_aharness.claim_loop import _Heartbeat
|
|
from rein_aharness.lease_monitor import LeaseMonitor
|
|
from rein_aharness.ops_run_client import OpsRunError
|
|
|
|
|
|
def test_loss_is_one_shot_and_bounded() -> None:
|
|
seen = []
|
|
monitor = LeaseMonitor(seen.append)
|
|
|
|
first = monitor.mark_lost("ProviderError" * 20)
|
|
second = monitor.mark_lost("OtherError")
|
|
|
|
assert monitor.lost is True
|
|
assert first == second
|
|
assert len(first.error_type) <= 80
|
|
assert first.error_type.startswith("ProviderError")
|
|
assert seen == [first]
|
|
|
|
|
|
def test_wait_lost_and_stop_are_distinct() -> None:
|
|
monitor = LeaseMonitor()
|
|
|
|
assert monitor.wait_lost(timeout=0) is False
|
|
monitor.stop()
|
|
assert monitor.stopped is True
|
|
assert monitor.lost is False
|
|
monitor.mark_lost("OpsRunError")
|
|
assert monitor.wait_lost(timeout=0) is True
|
|
|
|
|
|
def test_concurrent_loss_still_notifies_once() -> None:
|
|
seen = []
|
|
monitor = LeaseMonitor(seen.append)
|
|
|
|
import threading
|
|
|
|
threads = [threading.Thread(target=monitor.mark_lost, args=("OpsRunError",)) for _ in range(8)]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join()
|
|
|
|
assert len(seen) == 1
|
|
|
|
|
|
def test_heartbeat_publishes_loss_without_raw_exception_text() -> None:
|
|
client = MagicMock()
|
|
client.heartbeat.side_effect = OpsRunError("secret provider response", status_code=409)
|
|
heartbeat = _Heartbeat(client, "run-1", lease_seconds=90)
|
|
|
|
with patch("rein_aharness.claim_loop._heartbeat_interval", return_value=0.01):
|
|
heartbeat.start()
|
|
deadline = time.monotonic() + 1.0
|
|
while not heartbeat.monitor.lost and time.monotonic() < deadline:
|
|
time.sleep(0.005)
|
|
heartbeat.stop()
|
|
|
|
assert heartbeat.monitor.lost is True
|
|
assert heartbeat.monitor.evidence is not None
|
|
assert "secret" not in heartbeat.monitor.evidence.error_type
|
|
|
|
|
|
def test_heartbeat_transport_error_does_not_immediately_mark_loss() -> None:
|
|
client = MagicMock()
|
|
client.heartbeat.side_effect = OpsRunError("connection reset")
|
|
heartbeat = _Heartbeat(client, "run-1", lease_seconds=90)
|
|
|
|
with patch("rein_aharness.claim_loop._heartbeat_interval", return_value=0.01):
|
|
heartbeat.start()
|
|
time.sleep(0.04)
|
|
heartbeat.stop()
|
|
|
|
assert heartbeat.monitor.lost is False
|