feat: bridge_self bot commands — status, thresholds, reset, health
Adds bot commands for the bridge_self provider so operators can inspect and manage bridge health from chat: /status, /thresholds, /reset, /health. Includes Jinja2 templates for both locales, seed data, capability slots, and a handler that exposes pending deferred backlog plus per-counter reset. Also adds .claude/skills/ for project-scoped graph-aware skills.
This commit is contained in:
@@ -263,3 +263,386 @@ def test_default_template_loader_returns_bridge_self_slots() -> None:
|
||||
# Sanity: each template references at least one of the bridge_self vars.
|
||||
for tpl in list(en.values()) + list(ru.values()):
|
||||
assert "{{" in tpl
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Bot commands — context builders
|
||||
#
|
||||
# These tests run against the real (temp-dir) DB via the FastAPI lifespan so
|
||||
# that ``services.bridge_self`` helpers using ``get_engine()`` resolve to the
|
||||
# same DB the test seeds rows into. We follow the pattern used by
|
||||
# test_webhook_status_handler.py — bootstrap once, seed under TestClient.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _bootstrap_app():
|
||||
"""Bring up the app once so migrations run against the temp DB."""
|
||||
from notify_bridge_server.main import app
|
||||
|
||||
return app
|
||||
|
||||
|
||||
async def _seed_user_and_provider(
|
||||
*, username: str, config: dict[str, int],
|
||||
) -> tuple[int, int]:
|
||||
"""Create a fresh ``(user_id, provider_id)`` against the live engine.
|
||||
|
||||
Uses two short-lived sessions to avoid SQLAlchemy auto-expiring the
|
||||
first-committed object once a second commit fires on the same session.
|
||||
"""
|
||||
from notify_bridge_server.database.engine import get_engine
|
||||
from notify_bridge_server.database.models import ServiceProvider, User
|
||||
|
||||
engine = get_engine()
|
||||
async with AsyncSession(engine) as db:
|
||||
user = User(
|
||||
username=f"{username}_{datetime.now(timezone.utc).timestamp()}",
|
||||
hashed_password="x", role="user",
|
||||
)
|
||||
db.add(user)
|
||||
await db.commit()
|
||||
await db.refresh(user)
|
||||
user_id = int(user.id)
|
||||
async with AsyncSession(engine) as db:
|
||||
provider = ServiceProvider(
|
||||
user_id=user_id, type="bridge_self", name="Bridge",
|
||||
config=dict(config),
|
||||
)
|
||||
db.add(provider)
|
||||
await db.commit()
|
||||
await db.refresh(provider)
|
||||
provider_id = int(provider.id)
|
||||
return user_id, provider_id
|
||||
|
||||
|
||||
async def _load_provider(provider_id: int):
|
||||
from notify_bridge_server.database.engine import get_engine
|
||||
from notify_bridge_server.database.models import ServiceProvider
|
||||
|
||||
engine = get_engine()
|
||||
async with AsyncSession(engine) as db:
|
||||
return await db.get(ServiceProvider, provider_id)
|
||||
|
||||
|
||||
def test_command_status_returns_empty_lists_when_no_failures(tmp_data_dir) -> None: # noqa: ARG001
|
||||
import asyncio
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from notify_bridge_server.commands.bridge_self_handler import (
|
||||
_build_status_context,
|
||||
)
|
||||
from notify_bridge_server.services import bridge_self as bs
|
||||
|
||||
app = _bootstrap_app()
|
||||
with TestClient(app):
|
||||
async def run() -> None:
|
||||
_user_id, provider_id = await _seed_user_and_provider(
|
||||
username="status_user",
|
||||
config={
|
||||
"poll_failure_threshold": 3,
|
||||
"deferred_backlog_threshold": 100,
|
||||
"target_failure_threshold": 5,
|
||||
},
|
||||
)
|
||||
provider = await _load_provider(provider_id)
|
||||
|
||||
# Make sure the in-memory dicts contain nothing.
|
||||
bs._poll_failure_counts.clear()
|
||||
bs._target_failure_counts.clear()
|
||||
|
||||
ctx = await _build_status_context(provider)
|
||||
|
||||
assert ctx["poll_failures"] == []
|
||||
assert ctx["target_failures"] == []
|
||||
assert ctx["deferred_pending"] == 0
|
||||
assert ctx["deferred_threshold"] == 100
|
||||
|
||||
asyncio.run(run())
|
||||
|
||||
|
||||
def test_command_thresholds_returns_user_config(tmp_data_dir) -> None: # noqa: ARG001
|
||||
import asyncio
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from notify_bridge_server.commands.bridge_self_handler import (
|
||||
_build_thresholds_context,
|
||||
)
|
||||
|
||||
app = _bootstrap_app()
|
||||
with TestClient(app):
|
||||
async def run() -> None:
|
||||
_user_id, provider_id = await _seed_user_and_provider(
|
||||
username="thresholds_user",
|
||||
config={
|
||||
"poll_failure_threshold": 7,
|
||||
"deferred_backlog_threshold": 250,
|
||||
"target_failure_threshold": 11,
|
||||
},
|
||||
)
|
||||
provider = await _load_provider(provider_id)
|
||||
|
||||
ctx = await _build_thresholds_context(provider)
|
||||
|
||||
assert ctx == {
|
||||
"poll_failure_threshold": 7,
|
||||
"deferred_backlog_threshold": 250,
|
||||
"target_failure_threshold": 11,
|
||||
}
|
||||
|
||||
asyncio.run(run())
|
||||
|
||||
|
||||
def test_command_reset_clears_named_counter_and_is_idempotent(tmp_data_dir) -> None: # noqa: ARG001
|
||||
"""``/reset`` clears the in-memory counter and is idempotent.
|
||||
|
||||
Now ownership-aware: we seed a real NotificationTracker owned by the
|
||||
test user so ``find_tracker_owner`` returns a matching user_id and the
|
||||
reset proceeds. The cross-user-rejection case is covered by
|
||||
:func:`test_command_reset_rejects_cross_user_subject` below.
|
||||
"""
|
||||
import asyncio
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from notify_bridge_server.commands.bridge_self_handler import (
|
||||
_build_reset_context, _parse_reset_subject,
|
||||
)
|
||||
from notify_bridge_server.database.engine import get_engine
|
||||
from notify_bridge_server.database.models import NotificationTracker
|
||||
from notify_bridge_server.services import bridge_self as bs
|
||||
|
||||
app = _bootstrap_app()
|
||||
with TestClient(app):
|
||||
async def run() -> None:
|
||||
user_id, provider_id = await _seed_user_and_provider(
|
||||
username="reset_user",
|
||||
config={
|
||||
"poll_failure_threshold": 3,
|
||||
"deferred_backlog_threshold": 100,
|
||||
"target_failure_threshold": 5,
|
||||
},
|
||||
)
|
||||
provider = await _load_provider(provider_id)
|
||||
|
||||
# Seed an owned NotificationTracker so the ownership check
|
||||
# in _build_reset_context can match it back to user_id.
|
||||
engine = get_engine()
|
||||
async with AsyncSession(engine) as db:
|
||||
tracker = NotificationTracker(
|
||||
user_id=user_id, provider_id=provider_id,
|
||||
name="reset-test", enabled=True,
|
||||
)
|
||||
db.add(tracker)
|
||||
await db.commit()
|
||||
await db.refresh(tracker)
|
||||
tid = int(tracker.id)
|
||||
|
||||
bs.reset_poll_counter(tid)
|
||||
bs.record_poll_failure(tid, "boom")
|
||||
bs.record_poll_failure(tid, "boom")
|
||||
assert bs.get_poll_failure_count(tid) == 2
|
||||
|
||||
ctx = await _build_reset_context(f"tracker:{tid}", provider)
|
||||
assert ctx["success"] is True
|
||||
assert ctx["subject_type"] == "tracker"
|
||||
assert ctx["subject_id"] == tid
|
||||
assert ctx["previous_count"] == 2
|
||||
assert ctx["error_message"] is None
|
||||
assert bs.get_poll_failure_count(tid) == 0
|
||||
|
||||
# Idempotent — second call still succeeds with previous=0.
|
||||
ctx2 = await _build_reset_context(f"tracker:{tid}", provider)
|
||||
assert ctx2["success"] is True
|
||||
assert ctx2["previous_count"] == 0
|
||||
|
||||
# Parse error → templated error, no exception.
|
||||
bad_ctx = await _build_reset_context("not a subject", provider)
|
||||
assert bad_ctx["success"] is False
|
||||
assert bad_ctx["error_message"]
|
||||
|
||||
asyncio.run(run())
|
||||
|
||||
# Parser direct sanity-checks (pure function, no DB needed).
|
||||
assert _parse_reset_subject("all") == ("all", None, None)
|
||||
assert _parse_reset_subject("tracker:42") == ("tracker", 42, None)
|
||||
assert _parse_reset_subject("target:7") == ("target", 7, None)
|
||||
_, _, err = _parse_reset_subject("rocket:1")
|
||||
assert err is not None
|
||||
|
||||
|
||||
def test_command_reset_rejects_cross_user_subject(tmp_data_dir) -> None: # noqa: ARG001
|
||||
"""User A cannot reset a counter belonging to user B's tracker.
|
||||
|
||||
Regression guard for the multi-tenant data-leak hole the original
|
||||
handler had — ``reset_counter`` was called without verifying the
|
||||
subject's ``user_id`` matched ``provider.user_id``.
|
||||
"""
|
||||
import asyncio
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from notify_bridge_server.commands.bridge_self_handler import _build_reset_context
|
||||
from notify_bridge_server.database.engine import get_engine
|
||||
from notify_bridge_server.database.models import NotificationTracker
|
||||
from notify_bridge_server.services import bridge_self as bs
|
||||
|
||||
app = _bootstrap_app()
|
||||
with TestClient(app):
|
||||
async def run() -> None:
|
||||
user_a_id, provider_a_id = await _seed_user_and_provider(
|
||||
username="user_a", config={
|
||||
"poll_failure_threshold": 3,
|
||||
"deferred_backlog_threshold": 100,
|
||||
"target_failure_threshold": 5,
|
||||
},
|
||||
)
|
||||
user_b_id, provider_b_id = await _seed_user_and_provider(
|
||||
username="user_b", config={
|
||||
"poll_failure_threshold": 3,
|
||||
"deferred_backlog_threshold": 100,
|
||||
"target_failure_threshold": 5,
|
||||
},
|
||||
)
|
||||
provider_a = await _load_provider(provider_a_id)
|
||||
|
||||
# Seed a tracker owned by user B and increment its counter.
|
||||
engine = get_engine()
|
||||
async with AsyncSession(engine) as db:
|
||||
tracker_b = NotificationTracker(
|
||||
user_id=user_b_id, provider_id=provider_b_id,
|
||||
name="b-only", enabled=True,
|
||||
)
|
||||
db.add(tracker_b)
|
||||
await db.commit()
|
||||
await db.refresh(tracker_b)
|
||||
tid_b = int(tracker_b.id)
|
||||
|
||||
bs.reset_poll_counter(tid_b)
|
||||
bs.record_poll_failure(tid_b, "boom")
|
||||
assert bs.get_poll_failure_count(tid_b) == 1
|
||||
|
||||
# User A tries to reset user B's tracker — must fail.
|
||||
ctx = await _build_reset_context(f"tracker:{tid_b}", provider_a)
|
||||
assert ctx["success"] is False
|
||||
assert "not found" in (ctx["error_message"] or "").lower()
|
||||
# Counter must remain untouched.
|
||||
assert bs.get_poll_failure_count(tid_b) == 1
|
||||
|
||||
asyncio.run(run())
|
||||
|
||||
|
||||
def test_command_health_is_healthy_when_counters_zero(tmp_data_dir) -> None: # noqa: ARG001
|
||||
import asyncio
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from notify_bridge_server.commands.bridge_self_handler import (
|
||||
_build_health_context,
|
||||
)
|
||||
from notify_bridge_server.services import bridge_self as bs
|
||||
|
||||
app = _bootstrap_app()
|
||||
with TestClient(app):
|
||||
async def run() -> None:
|
||||
_user_id, provider_id = await _seed_user_and_provider(
|
||||
username="health_user",
|
||||
config={
|
||||
"poll_failure_threshold": 3,
|
||||
"deferred_backlog_threshold": 100,
|
||||
"target_failure_threshold": 5,
|
||||
},
|
||||
)
|
||||
provider = await _load_provider(provider_id)
|
||||
|
||||
# Empty counters and no deferred rows for this user.
|
||||
bs._poll_failure_counts.clear()
|
||||
bs._target_failure_counts.clear()
|
||||
|
||||
ctx = await _build_health_context(provider)
|
||||
|
||||
assert ctx["healthy"] is True
|
||||
assert ctx["failing_tracker_count"] == 0
|
||||
assert ctx["failing_target_count"] == 0
|
||||
assert ctx["deferred_pending"] == 0
|
||||
assert "healthy" in ctx["summary"].lower()
|
||||
|
||||
asyncio.run(run())
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# reset_counter direct unit test (covers the "all" branch)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_reset_counter_all_clears_every_dict() -> None:
|
||||
from notify_bridge_server.services import bridge_self as bs
|
||||
|
||||
# Seed both dicts with a couple of entries.
|
||||
bs._poll_failure_counts.clear()
|
||||
bs._target_failure_counts.clear()
|
||||
bs.record_poll_failure(9_401, "boom")
|
||||
bs.record_poll_failure(9_402, "boom")
|
||||
bs.record_target_failure(9_501, "503")
|
||||
|
||||
cleared = bs.reset_counter("all")
|
||||
|
||||
# 2 poll + 1 target = 3 entries cleared.
|
||||
assert cleared == 3
|
||||
assert bs._poll_failure_counts == {}
|
||||
assert bs._target_failure_counts == {}
|
||||
|
||||
|
||||
def test_reset_counter_unknown_failure_type_is_noop() -> None:
|
||||
from notify_bridge_server.services import bridge_self as bs
|
||||
|
||||
bs._poll_failure_counts.clear()
|
||||
bs.record_poll_failure(9_601, "boom")
|
||||
|
||||
# Unknown type returns 0 and leaves dicts intact.
|
||||
assert bs.reset_counter("rocket_failures", 9_601) == 0
|
||||
assert bs.get_poll_failure_count(9_601) == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Capability / handler registration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_capability_registry_lists_bridge_self_commands() -> None:
|
||||
from notify_bridge_core.providers.capabilities import get_capabilities
|
||||
|
||||
caps = get_capabilities("bridge_self")
|
||||
assert caps is not None
|
||||
|
||||
cmd_names = {c["name"] for c in caps.commands}
|
||||
assert {"status", "thresholds", "reset", "health", "help"}.issubset(cmd_names)
|
||||
|
||||
slot_names = {s["name"] for s in caps.command_slots}
|
||||
# Response slots
|
||||
assert {"status", "thresholds", "reset", "health"}.issubset(slot_names)
|
||||
# Description slots — needed so the menu sync registers a description
|
||||
# for every operator-facing command.
|
||||
assert {
|
||||
"desc_status", "desc_thresholds", "desc_reset", "desc_health",
|
||||
}.issubset(slot_names)
|
||||
|
||||
|
||||
def test_command_handler_registered_for_bridge_self() -> None:
|
||||
"""Auto-registration wires the bridge_self handler into dispatch."""
|
||||
from notify_bridge_server.commands.dispatch import get_handler
|
||||
|
||||
handler = get_handler("bridge_self")
|
||||
assert handler is not None
|
||||
assert handler.provider_type == "bridge_self"
|
||||
assert {"status", "thresholds", "reset", "health"} == handler.get_provider_commands()
|
||||
|
||||
|
||||
def test_default_command_template_loader_returns_bridge_self_slots() -> None:
|
||||
"""All shipped command-template slots load for both locales."""
|
||||
from notify_bridge_core.templates.command_defaults.loader import (
|
||||
load_default_command_templates,
|
||||
)
|
||||
|
||||
en = load_default_command_templates("en", "bridge_self")
|
||||
ru = load_default_command_templates("ru", "bridge_self")
|
||||
|
||||
required = {"status", "thresholds", "reset", "health", "help", "start"}
|
||||
assert required.issubset(en.keys())
|
||||
assert required.issubset(ru.keys())
|
||||
|
||||
Reference in New Issue
Block a user