feat: medic revival + concurrent monitor viewers — pytest logs route to tmp (PYTEST_CURRENT_TEST, fixture storms cant pollute prod logs), breaker self-heals (half-open on read, close on probe, cooldown decay), TTL mutes (mute/off auto-expire 24h, --for/--forever, temp off keeps detection), footer+navmap mute breadcrumbs; prax monitor lock scoped to TG relay role (relay.pid) so viewers always start — Patrick ruling: processes are not agents. Loop proven live: planted commons bug fixed by medic dispatch in 105s byte-identical. 993 prax + 603 trigger green
This commit is contained in:
AIOSAI
@@ -65,6 +65,7 @@ drone @ai_mail inbox # check mail → view <id>
|
||||
drone @flow create . "Subject" [dplan] # new plan (default FPLAN)
|
||||
drone @seedgo audit aipass @branch # standards audit (drop @branch = all)
|
||||
drone @seedgo checklist <file|dir> # quick standards check
|
||||
drone @trigger medic mute @<self> # BEFORE build/edit work — auto-expires 24h
|
||||
drone @git status / diff / log # read-only git awareness
|
||||
drone @memory search "query" # recall archived context
|
||||
```
|
||||
|
||||
@@ -13,6 +13,38 @@ PyPI version — not the changelog header.
|
||||
|
||||
### Added
|
||||
|
||||
- **Medic is back on — and the loop is proven live.** Off since 2026-05-10 (a
|
||||
pytest fixture storm flooded the error registry; the off switch was pulled to
|
||||
stop the noise and forgotten for 65 days). Three fixes made re-enable safe:
|
||||
(1) @prax: pytest logging routes to a temp dir when `PYTEST_CURRENT_TEST` is
|
||||
set — test fixtures can never pollute production `logs/` again (the storm
|
||||
class that caused the shutdown); (2) @trigger: circuit breaker self-heals —
|
||||
open breakers half-open on read, close on a successful probe, cooldown decays
|
||||
to base (previously `half_open` was a terminal trap and only manual reset
|
||||
recovered); (3) @trigger: **TTL mutes** — `medic mute @branch` and `medic off`
|
||||
now auto-expire after 24h by default (`--for 48h/7d` custom, `--forever`
|
||||
explicit kill switch; temp `off` keeps detection running). Agents doing build
|
||||
work mute themselves and never have to remember to unmute — the permanent
|
||||
switch that got medic forgotten no longer exists. Breadcrumbs shipped: ai_mail
|
||||
footer + navmap tell every agent to mute before build work. Live-fire proof:
|
||||
a planted commons SQL bug was detected, dispatched, and fixed byte-identical
|
||||
by @commons in 105 seconds (15/15 tests green); a real TG poll error was
|
||||
correctly triaged NOT ACTIONABLE; organic instance-lock noise was correctly
|
||||
triaged LOW/expected. @skills/@api on 7-day mutes until the TG poll-level fix
|
||||
lands. 993 prax + 603 trigger tests green.
|
||||
|
||||
- **Prax monitor: concurrent viewers — laptop and Telegram mirror side by
|
||||
side.** Patrick's ruling after being locked out of his own monitor three
|
||||
times: *processes are not agents; display processes must never be
|
||||
single-instance.* The instance lock is gone from the display path — any
|
||||
number of `monitor run` viewers start and render concurrently. The lock is
|
||||
scoped to the one true single-writer responsibility: the Telegram relay
|
||||
(`relay.pid`, held by `prax-monitor.service`); extra instances run
|
||||
viewer-only, so no TG double-sends. The misleading "kill the existing
|
||||
process" error is dead. 998 prax tests green, 3 new concurrent-viewer tests;
|
||||
live-verified: interactive Mission Control rendering while the TG relay
|
||||
service runs untouched.
|
||||
|
||||
- **Telegram user-comment mirror: the TG chat now shows the whole conversation,
|
||||
whichever door you speak through.** Patrick's spec from the live cross-door
|
||||
drill: his own messages typed in the terminal or claude.ai remote never
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: footer.py
|
||||
# Description: Email Footer Handler
|
||||
# Version: 1.0.0
|
||||
# Version: 1.1.0
|
||||
# Created: 2026-01-29
|
||||
# Modified: 2026-01-29
|
||||
# Modified: 2026-07-14
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
@@ -19,6 +19,8 @@ from aipass.ai_mail.apps.handlers.json import json_handler
|
||||
# Standard footer for all outgoing emails
|
||||
STANDARD_FOOTER = """
|
||||
---
|
||||
▶ BUILD/EDIT task? First: drone @trigger medic mute @<your-branch> — auto-expires 24h, no unmute.
|
||||
|
||||
⚠️ TASK CHECKLIST (before marking complete):
|
||||
□ SEEDGO CHECK → drone @seedgo audit @branch (80%+)
|
||||
□ UPDATE MEMORIES → Your .trinity/local.json records this work
|
||||
|
||||
@@ -69,6 +69,20 @@
|
||||
"subject": "STATUS decommission — prax dashboard todo_count + engine dormant",
|
||||
"date_closed": "2026-06-08",
|
||||
"location": "prax"
|
||||
},
|
||||
{
|
||||
"plan_id": "FPLAN-0321",
|
||||
"type": "FPLAN",
|
||||
"subject": "Pytest log routing: detect test context, route away from production logs",
|
||||
"date_closed": "2026-07-14",
|
||||
"location": "prax"
|
||||
},
|
||||
{
|
||||
"plan_id": "FPLAN-0323",
|
||||
"type": "FPLAN",
|
||||
"subject": "Remove single-instance lock from monitor display path — concurrent viewers",
|
||||
"date_closed": "2026-07-14",
|
||||
"location": "prax"
|
||||
}
|
||||
],
|
||||
"document_metadata": {
|
||||
|
||||
@@ -31,13 +31,14 @@ import inspect
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import Dict, Any, Optional
|
||||
|
||||
from aipass.prax.apps.handlers.json import json_handler
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# =============================================
|
||||
# CONFIGURATION
|
||||
# =============================================
|
||||
@@ -75,6 +76,10 @@ def get_system_logs_dir() -> Path:
|
||||
p = Path(test_log_dir) / "system"
|
||||
p.mkdir(parents=True, exist_ok=True)
|
||||
return p
|
||||
if os.environ.get("PYTEST_CURRENT_TEST"):
|
||||
p = Path(tempfile.gettempdir()) / "aipass_test_logs" / "system"
|
||||
p.mkdir(parents=True, exist_ok=True)
|
||||
return p
|
||||
global _system_logs_dir_cache
|
||||
if _system_logs_dir_cache is None:
|
||||
repo_root = _find_repo_root()
|
||||
@@ -133,6 +138,10 @@ def get_module_logs_dir(module_name: Optional[str] = None) -> Path:
|
||||
p = Path(test_log_dir) / module_name
|
||||
p.mkdir(parents=True, exist_ok=True)
|
||||
return p
|
||||
if os.environ.get("PYTEST_CURRENT_TEST"):
|
||||
p = Path(tempfile.gettempdir()) / "aipass_test_logs" / module_name
|
||||
p.mkdir(parents=True, exist_ok=True)
|
||||
return p
|
||||
|
||||
# Standard: src/aipass/{module}/logs
|
||||
branch_dir = ECOSYSTEM_ROOT / module_name
|
||||
|
||||
@@ -6,13 +6,14 @@
|
||||
# Modified: 2026-07-10
|
||||
# =============================================
|
||||
|
||||
"""Single-instance lock for the prax monitor.
|
||||
"""Relay-scoped lock for the prax monitor Telegram relay.
|
||||
|
||||
Prevents duplicate monitor processes from running concurrently (and
|
||||
double-sending Telegram relay messages). Uses a pidfile with liveness
|
||||
check — cross-platform (Linux / macOS / Windows).
|
||||
Prevents duplicate Telegram sends when multiple monitor viewers run
|
||||
concurrently. The display path is never blocked — only the TG relay
|
||||
acquires this lock, so interactive viewers always start.
|
||||
|
||||
Lock file lives in prax_json/monitor.pid (outside system_logs/ to avoid
|
||||
Uses a pidfile with liveness check — cross-platform (Linux / macOS / Windows).
|
||||
Lock file lives in prax_json/relay.pid (outside system_logs/ to avoid
|
||||
the tailed-directory feedback loop).
|
||||
"""
|
||||
|
||||
@@ -82,14 +83,14 @@ def _is_pid_alive(pid: int) -> bool:
|
||||
|
||||
|
||||
def get_lock_path() -> Path:
|
||||
"""Return the path for the monitor single-instance lock file."""
|
||||
"""Return the path for the relay lock file."""
|
||||
if _lock_path_override is not None:
|
||||
return _lock_path_override
|
||||
return Path(__file__).resolve().parent.parent.parent / "prax_json" / "monitor.pid"
|
||||
return Path(__file__).resolve().parent.parent.parent / "prax_json" / "relay.pid"
|
||||
|
||||
|
||||
def acquire(error_fn=None) -> None:
|
||||
"""Acquire single-instance lock. Raises SystemExit(1) if another live instance holds it."""
|
||||
def try_acquire() -> bool:
|
||||
"""Try to acquire the relay lock. Returns True if acquired, False if held by a live process."""
|
||||
global _held_lock
|
||||
lock_path = get_lock_path()
|
||||
json_handler.log_operation("instance_lock_acquire", {"pid": os.getpid()})
|
||||
@@ -99,11 +100,8 @@ def acquire(error_fn=None) -> None:
|
||||
data = _json.loads(lock_path.read_text(encoding="utf-8"))
|
||||
existing_pid = data.get("pid", 0)
|
||||
if existing_pid and _is_pid_alive(existing_pid):
|
||||
msg = f"Monitor already running (PID {existing_pid}). Kill the existing process or remove {lock_path}"
|
||||
if error_fn:
|
||||
error_fn(msg)
|
||||
logger.error("[instance_lock] %s", msg)
|
||||
raise SystemExit(1)
|
||||
logger.info("[instance_lock] Relay lock held by PID %d — skipping TG relay", existing_pid)
|
||||
return False
|
||||
logger.info("[instance_lock] Reclaiming stale lock (PID %d is dead)", existing_pid)
|
||||
except (ValueError, OSError) as exc:
|
||||
logger.info("[instance_lock] Removing corrupt lock file: %s", exc)
|
||||
@@ -111,7 +109,8 @@ def acquire(error_fn=None) -> None:
|
||||
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
lock_path.write_text(_json.dumps({"pid": os.getpid()}), encoding="utf-8")
|
||||
_held_lock = lock_path
|
||||
logger.info("[instance_lock] Acquired (PID %d)", os.getpid())
|
||||
logger.info("[instance_lock] Acquired relay lock (PID %d)", os.getpid())
|
||||
return True
|
||||
|
||||
|
||||
def release() -> None:
|
||||
|
||||
@@ -26,6 +26,7 @@ from urllib.request import urlopen as _http_fetch
|
||||
|
||||
from aipass.prax.apps.modules.logger import get_direct_logger
|
||||
from aipass.prax.apps.handlers.json import json_handler
|
||||
from aipass.prax.apps.handlers.monitoring import instance_lock
|
||||
|
||||
logger = get_direct_logger()
|
||||
|
||||
@@ -72,6 +73,10 @@ def init_relay(enabled: bool, config: Optional[dict] = None) -> None:
|
||||
logger.info("[telegram_relay] Incomplete config (missing bot_token or chat_id) — relay inactive")
|
||||
return
|
||||
|
||||
if not instance_lock.try_acquire():
|
||||
logger.info("[telegram_relay] Another process owns the TG relay — viewer-only mode")
|
||||
return
|
||||
|
||||
_bot_token = token
|
||||
_chat_id = int(chat)
|
||||
_RELAY_ACTIVE = True
|
||||
@@ -111,6 +116,7 @@ def stop_relay() -> None:
|
||||
_thread.join(timeout=BATCH_INTERVAL + 2)
|
||||
_thread = None
|
||||
|
||||
instance_lock.release()
|
||||
json_handler.log_operation("relay_stopped", {})
|
||||
logger.info("[telegram_relay] Relay stopped")
|
||||
|
||||
|
||||
@@ -48,7 +48,6 @@ from aipass.prax.apps.handlers.monitoring.telegram_relay import (
|
||||
is_relay_enabled_by_env,
|
||||
)
|
||||
from aipass.prax.apps.handlers.monitoring.pid_cache import get_pid_for_branch as _get_pid_for_branch
|
||||
from aipass.prax.apps.handlers.monitoring import instance_lock
|
||||
|
||||
import json as _json
|
||||
|
||||
@@ -177,8 +176,6 @@ def _run_monitor(args: List[str]) -> bool:
|
||||
global _event_queue, _module_tracker
|
||||
global _display_thread, _file_watcher_thread, _log_watcher_thread
|
||||
|
||||
instance_lock.acquire(error_fn=error)
|
||||
|
||||
json_handler.log_operation("monitor_started", {"args": args})
|
||||
logger.info(f"Starting unified monitoring (args: {args})")
|
||||
|
||||
@@ -258,7 +255,6 @@ def _stop_threads():
|
||||
if t is not None and t.is_alive():
|
||||
t.join(timeout=2.0)
|
||||
|
||||
instance_lock.release()
|
||||
logger.info("All monitoring threads stopped")
|
||||
|
||||
|
||||
|
||||
@@ -47,8 +47,9 @@ def _fresh_import_load(monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(load_mod, "PRAX_LOGGER_CONFIG_FILE", prax_json_dir / "prax_logger_config.json")
|
||||
# Reset the lazy cache so get_system_logs_dir() re-resolves
|
||||
monkeypatch.setattr(load_mod, "_system_logs_dir_cache", None)
|
||||
# Clear test log redirect so tests exercise real path resolution
|
||||
# Clear test log redirects so tests exercise real path resolution
|
||||
monkeypatch.delenv("AIPASS_TEST_LOG_DIR", raising=False)
|
||||
monkeypatch.delenv("PYTEST_CURRENT_TEST", raising=False)
|
||||
|
||||
return load_mod
|
||||
|
||||
@@ -181,6 +182,50 @@ class TestGetModuleLogsDir:
|
||||
assert not (tmp_path / "polyglot").exists()
|
||||
|
||||
|
||||
# =============================================
|
||||
# TESTS: PYTEST_CURRENT_TEST routing
|
||||
# =============================================
|
||||
|
||||
|
||||
class TestPytestCurrentTestRouting:
|
||||
"""PYTEST_CURRENT_TEST env var routes logs to temp dir, not production."""
|
||||
|
||||
def test_system_logs_routed_to_temp(self, mock_prax_infrastructure, monkeypatch, tmp_path):
|
||||
load_mod = _fresh_import_load(monkeypatch, tmp_path)
|
||||
monkeypatch.setenv("PYTEST_CURRENT_TEST", "tests/test_foo.py::test_bar (call)")
|
||||
result = load_mod.get_system_logs_dir()
|
||||
assert "aipass_test_logs" in str(result)
|
||||
assert result.name == "system"
|
||||
assert result.exists()
|
||||
assert not (tmp_path / "system_logs").exists()
|
||||
|
||||
def test_module_logs_routed_to_temp(self, mock_prax_infrastructure, monkeypatch, tmp_path):
|
||||
load_mod = _fresh_import_load(monkeypatch, tmp_path)
|
||||
monkeypatch.setenv("PYTEST_CURRENT_TEST", "tests/test_foo.py::test_bar (call)")
|
||||
(tmp_path / "flow").mkdir()
|
||||
result = load_mod.get_module_logs_dir("flow")
|
||||
assert "aipass_test_logs" in str(result)
|
||||
assert result.name == "flow"
|
||||
assert result.exists()
|
||||
assert not (tmp_path / "flow" / "logs").exists()
|
||||
|
||||
def test_aipass_test_log_dir_takes_precedence(self, mock_prax_infrastructure, monkeypatch, tmp_path):
|
||||
load_mod = _fresh_import_load(monkeypatch, tmp_path)
|
||||
override = tmp_path / "custom_test_logs"
|
||||
override.mkdir()
|
||||
monkeypatch.setenv("AIPASS_TEST_LOG_DIR", str(override))
|
||||
monkeypatch.setenv("PYTEST_CURRENT_TEST", "tests/test_foo.py::test_bar (call)")
|
||||
result = load_mod.get_system_logs_dir()
|
||||
assert result == override / "system"
|
||||
|
||||
def test_no_pytest_env_uses_production_path(self, mock_prax_infrastructure, monkeypatch, tmp_path):
|
||||
load_mod = _fresh_import_load(monkeypatch, tmp_path)
|
||||
monkeypatch.delenv("PYTEST_CURRENT_TEST", raising=False)
|
||||
monkeypatch.setattr(load_mod, "_find_repo_root", lambda: tmp_path)
|
||||
result = load_mod.get_system_logs_dir()
|
||||
assert result == tmp_path / "system_logs"
|
||||
|
||||
|
||||
# =============================================
|
||||
# TESTS: lines_to_bytes
|
||||
# =============================================
|
||||
|
||||
@@ -10,8 +10,9 @@
|
||||
|
||||
Covers:
|
||||
- _is_pid_alive() cross-platform liveness check
|
||||
- acquire() creates lock, refuses live duplicate, reclaims stale
|
||||
- try_acquire() creates lock, returns False for live duplicate, reclaims stale
|
||||
- release() removes lock file on clean shutdown
|
||||
- Concurrent viewer: relay lock scoped to TG sends, never blocks display
|
||||
"""
|
||||
|
||||
import json
|
||||
@@ -94,86 +95,78 @@ class TestIsPidAlive:
|
||||
assert mod._is_pid_alive(1234) is True
|
||||
|
||||
|
||||
class TestAcquire:
|
||||
"""Test single-instance lock acquisition."""
|
||||
class TestTryAcquire:
|
||||
"""Test relay lock acquisition."""
|
||||
|
||||
def test_creates_lock_file(self, tmp_path):
|
||||
"""acquire() creates a lock file with the current PID."""
|
||||
"""try_acquire() creates a lock file with the current PID."""
|
||||
mod = _import_lock()
|
||||
lock_path = tmp_path / "monitor.pid"
|
||||
lock_path = tmp_path / "relay.pid"
|
||||
setattr(mod, "_lock_path_override", lock_path)
|
||||
|
||||
mod.acquire()
|
||||
assert mod.try_acquire() is True
|
||||
|
||||
assert lock_path.exists()
|
||||
data = json.loads(lock_path.read_text(encoding="utf-8"))
|
||||
assert data["pid"] == os.getpid()
|
||||
|
||||
def test_refuses_when_live_instance_holds_lock(self, tmp_path):
|
||||
"""acquire() exits with SystemExit(1) when another live process holds the lock."""
|
||||
def test_returns_false_when_live_holder(self, tmp_path):
|
||||
"""try_acquire() returns False when another live process holds the lock."""
|
||||
mod = _import_lock()
|
||||
lock_path = tmp_path / "monitor.pid"
|
||||
lock_path = tmp_path / "relay.pid"
|
||||
setattr(mod, "_lock_path_override", lock_path)
|
||||
|
||||
lock_path.write_text(json.dumps({"pid": os.getpid()}), encoding="utf-8")
|
||||
|
||||
import pytest
|
||||
|
||||
mock_error = MagicMock()
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
mod.acquire(error_fn=mock_error)
|
||||
assert exc_info.value.code == 1
|
||||
mock_error.assert_called_once()
|
||||
assert str(os.getpid()) in mock_error.call_args[0][0]
|
||||
assert mod.try_acquire() is False
|
||||
|
||||
def test_reclaims_stale_lock(self, tmp_path):
|
||||
"""acquire() reclaims the lock when the recorded PID is dead."""
|
||||
"""try_acquire() reclaims the lock when the recorded PID is dead."""
|
||||
mod = _import_lock()
|
||||
lock_path = tmp_path / "monitor.pid"
|
||||
lock_path = tmp_path / "relay.pid"
|
||||
setattr(mod, "_lock_path_override", lock_path)
|
||||
|
||||
lock_path.write_text(json.dumps({"pid": 99999999}), encoding="utf-8")
|
||||
|
||||
with patch.object(mod, "_is_pid_alive", return_value=False):
|
||||
mod.acquire()
|
||||
assert mod.try_acquire() is True
|
||||
|
||||
data = json.loads(lock_path.read_text(encoding="utf-8"))
|
||||
assert data["pid"] == os.getpid()
|
||||
|
||||
def test_reclaims_corrupt_lock_file(self, tmp_path):
|
||||
"""acquire() overwrites a corrupt lock file."""
|
||||
"""try_acquire() overwrites a corrupt lock file."""
|
||||
mod = _import_lock()
|
||||
lock_path = tmp_path / "monitor.pid"
|
||||
lock_path = tmp_path / "relay.pid"
|
||||
setattr(mod, "_lock_path_override", lock_path)
|
||||
|
||||
lock_path.write_text("{corrupt json", encoding="utf-8")
|
||||
|
||||
mod.acquire()
|
||||
assert mod.try_acquire() is True
|
||||
|
||||
data = json.loads(lock_path.read_text(encoding="utf-8"))
|
||||
assert data["pid"] == os.getpid()
|
||||
|
||||
def test_creates_parent_directories(self, tmp_path):
|
||||
"""acquire() creates parent directories if they don't exist."""
|
||||
"""try_acquire() creates parent directories if they don't exist."""
|
||||
mod = _import_lock()
|
||||
lock_path = tmp_path / "nested" / "dir" / "monitor.pid"
|
||||
lock_path = tmp_path / "nested" / "dir" / "relay.pid"
|
||||
setattr(mod, "_lock_path_override", lock_path)
|
||||
|
||||
mod.acquire()
|
||||
|
||||
assert mod.try_acquire() is True
|
||||
assert lock_path.exists()
|
||||
|
||||
|
||||
class TestRelease:
|
||||
"""Test single-instance lock release."""
|
||||
"""Test relay lock release."""
|
||||
|
||||
def test_removes_lock_file(self, tmp_path):
|
||||
"""release() removes the lock file."""
|
||||
mod = _import_lock()
|
||||
lock_path = tmp_path / "monitor.pid"
|
||||
lock_path = tmp_path / "relay.pid"
|
||||
setattr(mod, "_lock_path_override", lock_path)
|
||||
|
||||
mod.acquire()
|
||||
mod.try_acquire()
|
||||
assert lock_path.exists()
|
||||
|
||||
mod.release()
|
||||
@@ -182,10 +175,10 @@ class TestRelease:
|
||||
def test_clears_held_lock_state(self, tmp_path):
|
||||
"""release() clears the _held_lock global."""
|
||||
mod = _import_lock()
|
||||
lock_path = tmp_path / "monitor.pid"
|
||||
lock_path = tmp_path / "relay.pid"
|
||||
setattr(mod, "_lock_path_override", lock_path)
|
||||
|
||||
mod.acquire()
|
||||
mod.try_acquire()
|
||||
mod.release()
|
||||
assert mod._held_lock is None
|
||||
|
||||
@@ -198,10 +191,39 @@ class TestRelease:
|
||||
def test_release_handles_already_deleted_file(self, tmp_path):
|
||||
"""release() handles the case where the lock file was already deleted."""
|
||||
mod = _import_lock()
|
||||
lock_path = tmp_path / "monitor.pid"
|
||||
lock_path = tmp_path / "relay.pid"
|
||||
setattr(mod, "_lock_path_override", lock_path)
|
||||
|
||||
mod.acquire()
|
||||
mod.try_acquire()
|
||||
lock_path.unlink()
|
||||
mod.release()
|
||||
assert mod._held_lock is None
|
||||
|
||||
|
||||
class TestConcurrentViewers:
|
||||
"""Concurrent monitor viewers: relay lock scoped, display never blocked."""
|
||||
|
||||
def test_second_acquire_returns_false(self, tmp_path):
|
||||
"""Second try_acquire() returns False when first holds the lock."""
|
||||
mod = _import_lock()
|
||||
lock_path = tmp_path / "relay.pid"
|
||||
setattr(mod, "_lock_path_override", lock_path)
|
||||
|
||||
assert mod.try_acquire() is True
|
||||
assert mod.try_acquire() is False
|
||||
|
||||
def test_release_then_reacquire(self, tmp_path):
|
||||
"""After release(), another process can acquire the relay lock."""
|
||||
mod = _import_lock()
|
||||
lock_path = tmp_path / "relay.pid"
|
||||
setattr(mod, "_lock_path_override", lock_path)
|
||||
|
||||
assert mod.try_acquire() is True
|
||||
mod.release()
|
||||
assert mod.try_acquire() is True
|
||||
|
||||
def test_lock_path_is_relay_pid(self):
|
||||
"""Default lock file is relay.pid, not monitor.pid."""
|
||||
mod = _import_lock()
|
||||
setattr(mod, "_lock_path_override", None)
|
||||
assert mod.get_lock_path().name == "relay.pid"
|
||||
|
||||
@@ -551,6 +551,12 @@
|
||||
"lines": [40],
|
||||
"pattern": "handler imported directly",
|
||||
"reason": "Test helper _import_module() must import the handler module directly to test it. All trigger test files follow this pattern."
|
||||
},
|
||||
{
|
||||
"file": "apps/handlers/medic_state.py",
|
||||
"standard": "unused_function",
|
||||
"pattern": "get_muted_branches",
|
||||
"reason": "Public API returning List[str] of active muted branches. Used by 15+ existing tests and part of the medic_state interface. get_muted_branches_detail() supplements it for status display — this is the simple accessor."
|
||||
}
|
||||
],
|
||||
"notes": {
|
||||
|
||||
@@ -227,6 +227,25 @@ _circuit_breaker = _load_circuit_breaker_state()
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _evaluate_state() -> None:
|
||||
"""Evaluate circuit breaker state transitions on read.
|
||||
|
||||
If state is 'open' and cooldown has expired, transition to 'half_open'
|
||||
with the probe slot available. Called by both circuit_breaker_allows()
|
||||
and get_circuit_breaker_status() so the breaker self-heals even when
|
||||
medic is off and no dispatches are running.
|
||||
"""
|
||||
global _circuit_breaker
|
||||
if _circuit_breaker.state != "open":
|
||||
return
|
||||
elapsed = time.time() - _circuit_breaker.opened_at
|
||||
if elapsed >= _circuit_breaker.cooldown_seconds:
|
||||
_circuit_breaker.state = "half_open"
|
||||
_circuit_breaker.half_open_allow = True
|
||||
_circuit_breaker.summary_sent = False
|
||||
_save_circuit_breaker_state()
|
||||
|
||||
|
||||
def circuit_breaker_allows() -> bool:
|
||||
"""Check if the circuit breaker allows dispatch.
|
||||
|
||||
@@ -234,34 +253,25 @@ def circuit_breaker_allows() -> bool:
|
||||
- Closed (normal): All dispatches allowed. Records are checked against
|
||||
trip_threshold to determine if breaker should open.
|
||||
- Open (paused): No dispatches. Transitions to half_open after cooldown
|
||||
period expires.
|
||||
- Half-Open (testing): Allow ONE dispatch to test recovery. If it
|
||||
resolves, caller should reset to Closed. If another error comes,
|
||||
circuit_breaker_record_error() will re-open with doubled cooldown.
|
||||
period expires (evaluated on read via _evaluate_state).
|
||||
- Half-Open (testing): Allow ONE dispatch to test recovery. On success
|
||||
the caller must call circuit_breaker_probe_succeeded() to close.
|
||||
If another error comes, circuit_breaker_record_error() re-opens
|
||||
with doubled cooldown.
|
||||
|
||||
Returns:
|
||||
True if dispatch is allowed, False if breaker is blocking
|
||||
"""
|
||||
global _circuit_breaker
|
||||
now = time.time()
|
||||
_evaluate_state()
|
||||
|
||||
if _circuit_breaker.state == "closed":
|
||||
return True
|
||||
|
||||
if _circuit_breaker.state == "open":
|
||||
elapsed = now - _circuit_breaker.opened_at
|
||||
if elapsed >= _circuit_breaker.cooldown_seconds:
|
||||
# Cooldown expired - transition to half_open
|
||||
# This call IS the probe dispatch, so mark probe as used
|
||||
_circuit_breaker.state = "half_open"
|
||||
_circuit_breaker.half_open_allow = False
|
||||
_circuit_breaker.summary_sent = False
|
||||
return True
|
||||
return False
|
||||
|
||||
if _circuit_breaker.state == "half_open":
|
||||
if _circuit_breaker.half_open_allow:
|
||||
_circuit_breaker.half_open_allow = False
|
||||
_save_circuit_breaker_state()
|
||||
return True
|
||||
return False
|
||||
|
||||
@@ -319,6 +329,25 @@ def circuit_breaker_trip(reason: str = "") -> None:
|
||||
_save_circuit_breaker_state()
|
||||
|
||||
|
||||
def circuit_breaker_probe_succeeded() -> None:
|
||||
"""Close the breaker after a successful dispatch during half_open probe.
|
||||
|
||||
Transitions half_open -> closed and resets cooldown to base_cooldown
|
||||
so future trips start with the short cooldown again. No-op if the
|
||||
breaker is not in half_open state.
|
||||
"""
|
||||
global _circuit_breaker
|
||||
if _circuit_breaker.state != "half_open":
|
||||
return
|
||||
_circuit_breaker.state = "closed"
|
||||
_circuit_breaker.opened_at = 0.0
|
||||
_circuit_breaker.cooldown_seconds = _circuit_breaker.base_cooldown
|
||||
_circuit_breaker.recent_errors = []
|
||||
_circuit_breaker.summary_sent = False
|
||||
_circuit_breaker.half_open_allow = True
|
||||
_clear_circuit_breaker_state()
|
||||
|
||||
|
||||
def circuit_breaker_reset() -> None:
|
||||
"""Reset circuit breaker to closed state.
|
||||
|
||||
@@ -339,16 +368,24 @@ def circuit_breaker_reset() -> None:
|
||||
def get_circuit_breaker_status() -> dict:
|
||||
"""Get current circuit breaker state as a dictionary.
|
||||
|
||||
Evaluates state transitions first so the returned state is always
|
||||
up-to-date (e.g. an expired open breaker will report as half_open).
|
||||
|
||||
Returns:
|
||||
Dict with keys: state, opened_at, cooldown_seconds,
|
||||
recent_error_count, summary_sent
|
||||
recent_error_count, summary_sent, remaining_seconds
|
||||
"""
|
||||
_evaluate_state()
|
||||
remaining = 0
|
||||
if _circuit_breaker.state == "open":
|
||||
remaining = max(0, int(_circuit_breaker.cooldown_seconds - (time.time() - _circuit_breaker.opened_at)))
|
||||
return {
|
||||
"state": _circuit_breaker.state,
|
||||
"opened_at": _circuit_breaker.opened_at,
|
||||
"cooldown_seconds": _circuit_breaker.cooldown_seconds,
|
||||
"recent_error_count": len(_circuit_breaker.recent_errors),
|
||||
"summary_sent": _circuit_breaker.summary_sent,
|
||||
"remaining_seconds": remaining,
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -81,6 +81,7 @@ try:
|
||||
from aipass.trigger.apps.handlers.error_registry import (
|
||||
circuit_breaker_allows,
|
||||
circuit_breaker_record_error,
|
||||
circuit_breaker_probe_succeeded,
|
||||
should_dispatch as registry_should_dispatch,
|
||||
record_dispatch as registry_record_dispatch,
|
||||
)
|
||||
@@ -97,6 +98,10 @@ except ImportError:
|
||||
"""Fallback no-op error recording when error_registry is unavailable."""
|
||||
pass
|
||||
|
||||
def circuit_breaker_probe_succeeded() -> None:
|
||||
"""Fallback no-op when error_registry is unavailable."""
|
||||
pass
|
||||
|
||||
def registry_should_dispatch(fingerprint: str) -> bool:
|
||||
"""Fallback dispatch check that always allows dispatch for any fingerprint."""
|
||||
return True
|
||||
@@ -116,44 +121,70 @@ def _is_medic_enabled() -> bool:
|
||||
"""
|
||||
Check if medic (auto-healing dispatch) is enabled.
|
||||
|
||||
Reads medic_enabled from trigger_config.json.
|
||||
Reads medic_enabled from trigger_config.json. If disabled with a TTL
|
||||
(medic_disabled_until timestamp), treats an expired TTL as enabled.
|
||||
Defaults to True if config is missing or unreadable.
|
||||
|
||||
Returns:
|
||||
True if medic dispatch is enabled
|
||||
"""
|
||||
try:
|
||||
if TRIGGER_CONFIG_FILE.exists():
|
||||
data = json.loads(TRIGGER_CONFIG_FILE.read_text(encoding="utf-8"))
|
||||
return bool(data.get("config", {}).get("medic_enabled", True))
|
||||
if not TRIGGER_CONFIG_FILE.exists():
|
||||
return True
|
||||
data = json.loads(TRIGGER_CONFIG_FILE.read_text(encoding="utf-8"))
|
||||
config = data.get("config", {})
|
||||
enabled = bool(config.get("medic_enabled", True))
|
||||
if enabled:
|
||||
return True
|
||||
disabled_until = config.get("medic_disabled_until")
|
||||
if disabled_until and datetime.fromisoformat(disabled_until) <= datetime.now():
|
||||
return True
|
||||
return False
|
||||
except Exception as exc:
|
||||
_log_warning(f"_is_medic_enabled config read failed: {exc}")
|
||||
return True # Default to enabled on read failure
|
||||
return True
|
||||
return True
|
||||
|
||||
|
||||
def _mute_entry_matches(entry, branch_lower: str, now: datetime) -> bool:
|
||||
"""Check if a single mute entry matches the branch and is still active."""
|
||||
if isinstance(entry, str):
|
||||
return entry.lower() == branch_lower
|
||||
if not isinstance(entry, dict):
|
||||
return False
|
||||
if entry.get("name", "").lower() != branch_lower:
|
||||
return False
|
||||
expires_at = entry.get("expires_at")
|
||||
if expires_at is None:
|
||||
return True
|
||||
return datetime.fromisoformat(expires_at) > now
|
||||
|
||||
|
||||
def _is_branch_muted(branch_name: str) -> bool:
|
||||
"""
|
||||
Check if a specific branch is muted for medic dispatch.
|
||||
|
||||
Reads muted_branches list from trigger_config.json.
|
||||
Muted branches have errors detected but NOT dispatched.
|
||||
Reads muted_branches list from trigger_config.json. Supports both
|
||||
legacy plain-string entries (permanent) and new dict entries with
|
||||
optional expires_at timestamp. Expired TTL mutes are treated as
|
||||
unmuted.
|
||||
|
||||
Args:
|
||||
branch_name: Branch name (case-insensitive)
|
||||
|
||||
Returns:
|
||||
True if branch is in the muted list
|
||||
True if branch is actively muted
|
||||
"""
|
||||
try:
|
||||
if TRIGGER_CONFIG_FILE.exists():
|
||||
data = json.loads(TRIGGER_CONFIG_FILE.read_text(encoding="utf-8"))
|
||||
muted = data.get("config", {}).get("muted_branches", [])
|
||||
return branch_name.lower() in [b.lower() for b in muted]
|
||||
if not TRIGGER_CONFIG_FILE.exists():
|
||||
return False
|
||||
data = json.loads(TRIGGER_CONFIG_FILE.read_text(encoding="utf-8"))
|
||||
muted = data.get("config", {}).get("muted_branches", [])
|
||||
branch_lower = branch_name.lower()
|
||||
now = datetime.now()
|
||||
return any(_mute_entry_matches(e, branch_lower, now) for e in muted)
|
||||
except Exception as exc:
|
||||
_log_warning(f"_is_branch_muted config read failed: {exc}")
|
||||
return False
|
||||
return False
|
||||
|
||||
|
||||
def set_send_email_callback(callback: Callable[..., bool]) -> None:
|
||||
@@ -550,6 +581,7 @@ def handle_error_detected(
|
||||
if _REGISTRY_DISPATCH_AVAILABLE and fingerprint:
|
||||
# Medic v2: per-fingerprint dispatch tracking
|
||||
registry_record_dispatch(fingerprint)
|
||||
circuit_breaker_probe_succeeded()
|
||||
else:
|
||||
# Legacy: per-branch rate limiting
|
||||
_record_dispatch(recipient)
|
||||
|
||||
@@ -17,9 +17,10 @@ Architecture:
|
||||
"""
|
||||
|
||||
import json
|
||||
from datetime import datetime
|
||||
import re
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from aipass.prax.apps.modules.logger import get_direct_logger
|
||||
from aipass.trigger.apps.config import TRIGGER_ROOT, atomic_write_json, json_file_lock
|
||||
@@ -31,6 +32,51 @@ TRIGGER_CONFIG_FILE = TRIGGER_ROOT / "trigger_json" / "trigger_config.json"
|
||||
MEDIC_SUPPRESSED_LOG = TRIGGER_ROOT / "logs" / "medic_suppressed.jsonl"
|
||||
RATE_LIMITED_LOG = TRIGGER_ROOT / "logs" / "rate_limited.jsonl"
|
||||
|
||||
_DURATION_RE = re.compile(r"^(\d+)(h|d)$")
|
||||
|
||||
DEFAULT_MUTE_SECONDS = 86400 # 24 hours
|
||||
DEFAULT_OFF_SECONDS = 86400 # 24 hours
|
||||
|
||||
|
||||
def parse_duration(duration_str: str) -> Optional[float]:
|
||||
"""Parse a duration string like '24h', '48h', '7d' into seconds.
|
||||
|
||||
Args:
|
||||
duration_str: Duration with unit suffix (h=hours, d=days)
|
||||
|
||||
Returns:
|
||||
Seconds as float, or None if unparseable
|
||||
"""
|
||||
m = _DURATION_RE.match(duration_str.strip())
|
||||
if not m:
|
||||
return None
|
||||
value, unit = int(m.group(1)), m.group(2)
|
||||
if unit == "h":
|
||||
return float(value * 3600)
|
||||
return float(value * 86400)
|
||||
|
||||
|
||||
def _is_mute_active(entry, now: datetime) -> bool:
|
||||
"""Check if a single mute entry is still active."""
|
||||
if isinstance(entry, str):
|
||||
return True
|
||||
if not isinstance(entry, dict):
|
||||
return False
|
||||
expires_at = entry.get("expires_at")
|
||||
if expires_at is None:
|
||||
return True
|
||||
return datetime.fromisoformat(expires_at) > now
|
||||
|
||||
|
||||
def _clean_expired_mutes(data: dict) -> None:
|
||||
"""Remove expired mute entries from config data in-place."""
|
||||
config = data.get("config", {})
|
||||
muted = config.get("muted_branches", [])
|
||||
if not muted:
|
||||
return
|
||||
now = datetime.now()
|
||||
config["muted_branches"] = [e for e in muted if _is_mute_active(e, now)]
|
||||
|
||||
|
||||
def read_config() -> dict:
|
||||
"""
|
||||
@@ -50,7 +96,7 @@ def read_config() -> dict:
|
||||
|
||||
def write_config(data: dict) -> bool:
|
||||
"""
|
||||
Write trigger_config.json.
|
||||
Write trigger_config.json. Cleans expired mute entries before writing.
|
||||
|
||||
Args:
|
||||
data: Config dict to persist
|
||||
@@ -59,6 +105,7 @@ def write_config(data: dict) -> bool:
|
||||
True on success, False on failure
|
||||
"""
|
||||
try:
|
||||
_clean_expired_mutes(data)
|
||||
atomic_write_json(TRIGGER_CONFIG_FILE, data)
|
||||
return True
|
||||
except Exception as exc:
|
||||
@@ -70,19 +117,43 @@ def is_enabled() -> bool:
|
||||
"""
|
||||
Check if Medic is currently enabled.
|
||||
|
||||
If disabled with a TTL (medic_disabled_until), treats an expired
|
||||
TTL as enabled — evaluate on read, no timers.
|
||||
|
||||
Returns:
|
||||
True if medic_enabled is True in config (defaults to True)
|
||||
True if medic_enabled is True or its TTL has expired
|
||||
"""
|
||||
data = read_config()
|
||||
return bool(data.get("config", {}).get("medic_enabled", True))
|
||||
config = data.get("config", {})
|
||||
enabled = bool(config.get("medic_enabled", True))
|
||||
if not enabled:
|
||||
disabled_until = config.get("medic_disabled_until")
|
||||
if disabled_until:
|
||||
if datetime.fromisoformat(disabled_until) <= datetime.now():
|
||||
return True
|
||||
return enabled
|
||||
|
||||
|
||||
def set_enabled(enabled: bool) -> bool:
|
||||
def get_disabled_until() -> Optional[str]:
|
||||
"""Get the medic_disabled_until timestamp if set.
|
||||
|
||||
Returns:
|
||||
ISO timestamp string, or None if not set or permanent off
|
||||
"""
|
||||
data = read_config()
|
||||
return data.get("config", {}).get("medic_disabled_until")
|
||||
|
||||
|
||||
def set_enabled(enabled: bool, duration_seconds: Optional[float] = None) -> bool:
|
||||
"""
|
||||
Set medic_enabled flag in config.
|
||||
|
||||
When disabling with a duration, stores medic_disabled_until so the
|
||||
off state auto-expires. When enabling, clears any stored expiry.
|
||||
|
||||
Args:
|
||||
enabled: True to enable, False to disable
|
||||
duration_seconds: TTL in seconds for disable (None = permanent)
|
||||
|
||||
Returns:
|
||||
True on success
|
||||
@@ -92,6 +163,11 @@ def set_enabled(enabled: bool) -> bool:
|
||||
if "config" not in data:
|
||||
data["config"] = {}
|
||||
data["config"]["medic_enabled"] = enabled
|
||||
if not enabled and duration_seconds is not None:
|
||||
expires = datetime.now() + timedelta(seconds=duration_seconds)
|
||||
data["config"]["medic_disabled_until"] = expires.isoformat()
|
||||
else:
|
||||
data["config"].pop("medic_disabled_until", None)
|
||||
data["timestamp"] = datetime.now().strftime("%Y-%m-%d")
|
||||
|
||||
if write_config(data):
|
||||
@@ -118,25 +194,74 @@ def _normalize_branch_name(name: str) -> str:
|
||||
|
||||
def get_muted_branches() -> List[str]:
|
||||
"""
|
||||
Get list of muted branch names.
|
||||
Get list of currently active muted branch names.
|
||||
|
||||
Evaluates TTL expiry on read — expired mutes are filtered out.
|
||||
|
||||
Returns:
|
||||
List of muted branch names (lowercase, e.g., ['speakeasy', 'api'])
|
||||
"""
|
||||
data = read_config()
|
||||
raw = data.get("config", {}).get("muted_branches", [])
|
||||
return [_normalize_branch_name(b) for b in raw]
|
||||
now = datetime.now()
|
||||
result = []
|
||||
for entry in raw:
|
||||
if isinstance(entry, str):
|
||||
result.append(_normalize_branch_name(entry))
|
||||
elif isinstance(entry, dict):
|
||||
expires_at = entry.get("expires_at")
|
||||
if expires_at is None or datetime.fromisoformat(expires_at) > now:
|
||||
result.append(_normalize_branch_name(entry.get("name", "")))
|
||||
return result
|
||||
|
||||
|
||||
def mute_branch(branch_name: str) -> bool:
|
||||
def get_muted_branches_detail() -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Add a branch to the muted list.
|
||||
Get muted branches with expiry info for status display.
|
||||
|
||||
Returns active mutes only (expired ones filtered out).
|
||||
|
||||
Returns:
|
||||
List of dicts with 'name' and 'expires_at' (None = permanent)
|
||||
"""
|
||||
data = read_config()
|
||||
raw = data.get("config", {}).get("muted_branches", [])
|
||||
now = datetime.now()
|
||||
result = []
|
||||
for entry in raw:
|
||||
if isinstance(entry, str):
|
||||
result.append({"name": _normalize_branch_name(entry), "expires_at": None})
|
||||
elif isinstance(entry, dict):
|
||||
expires_at = entry.get("expires_at")
|
||||
if expires_at is None or datetime.fromisoformat(expires_at) > now:
|
||||
result.append(
|
||||
{
|
||||
"name": _normalize_branch_name(entry.get("name", "")),
|
||||
"expires_at": expires_at,
|
||||
}
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def _mute_entry_name(entry) -> str:
|
||||
"""Extract the normalized branch name from a mute entry (string or dict)."""
|
||||
if isinstance(entry, str):
|
||||
return _normalize_branch_name(entry)
|
||||
if isinstance(entry, dict):
|
||||
return _normalize_branch_name(entry.get("name", ""))
|
||||
return ""
|
||||
|
||||
|
||||
def mute_branch(branch_name: str, duration_seconds: Optional[float] = None) -> bool:
|
||||
"""
|
||||
Add a branch to the muted list with optional TTL.
|
||||
|
||||
Muted branches will have errors detected but NOT dispatched.
|
||||
Persists in trigger_config.json.
|
||||
|
||||
Args:
|
||||
branch_name: Branch name (with or without @)
|
||||
duration_seconds: TTL in seconds (None = permanent/forever)
|
||||
|
||||
Returns:
|
||||
True on success
|
||||
@@ -146,10 +271,14 @@ def mute_branch(branch_name: str) -> bool:
|
||||
data = read_config()
|
||||
if "config" not in data:
|
||||
data["config"] = {}
|
||||
muted = [_normalize_branch_name(b) for b in data["config"].get("muted_branches", [])]
|
||||
if clean not in muted:
|
||||
muted.append(clean)
|
||||
data["config"]["muted_branches"] = muted
|
||||
raw_muted = data["config"].get("muted_branches", [])
|
||||
new_muted = [e for e in raw_muted if _mute_entry_name(e) != clean]
|
||||
if duration_seconds is not None:
|
||||
expires = datetime.now() + timedelta(seconds=duration_seconds)
|
||||
new_muted.append({"name": clean, "expires_at": expires.isoformat()})
|
||||
else:
|
||||
new_muted.append({"name": clean, "expires_at": None})
|
||||
data["config"]["muted_branches"] = new_muted
|
||||
data["timestamp"] = datetime.now().strftime("%Y-%m-%d")
|
||||
return write_config(data)
|
||||
|
||||
@@ -169,9 +298,8 @@ def unmute_branch(branch_name: str) -> bool:
|
||||
data = read_config()
|
||||
if "config" not in data:
|
||||
data["config"] = {}
|
||||
muted = [_normalize_branch_name(b) for b in data["config"].get("muted_branches", [])]
|
||||
muted = [b for b in muted if b != clean]
|
||||
data["config"]["muted_branches"] = muted
|
||||
raw_muted = data["config"].get("muted_branches", [])
|
||||
data["config"]["muted_branches"] = [e for e in raw_muted if _mute_entry_name(e) != clean]
|
||||
data["timestamp"] = datetime.now().strftime("%Y-%m-%d")
|
||||
return write_config(data)
|
||||
|
||||
|
||||
@@ -22,7 +22,6 @@ Architecture: Module orchestrates, error_registry handler manages data
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
from typing import Optional
|
||||
|
||||
|
||||
@@ -465,13 +464,8 @@ def _cmd_circuit_breaker(console, args: list) -> bool:
|
||||
if cb_st == "closed":
|
||||
console.print(" [dim]Normal operation - all dispatch allowed[/dim]")
|
||||
elif cb_st == "open":
|
||||
opened_at = cb.get("opened_at", 0)
|
||||
cooldown = cb.get("cooldown_seconds", 0)
|
||||
if opened_at > 0:
|
||||
remaining = max(0, cooldown - int(time.time() - opened_at))
|
||||
error(f"Dispatch paused - {remaining}s remaining until half-open")
|
||||
else:
|
||||
error("Dispatch paused")
|
||||
remaining = cb.get("remaining_seconds", 0)
|
||||
error(f"Dispatch paused - {remaining}s remaining until half-open")
|
||||
console.print()
|
||||
console.print(" [dim]Run 'drone @trigger errors circuit-breaker reset' to force close[/dim]")
|
||||
elif cb_st == "half_open":
|
||||
|
||||
@@ -30,11 +30,15 @@ from aipass.trigger.apps.handlers.json import json_handler
|
||||
from aipass.trigger.apps.handlers.medic_state import (
|
||||
is_enabled,
|
||||
set_enabled,
|
||||
get_muted_branches,
|
||||
get_muted_branches_detail,
|
||||
get_disabled_until,
|
||||
mute_branch,
|
||||
unmute_branch,
|
||||
get_suppression_stats,
|
||||
get_rate_limit_stats,
|
||||
parse_duration,
|
||||
DEFAULT_MUTE_SECONDS,
|
||||
DEFAULT_OFF_SECONDS,
|
||||
)
|
||||
|
||||
if sys.platform == "win32":
|
||||
@@ -188,21 +192,24 @@ def print_help() -> None:
|
||||
console.rule("COMMANDS")
|
||||
console.print()
|
||||
console.print(" [bold]on[/bold] Enable error dispatch (starts log watcher if needed)")
|
||||
console.print(" [bold]off[/bold] Disable error dispatch globally (errors still logged)")
|
||||
console.print(" [bold]off[/bold] Disable dispatch for 24h (detection continues)")
|
||||
console.print(" [bold]off --forever[/bold] Disable dispatch permanently (stops log watcher)")
|
||||
console.print(" [bold]status[/bold] Show current state, muted branches, and statistics")
|
||||
console.print(" [bold]mute[/bold] @branch Suppress dispatch for a specific branch")
|
||||
console.print(" [bold]mute[/bold] @branch Suppress dispatch for 24h (default)")
|
||||
console.print(" [bold]mute[/bold] @branch --for 48h Custom TTL (e.g. 48h, 7d)")
|
||||
console.print(" [bold]mute[/bold] @branch --forever Permanent mute")
|
||||
console.print(" [bold]unmute[/bold] @branch Resume dispatch for a muted branch")
|
||||
console.print(" [bold]help[/bold] Show this help")
|
||||
console.print()
|
||||
console.rule("OFF vs MUTE")
|
||||
console.print()
|
||||
console.print(" [yellow]off[/yellow] Global kill switch. ALL error dispatch stops. No branch")
|
||||
console.print(" receives auto-healing emails. Errors still logged to")
|
||||
console.print(" medic_suppressed.jsonl for review.")
|
||||
console.print(" [yellow]off[/yellow] 24h dispatch suppression (detection continues).")
|
||||
console.print(" Auto-resumes after 24 hours. Use --forever for")
|
||||
console.print(" permanent disable (stops log watcher too).")
|
||||
console.print()
|
||||
console.print(" [yellow]mute[/yellow] Per-branch suppress. Only the muted branch stops receiving")
|
||||
console.print(" dispatch. All other branches continue normally. Muted errors")
|
||||
console.print(" logged to medic_suppressed.jsonl.")
|
||||
console.print(" [yellow]mute[/yellow] Per-branch suppress for 24h (default).")
|
||||
console.print(" --for 48h or --for 7d for custom duration.")
|
||||
console.print(" --forever for permanent. Auto-expires — no need to unmute.")
|
||||
console.print()
|
||||
console.rule("EXAMPLES")
|
||||
console.print()
|
||||
@@ -231,20 +238,78 @@ def print_help() -> None:
|
||||
console.print()
|
||||
|
||||
|
||||
def _parse_duration_args(args: list) -> tuple:
|
||||
"""Extract --for <dur> and --forever from args.
|
||||
|
||||
Returns:
|
||||
(duration_seconds_or_None, is_forever, remaining_args)
|
||||
Default (no flags): duration=DEFAULT_MUTE_SECONDS, is_forever=False
|
||||
"""
|
||||
remaining = []
|
||||
duration = None
|
||||
is_forever = False
|
||||
i = 0
|
||||
while i < len(args):
|
||||
if args[i] == "--forever":
|
||||
is_forever = True
|
||||
i += 1
|
||||
elif args[i] == "--for" and i + 1 < len(args):
|
||||
parsed = parse_duration(args[i + 1])
|
||||
if parsed is not None:
|
||||
duration = parsed
|
||||
i += 2
|
||||
else:
|
||||
remaining.append(args[i])
|
||||
i += 1
|
||||
if is_forever:
|
||||
return None, True, remaining
|
||||
if duration is not None:
|
||||
return duration, False, remaining
|
||||
return float(DEFAULT_MUTE_SECONDS), False, remaining
|
||||
|
||||
|
||||
def _fmt_remaining(iso_expiry: str) -> str:
|
||||
"""Format time remaining from an ISO expiry timestamp."""
|
||||
from datetime import datetime
|
||||
|
||||
try:
|
||||
expires = datetime.fromisoformat(iso_expiry)
|
||||
remaining = expires - datetime.now()
|
||||
total_secs = max(0, int(remaining.total_seconds()))
|
||||
if total_secs <= 0:
|
||||
return "expired"
|
||||
hours, rem = divmod(total_secs, 3600)
|
||||
minutes = rem // 60
|
||||
if hours >= 24:
|
||||
days = hours // 24
|
||||
hours = hours % 24
|
||||
return f"{days}d {hours}h"
|
||||
return f"{hours}h {minutes}m"
|
||||
except Exception as exc:
|
||||
logger.warning("[MEDIC] _fmt_remaining parse failed: %s", exc)
|
||||
return "unknown"
|
||||
|
||||
|
||||
def _handle_mute(console, args: list) -> None:
|
||||
"""Handle 'medic mute @branch'."""
|
||||
"""Handle 'medic mute @branch [--for <dur>] [--forever]'."""
|
||||
from aipass.cli.apps.modules import error
|
||||
|
||||
if not args:
|
||||
error("Missing branch name", suggestion="Usage: medic mute @branch")
|
||||
duration_secs, is_forever, rest = _parse_duration_args(args)
|
||||
|
||||
if not rest:
|
||||
error("Missing branch name", suggestion="Usage: medic mute @branch [--for 48h] [--forever]")
|
||||
return
|
||||
branch_name = _extract_branch_name(args[0])
|
||||
branch_name = _extract_branch_name(rest[0])
|
||||
if not branch_name:
|
||||
error("Missing branch name", suggestion="Usage: medic mute @branch")
|
||||
error("Missing branch name", suggestion="Usage: medic mute @branch [--for 48h] [--forever]")
|
||||
return
|
||||
if mute_branch(branch_name):
|
||||
if mute_branch(branch_name, duration_seconds=duration_secs):
|
||||
logger.info(f"[MEDIC] Muted branch: {branch_name}")
|
||||
console.print(f" [yellow]Muted[/yellow] @{branch_name} — errors logged but not dispatched")
|
||||
if is_forever or duration_secs is None:
|
||||
console.print(f" [yellow]Muted[/yellow] @{branch_name} — permanent (use unmute to restore)")
|
||||
else:
|
||||
hours = int(duration_secs) // 3600
|
||||
console.print(f" [yellow]Muted[/yellow] @{branch_name} — auto-expires in {hours}h")
|
||||
else:
|
||||
error(f"Failed to mute @{branch_name}", suggestion="Check trigger_config.json")
|
||||
|
||||
@@ -274,17 +339,33 @@ def _handle_status(console) -> None:
|
||||
|
||||
suppression = get_suppression_stats()
|
||||
rate_limits = get_rate_limit_stats()
|
||||
muted = get_muted_branches()
|
||||
muted_detail = get_muted_branches_detail()
|
||||
|
||||
state_color = "green" if enabled else "yellow"
|
||||
state_text = "ENABLED" if enabled else "DISABLED"
|
||||
|
||||
disabled_until = get_disabled_until()
|
||||
if not enabled and disabled_until:
|
||||
remaining = _fmt_remaining(disabled_until)
|
||||
state_text = f"DISABLED (auto-resumes in {remaining})"
|
||||
|
||||
if watcher_active:
|
||||
watcher_text = "[green]running[/green] (systemd)"
|
||||
elif enabled:
|
||||
watcher_text = "[yellow]stopped[/yellow] — run [bold]medic on[/bold] to start"
|
||||
else:
|
||||
watcher_text = "stopped"
|
||||
muted_text = ", ".join(f"@{b}" for b in muted) if muted else "none"
|
||||
|
||||
if muted_detail:
|
||||
muted_parts = []
|
||||
for m in muted_detail:
|
||||
if m["expires_at"] is None:
|
||||
muted_parts.append(f"@{m['name']} [dim](permanent)[/dim]")
|
||||
else:
|
||||
muted_parts.append(f"@{m['name']} [dim]({_fmt_remaining(m['expires_at'])} left)[/dim]")
|
||||
muted_text = ", ".join(muted_parts)
|
||||
else:
|
||||
muted_text = "none"
|
||||
|
||||
console.print("Medic Status")
|
||||
console.print(f" State: [{state_color}]{state_text}[/{state_color}]")
|
||||
@@ -330,30 +411,47 @@ def _handle_on(console) -> None:
|
||||
)
|
||||
|
||||
|
||||
def _handle_off(console) -> None:
|
||||
"""Handle 'medic off' — disable dispatch and stop watcher."""
|
||||
def _handle_off(console, args: list | None = None) -> None:
|
||||
"""Handle 'medic off [--forever]' — disable dispatch with 24h TTL or permanently."""
|
||||
from rich.panel import Panel
|
||||
from aipass.cli.apps.modules import error
|
||||
|
||||
if not set_enabled(False):
|
||||
error("Failed to disable Medic", suggestion="Check trigger_config.json")
|
||||
return
|
||||
args = args or []
|
||||
is_forever = "--forever" in args
|
||||
|
||||
logger.info("[MEDIC] Medic DISABLED - error dispatch suppressed")
|
||||
if _is_service_active():
|
||||
_systemctl("stop")
|
||||
logger.info("[MEDIC] Log watcher service stopped")
|
||||
|
||||
console.print(
|
||||
Panel(
|
||||
"[bold yellow]Medic DISABLED[/bold yellow]\n\n"
|
||||
"Error dispatch is [yellow]suppressed[/yellow]. Errors are still detected\n"
|
||||
"and logged to [dim]medic_suppressed.jsonl[/dim] for review.\n"
|
||||
"Log watcher: [yellow]stopped[/yellow]",
|
||||
title="Medic",
|
||||
border_style="yellow",
|
||||
if is_forever:
|
||||
if not set_enabled(False):
|
||||
error("Failed to disable Medic", suggestion="Check trigger_config.json")
|
||||
return
|
||||
logger.info("[MEDIC] Medic DISABLED permanently")
|
||||
if _is_service_active():
|
||||
_systemctl("stop")
|
||||
logger.info("[MEDIC] Log watcher service stopped")
|
||||
console.print(
|
||||
Panel(
|
||||
"[bold yellow]Medic DISABLED (permanent)[/bold yellow]\n\n"
|
||||
"Error dispatch is [yellow]suppressed[/yellow]. Log watcher stopped.\n"
|
||||
"Use [bold]medic on[/bold] to re-enable.",
|
||||
title="Medic",
|
||||
border_style="yellow",
|
||||
)
|
||||
)
|
||||
else:
|
||||
hours = DEFAULT_OFF_SECONDS // 3600
|
||||
if not set_enabled(False, duration_seconds=float(DEFAULT_OFF_SECONDS)):
|
||||
error("Failed to disable Medic", suggestion="Check trigger_config.json")
|
||||
return
|
||||
logger.info("[MEDIC] Medic DISABLED for %dh", hours)
|
||||
console.print(
|
||||
Panel(
|
||||
f"[bold yellow]Medic DISABLED ({hours}h)[/bold yellow]\n\n"
|
||||
f"Error dispatch suppressed for {hours} hours, then auto-resumes.\n"
|
||||
"Detection continues (log watcher stays running).\n"
|
||||
"Use [bold]medic off --forever[/bold] for permanent disable.",
|
||||
title="Medic",
|
||||
border_style="yellow",
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _route_medic_module(args: list) -> bool:
|
||||
@@ -401,7 +499,7 @@ def handle_command(command: str, args: list) -> bool:
|
||||
"mute": lambda: _handle_mute(console, args),
|
||||
"unmute": lambda: _handle_unmute(console, args),
|
||||
"on": lambda: _handle_on(console),
|
||||
"off": lambda: _handle_off(console),
|
||||
"off": lambda: _handle_off(console, args),
|
||||
"status": lambda: _handle_status(console),
|
||||
}
|
||||
handler = handlers.get(command)
|
||||
|
||||
@@ -8,7 +8,9 @@
|
||||
|
||||
"""Tests for error_detected event handler: set_send_email_callback, handle_error_detected, and fallback stubs."""
|
||||
|
||||
import json
|
||||
import sys
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
@@ -399,3 +401,249 @@ class TestFallbackStubs:
|
||||
"""Module reports registry dispatch as unavailable."""
|
||||
mod = _import_module()
|
||||
assert mod._REGISTRY_DISPATCH_AVAILABLE is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TTL-aware medic enable/disable
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestMedicEnabledTTL:
|
||||
"""Tests for _is_medic_enabled TTL expiry behavior."""
|
||||
|
||||
def test_medic_enabled_ttl_expired(self) -> None:
|
||||
"""medic_enabled=False with expired TTL -> treated as enabled, dispatch proceeds."""
|
||||
mod = _import_module()
|
||||
real_is_medic_enabled = mod._is_medic_enabled
|
||||
send = _setup_happy_path(mod)
|
||||
mod._is_medic_enabled = real_is_medic_enabled # type: ignore[attr-defined]
|
||||
|
||||
config_file = mod.TRIGGER_CONFIG_FILE
|
||||
config_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
past = (datetime.now() - timedelta(hours=1)).isoformat()
|
||||
config_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"config": {
|
||||
"medic_enabled": False,
|
||||
"medic_disabled_until": past,
|
||||
}
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
mod.handle_error_detected(
|
||||
branch="flow",
|
||||
module="cfg",
|
||||
message="err",
|
||||
error_hash="h1",
|
||||
count=2,
|
||||
fingerprint="fp_ttl_exp",
|
||||
)
|
||||
|
||||
send.assert_called_once()
|
||||
|
||||
def test_medic_enabled_ttl_active(self) -> None:
|
||||
"""medic_enabled=False with future TTL -> medic still disabled, dispatch suppressed."""
|
||||
mod = _import_module()
|
||||
real_is_medic_enabled = mod._is_medic_enabled
|
||||
send = _setup_happy_path(mod)
|
||||
mod._is_medic_enabled = real_is_medic_enabled # type: ignore[attr-defined]
|
||||
|
||||
config_file = mod.TRIGGER_CONFIG_FILE
|
||||
config_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
future = (datetime.now() + timedelta(hours=1)).isoformat()
|
||||
config_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"config": {
|
||||
"medic_enabled": False,
|
||||
"medic_disabled_until": future,
|
||||
}
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
mod.handle_error_detected(
|
||||
branch="flow",
|
||||
module="cfg",
|
||||
message="err",
|
||||
error_hash="h1",
|
||||
count=2,
|
||||
fingerprint="fp_ttl_act",
|
||||
)
|
||||
|
||||
send.assert_not_called()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Branch mute dict/string format support
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBranchMutedFormats:
|
||||
"""Tests for _is_branch_muted dict and string format support."""
|
||||
|
||||
def test_branch_muted_dict_format_active(self) -> None:
|
||||
"""Dict entry with future expires_at -> branch IS muted, dispatch suppressed."""
|
||||
mod = _import_module()
|
||||
real_is_branch_muted = mod._is_branch_muted
|
||||
send = _setup_happy_path(mod)
|
||||
mod._is_branch_muted = real_is_branch_muted # type: ignore[attr-defined]
|
||||
mod._get_registered_emails = MagicMock(return_value={"@api", "@flow"}) # type: ignore[attr-defined]
|
||||
|
||||
config_file = mod.TRIGGER_CONFIG_FILE
|
||||
config_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
future = (datetime.now() + timedelta(hours=1)).isoformat()
|
||||
config_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"config": {
|
||||
"medic_enabled": True,
|
||||
"muted_branches": [{"name": "api", "expires_at": future}],
|
||||
}
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
mod.handle_error_detected(
|
||||
branch="api",
|
||||
module="cfg",
|
||||
message="err",
|
||||
error_hash="h1",
|
||||
count=2,
|
||||
fingerprint="fp_mute_act",
|
||||
)
|
||||
|
||||
send.assert_not_called()
|
||||
|
||||
def test_branch_muted_dict_format_expired(self) -> None:
|
||||
"""Dict entry with past expires_at -> branch NOT muted, dispatch proceeds."""
|
||||
mod = _import_module()
|
||||
real_is_branch_muted = mod._is_branch_muted
|
||||
send = _setup_happy_path(mod)
|
||||
mod._is_branch_muted = real_is_branch_muted # type: ignore[attr-defined]
|
||||
mod._get_registered_emails = MagicMock(return_value={"@api", "@flow"}) # type: ignore[attr-defined]
|
||||
|
||||
config_file = mod.TRIGGER_CONFIG_FILE
|
||||
config_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
past = (datetime.now() - timedelta(hours=1)).isoformat()
|
||||
config_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"config": {
|
||||
"medic_enabled": True,
|
||||
"muted_branches": [{"name": "api", "expires_at": past}],
|
||||
}
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
mod.handle_error_detected(
|
||||
branch="api",
|
||||
module="cfg",
|
||||
message="err",
|
||||
error_hash="h1",
|
||||
count=2,
|
||||
fingerprint="fp_mute_exp",
|
||||
)
|
||||
|
||||
send.assert_called_once()
|
||||
|
||||
def test_branch_muted_plain_string_backcompat(self) -> None:
|
||||
"""Plain string entry in muted_branches -> branch IS muted (permanent)."""
|
||||
mod = _import_module()
|
||||
real_is_branch_muted = mod._is_branch_muted
|
||||
send = _setup_happy_path(mod)
|
||||
mod._is_branch_muted = real_is_branch_muted # type: ignore[attr-defined]
|
||||
mod._get_registered_emails = MagicMock(return_value={"@api", "@flow"}) # type: ignore[attr-defined]
|
||||
|
||||
config_file = mod.TRIGGER_CONFIG_FILE
|
||||
config_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
config_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"config": {
|
||||
"medic_enabled": True,
|
||||
"muted_branches": ["api"],
|
||||
}
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
mod.handle_error_detected(
|
||||
branch="api",
|
||||
module="cfg",
|
||||
message="err",
|
||||
error_hash="h1",
|
||||
count=2,
|
||||
fingerprint="fp_str_perm",
|
||||
)
|
||||
|
||||
send.assert_not_called()
|
||||
|
||||
def test_branch_muted_dict_permanent(self) -> None:
|
||||
"""Dict entry with expires_at=null -> branch IS muted (permanent)."""
|
||||
mod = _import_module()
|
||||
real_is_branch_muted = mod._is_branch_muted
|
||||
send = _setup_happy_path(mod)
|
||||
mod._is_branch_muted = real_is_branch_muted # type: ignore[attr-defined]
|
||||
mod._get_registered_emails = MagicMock(return_value={"@api", "@flow"}) # type: ignore[attr-defined]
|
||||
|
||||
config_file = mod.TRIGGER_CONFIG_FILE
|
||||
config_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
config_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"config": {
|
||||
"medic_enabled": True,
|
||||
"muted_branches": [{"name": "api", "expires_at": None}],
|
||||
}
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
mod.handle_error_detected(
|
||||
branch="api",
|
||||
module="cfg",
|
||||
message="err",
|
||||
error_hash="h1",
|
||||
count=2,
|
||||
fingerprint="fp_dict_perm",
|
||||
)
|
||||
|
||||
send.assert_not_called()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# circuit_breaker_probe_succeeded after dispatch
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestProbeSucceeded:
|
||||
"""Tests for circuit_breaker_probe_succeeded called after dispatch."""
|
||||
|
||||
def test_probe_succeeded_called_after_dispatch(self) -> None:
|
||||
"""circuit_breaker_probe_succeeded is called after successful dispatch with fingerprint."""
|
||||
mod = _import_module()
|
||||
send = _setup_happy_path(mod)
|
||||
mod.circuit_breaker_probe_succeeded = MagicMock() # type: ignore[attr-defined]
|
||||
|
||||
mod.handle_error_detected(
|
||||
branch="flow",
|
||||
module="cfg",
|
||||
message="err",
|
||||
error_hash="h1",
|
||||
count=2,
|
||||
fingerprint="fp_probe",
|
||||
)
|
||||
|
||||
send.assert_called_once()
|
||||
mod.registry_record_dispatch.assert_called_once_with("fp_probe") # type: ignore[attr-defined]
|
||||
mod.circuit_breaker_probe_succeeded.assert_called_once() # type: ignore[attr-defined]
|
||||
|
||||
@@ -1012,3 +1012,165 @@ def test_purge_stale_custom_days(tmp_path: Path) -> None:
|
||||
|
||||
# 7-day cutoff should remove it
|
||||
assert er.purge_stale(days=7) == 1
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 17. Circuit breaker self-heal (_evaluate_state, probe_succeeded, status)
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
def test_evaluate_state_transitions_open_to_half_open_after_cooldown(tmp_path: Path) -> None:
|
||||
"""_evaluate_state transitions open -> half_open when cooldown has expired."""
|
||||
_seed_registry(tmp_path)
|
||||
er = _import_registry()
|
||||
er.circuit_breaker_reset()
|
||||
|
||||
er.circuit_breaker_trip(reason="test")
|
||||
# Backdate opened_at so cooldown is expired
|
||||
er._circuit_breaker.opened_at = time.time() - er._circuit_breaker.cooldown_seconds - 10
|
||||
|
||||
status = er.get_circuit_breaker_status()
|
||||
assert status["state"] == "half_open"
|
||||
|
||||
|
||||
def test_evaluate_state_no_transition_before_cooldown(tmp_path: Path) -> None:
|
||||
"""_evaluate_state keeps state open when cooldown has not yet expired."""
|
||||
_seed_registry(tmp_path)
|
||||
er = _import_registry()
|
||||
er.circuit_breaker_reset()
|
||||
|
||||
er.circuit_breaker_trip(reason="test")
|
||||
# opened_at is now (cooldown is 300s), so it should stay open
|
||||
assert er._circuit_breaker.state == "open"
|
||||
|
||||
status = er.get_circuit_breaker_status()
|
||||
assert status["state"] == "open"
|
||||
|
||||
|
||||
def test_evaluate_state_no_op_when_closed(tmp_path: Path) -> None:
|
||||
"""_evaluate_state is a no-op when breaker is already closed."""
|
||||
_seed_registry(tmp_path)
|
||||
er = _import_registry()
|
||||
er.circuit_breaker_reset()
|
||||
|
||||
assert er._circuit_breaker.state == "closed"
|
||||
|
||||
status = er.get_circuit_breaker_status()
|
||||
assert status["state"] == "closed"
|
||||
|
||||
|
||||
def test_probe_succeeded_closes_breaker(tmp_path: Path) -> None:
|
||||
"""circuit_breaker_probe_succeeded transitions half_open -> closed and resets cooldown."""
|
||||
_seed_registry(tmp_path)
|
||||
er = _import_registry()
|
||||
er.circuit_breaker_reset()
|
||||
|
||||
# Trip the breaker and expire cooldown to get to half_open
|
||||
er.circuit_breaker_trip(reason="test")
|
||||
er._circuit_breaker.opened_at = time.time() - er._circuit_breaker.cooldown_seconds - 1
|
||||
er.circuit_breaker_allows() # Transitions to half_open
|
||||
assert er._circuit_breaker.state == "half_open"
|
||||
|
||||
er.circuit_breaker_probe_succeeded()
|
||||
|
||||
assert er._circuit_breaker.state == "closed"
|
||||
assert er._circuit_breaker.cooldown_seconds == er._circuit_breaker.base_cooldown
|
||||
assert er._circuit_breaker.opened_at == 0.0
|
||||
assert er._circuit_breaker.recent_errors == []
|
||||
|
||||
|
||||
def test_probe_succeeded_noop_when_closed(tmp_path: Path) -> None:
|
||||
"""circuit_breaker_probe_succeeded is a no-op when breaker is closed."""
|
||||
_seed_registry(tmp_path)
|
||||
er = _import_registry()
|
||||
er.circuit_breaker_reset()
|
||||
|
||||
assert er._circuit_breaker.state == "closed"
|
||||
|
||||
er.circuit_breaker_probe_succeeded()
|
||||
|
||||
assert er._circuit_breaker.state == "closed"
|
||||
|
||||
|
||||
def test_probe_succeeded_noop_when_open(tmp_path: Path) -> None:
|
||||
"""circuit_breaker_probe_succeeded is a no-op when breaker is open."""
|
||||
_seed_registry(tmp_path)
|
||||
er = _import_registry()
|
||||
er.circuit_breaker_reset()
|
||||
|
||||
er.circuit_breaker_trip(reason="test")
|
||||
assert er._circuit_breaker.state == "open"
|
||||
|
||||
er.circuit_breaker_probe_succeeded()
|
||||
|
||||
assert er._circuit_breaker.state == "open"
|
||||
|
||||
|
||||
def test_status_returns_remaining_seconds(tmp_path: Path) -> None:
|
||||
"""get_circuit_breaker_status returns approximately correct remaining_seconds."""
|
||||
_seed_registry(tmp_path)
|
||||
er = _import_registry()
|
||||
er.circuit_breaker_reset()
|
||||
|
||||
er.circuit_breaker_trip(reason="test")
|
||||
cooldown = er._circuit_breaker.cooldown_seconds
|
||||
# Backdate opened_at by 100 seconds so remaining ~ cooldown - 100
|
||||
er._circuit_breaker.opened_at = time.time() - 100
|
||||
|
||||
status = er.get_circuit_breaker_status()
|
||||
expected_remaining = cooldown - 100
|
||||
# Allow 2-second tolerance for timing
|
||||
assert abs(status["remaining_seconds"] - expected_remaining) <= 2
|
||||
|
||||
|
||||
def test_status_remaining_zero_when_closed(tmp_path: Path) -> None:
|
||||
"""remaining_seconds is 0 when breaker is closed."""
|
||||
_seed_registry(tmp_path)
|
||||
er = _import_registry()
|
||||
er.circuit_breaker_reset()
|
||||
|
||||
status = er.get_circuit_breaker_status()
|
||||
assert status["remaining_seconds"] == 0
|
||||
|
||||
|
||||
def test_breaker_half_open_on_read_then_allows_probe(tmp_path: Path) -> None:
|
||||
"""Expired open breaker transitions to half_open on allows() and consumes probe slot."""
|
||||
_seed_registry(tmp_path)
|
||||
er = _import_registry()
|
||||
er.circuit_breaker_reset()
|
||||
|
||||
er.circuit_breaker_trip(reason="test")
|
||||
# Expire the cooldown
|
||||
er._circuit_breaker.opened_at = time.time() - er._circuit_breaker.cooldown_seconds - 1
|
||||
|
||||
# First call: transitions open -> half_open, returns True (probe allowed)
|
||||
result = er.circuit_breaker_allows()
|
||||
assert result is True
|
||||
assert er._circuit_breaker.state == "half_open"
|
||||
assert er._circuit_breaker.half_open_allow is False
|
||||
|
||||
|
||||
def test_breaker_closes_after_successful_probe_dispatch(tmp_path: Path) -> None:
|
||||
"""Full self-heal cycle: open -> half_open -> probe allowed -> probe_succeeded -> closed."""
|
||||
_seed_registry(tmp_path)
|
||||
er = _import_registry()
|
||||
er.circuit_breaker_reset()
|
||||
|
||||
base_cooldown = er._circuit_breaker.base_cooldown
|
||||
|
||||
# Trip the breaker
|
||||
er.circuit_breaker_trip(reason="test")
|
||||
assert er._circuit_breaker.state == "open"
|
||||
|
||||
# Expire the cooldown
|
||||
er._circuit_breaker.opened_at = time.time() - er._circuit_breaker.cooldown_seconds - 1
|
||||
|
||||
# Probe dispatch: transitions open -> half_open and allows
|
||||
assert er.circuit_breaker_allows() is True
|
||||
assert er._circuit_breaker.state == "half_open"
|
||||
|
||||
# Probe succeeded: transitions half_open -> closed
|
||||
er.circuit_breaker_probe_succeeded()
|
||||
assert er._circuit_breaker.state == "closed"
|
||||
assert er._circuit_breaker.cooldown_seconds == base_cooldown
|
||||
assert er._circuit_breaker.opened_at == 0.0
|
||||
|
||||
@@ -47,6 +47,8 @@ def _mock_infrastructure(monkeypatch):
|
||||
medic_state_mod.is_enabled = MagicMock(return_value=True)
|
||||
medic_state_mod.set_enabled = MagicMock(return_value=True)
|
||||
medic_state_mod.get_muted_branches = MagicMock(return_value=[])
|
||||
medic_state_mod.get_muted_branches_detail = MagicMock(return_value=[])
|
||||
medic_state_mod.get_disabled_until = MagicMock(return_value=None)
|
||||
medic_state_mod.mute_branch = MagicMock(return_value=True)
|
||||
medic_state_mod.unmute_branch = MagicMock(return_value=True)
|
||||
medic_state_mod.get_suppression_stats = MagicMock(
|
||||
@@ -61,6 +63,9 @@ def _mock_infrastructure(monkeypatch):
|
||||
"last_rate_limited": "never",
|
||||
}
|
||||
)
|
||||
medic_state_mod.parse_duration = MagicMock(return_value=None)
|
||||
medic_state_mod.DEFAULT_MUTE_SECONDS = 86400
|
||||
medic_state_mod.DEFAULT_OFF_SECONDS = 86400
|
||||
monkeypatch.setitem(sys.modules, "aipass.trigger.apps.handlers.medic_state", medic_state_mod)
|
||||
|
||||
# -- CLI console (lazy import inside handle_command) --------------------
|
||||
@@ -189,7 +194,7 @@ def test_handle_command_on_failure_prints_error():
|
||||
|
||||
|
||||
def test_handle_command_off_disables_medic():
|
||||
"""handle_command('off', []) calls set_enabled(False), prints Panel, returns True."""
|
||||
"""handle_command('off', []) calls set_enabled with 24h TTL, prints Panel, returns True."""
|
||||
medic = _import_medic()
|
||||
|
||||
with patch.object(medic, "_systemctl", return_value=True):
|
||||
@@ -198,21 +203,34 @@ def test_handle_command_off_disables_medic():
|
||||
|
||||
assert result is True
|
||||
state = _get_medic_state()
|
||||
state.set_enabled.assert_called_with(False)
|
||||
# Verify console.print was called (Panel is a mock object for success output)
|
||||
state.set_enabled.assert_called_with(False, duration_seconds=86400.0)
|
||||
console = _get_console()
|
||||
assert console.print.call_count >= 1, "console.print should be called with success Panel"
|
||||
|
||||
|
||||
def test_handle_command_off_stops_active_service():
|
||||
"""handle_command('off', []) stops the service when it is active."""
|
||||
def test_handle_command_off_forever_stops_service():
|
||||
"""handle_command('off', ['--forever']) stops the service and disables permanently."""
|
||||
medic = _import_medic()
|
||||
|
||||
with patch.object(medic, "_systemctl", return_value=True) as mock_ctl:
|
||||
with patch.object(medic, "_is_service_active", return_value=True):
|
||||
result = medic.handle_command("off", ["--forever"])
|
||||
|
||||
assert result is True
|
||||
state = _get_medic_state()
|
||||
state.set_enabled.assert_called_with(False)
|
||||
mock_ctl.assert_called_with("stop")
|
||||
|
||||
|
||||
def test_handle_command_off_ttl_keeps_watcher():
|
||||
"""handle_command('off', []) with default TTL does NOT stop the log watcher."""
|
||||
medic = _import_medic()
|
||||
|
||||
with patch.object(medic, "_systemctl", return_value=True) as mock_ctl:
|
||||
with patch.object(medic, "_is_service_active", return_value=True):
|
||||
medic.handle_command("off", [])
|
||||
|
||||
mock_ctl.assert_called_with("stop")
|
||||
mock_ctl.assert_not_called()
|
||||
|
||||
|
||||
def test_handle_command_off_failure_prints_error():
|
||||
@@ -249,7 +267,7 @@ def test_handle_command_status_returns_current_state():
|
||||
assert result is True
|
||||
state = _get_medic_state()
|
||||
state.is_enabled.assert_called_once()
|
||||
state.get_muted_branches.assert_called_once()
|
||||
state.get_muted_branches_detail.assert_called_once()
|
||||
state.get_suppression_stats.assert_called_once()
|
||||
state.get_rate_limit_stats.assert_called_once()
|
||||
|
||||
@@ -283,19 +301,22 @@ def test_handle_command_status_shows_disabled():
|
||||
|
||||
|
||||
def test_handle_command_status_shows_muted_branches():
|
||||
"""When branches are muted, status lists them in the muted branches line."""
|
||||
"""When branches are muted, status lists them with expiry info."""
|
||||
medic = _import_medic()
|
||||
state = _get_medic_state()
|
||||
state.get_muted_branches.return_value = ["speakeasy", "api"]
|
||||
state.get_muted_branches_detail.return_value = [
|
||||
{"name": "speakeasy", "expires_at": None},
|
||||
{"name": "api", "expires_at": None},
|
||||
]
|
||||
|
||||
with patch.object(medic, "_is_service_active", return_value=True):
|
||||
medic.handle_command("status", [])
|
||||
|
||||
console = _get_console()
|
||||
printed = _get_print_str_args(console)
|
||||
# Source builds: " Muted branches: @speakeasy, @api"
|
||||
muted_line = " Muted branches: @speakeasy, @api"
|
||||
assert muted_line in printed, f"Expected muted line '{muted_line}' in printed args: {printed}"
|
||||
muted_lines = [p for p in printed if "Muted branches:" in p]
|
||||
assert muted_lines, f"Expected muted branches line in printed args: {printed}"
|
||||
assert "@speakeasy" in muted_lines[0] and "@api" in muted_lines[0]
|
||||
|
||||
|
||||
def test_handle_command_status_suppression_hint_when_disabled():
|
||||
@@ -319,13 +340,13 @@ def test_handle_command_status_suppression_hint_when_disabled():
|
||||
|
||||
|
||||
def test_handle_command_mute_branch():
|
||||
"""handle_command('mute', ['@speakeasy']) mutes the branch."""
|
||||
"""handle_command('mute', ['@speakeasy']) mutes the branch with 24h default TTL."""
|
||||
medic = _import_medic()
|
||||
result = medic.handle_command("mute", ["@speakeasy"])
|
||||
|
||||
assert result is True
|
||||
state = _get_medic_state()
|
||||
state.mute_branch.assert_called_once_with("speakeasy")
|
||||
state.mute_branch.assert_called_once_with("speakeasy", duration_seconds=86400.0)
|
||||
|
||||
|
||||
def test_handle_command_mute_branch_without_at():
|
||||
@@ -334,18 +355,18 @@ def test_handle_command_mute_branch_without_at():
|
||||
medic.handle_command("mute", ["speakeasy"])
|
||||
|
||||
state = _get_medic_state()
|
||||
state.mute_branch.assert_called_once_with("speakeasy")
|
||||
state.mute_branch.assert_called_once_with("speakeasy", duration_seconds=86400.0)
|
||||
|
||||
|
||||
def test_handle_command_mute_prints_confirmation():
|
||||
"""Successful mute prints the exact confirmation message with the branch name."""
|
||||
"""Successful mute prints confirmation with TTL info."""
|
||||
medic = _import_medic()
|
||||
medic.handle_command("mute", ["@api"])
|
||||
|
||||
console = _get_console()
|
||||
printed = _get_print_str_args(console)
|
||||
expected = " [yellow]Muted[/yellow] @api — errors logged but not dispatched"
|
||||
assert expected in printed, f"Expected mute confirmation '{expected}' in printed args: {printed}"
|
||||
mute_lines = [p for p in printed if "Muted" in p and "@api" in p]
|
||||
assert mute_lines, f"Expected mute confirmation for @api in printed args: {printed}"
|
||||
|
||||
|
||||
def test_handle_command_mute_failure_prints_error():
|
||||
@@ -522,7 +543,7 @@ def test_handle_command_medic_routes_mute_with_args():
|
||||
|
||||
assert result is True
|
||||
state = _get_medic_state()
|
||||
state.mute_branch.assert_called_once_with("speakeasy")
|
||||
state.mute_branch.assert_called_once_with("speakeasy", duration_seconds=86400.0)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -502,3 +502,441 @@ class TestGetRateLimitStats:
|
||||
|
||||
assert result["rate_limited_count"] == 1
|
||||
assert result["last_rate_limited"] == "unknown"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests -- parse_duration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestParseDuration:
|
||||
"""Tests for parse_duration."""
|
||||
|
||||
def test_parse_duration_hours(self, state_mod):
|
||||
"""parse_duration converts '24h' to 86400.0 seconds."""
|
||||
result = state_mod.parse_duration("24h")
|
||||
|
||||
assert result == 86400.0
|
||||
|
||||
def test_parse_duration_days(self, state_mod):
|
||||
"""parse_duration converts '7d' to 604800.0 seconds."""
|
||||
result = state_mod.parse_duration("7d")
|
||||
|
||||
assert result == 604800.0
|
||||
|
||||
def test_parse_duration_invalid(self, state_mod):
|
||||
"""parse_duration returns None for invalid input."""
|
||||
result = state_mod.parse_duration("abc")
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests -- mute_branch TTL
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestMuteBranchTTL:
|
||||
"""Tests for mute_branch with TTL support."""
|
||||
|
||||
def test_mute_branch_default_permanent(self, state_mod):
|
||||
"""mute_branch with no duration stores dict with expires_at null."""
|
||||
state_mod.mute_branch("api")
|
||||
|
||||
config_file = state_mod.TRIGGER_CONFIG_FILE
|
||||
data = json.loads(config_file.read_text(encoding="utf-8"))
|
||||
muted = data["config"]["muted_branches"]
|
||||
assert len(muted) == 1
|
||||
assert muted[0] == {"name": "api", "expires_at": None}
|
||||
|
||||
def test_mute_branch_with_ttl(self, state_mod):
|
||||
"""mute_branch with duration stores expires_at roughly 1h from now."""
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
before = datetime.now()
|
||||
state_mod.mute_branch("api", duration_seconds=3600)
|
||||
after = datetime.now()
|
||||
|
||||
config_file = state_mod.TRIGGER_CONFIG_FILE
|
||||
data = json.loads(config_file.read_text(encoding="utf-8"))
|
||||
muted = data["config"]["muted_branches"]
|
||||
assert len(muted) == 1
|
||||
assert muted[0]["name"] == "api"
|
||||
expires = datetime.fromisoformat(muted[0]["expires_at"])
|
||||
assert expires >= before + timedelta(seconds=3600)
|
||||
assert expires <= after + timedelta(seconds=3600)
|
||||
|
||||
def test_mute_forever_null_expires(self, state_mod):
|
||||
"""mute_branch with duration_seconds=None stores expires_at as null."""
|
||||
state_mod.mute_branch("api", duration_seconds=None)
|
||||
|
||||
config_file = state_mod.TRIGGER_CONFIG_FILE
|
||||
data = json.loads(config_file.read_text(encoding="utf-8"))
|
||||
muted = data["config"]["muted_branches"]
|
||||
assert len(muted) == 1
|
||||
assert muted[0]["expires_at"] is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests -- get_muted_branches TTL filtering
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestGetMutedBranchesTTL:
|
||||
"""Tests for get_muted_branches with TTL-aware filtering."""
|
||||
|
||||
def test_get_muted_branches_filters_expired(self, state_mod):
|
||||
"""get_muted_branches excludes dict entries whose expires_at is in the past."""
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
expired_ts = (datetime.now() - timedelta(hours=1)).isoformat()
|
||||
config_file = state_mod.TRIGGER_CONFIG_FILE
|
||||
config_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
config_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"config": {
|
||||
"muted_branches": [
|
||||
{"name": "api", "expires_at": expired_ts},
|
||||
]
|
||||
}
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
result = state_mod.get_muted_branches()
|
||||
|
||||
assert result == []
|
||||
|
||||
def test_get_muted_branches_keeps_active(self, state_mod):
|
||||
"""get_muted_branches includes dict entries whose expires_at is in the future."""
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
future_ts = (datetime.now() + timedelta(hours=1)).isoformat()
|
||||
config_file = state_mod.TRIGGER_CONFIG_FILE
|
||||
config_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
config_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"config": {
|
||||
"muted_branches": [
|
||||
{"name": "api", "expires_at": future_ts},
|
||||
]
|
||||
}
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
result = state_mod.get_muted_branches()
|
||||
|
||||
assert result == ["api"]
|
||||
|
||||
def test_get_muted_branches_plain_string_backcompat(self, state_mod):
|
||||
"""get_muted_branches returns plain string entries as permanent mutes."""
|
||||
config_file = state_mod.TRIGGER_CONFIG_FILE
|
||||
config_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
config_file.write_text(
|
||||
json.dumps({"config": {"muted_branches": ["speakeasy"]}}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
result = state_mod.get_muted_branches()
|
||||
|
||||
assert result == ["speakeasy"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests -- get_muted_branches_detail
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestGetMutedBranchesDetail:
|
||||
"""Tests for get_muted_branches_detail."""
|
||||
|
||||
def test_get_muted_branches_detail_returns_expiry(self, state_mod):
|
||||
"""get_muted_branches_detail returns dicts with name and expires_at."""
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
future_ts = (datetime.now() + timedelta(hours=2)).isoformat()
|
||||
config_file = state_mod.TRIGGER_CONFIG_FILE
|
||||
config_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
config_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"config": {
|
||||
"muted_branches": [
|
||||
{"name": "api", "expires_at": future_ts},
|
||||
"speakeasy",
|
||||
]
|
||||
}
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
result = state_mod.get_muted_branches_detail()
|
||||
|
||||
assert len(result) == 2
|
||||
assert result[0] == {"name": "api", "expires_at": future_ts}
|
||||
assert result[1] == {"name": "speakeasy", "expires_at": None}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests -- is_enabled TTL
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestIsEnabledTTL:
|
||||
"""Tests for is_enabled with TTL-based disable."""
|
||||
|
||||
def test_is_enabled_ttl_expired_returns_true(self, state_mod):
|
||||
"""is_enabled returns True when disabled but medic_disabled_until is in the past."""
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
past_ts = (datetime.now() - timedelta(hours=1)).isoformat()
|
||||
config_file = state_mod.TRIGGER_CONFIG_FILE
|
||||
config_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
config_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"config": {
|
||||
"medic_enabled": False,
|
||||
"medic_disabled_until": past_ts,
|
||||
}
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
result = state_mod.is_enabled()
|
||||
|
||||
assert result is True
|
||||
|
||||
def test_is_enabled_ttl_active_returns_false(self, state_mod):
|
||||
"""is_enabled returns False when disabled and medic_disabled_until is in the future."""
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
future_ts = (datetime.now() + timedelta(hours=1)).isoformat()
|
||||
config_file = state_mod.TRIGGER_CONFIG_FILE
|
||||
config_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
config_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"config": {
|
||||
"medic_enabled": False,
|
||||
"medic_disabled_until": future_ts,
|
||||
}
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
result = state_mod.is_enabled()
|
||||
|
||||
assert result is False
|
||||
|
||||
def test_is_enabled_permanent_off(self, state_mod):
|
||||
"""is_enabled returns False when permanently disabled (no medic_disabled_until)."""
|
||||
config_file = state_mod.TRIGGER_CONFIG_FILE
|
||||
config_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
config_file.write_text(
|
||||
json.dumps({"config": {"medic_enabled": False}}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
result = state_mod.is_enabled()
|
||||
|
||||
assert result is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests -- set_enabled with duration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestSetEnabledDuration:
|
||||
"""Tests for set_enabled with duration_seconds parameter."""
|
||||
|
||||
def test_set_enabled_off_with_duration(self, state_mod):
|
||||
"""set_enabled(False, duration) stores medic_disabled_until timestamp."""
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
before = datetime.now()
|
||||
state_mod.set_enabled(False, duration_seconds=86400)
|
||||
after = datetime.now()
|
||||
|
||||
config_file = state_mod.TRIGGER_CONFIG_FILE
|
||||
data = json.loads(config_file.read_text(encoding="utf-8"))
|
||||
assert data["config"]["medic_enabled"] is False
|
||||
disabled_until = data["config"]["medic_disabled_until"]
|
||||
ts = datetime.fromisoformat(disabled_until)
|
||||
assert ts >= before + timedelta(seconds=86400)
|
||||
assert ts <= after + timedelta(seconds=86400)
|
||||
|
||||
def test_set_enabled_on_clears_disabled_until(self, state_mod):
|
||||
"""set_enabled(True) clears any existing medic_disabled_until."""
|
||||
# First disable with TTL
|
||||
state_mod.set_enabled(False, duration_seconds=3600)
|
||||
config_file = state_mod.TRIGGER_CONFIG_FILE
|
||||
data = json.loads(config_file.read_text(encoding="utf-8"))
|
||||
assert "medic_disabled_until" in data["config"]
|
||||
|
||||
# Then re-enable
|
||||
state_mod.set_enabled(True)
|
||||
data = json.loads(config_file.read_text(encoding="utf-8"))
|
||||
assert data["config"]["medic_enabled"] is True
|
||||
assert "medic_disabled_until" not in data["config"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests -- get_disabled_until
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestGetDisabledUntil:
|
||||
"""Tests for get_disabled_until."""
|
||||
|
||||
def test_get_disabled_until_returns_timestamp(self, state_mod):
|
||||
"""get_disabled_until returns the ISO timestamp when set."""
|
||||
from datetime import datetime
|
||||
|
||||
state_mod.set_enabled(False, duration_seconds=86400)
|
||||
|
||||
result = state_mod.get_disabled_until()
|
||||
|
||||
assert result is not None
|
||||
ts = datetime.fromisoformat(result)
|
||||
assert ts > datetime.now()
|
||||
|
||||
def test_get_disabled_until_returns_none(self, state_mod):
|
||||
"""get_disabled_until returns None when no TTL is set."""
|
||||
state_mod.set_enabled(False)
|
||||
|
||||
result = state_mod.get_disabled_until()
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests -- unmute_branch with dict entries
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestUnmuteBranchDict:
|
||||
"""Tests for unmute_branch handling dict-format entries."""
|
||||
|
||||
def test_unmute_handles_dict_entries(self, state_mod):
|
||||
"""unmute_branch removes a dict-format mute entry."""
|
||||
state_mod.mute_branch("api", duration_seconds=3600)
|
||||
assert "api" in state_mod.get_muted_branches()
|
||||
|
||||
result = state_mod.unmute_branch("api")
|
||||
|
||||
assert result is True
|
||||
assert "api" not in state_mod.get_muted_branches()
|
||||
|
||||
def test_unmute_handles_mixed_entries(self, state_mod):
|
||||
"""unmute_branch removes target from list with both string and dict entries."""
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
future_ts = (datetime.now() + timedelta(hours=2)).isoformat()
|
||||
config_file = state_mod.TRIGGER_CONFIG_FILE
|
||||
config_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
config_file.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"config": {
|
||||
"muted_branches": [
|
||||
"speakeasy",
|
||||
{"name": "api", "expires_at": future_ts},
|
||||
{"name": "drone", "expires_at": None},
|
||||
]
|
||||
}
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
state_mod.unmute_branch("api")
|
||||
|
||||
muted = state_mod.get_muted_branches()
|
||||
assert "speakeasy" in muted
|
||||
assert "api" not in muted
|
||||
assert "drone" in muted
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests -- _clean_expired_mutes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCleanExpiredMutes:
|
||||
"""Tests for _clean_expired_mutes."""
|
||||
|
||||
def test_clean_expired_mutes_removes_old(self, state_mod):
|
||||
"""_clean_expired_mutes removes expired dict entries, keeps strings and active dicts."""
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
expired_ts = (datetime.now() - timedelta(hours=1)).isoformat()
|
||||
future_ts = (datetime.now() + timedelta(hours=1)).isoformat()
|
||||
|
||||
data = {
|
||||
"config": {
|
||||
"muted_branches": [
|
||||
"speakeasy",
|
||||
{"name": "api", "expires_at": expired_ts},
|
||||
{"name": "drone", "expires_at": future_ts},
|
||||
{"name": "flow", "expires_at": None},
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
state_mod._clean_expired_mutes(data)
|
||||
|
||||
remaining = data["config"]["muted_branches"]
|
||||
names = []
|
||||
for entry in remaining:
|
||||
if isinstance(entry, str):
|
||||
names.append(entry)
|
||||
else:
|
||||
names.append(entry["name"])
|
||||
assert "speakeasy" in names
|
||||
assert "api" not in names
|
||||
assert "drone" in names
|
||||
assert "flow" in names
|
||||
assert len(remaining) == 3
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests -- write_config cleans expired mutes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestWriteConfigCleansMutes:
|
||||
"""Tests for write_config calling _clean_expired_mutes."""
|
||||
|
||||
def test_write_config_cleans_expired_mutes(self, state_mod):
|
||||
"""write_config removes expired mute entries before persisting."""
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
expired_ts = (datetime.now() - timedelta(hours=1)).isoformat()
|
||||
future_ts = (datetime.now() + timedelta(hours=1)).isoformat()
|
||||
|
||||
data = {
|
||||
"config": {
|
||||
"muted_branches": [
|
||||
{"name": "api", "expires_at": expired_ts},
|
||||
{"name": "drone", "expires_at": future_ts},
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
state_mod.write_config(data)
|
||||
|
||||
config_file = state_mod.TRIGGER_CONFIG_FILE
|
||||
written = json.loads(config_file.read_text(encoding="utf-8"))
|
||||
muted = written["config"]["muted_branches"]
|
||||
assert len(muted) == 1
|
||||
assert muted[0]["name"] == "drone"
|
||||
|
||||
Reference in New Issue
Block a user