759 lines
25 KiB
Python
759 lines
25 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
from uuid import UUID
|
|
|
|
import pytest
|
|
|
|
import src.repositories.drift_repository as drift_repository
|
|
import src.repositories.incident_repository as incident_repository
|
|
import src.services.ai_advisory_helpers as ai_advisory_helpers
|
|
import src.services.approval_db as approval_db
|
|
import src.services.callback_dispatcher as callback_dispatcher
|
|
import src.services.telegram_gateway as gateway_module
|
|
from src.models.approval import ApprovalStatus
|
|
from src.models.incident import IncidentStatus
|
|
from src.services.callback_dispatcher import DispatchResult
|
|
from src.services.telegram_gateway import TelegramGateway
|
|
|
|
_CHAT_ID = "verified-chat"
|
|
_MESSAGE_ID = 77
|
|
_DESTINATION_BINDING = gateway_module._telegram_destination_binding(_CHAT_ID)
|
|
_SENT_RECEIPT = {
|
|
"ok": True,
|
|
"_awooop_delivery_status": "sent",
|
|
"_awooop_provider_send_performed": True,
|
|
"result": {"message_id": _MESSAGE_ID, "chat": {"id": _CHAT_ID}},
|
|
"_awoooi_canonical_route_receipt": {
|
|
"schema_version": "telegram_canonical_egress_receipt_v1",
|
|
"decision": "allowed",
|
|
"destination_alias": "inbound_interaction_reply_target",
|
|
"destination_binding": _DESTINATION_BINDING,
|
|
"sender_bot_alias": "tsenyang_bot",
|
|
"provider_send_performed": True,
|
|
},
|
|
gateway_module._DELIVERY_CONTEXT_KEY: {
|
|
"schema_version": "telegram_delivery_context_v1",
|
|
"method": "sendMessage",
|
|
"sender_bot_alias": "tsenyang_bot",
|
|
"destination_alias": "inbound_interaction_reply_target",
|
|
"destination_binding": _DESTINATION_BINDING,
|
|
"payload_destination_binding": _DESTINATION_BINDING,
|
|
"provider_destination_binding": _DESTINATION_BINDING,
|
|
"destination_binding_verified": True,
|
|
},
|
|
}
|
|
|
|
|
|
def _assert_verified_callback_payload(payload: dict) -> None:
|
|
assert payload["chat_id"] == _CHAT_ID
|
|
assert payload["reply_to_message_id"] == _MESSAGE_ID
|
|
assert payload[gateway_module._NON_MONITORING_INTERACTION_KEY] == {
|
|
"interaction_kind": "callback_reply",
|
|
"inbound_chat_id": _CHAT_ID,
|
|
"inbound_message_id": _MESSAGE_ID,
|
|
}
|
|
|
|
|
|
class _DriftRepo:
|
|
def __init__(self, value=None, error: Exception | None = None) -> None:
|
|
self.value = value
|
|
self.error = error
|
|
|
|
async def get_by_id(self, _report_id: str):
|
|
if self.error:
|
|
raise self.error
|
|
return self.value
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"repo",
|
|
[
|
|
_DriftRepo(),
|
|
_DriftRepo(SimpleNamespace(items=[], namespace="awoooi-prod")),
|
|
_DriftRepo(error=RuntimeError("read failed")),
|
|
],
|
|
)
|
|
async def test_drift_detail_page_and_error_keep_verified_callback_context(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
repo: _DriftRepo,
|
|
) -> None:
|
|
gateway = TelegramGateway()
|
|
send_request = AsyncMock(return_value={"ok": True})
|
|
monkeypatch.setattr(gateway, "_send_request", send_request)
|
|
monkeypatch.setattr(drift_repository, "get_drift_repository", lambda: repo)
|
|
|
|
await gateway._send_drift_diff_detail(
|
|
"drift-1",
|
|
inbound_chat_id=_CHAT_ID,
|
|
inbound_message_id=_MESSAGE_ID,
|
|
)
|
|
|
|
send_request.assert_awaited_once()
|
|
method, payload = send_request.await_args.args
|
|
assert method == "sendMessage"
|
|
_assert_verified_callback_payload(payload)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_advisory_and_drift_outcome_keep_verified_callback_context(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
gateway = TelegramGateway()
|
|
send_request = AsyncMock(return_value={"ok": True})
|
|
monkeypatch.setattr(gateway, "_send_request", send_request)
|
|
monkeypatch.setattr(gateway, "_answer_callback", AsyncMock())
|
|
monkeypatch.setattr(
|
|
ai_advisory_helpers,
|
|
"handle_ai_advisory_callback",
|
|
AsyncMock(return_value={"feedback_text": "done", "success": True}),
|
|
)
|
|
|
|
await gateway._handle_ai_advisory_action(
|
|
action="ai_advisory_view",
|
|
advisory_payload="capacity:item-1",
|
|
callback_query_id="callback-1",
|
|
user_id=42,
|
|
username="operator",
|
|
user={"id": 42},
|
|
inbound_chat_id=_CHAT_ID,
|
|
inbound_message_id=_MESSAGE_ID,
|
|
)
|
|
_assert_verified_callback_payload(send_request.await_args.args[1])
|
|
|
|
send_request.reset_mock()
|
|
|
|
class _Redis:
|
|
async def get(self, _key: str) -> str:
|
|
return str(_MESSAGE_ID)
|
|
|
|
monkeypatch.setattr(gateway_module, "get_redis", lambda: _Redis())
|
|
await gateway._edit_drift_card_outcome(
|
|
report_id="drift-1",
|
|
verb="adopted",
|
|
by="operator",
|
|
ok=True,
|
|
inbound_chat_id=_CHAT_ID,
|
|
inbound_message_id=_MESSAGE_ID,
|
|
)
|
|
send_payload = next(
|
|
call.args[1]
|
|
for call in send_request.await_args_list
|
|
if call.args[0] == "sendMessage"
|
|
)
|
|
_assert_verified_callback_payload(send_payload)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_drift_outcome_ignores_stale_redis_parent_receipt(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
gateway = TelegramGateway()
|
|
send_request = AsyncMock(return_value={"ok": True})
|
|
redis_reads: list[str] = []
|
|
|
|
class _Redis:
|
|
async def get(self, key: str) -> str:
|
|
redis_reads.append(key)
|
|
return "999"
|
|
|
|
monkeypatch.setattr(gateway, "_send_request", send_request)
|
|
monkeypatch.setattr(gateway_module, "get_redis", lambda: _Redis())
|
|
|
|
await gateway._edit_drift_card_outcome(
|
|
report_id="drift-stale-parent",
|
|
verb="adopted",
|
|
by="operator",
|
|
ok=True,
|
|
inbound_chat_id=_CHAT_ID,
|
|
inbound_message_id=_MESSAGE_ID,
|
|
)
|
|
|
|
assert redis_reads == ["tg_drift:drift-stale-parent"]
|
|
assert [call.args[0] for call in send_request.await_args_list] == [
|
|
"editMessageReplyMarkup",
|
|
"sendMessage",
|
|
]
|
|
edit_payload = send_request.await_args_list[0].args[1]
|
|
assert edit_payload["chat_id"] == _CHAT_ID
|
|
assert edit_payload["message_id"] == _MESSAGE_ID
|
|
send_payload = send_request.await_args_list[1].args[1]
|
|
_assert_verified_callback_payload(send_payload)
|
|
for call in send_request.await_args_list:
|
|
provider_payload = call.args[1]
|
|
assert provider_payload.get("message_id") != 999
|
|
assert provider_payload.get("reply_to_message_id") != 999
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_llm_and_category_results_keep_verified_callback_context(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
gateway = TelegramGateway()
|
|
send_request = AsyncMock(return_value={"ok": True})
|
|
monkeypatch.setattr(gateway, "_send_request", send_request)
|
|
monkeypatch.setattr(gateway._security, "is_whitelisted", lambda _user_id: True)
|
|
|
|
class _Redis:
|
|
async def get(self, _key: str) -> str:
|
|
return json.dumps(
|
|
{
|
|
"name": "inspect",
|
|
"provider": "internal",
|
|
"tool": "status",
|
|
"risk": "low",
|
|
"incident_id": "INC-1",
|
|
}
|
|
)
|
|
|
|
monkeypatch.setattr(gateway_module, "get_redis", lambda: _Redis())
|
|
monkeypatch.setattr(
|
|
callback_dispatcher,
|
|
"dispatch_llm_action",
|
|
lambda _action, _context: {"ok": True},
|
|
)
|
|
await gateway._handle_llm_action_callback(
|
|
callback_query_id="callback-1",
|
|
callback_data="la:short",
|
|
user_id=42,
|
|
username="operator",
|
|
inbound_chat_id=_CHAT_ID,
|
|
inbound_message_id=_MESSAGE_ID,
|
|
)
|
|
llm_payload = next(
|
|
call.args[1]
|
|
for call in send_request.await_args_list
|
|
if call.args[0] == "sendMessage"
|
|
)
|
|
_assert_verified_callback_payload(llm_payload)
|
|
|
|
send_request.reset_mock()
|
|
monkeypatch.setattr(gateway, "_answer_callback", AsyncMock())
|
|
monkeypatch.setattr(
|
|
callback_dispatcher,
|
|
"get_action_spec",
|
|
lambda _action: SimpleNamespace(emoji="🔎"),
|
|
)
|
|
monkeypatch.setattr(
|
|
callback_dispatcher,
|
|
"dispatch_action",
|
|
AsyncMock(
|
|
return_value=DispatchResult(
|
|
success=True,
|
|
action="inspect",
|
|
incident_id="INC-1",
|
|
user_id=42,
|
|
result_text="result",
|
|
)
|
|
),
|
|
)
|
|
|
|
class _IncidentRepo:
|
|
async def get_by_id(self, _incident_id: str):
|
|
return SimpleNamespace(signals=[])
|
|
|
|
monkeypatch.setattr(
|
|
incident_repository,
|
|
"get_incident_repository",
|
|
lambda: _IncidentRepo(),
|
|
)
|
|
await gateway._dispatch_category_action(
|
|
callback_query_id="callback-2",
|
|
action="inspect",
|
|
incident_id="INC-1",
|
|
user_id=42,
|
|
inbound_chat_id=_CHAT_ID,
|
|
inbound_message_id=_MESSAGE_ID,
|
|
)
|
|
_assert_verified_callback_payload(send_request.await_args.args[1])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_category_callback_keeps_verified_context(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
gateway = TelegramGateway()
|
|
send_request = AsyncMock(return_value={"ok": True})
|
|
monkeypatch.setattr(gateway, "_send_request", send_request)
|
|
monkeypatch.setattr(
|
|
gateway._security,
|
|
"parse_callback_data",
|
|
lambda _data: {
|
|
"action": "restart_safe",
|
|
"approval_id": "INC-1",
|
|
"is_info_action": False,
|
|
"nonce": "nonce",
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway._security,
|
|
"verify_callback",
|
|
AsyncMock(return_value={"id": 42}),
|
|
)
|
|
monkeypatch.setattr(
|
|
gateway, "_check_incident_state_guard", AsyncMock(return_value=None)
|
|
)
|
|
monkeypatch.setattr(gateway, "_answer_callback", AsyncMock())
|
|
spec = SimpleNamespace(
|
|
requires_multi_sig=False,
|
|
risk="medium",
|
|
mcp_provider="internal",
|
|
mcp_tool="restart_safe",
|
|
emoji="▶️",
|
|
label="restart",
|
|
)
|
|
monkeypatch.setattr(callback_dispatcher, "get_action_spec", lambda _action: spec)
|
|
monkeypatch.setattr(
|
|
callback_dispatcher,
|
|
"dispatch_action",
|
|
AsyncMock(
|
|
return_value=DispatchResult(
|
|
success=True,
|
|
action="restart_safe",
|
|
incident_id="INC-1",
|
|
user_id=42,
|
|
result_text="done",
|
|
)
|
|
),
|
|
)
|
|
|
|
class _IncidentRepo:
|
|
async def get_by_id(self, _incident_id: str):
|
|
return SimpleNamespace(incident_id="INC-1", signals=[])
|
|
|
|
monkeypatch.setattr(
|
|
incident_repository,
|
|
"get_incident_repository",
|
|
lambda: _IncidentRepo(),
|
|
)
|
|
|
|
result = await gateway.handle_callback(
|
|
callback_query_id="callback-1",
|
|
callback_data="restart_safe:INC-1:nonce",
|
|
user_id=42,
|
|
message_id=_MESSAGE_ID,
|
|
username="operator",
|
|
chat_id=_CHAT_ID,
|
|
)
|
|
|
|
assert result["success"] is True
|
|
_assert_verified_callback_payload(send_request.await_args.args[1])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approval_replies_keep_verified_callback_context(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
gateway = TelegramGateway()
|
|
send_request = AsyncMock(return_value={"ok": True})
|
|
monkeypatch.setattr(gateway, "_send_request", send_request)
|
|
|
|
await gateway._notify_approval_result(
|
|
message_id=_MESSAGE_ID,
|
|
incident_id="INC-1",
|
|
action="approve",
|
|
username="operator",
|
|
execution_triggered=False,
|
|
inbound_chat_id=_CHAT_ID,
|
|
inbound_message_id=_MESSAGE_ID,
|
|
)
|
|
status_payload = next(
|
|
call.args[1]
|
|
for call in send_request.await_args_list
|
|
if call.args[0] == "sendMessage"
|
|
)
|
|
_assert_verified_callback_payload(status_payload)
|
|
source_extra = status_payload[gateway_module._AWOOOP_SOURCE_ENVELOPE_EXTRA_KEY]
|
|
merged = gateway_module._merge_outbound_source_envelope_extra({}, source_extra)
|
|
assert merged["callback_reply"]["parent_provider_message_id"] == str(_MESSAGE_ID)
|
|
|
|
send_request.reset_mock()
|
|
approval_id = UUID("44444444-4444-4444-4444-444444444444")
|
|
terminal = SimpleNamespace(
|
|
id=approval_id,
|
|
status=ApprovalStatus.EXECUTION_SUCCESS,
|
|
action="restart_safe",
|
|
metadata={},
|
|
)
|
|
|
|
class _ApprovalService:
|
|
async def sign_approval(self, **_kwargs):
|
|
return terminal, "already done", False
|
|
|
|
monkeypatch.setattr(approval_db, "get_approval_service", lambda: _ApprovalService())
|
|
await gateway._execute_approval_action(
|
|
action="approve",
|
|
approval_id=str(approval_id),
|
|
user_id=42,
|
|
username="operator",
|
|
message_id=_MESSAGE_ID,
|
|
inbound_chat_id=_CHAT_ID,
|
|
)
|
|
_assert_verified_callback_payload(send_request.await_args.args[1])
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("result", "expected"),
|
|
[
|
|
(_SENT_RECEIPT, True),
|
|
({"ok": True}, False),
|
|
({"ok": True, "_awooop_delivery_status": "sent_reused"}, False),
|
|
({"ok": False}, False),
|
|
(
|
|
{"ok": True, "_awooop_delivery_status": "blocked_no_egress"},
|
|
False,
|
|
),
|
|
({"ok": True, "_awooop_provider_send_performed": False}, False),
|
|
(
|
|
{
|
|
"ok": True,
|
|
"_awoooi_canonical_route_receipt": {
|
|
"provider_send_performed": False,
|
|
},
|
|
},
|
|
False,
|
|
),
|
|
],
|
|
)
|
|
def test_telegram_send_delivery_requires_affirmative_provider_receipt(
|
|
result: dict,
|
|
expected: bool,
|
|
) -> None:
|
|
assert gateway_module._telegram_send_delivery_succeeded(result) is expected
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"primary_failure",
|
|
[
|
|
{"ok": False},
|
|
{"ok": True, "_awooop_delivery_status": "blocked_no_egress"},
|
|
{"ok": True, "_awooop_provider_send_performed": False},
|
|
{
|
|
"ok": True,
|
|
"_awoooi_canonical_route_receipt": {
|
|
"provider_send_performed": False,
|
|
},
|
|
},
|
|
],
|
|
)
|
|
async def test_approval_structured_no_send_retries_plain_on_exact_parent(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
primary_failure: dict,
|
|
) -> None:
|
|
gateway = TelegramGateway()
|
|
attempts: list[tuple[str, dict]] = []
|
|
send_results = iter([primary_failure, dict(_SENT_RECEIPT)])
|
|
|
|
async def structured_primary_failure(method: str, payload: dict) -> dict:
|
|
attempts.append((method, payload))
|
|
if method == "sendMessage":
|
|
return next(send_results)
|
|
return {"ok": True}
|
|
|
|
record_failure = AsyncMock(return_value=None)
|
|
shared_fallback = AsyncMock(return_value=dict(_SENT_RECEIPT))
|
|
monkeypatch.setattr(gateway, "_send_request", structured_primary_failure)
|
|
monkeypatch.setattr(gateway, "_record_callback_reply_failure", record_failure)
|
|
monkeypatch.setattr(gateway, "send_alert_notification", shared_fallback)
|
|
|
|
await gateway._notify_approval_result(
|
|
message_id=_MESSAGE_ID,
|
|
incident_id="INC-1",
|
|
action="approve",
|
|
username="operator",
|
|
execution_triggered=False,
|
|
inbound_chat_id=_CHAT_ID,
|
|
inbound_message_id=_MESSAGE_ID,
|
|
)
|
|
|
|
send_attempts = [payload for method, payload in attempts if method == "sendMessage"]
|
|
assert len(send_attempts) == 2
|
|
for payload in send_attempts:
|
|
_assert_verified_callback_payload(payload)
|
|
assert send_attempts[0]["parse_mode"] == "HTML"
|
|
assert "parse_mode" not in send_attempts[1]
|
|
record_failure.assert_not_awaited()
|
|
shared_fallback.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approval_structured_double_no_send_records_durable_failure(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
gateway = TelegramGateway()
|
|
attempts: list[tuple[str, dict]] = []
|
|
send_results = iter(
|
|
[
|
|
{"ok": True, "_awooop_delivery_status": "blocked_no_egress"},
|
|
{"ok": True, "_awooop_provider_send_performed": False},
|
|
]
|
|
)
|
|
|
|
async def structured_no_send(method: str, payload: dict) -> dict:
|
|
attempts.append((method, payload))
|
|
if method == "sendMessage":
|
|
return next(send_results)
|
|
return {"ok": True}
|
|
|
|
record_failure = AsyncMock(return_value=None)
|
|
shared_fallback = AsyncMock(return_value={"ok": True})
|
|
monkeypatch.setattr(gateway, "_send_request", structured_no_send)
|
|
monkeypatch.setattr(gateway, "_record_callback_reply_failure", record_failure)
|
|
monkeypatch.setattr(gateway, "send_alert_notification", shared_fallback)
|
|
|
|
await gateway._notify_approval_result(
|
|
message_id=_MESSAGE_ID,
|
|
incident_id="INC-1",
|
|
action="reject",
|
|
username="operator",
|
|
execution_triggered=False,
|
|
inbound_chat_id=_CHAT_ID,
|
|
inbound_message_id=_MESSAGE_ID,
|
|
)
|
|
|
|
send_attempts = [payload for method, payload in attempts if method == "sendMessage"]
|
|
assert len(send_attempts) == 2
|
|
for payload in send_attempts:
|
|
_assert_verified_callback_payload(payload)
|
|
shared_fallback.assert_not_awaited()
|
|
record_failure.assert_awaited_once()
|
|
assert record_failure.await_args.kwargs["parent_message_id"] == _MESSAGE_ID
|
|
assert record_failure.await_args.kwargs["incident_id"] == "INC-1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"action",
|
|
["tune", "snooze", "silence", "approve", "reject"],
|
|
)
|
|
async def test_callback_card_edits_use_only_verified_inbound_identity(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
action: str,
|
|
) -> None:
|
|
gateway = TelegramGateway()
|
|
send_request = AsyncMock(return_value=dict(_SENT_RECEIPT))
|
|
monkeypatch.setattr(gateway, "_send_request", send_request)
|
|
|
|
await gateway._update_message_after_action(
|
|
action=action,
|
|
username="operator",
|
|
original_text="original",
|
|
inbound_chat_id=_CHAT_ID,
|
|
inbound_message_id=_MESSAGE_ID,
|
|
)
|
|
|
|
assert [call.args[0] for call in send_request.await_args_list] == [
|
|
"editMessageReplyMarkup",
|
|
"editMessageText",
|
|
]
|
|
for call in send_request.await_args_list:
|
|
payload = call.args[1]
|
|
assert payload["chat_id"] == _CHAT_ID
|
|
assert payload["message_id"] == _MESSAGE_ID
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolved_state_guard_edits_only_verified_inbound_message(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
gateway = TelegramGateway()
|
|
send_request = AsyncMock(return_value={"ok": True})
|
|
monkeypatch.setattr(gateway, "_send_request", send_request)
|
|
monkeypatch.setattr(gateway, "_answer_callback", AsyncMock())
|
|
approval_id = UUID("55555555-5555-5555-5555-555555555555")
|
|
|
|
class _ApprovalService:
|
|
async def get_approval_by_id(self, _approval_id: UUID):
|
|
return SimpleNamespace(incident_id="INC-1")
|
|
|
|
class _IncidentRepo:
|
|
async def get_by_id(self, _incident_id: str):
|
|
return SimpleNamespace(
|
|
status=IncidentStatus.RESOLVED,
|
|
resolved_at=None,
|
|
)
|
|
|
|
monkeypatch.setattr(approval_db, "get_approval_service", lambda: _ApprovalService())
|
|
monkeypatch.setattr(
|
|
incident_repository,
|
|
"get_incident_repository",
|
|
lambda: _IncidentRepo(),
|
|
)
|
|
|
|
result = await gateway._check_incident_state_guard(
|
|
approval_id=str(approval_id),
|
|
callback_query_id="callback-1",
|
|
inbound_chat_id=_CHAT_ID,
|
|
inbound_message_id=_MESSAGE_ID,
|
|
original_text="original",
|
|
)
|
|
|
|
assert result == {
|
|
"blocked": True,
|
|
"reason": "already_resolved",
|
|
"approval_id": str(approval_id),
|
|
}
|
|
method, payload = send_request.await_args.args
|
|
assert method == "editMessageText"
|
|
assert payload["chat_id"] == _CHAT_ID
|
|
assert payload["message_id"] == _MESSAGE_ID
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approval_reply_failure_retries_exact_parent_then_records_no_send(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
gateway = TelegramGateway()
|
|
attempts: list[tuple[str, dict]] = []
|
|
|
|
async def fail_replies(method: str, payload: dict) -> dict:
|
|
attempts.append((method, payload))
|
|
if method == "sendMessage":
|
|
raise gateway_module.TelegramGatewayError("provider unavailable")
|
|
return {"ok": True}
|
|
|
|
record_failure = AsyncMock(return_value=None)
|
|
shared_fallback = AsyncMock(return_value={"ok": True})
|
|
monkeypatch.setattr(gateway, "_send_request", fail_replies)
|
|
monkeypatch.setattr(gateway, "_record_callback_reply_failure", record_failure)
|
|
monkeypatch.setattr(gateway, "send_alert_notification", shared_fallback)
|
|
|
|
await gateway._notify_approval_result(
|
|
message_id=_MESSAGE_ID,
|
|
incident_id="INC-1",
|
|
action="reject",
|
|
username="operator",
|
|
execution_triggered=False,
|
|
inbound_chat_id=_CHAT_ID,
|
|
inbound_message_id=_MESSAGE_ID,
|
|
)
|
|
|
|
send_attempts = [payload for method, payload in attempts if method == "sendMessage"]
|
|
assert len(send_attempts) == 2
|
|
for payload in send_attempts:
|
|
_assert_verified_callback_payload(payload)
|
|
extra = payload[gateway_module._AWOOOP_SOURCE_ENVELOPE_EXTRA_KEY]
|
|
merged = gateway_module._merge_outbound_source_envelope_extra({}, extra)
|
|
assert merged["callback_reply"]["parent_provider_message_id"] == str(
|
|
_MESSAGE_ID
|
|
)
|
|
assert "parse_mode" in send_attempts[0]
|
|
assert "parse_mode" not in send_attempts[1]
|
|
shared_fallback.assert_not_awaited()
|
|
record_failure.assert_awaited_once()
|
|
assert record_failure.await_args.kwargs["parent_message_id"] == _MESSAGE_ID
|
|
assert record_failure.await_args.kwargs["incident_id"] == "INC-1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approval_reply_parent_mismatch_is_durable_no_send_terminal(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
gateway = TelegramGateway()
|
|
send_request = AsyncMock(return_value={"ok": True})
|
|
record_failure = AsyncMock(return_value=None)
|
|
shared_fallback = AsyncMock(return_value={"ok": True})
|
|
monkeypatch.setattr(gateway, "_send_request", send_request)
|
|
monkeypatch.setattr(gateway, "_record_callback_reply_failure", record_failure)
|
|
monkeypatch.setattr(gateway, "send_alert_notification", shared_fallback)
|
|
|
|
await gateway._notify_approval_result(
|
|
message_id=_MESSAGE_ID + 1,
|
|
incident_id="INC-1",
|
|
action="approve",
|
|
username="operator",
|
|
execution_triggered=False,
|
|
inbound_chat_id=_CHAT_ID,
|
|
inbound_message_id=_MESSAGE_ID,
|
|
)
|
|
|
|
send_request.assert_not_awaited()
|
|
shared_fallback.assert_not_awaited()
|
|
record_failure.assert_awaited_once()
|
|
assert record_failure.await_args.kwargs["parent_message_id"] == _MESSAGE_ID
|
|
|
|
|
|
class _LifecycleRedis:
|
|
async def get(self, key: str) -> str:
|
|
if key.startswith("tg_approval:") or key.startswith("tg_msg:"):
|
|
return str(_MESSAGE_ID)
|
|
raise AssertionError(key)
|
|
|
|
async def set(self, *_args, **_kwargs) -> bool:
|
|
return True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_autonomous_lifecycle_threads_use_canonical_route_and_parent_receipt(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
gateway = TelegramGateway()
|
|
send_request = AsyncMock(return_value=dict(_SENT_RECEIPT))
|
|
monkeypatch.setattr(gateway, "_send_request", send_request)
|
|
monkeypatch.setattr(gateway_module, "get_redis", lambda: _LifecycleRedis())
|
|
|
|
assert await gateway.mark_auto_repaired("INC-1", "playbook", 10) is True
|
|
assert await gateway.append_incident_update("INC-2", "healthy") is True
|
|
assert (
|
|
await gateway.append_grouped_alert_digest(
|
|
incident_id="INC-3",
|
|
group_key="group-1",
|
|
digest_text="digest",
|
|
)
|
|
is True
|
|
)
|
|
|
|
send_payloads = [
|
|
call.args[1]
|
|
for call in send_request.await_args_list
|
|
if call.args[0] == "sendMessage"
|
|
]
|
|
assert len(send_payloads) == 3
|
|
parent_lookup_keys = []
|
|
for payload in send_payloads:
|
|
assert gateway_module._NON_MONITORING_INTERACTION_KEY not in payload
|
|
assert payload[gateway_module._CANONICAL_ROUTE_CONTEXT_KEY] == {
|
|
"product_id": "awoooi",
|
|
"signal_family": "incident_lifecycle",
|
|
"severity": "P1",
|
|
}
|
|
extra = payload[gateway_module._AWOOOP_SOURCE_ENVELOPE_EXTRA_KEY]
|
|
merged = gateway_module._merge_outbound_source_envelope_extra({}, extra)
|
|
receipt = merged["incident_lifecycle_parent_receipt"]
|
|
assert receipt["parent_provider_message_id"] == str(_MESSAGE_ID)
|
|
assert receipt["durable_mirror_required"] is True
|
|
parent_lookup_keys.append(receipt["parent_lookup_key"])
|
|
assert merged["source_refs"]["parent_provider_message_ids"] == [
|
|
str(_MESSAGE_ID)
|
|
]
|
|
assert parent_lookup_keys == [
|
|
"tg_approval:INC-1",
|
|
"tg_msg:INC-2",
|
|
"tg_msg:INC-3",
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_callback_fails_closed_before_dispatch_without_inbound_identity(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
gateway = TelegramGateway()
|
|
llm_handler = AsyncMock()
|
|
monkeypatch.setattr(gateway, "_handle_llm_action_callback", llm_handler)
|
|
|
|
result = await gateway.handle_callback(
|
|
callback_query_id="callback-1",
|
|
callback_data="la:short",
|
|
user_id=42,
|
|
message_id=_MESSAGE_ID,
|
|
chat_id=None,
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["reason"] == "missing_verified_interaction_context"
|
|
llm_handler.assert_not_awaited()
|