367 lines
12 KiB
Python
367 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from src.api.v1 import telegram as telegram_api
|
|
from src.api.v1 import telegram_webhook as telegram_webhook_api
|
|
from src.services import telegram_gateway as telegram_gateway_module
|
|
from src.services.callback_dispatcher import load_action_registry
|
|
from src.services.telegram_button_registry import (
|
|
callback_action_is_registered,
|
|
registered_info_callback_actions,
|
|
resolve_callback_action_contract,
|
|
telegram_button_registry_snapshot,
|
|
)
|
|
from src.services.telegram_gateway import (
|
|
CALLBACK_INGRESS_RECEIPT_MAX_AGE_SECONDS,
|
|
TelegramGateway,
|
|
filter_telegram_reply_markup,
|
|
)
|
|
|
|
|
|
class _ConfiguredGateway(TelegramGateway):
|
|
@property
|
|
def bot_token(self) -> str:
|
|
return "test-configured"
|
|
|
|
@property
|
|
def chat_id(self) -> str:
|
|
return "test-configured"
|
|
|
|
|
|
class _Security:
|
|
@staticmethod
|
|
def generate_callback_nonce(approval_id: str, action: str) -> str:
|
|
return f"{action}:{approval_id}:1:nonce"
|
|
|
|
|
|
def _callback_rows(markup: dict | None) -> list[dict]:
|
|
return [
|
|
button
|
|
for row in (markup or {}).get("inline_keyboard", [])
|
|
for button in row
|
|
]
|
|
|
|
|
|
def test_callback_ingress_defaults_to_unverified_fail_closed() -> None:
|
|
gateway = _ConfiguredGateway()
|
|
|
|
status = gateway.callback_ingress_status()
|
|
|
|
assert status["mode"] == "external_callback_forward_unverified"
|
|
assert status["polling_active"] is False
|
|
assert status["receipt_available"] is False
|
|
assert status["interactive_buttons_ready"] is False
|
|
assert status["fail_closed_reason"] == "callback_ingress_not_verified"
|
|
|
|
|
|
def test_callback_ingress_receipt_has_24_hour_freshness_contract() -> None:
|
|
gateway = _ConfiguredGateway()
|
|
assert CALLBACK_INGRESS_RECEIPT_MAX_AGE_SECONDS == 24 * 60 * 60
|
|
|
|
gateway._last_callback_ingress_source = "canonical_webhook"
|
|
gateway._last_callback_ingress_receipt_at = datetime.now(UTC) - timedelta(hours=23)
|
|
fresh = gateway.callback_ingress_status()
|
|
assert fresh["receipt_fresh"] is True
|
|
assert fresh["interactive_buttons_ready"] is True
|
|
|
|
gateway._last_callback_ingress_receipt_at = datetime.now(UTC) - timedelta(hours=25)
|
|
stale = gateway.callback_ingress_status()
|
|
assert stale["stale"] is True
|
|
assert stale["interactive_buttons_ready"] is False
|
|
|
|
|
|
def test_polling_flag_without_live_task_is_stale_not_ready() -> None:
|
|
gateway = _ConfiguredGateway()
|
|
gateway._polling_active = True
|
|
gateway._polling_task = None
|
|
|
|
status = gateway.callback_ingress_status()
|
|
|
|
assert status["mode"] == "api_long_polling_stale"
|
|
assert status["polling_declared_active"] is True
|
|
assert status["polling_task_active"] is False
|
|
assert status["polling_active"] is False
|
|
assert status["interactive_buttons_ready"] is False
|
|
|
|
|
|
def test_fail_closed_markup_keeps_only_verified_https_urls() -> None:
|
|
markup = {
|
|
"inline_keyboard": [[
|
|
{"text": "詳情", "callback_data": "detail:INC-20260716-TEST"},
|
|
{"text": "Runs", "url": "https://awoooi.wooo.work/zh-TW/awooop/runs"},
|
|
{"text": "Unsafe", "url": "http://192.0.2.1/internal"},
|
|
{"text": "External", "url": "https://example.com/not-canonical"},
|
|
]]
|
|
}
|
|
|
|
filtered = filter_telegram_reply_markup(
|
|
markup,
|
|
interactive_buttons_ready=False,
|
|
)
|
|
|
|
assert _callback_rows(filtered) == [{
|
|
"text": "Runs",
|
|
"url": "https://awoooi.wooo.work/zh-TW/awooop/runs",
|
|
}]
|
|
|
|
|
|
def test_gateway_render_gate_uses_live_ingress_status() -> None:
|
|
gateway = _ConfiguredGateway()
|
|
payload = {
|
|
"text": "incident",
|
|
"reply_markup": {
|
|
"inline_keyboard": [[
|
|
{"text": "詳情", "callback_data": "detail:INC-20260716-TEST"},
|
|
{"text": "Runs", "url": "https://awoooi.wooo.work/zh-TW/awooop/runs"},
|
|
]]
|
|
},
|
|
}
|
|
|
|
guarded = gateway._apply_callback_ingress_render_gate(payload)
|
|
|
|
assert _callback_rows(guarded["reply_markup"]) == [{
|
|
"text": "Runs",
|
|
"url": "https://awoooi.wooo.work/zh-TW/awooop/runs",
|
|
}]
|
|
|
|
|
|
def test_ready_markup_still_removes_unknown_ghost_callbacks() -> None:
|
|
markup = {
|
|
"inline_keyboard": [[
|
|
{"text": "詳情", "callback_data": "detail:INC-20260716-TEST"},
|
|
{"text": "Ghost view", "callback_data": "view:ERR:1:nonce"},
|
|
{"text": "Ghost scale", "callback_data": "scale:RES:1:nonce"},
|
|
{"text": "Runs", "url": "https://awoooi.wooo.work/zh-TW/awooop/runs"},
|
|
]]
|
|
}
|
|
|
|
filtered = filter_telegram_reply_markup(
|
|
markup,
|
|
interactive_buttons_ready=True,
|
|
)
|
|
rendered = _callback_rows(filtered)
|
|
|
|
assert {button.get("callback_data") for button in rendered} == {
|
|
"detail:INC-20260716-TEST",
|
|
None,
|
|
}
|
|
assert all("view:" not in str(button) for button in rendered)
|
|
assert all("scale:" not in str(button) for button in rendered)
|
|
|
|
|
|
def test_machine_registry_excludes_ghosts_and_describes_shadow_tune() -> None:
|
|
assert callback_action_is_registered("detail:INC-20260716-TEST") is True
|
|
assert callback_action_is_registered("view:ERR:1:nonce") is False
|
|
assert callback_action_is_registered("scale:RES:1:nonce") is False
|
|
tune = resolve_callback_action_contract("tune:APR-1:1:nonce")
|
|
assert tune is not None
|
|
assert tune.state == "shadow_candidate_only"
|
|
|
|
snapshot = telegram_button_registry_snapshot()
|
|
action_names = {row["action"] for row in snapshot["actions"]}
|
|
assert snapshot["schema_version"] == "telegram_button_action_registry_v1"
|
|
assert snapshot["unknown_action_policy"] == "do_not_render"
|
|
assert "view" not in action_names
|
|
assert "scale" not in action_names
|
|
|
|
|
|
def test_every_yaml_action_has_a_renderable_registry_contract() -> None:
|
|
registry = load_action_registry()
|
|
assert registry
|
|
|
|
for action, spec in registry.items():
|
|
callback_data = (
|
|
f"{action}:APR-1:1:nonce"
|
|
if spec.callback_format == "nonce"
|
|
else f"{action}:INC-20260716-TEST"
|
|
)
|
|
contract = resolve_callback_action_contract(callback_data)
|
|
assert contract is not None, action
|
|
assert contract.handler == "callback_dispatcher.dispatch_action"
|
|
assert contract.executor.startswith("mcp:")
|
|
|
|
|
|
def test_security_parser_accepts_every_registered_info_action() -> None:
|
|
gateway = _ConfiguredGateway()
|
|
info_actions = registered_info_callback_actions()
|
|
assert {"detail", "history", "reanalyze", "drift_view_page"} <= info_actions
|
|
|
|
for action in info_actions:
|
|
parsed = gateway._security.parse_callback_data(
|
|
f"{action}:INC-20260716-TEST"
|
|
)
|
|
assert parsed["action"] == action
|
|
assert parsed["is_info_action"] is True
|
|
|
|
|
|
def test_legacy_builders_no_longer_emit_view_or_scale() -> None:
|
|
gateway = _ConfiguredGateway()
|
|
gateway._security = _Security()
|
|
|
|
sentry = _callback_rows(gateway._build_sentry_keyboard("ERR-1"))
|
|
resource = _callback_rows(gateway._build_resource_keyboard("RES-1"))
|
|
|
|
assert [button["callback_data"].split(":", 1)[0] for button in sentry] == [
|
|
"silence"
|
|
]
|
|
assert [button["callback_data"].split(":", 1)[0] for button in resource] == [
|
|
"silence"
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tune_button_truthfully_says_it_records_a_candidate() -> None:
|
|
gateway = _ConfiguredGateway()
|
|
gateway._security = _Security()
|
|
|
|
markup = await gateway._build_inline_keyboard(
|
|
approval_id="APR-1",
|
|
incident_id="INC-20260716-TEST",
|
|
include_auto_tuning=True,
|
|
auto_tuning_command="candidate-only",
|
|
)
|
|
labels = {button["text"] for button in _callback_rows(markup)}
|
|
|
|
assert "📝 記錄調優候選" in labels
|
|
assert "⚡ 執行自動調優" not in labels
|
|
|
|
|
|
class _InfoCallbackGateway(_ConfiguredGateway):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.approval_execution_called = False
|
|
|
|
async def mirror_callback_query_received(self, **_kwargs) -> str:
|
|
return "receipt-1"
|
|
|
|
async def handle_callback(self, **_kwargs) -> dict:
|
|
return {
|
|
"success": True,
|
|
"info_action": True,
|
|
"action": "detail",
|
|
"approval_id": "INC-20260716-TEST",
|
|
}
|
|
|
|
async def _execute_approval_action(self, **_kwargs) -> None:
|
|
self.approval_execution_called = True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_polling_info_action_does_not_execute_approval_stage() -> None:
|
|
gateway = _InfoCallbackGateway()
|
|
|
|
await gateway._handle_callback_query(
|
|
42,
|
|
{
|
|
"id": "callback-1",
|
|
"data": "detail:INC-20260716-TEST",
|
|
"from": {"id": 7, "username": "operator"},
|
|
"message": {
|
|
"message_id": 99,
|
|
"text": "incident",
|
|
"chat": {"id": -1001},
|
|
},
|
|
},
|
|
)
|
|
|
|
assert gateway.approval_execution_called is False
|
|
assert gateway.callback_ingress_status()["receipt_source"] == "api_long_polling"
|
|
|
|
|
|
def test_webhook_route_has_one_canonical_owner() -> None:
|
|
route_paths = [route.path for route in telegram_api.router.routes]
|
|
route_paths.extend(route.path for route in telegram_webhook_api.router.routes)
|
|
|
|
assert route_paths.count("/telegram/webhook") == 1
|
|
|
|
|
|
def test_production_webhook_without_secret_fails_closed(monkeypatch) -> None:
|
|
monkeypatch.setattr(
|
|
telegram_webhook_api,
|
|
"settings",
|
|
SimpleNamespace(ENVIRONMENT="prod", TELEGRAM_WEBHOOK_SECRET=""),
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
telegram_webhook_api._verify_secret_token(None)
|
|
|
|
assert exc_info.value.status_code == 503
|
|
|
|
|
|
class _CallbackRequest:
|
|
async def json(self) -> dict:
|
|
return {
|
|
"update_id": 77,
|
|
"callback_query": {
|
|
"id": "callback-77",
|
|
"data": "detail:INC-20260716-TEST",
|
|
"from": {"id": 7, "username": "operator"},
|
|
"message": {"message_id": 9, "chat": {"id": -1001}},
|
|
},
|
|
}
|
|
|
|
|
|
class _CanonicalWebhookGateway:
|
|
def __init__(self) -> None:
|
|
self.receipts: list[dict] = []
|
|
|
|
async def _handle_callback_query(
|
|
self,
|
|
update_id: int,
|
|
callback_query: dict,
|
|
*,
|
|
ingress_source: str,
|
|
) -> None:
|
|
self.receipts.append({
|
|
"update_id": update_id,
|
|
"callback_query": callback_query,
|
|
"ingress_source": ingress_source,
|
|
})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_canonical_webhook_forwards_callback_to_gateway_handler(monkeypatch) -> None:
|
|
gateway = _CanonicalWebhookGateway()
|
|
monkeypatch.setattr(
|
|
telegram_gateway_module,
|
|
"get_telegram_gateway",
|
|
lambda: gateway,
|
|
)
|
|
|
|
result = await telegram_webhook_api.telegram_webhook(_CallbackRequest())
|
|
|
|
assert result == {"ok": True}
|
|
assert len(gateway.receipts) == 1
|
|
assert gateway.receipts[0]["update_id"] == 77
|
|
assert gateway.receipts[0]["ingress_source"] == "canonical_webhook"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_health_is_degraded_when_callback_ingress_is_unverified(monkeypatch) -> None:
|
|
gateway = _ConfiguredGateway()
|
|
monkeypatch.setattr(telegram_api, "get_telegram_gateway", lambda: gateway)
|
|
monkeypatch.setattr(
|
|
telegram_api,
|
|
"settings",
|
|
SimpleNamespace(
|
|
OPENCLAW_TG_BOT_TOKEN="test-configured",
|
|
SRE_GROUP_CHAT_ID="test-configured",
|
|
OPENCLAW_TG_USER_WHITELIST=[],
|
|
ENVIRONMENT="test",
|
|
),
|
|
)
|
|
|
|
health = await telegram_api.telegram_health()
|
|
|
|
assert health["status"] == "degraded_callback_ingress_unverified"
|
|
assert health["mode"] == "external_callback_forward_unverified"
|
|
assert health["interactive_buttons_ready"] is False
|
|
assert health["callback_ingress"]["fail_closed_reason"] == (
|
|
"callback_ingress_not_verified"
|
|
)
|