848 lines
30 KiB
Python
848 lines
30 KiB
Python
import threading
|
|
|
|
from fastapi import HTTPException
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
|
|
from services.interaction_service.app import app as interaction_app
|
|
from services.shared.db import get_session
|
|
from services.shared.sql_models import Customer, CustomerExternalIdentity, Interaction, TelegramMessageRow, TelegramThreadRow
|
|
from services.telegram_adapter_service import app as telegram_module
|
|
from services.telegram_adapter_service.app import app as telegram_app
|
|
|
|
|
|
def admin_headers():
|
|
return {"X-User": "admin", "X-Role": "admin"}
|
|
|
|
|
|
def operator_headers(user="operator"):
|
|
return {"X-User": user, "X-Role": "operator"}
|
|
|
|
|
|
def patch_interaction_request(monkeypatch):
|
|
def fake_request(method: str, path: str, *, payload: dict | None = None) -> dict:
|
|
session = get_session()
|
|
try:
|
|
interaction_id = path.split("/")[2]
|
|
interaction = session.execute(
|
|
select(Interaction).where(Interaction.interaction_id == interaction_id)
|
|
).scalar_one()
|
|
if path.endswith("/assign"):
|
|
interaction.assigned_to = payload["assignee"]
|
|
interaction.status = "in_progress"
|
|
elif path.endswith("/status"):
|
|
interaction.status = payload["status"]
|
|
elif path.endswith("/escalate"):
|
|
interaction.status = "escalated"
|
|
interaction.queue_id = payload["target_queue_id"]
|
|
interaction.updated_at = telegram_module.utc_now_iso()
|
|
session.commit()
|
|
return {
|
|
"interaction_id": interaction.interaction_id,
|
|
"status": interaction.status,
|
|
"assigned_to": interaction.assigned_to,
|
|
"queue_id": interaction.queue_id,
|
|
}
|
|
finally:
|
|
session.close()
|
|
|
|
monkeypatch.setattr(telegram_module, "_interaction_request", fake_request)
|
|
|
|
|
|
def test_manual_webhook_creates_thread_and_linked_interaction():
|
|
telegram_client = TestClient(telegram_app)
|
|
interaction_client = TestClient(interaction_app)
|
|
|
|
response = telegram_client.post(
|
|
"/integrations/telegram/webhook",
|
|
json={
|
|
"chat_id": "chat_1",
|
|
"text": "Привет из Telegram",
|
|
"customer_external_id": "telegram:user_1",
|
|
"payload": {
|
|
"username": "lead_one",
|
|
"first_name": "Lead",
|
|
"last_name": "One",
|
|
"telegram_user_id": "tg_user_1",
|
|
},
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["thread_id"].startswith("tgt_")
|
|
assert payload["interaction_id"].startswith("int_")
|
|
assert payload["direction"] == "inbound"
|
|
|
|
thread_list = telegram_client.get("/integrations/telegram/threads", headers=admin_headers())
|
|
assert thread_list.status_code == 200
|
|
listed_thread = next(item for item in thread_list.json() if item["thread_id"] == payload["thread_id"])
|
|
assert listed_thread["chat_id"] == "chat_1"
|
|
assert listed_thread["interaction_id"] == payload["interaction_id"]
|
|
|
|
interaction = interaction_client.get(f"/interactions/{payload['interaction_id']}")
|
|
assert interaction.status_code == 200
|
|
assert interaction.json()["channel"] == "telegram"
|
|
assert interaction.json()["customer_id"].startswith("cus_")
|
|
|
|
session = get_session()
|
|
try:
|
|
identity = session.execute(
|
|
select(CustomerExternalIdentity).where(
|
|
CustomerExternalIdentity.channel == "telegram",
|
|
CustomerExternalIdentity.external_subject == "tg_user_1",
|
|
)
|
|
).scalar_one_or_none()
|
|
assert identity is not None
|
|
assert identity.customer_id == interaction.json()["customer_id"]
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def test_repeated_inbound_reuses_same_thread_and_interaction():
|
|
telegram_client = TestClient(telegram_app)
|
|
|
|
first = telegram_client.post(
|
|
"/integrations/telegram/webhook",
|
|
json={"chat_id": "chat_reuse", "text": "Первое", "payload": {"username": "reuse_user"}},
|
|
)
|
|
second = telegram_client.post(
|
|
"/integrations/telegram/webhook",
|
|
json={"chat_id": "chat_reuse", "text": "Второе", "payload": {"username": "reuse_user"}},
|
|
)
|
|
|
|
assert first.status_code == 200
|
|
assert second.status_code == 200
|
|
assert first.json()["thread_id"] == second.json()["thread_id"]
|
|
assert first.json()["interaction_id"] == second.json()["interaction_id"]
|
|
|
|
|
|
def test_list_messages_requires_roles():
|
|
telegram_client = TestClient(telegram_app)
|
|
created = telegram_client.post(
|
|
"/integrations/telegram/webhook",
|
|
json={"chat_id": "chat_messages_auth", "text": "Auth check"},
|
|
)
|
|
assert created.status_code == 200
|
|
|
|
anonymous = telegram_client.get(
|
|
"/integrations/telegram/messages",
|
|
params={"chat_id": "chat_messages_auth"},
|
|
)
|
|
assert anonymous.status_code == 403
|
|
|
|
admin = telegram_client.get(
|
|
"/integrations/telegram/messages",
|
|
params={"chat_id": "chat_messages_auth"},
|
|
headers=admin_headers(),
|
|
)
|
|
assert admin.status_code == 200
|
|
assert len(admin.json()) == 1
|
|
|
|
operator = telegram_client.get(
|
|
"/integrations/telegram/messages",
|
|
params={"chat_id": "chat_messages_auth"},
|
|
headers=operator_headers(),
|
|
)
|
|
assert operator.status_code == 200
|
|
assert len(operator.json()) == 1
|
|
|
|
|
|
def test_closed_interaction_is_reactivated_on_new_inbound():
|
|
telegram_client = TestClient(telegram_app)
|
|
interaction_client = TestClient(interaction_app)
|
|
|
|
created = telegram_client.post(
|
|
"/integrations/telegram/webhook",
|
|
json={"chat_id": "chat_reactivate", "text": "Первое сообщение"},
|
|
)
|
|
assert created.status_code == 200
|
|
interaction_id = created.json()["interaction_id"]
|
|
thread_id = created.json()["thread_id"]
|
|
|
|
session = get_session()
|
|
try:
|
|
thread = session.execute(select(TelegramThreadRow).where(TelegramThreadRow.thread_id == thread_id)).scalar_one()
|
|
interaction = session.execute(select(Interaction).where(Interaction.interaction_id == interaction_id)).scalar_one()
|
|
thread.status = "closed"
|
|
interaction.status = "closed"
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
reopened = telegram_client.post(
|
|
"/integrations/telegram/webhook",
|
|
json={"chat_id": "chat_reactivate", "text": "Новое сообщение после close"},
|
|
)
|
|
assert reopened.status_code == 200
|
|
assert reopened.json()["thread_id"] == thread_id
|
|
assert reopened.json()["interaction_id"] == interaction_id
|
|
|
|
interaction = interaction_client.get(f"/interactions/{interaction_id}")
|
|
assert interaction.status_code == 200
|
|
assert interaction.json()["status"] == "new"
|
|
|
|
|
|
def test_operator_can_claim_thread_and_foreign_operator_cannot_reply(monkeypatch):
|
|
patch_interaction_request(monkeypatch)
|
|
telegram_client = TestClient(telegram_app)
|
|
|
|
created = telegram_client.post(
|
|
"/integrations/telegram/webhook",
|
|
json={"chat_id": "chat_claim", "text": "Нужна помощь"},
|
|
)
|
|
thread_id = created.json()["thread_id"]
|
|
|
|
claimed = telegram_client.post(
|
|
f"/integrations/telegram/threads/{thread_id}/claim",
|
|
headers=operator_headers("operator_a"),
|
|
)
|
|
assert claimed.status_code == 200
|
|
assert claimed.json()["claimed_by_user"] == "operator_a"
|
|
assert claimed.json()["status"] == "in_progress"
|
|
|
|
forbidden = telegram_client.post(
|
|
f"/integrations/telegram/threads/{thread_id}/messages",
|
|
headers=operator_headers("operator_b"),
|
|
json={"text": "Чужой reply"},
|
|
)
|
|
assert forbidden.status_code == 403
|
|
|
|
|
|
def test_operator_reply_sends_outbound_message_and_persists_row(monkeypatch):
|
|
patch_interaction_request(monkeypatch)
|
|
started: list[str] = []
|
|
monkeypatch.setattr(telegram_module, "_start_telegram_reply_delivery", lambda message_id: started.append(message_id))
|
|
|
|
telegram_client = TestClient(telegram_app)
|
|
created = telegram_client.post(
|
|
"/integrations/telegram/webhook",
|
|
json={"chat_id": "chat_reply", "text": "Вопрос клиента"},
|
|
)
|
|
thread_id = created.json()["thread_id"]
|
|
|
|
claimed = telegram_client.post(
|
|
f"/integrations/telegram/threads/{thread_id}/claim",
|
|
headers=operator_headers("operator"),
|
|
)
|
|
assert claimed.status_code == 200
|
|
|
|
sent = telegram_client.post(
|
|
f"/integrations/telegram/threads/{thread_id}/messages",
|
|
headers=operator_headers("operator"),
|
|
json={"text": "Ответ оператора"},
|
|
)
|
|
assert sent.status_code == 200
|
|
payload = sent.json()
|
|
assert payload["direction"] == "outbound"
|
|
assert payload["delivery_status"] == "pending"
|
|
assert payload["telegram_message_id_external"] is None
|
|
assert payload["operator_user"] == "operator"
|
|
assert payload["author_type"] == "human"
|
|
assert payload["author_id"] == "operator"
|
|
assert started == [payload["message_id"]]
|
|
|
|
messages = telegram_client.get(
|
|
f"/integrations/telegram/threads/{thread_id}/messages",
|
|
headers=admin_headers(),
|
|
)
|
|
assert messages.status_code == 200
|
|
assert len(messages.json()) == 2
|
|
assert messages.json()[-1]["delivery_status"] == "pending"
|
|
assert messages.json()[-1]["text"] == "Ответ оператора"
|
|
monkeypatch.setattr(
|
|
telegram_module,
|
|
"_send_telegram_message",
|
|
lambda chat_id, text: {"ok": True, "result": {"message_id": 7788, "chat": {"id": chat_id}, "text": text}},
|
|
)
|
|
telegram_module._deliver_pending_telegram_reply(payload["message_id"])
|
|
|
|
delivered = telegram_client.get(
|
|
f"/integrations/telegram/threads/{thread_id}/messages",
|
|
headers=admin_headers(),
|
|
)
|
|
assert delivered.status_code == 200
|
|
assert delivered.json()[-1]["delivery_status"] == "sent"
|
|
assert delivered.json()[-1]["telegram_message_id_external"] == "7788"
|
|
|
|
|
|
def test_returning_human_owned_thread_to_ai_reopens_ai_flow(monkeypatch):
|
|
monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
|
|
enqueue_calls: list[tuple[str, str | None]] = []
|
|
monkeypatch.setattr(
|
|
telegram_module,
|
|
"_ai_enqueue_request",
|
|
lambda thread_id, trigger_message_id: enqueue_calls.append((thread_id, trigger_message_id)),
|
|
)
|
|
patch_interaction_request(monkeypatch)
|
|
|
|
telegram_client = TestClient(telegram_app)
|
|
created = telegram_client.post(
|
|
"/integrations/telegram/webhook",
|
|
json={
|
|
"chat_id": "chat_return_ai",
|
|
"text": "Первое сообщение",
|
|
"payload": {"telegram_user_id": "tg_return_ai", "username": "return_ai_user"},
|
|
},
|
|
)
|
|
assert created.status_code == 200
|
|
thread_id = created.json()["thread_id"]
|
|
first_message_id = created.json()["message_id"]
|
|
enqueue_calls.clear()
|
|
|
|
claimed = telegram_client.post(
|
|
f"/integrations/telegram/threads/{thread_id}/claim",
|
|
headers=operator_headers("operator_ai"),
|
|
)
|
|
assert claimed.status_code == 200
|
|
assert claimed.json()["ai_state"] == "human_owned"
|
|
|
|
returned = telegram_client.post(
|
|
f"/integrations/telegram/threads/{thread_id}/return-to-ai",
|
|
headers=operator_headers("operator_ai"),
|
|
)
|
|
assert returned.status_code == 200
|
|
assert returned.json()["ai_state"] == "queued"
|
|
assert returned.json()["claimed_by_user"] is None
|
|
assert returned.json()["status"] == "new"
|
|
assert enqueue_calls == []
|
|
|
|
follow_up = telegram_client.post(
|
|
"/integrations/telegram/webhook",
|
|
json={
|
|
"chat_id": "chat_return_ai",
|
|
"text": "Новое сообщение после возврата AI",
|
|
"payload": {"telegram_user_id": "tg_return_ai", "username": "return_ai_user"},
|
|
},
|
|
)
|
|
assert follow_up.status_code == 200
|
|
assert follow_up.json()["thread_id"] == thread_id
|
|
assert follow_up.json()["message_id"] != first_message_id
|
|
assert enqueue_calls == [(thread_id, follow_up.json()["message_id"])]
|
|
|
|
session = get_session()
|
|
try:
|
|
thread = session.execute(
|
|
select(TelegramThreadRow).where(TelegramThreadRow.thread_id == thread_id)
|
|
).scalar_one()
|
|
interaction = session.execute(
|
|
select(Interaction).where(Interaction.interaction_id == thread.interaction_id)
|
|
).scalar_one()
|
|
assert thread.ai_state == "queued"
|
|
assert thread.claimed_by_user is None
|
|
assert thread.status == "new"
|
|
assert interaction.status == "new"
|
|
assert interaction.assigned_to is None
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def test_failed_reply_is_scheduled_for_retry_and_then_sent(monkeypatch):
|
|
patch_interaction_request(monkeypatch)
|
|
monkeypatch.setenv("TELEGRAM_REPLY_DELIVERY_POLL_SECONDS", "3600")
|
|
monkeypatch.setattr(telegram_module, "_start_telegram_reply_delivery", lambda message_id: None)
|
|
|
|
telegram_client = TestClient(telegram_app)
|
|
created = telegram_client.post(
|
|
"/integrations/telegram/webhook",
|
|
json={"chat_id": "chat_retry", "text": "Retry me"},
|
|
)
|
|
thread_id = created.json()["thread_id"]
|
|
|
|
claimed = telegram_client.post(
|
|
f"/integrations/telegram/threads/{thread_id}/claim",
|
|
headers=operator_headers("operator_retry"),
|
|
)
|
|
assert claimed.status_code == 200
|
|
|
|
reply = telegram_client.post(
|
|
f"/integrations/telegram/threads/{thread_id}/messages",
|
|
headers=operator_headers("operator_retry"),
|
|
json={"text": "Please retry"},
|
|
)
|
|
assert reply.status_code == 200
|
|
message_id = reply.json()["message_id"]
|
|
|
|
attempts: list[int] = []
|
|
|
|
def flaky_send(chat_id: str, text: str) -> dict:
|
|
attempts.append(len(attempts) + 1)
|
|
if len(attempts) == 1:
|
|
raise HTTPException(status_code=502, detail="temporary send failure")
|
|
return {"ok": True, "result": {"message_id": 8899, "chat": {"id": chat_id}, "text": text}}
|
|
|
|
monkeypatch.setattr(telegram_module, "_send_telegram_message", flaky_send)
|
|
|
|
assert telegram_module._deliver_pending_telegram_reply(message_id) is False
|
|
|
|
session = get_session()
|
|
try:
|
|
row = session.execute(
|
|
select(TelegramMessageRow).where(TelegramMessageRow.message_id == message_id)
|
|
).scalar_one()
|
|
assert row.delivery_status == "retrying"
|
|
assert row.delivery_attempts == 1
|
|
assert row.next_delivery_attempt_at is not None
|
|
assert row.delivery_locked_until is None
|
|
assert row.last_delivery_error == "temporary send failure"
|
|
row.next_delivery_attempt_at = telegram_module.utc_now_iso()
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
assert telegram_module._deliver_pending_telegram_reply(message_id) is True
|
|
assert attempts == [1, 2]
|
|
|
|
session = get_session()
|
|
try:
|
|
row = session.execute(
|
|
select(TelegramMessageRow).where(TelegramMessageRow.message_id == message_id)
|
|
).scalar_one()
|
|
assert row.delivery_status == "sent"
|
|
assert row.delivery_attempts == 2
|
|
assert row.telegram_message_id_external == "8899"
|
|
assert row.next_delivery_attempt_at is None
|
|
assert row.delivery_locked_until is None
|
|
assert row.last_delivery_error is None
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def test_worker_batch_recovers_stale_sending_reply(monkeypatch):
|
|
patch_interaction_request(monkeypatch)
|
|
monkeypatch.setenv("TELEGRAM_REPLY_DELIVERY_POLL_SECONDS", "3600")
|
|
monkeypatch.setattr(telegram_module, "_start_telegram_reply_delivery", lambda message_id: None)
|
|
monkeypatch.setattr(
|
|
telegram_module,
|
|
"_send_telegram_message",
|
|
lambda chat_id, text: {"ok": True, "result": {"message_id": 9901, "chat": {"id": chat_id}, "text": text}},
|
|
)
|
|
|
|
telegram_client = TestClient(telegram_app)
|
|
created = telegram_client.post(
|
|
"/integrations/telegram/webhook",
|
|
json={"chat_id": "chat_stale_worker", "text": "Recover me"},
|
|
)
|
|
thread_id = created.json()["thread_id"]
|
|
|
|
claimed = telegram_client.post(
|
|
f"/integrations/telegram/threads/{thread_id}/claim",
|
|
headers=operator_headers("operator_stale"),
|
|
)
|
|
assert claimed.status_code == 200
|
|
|
|
reply = telegram_client.post(
|
|
f"/integrations/telegram/threads/{thread_id}/messages",
|
|
headers=operator_headers("operator_stale"),
|
|
json={"text": "Stale sending"},
|
|
)
|
|
assert reply.status_code == 200
|
|
message_id = reply.json()["message_id"]
|
|
|
|
session = get_session()
|
|
try:
|
|
row = session.execute(
|
|
select(TelegramMessageRow).where(TelegramMessageRow.message_id == message_id)
|
|
).scalar_one()
|
|
row.delivery_status = "sending"
|
|
row.delivery_locked_until = "2000-01-01T00:00:00+00:00"
|
|
row.next_delivery_attempt_at = telegram_module.utc_now_iso()
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
processed = telegram_module._process_due_telegram_reply_batch(batch_size=1)
|
|
assert processed == 1
|
|
|
|
session = get_session()
|
|
try:
|
|
row = session.execute(
|
|
select(TelegramMessageRow).where(TelegramMessageRow.message_id == message_id)
|
|
).scalar_one()
|
|
assert row.delivery_status == "sent"
|
|
assert row.delivery_attempts == 1
|
|
assert row.telegram_message_id_external == "9901"
|
|
assert row.delivery_locked_until is None
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def test_bot_webhook_secret_and_unsupported_content_placeholder(monkeypatch):
|
|
monkeypatch.setenv("TELEGRAM_WEBHOOK_SECRET", "secret-15")
|
|
telegram_client = TestClient(telegram_app)
|
|
|
|
forbidden = telegram_client.post(
|
|
"/integrations/telegram/bot/webhook",
|
|
json={"message": {"message_id": 1, "chat": {"id": 44}}},
|
|
)
|
|
assert forbidden.status_code == 403
|
|
|
|
accepted = telegram_client.post(
|
|
"/integrations/telegram/bot/webhook",
|
|
headers={"X-Telegram-Bot-Api-Secret-Token": "secret-15"},
|
|
json={
|
|
"message": {
|
|
"message_id": 9,
|
|
"chat": {"id": 555},
|
|
"from": {"id": 77, "username": "photo_user"},
|
|
"photo": [{"file_id": "abc"}],
|
|
}
|
|
},
|
|
)
|
|
assert accepted.status_code == 200
|
|
thread_id = accepted.json()["thread_id"]
|
|
|
|
messages = telegram_client.get(
|
|
f"/integrations/telegram/threads/{thread_id}/messages",
|
|
headers=admin_headers(),
|
|
)
|
|
assert messages.status_code == 200
|
|
assert messages.json()[0]["direction"] == "system"
|
|
assert "[Unsupported Telegram content: photo]" in messages.json()[0]["text"]
|
|
|
|
|
|
def test_bot_webhook_unsupported_content_does_not_enqueue_ai(monkeypatch):
|
|
monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
|
|
enqueue_calls: list[tuple[str, str | None]] = []
|
|
monkeypatch.setattr(
|
|
telegram_module,
|
|
"_ai_enqueue_request",
|
|
lambda thread_id, trigger_message_id: enqueue_calls.append((thread_id, trigger_message_id)),
|
|
)
|
|
telegram_client = TestClient(telegram_app)
|
|
|
|
accepted = telegram_client.post(
|
|
"/integrations/telegram/bot/webhook",
|
|
json={
|
|
"message": {
|
|
"message_id": 19,
|
|
"chat": {"id": 559},
|
|
"from": {"id": 79, "username": "photo_user"},
|
|
"photo": [{"file_id": "abc"}],
|
|
}
|
|
},
|
|
)
|
|
|
|
assert accepted.status_code == 200
|
|
assert enqueue_calls == []
|
|
|
|
session = get_session()
|
|
try:
|
|
thread = session.execute(
|
|
select(TelegramThreadRow).where(TelegramThreadRow.thread_id == accepted.json()["thread_id"])
|
|
).scalar_one()
|
|
assert thread.ai_state is None
|
|
assert thread.ai_handoff_reason is None
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def test_bot_webhook_contact_saves_customer_phone(monkeypatch):
|
|
monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
|
|
enqueue_calls: list[tuple[str, str | None]] = []
|
|
monkeypatch.setattr(
|
|
telegram_module,
|
|
"_ai_enqueue_request",
|
|
lambda thread_id, trigger_message_id: enqueue_calls.append((thread_id, trigger_message_id)),
|
|
)
|
|
telegram_client = TestClient(telegram_app)
|
|
|
|
accepted = telegram_client.post(
|
|
"/integrations/telegram/bot/webhook",
|
|
json={
|
|
"message": {
|
|
"message_id": 21,
|
|
"chat": {"id": 562},
|
|
"from": {"id": 81, "username": "contact_user", "first_name": "Contact"},
|
|
"contact": {"phone_number": "+7 (777) 123-45-67", "user_id": 81},
|
|
}
|
|
},
|
|
)
|
|
|
|
assert accepted.status_code == 200
|
|
assert enqueue_calls == []
|
|
|
|
session = get_session()
|
|
try:
|
|
thread = session.execute(
|
|
select(TelegramThreadRow).where(TelegramThreadRow.thread_id == accepted.json()["thread_id"])
|
|
).scalar_one()
|
|
interaction = session.execute(
|
|
select(Interaction).where(Interaction.interaction_id == thread.interaction_id)
|
|
).scalar_one()
|
|
customer = session.execute(
|
|
select(Customer).where(Customer.customer_id == interaction.customer_id)
|
|
).scalar_one_or_none()
|
|
finally:
|
|
session.close()
|
|
|
|
assert thread.ai_state is None
|
|
assert customer is not None
|
|
assert customer.preferred_phone == "+77771234567"
|
|
assert "+77771234567" in customer.phones_json
|
|
|
|
messages = telegram_client.get(
|
|
f"/integrations/telegram/threads/{accepted.json()['thread_id']}/messages",
|
|
headers=admin_headers(),
|
|
)
|
|
assert messages.status_code == 200
|
|
assert messages.json()[0]["direction"] == "system"
|
|
assert messages.json()[0]["text"] == "[Shared Telegram contact]"
|
|
|
|
|
|
def test_system_message_preserves_existing_handoff_state(monkeypatch):
|
|
monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
|
|
enqueue_calls: list[tuple[str, str | None]] = []
|
|
monkeypatch.setattr(
|
|
telegram_module,
|
|
"_ai_enqueue_request",
|
|
lambda thread_id, trigger_message_id: enqueue_calls.append((thread_id, trigger_message_id)),
|
|
)
|
|
telegram_client = TestClient(telegram_app)
|
|
|
|
created = telegram_client.post(
|
|
"/integrations/telegram/webhook",
|
|
json={"chat_id": "chat_system_handoff", "text": "Need human"},
|
|
)
|
|
assert created.status_code == 200
|
|
thread_id = created.json()["thread_id"]
|
|
|
|
session = get_session()
|
|
try:
|
|
thread = session.execute(select(TelegramThreadRow).where(TelegramThreadRow.thread_id == thread_id)).scalar_one()
|
|
thread.ai_state = "handoff_required"
|
|
thread.ai_handoff_reason = "Existing handoff"
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
enqueue_calls.clear()
|
|
accepted = telegram_client.post(
|
|
"/integrations/telegram/bot/webhook",
|
|
json={
|
|
"message": {
|
|
"message_id": 29,
|
|
"chat": {"id": "chat_system_handoff"},
|
|
"from": {"id": 93, "username": "photo_user"},
|
|
"photo": [{"file_id": "abc"}],
|
|
}
|
|
},
|
|
)
|
|
assert accepted.status_code == 200
|
|
assert enqueue_calls == []
|
|
|
|
session = get_session()
|
|
try:
|
|
thread = session.execute(select(TelegramThreadRow).where(TelegramThreadRow.thread_id == thread_id)).scalar_one()
|
|
assert thread.ai_state == "handoff_required"
|
|
assert thread.ai_handoff_reason == "Existing handoff"
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def test_bot_webhook_duplicate_message_id_is_idempotent():
|
|
telegram_client = TestClient(telegram_app)
|
|
payload = {
|
|
"update_id": 9015,
|
|
"message": {
|
|
"message_id": 5150,
|
|
"chat": {"id": 55150},
|
|
"from": {"id": 7715, "username": "duplicate_user"},
|
|
"text": "Duplicate payload",
|
|
},
|
|
}
|
|
|
|
first = telegram_client.post("/integrations/telegram/bot/webhook", json=payload)
|
|
second = telegram_client.post("/integrations/telegram/bot/webhook", json=payload)
|
|
|
|
assert first.status_code == 200
|
|
assert second.status_code == 200
|
|
assert first.json()["thread_id"] == second.json()["thread_id"]
|
|
assert first.json()["interaction_id"] == second.json()["interaction_id"]
|
|
assert first.json()["message_id"] == second.json()["message_id"]
|
|
assert first.json()["thread_created"] is True
|
|
assert second.json()["thread_created"] is False
|
|
|
|
messages = telegram_client.get(
|
|
f"/integrations/telegram/threads/{first.json()['thread_id']}/messages",
|
|
headers=admin_headers(),
|
|
)
|
|
assert messages.status_code == 200
|
|
assert len(messages.json()) == 1
|
|
|
|
|
|
def test_reply_after_close_is_conflict_and_close_clears_claim(monkeypatch):
|
|
patch_interaction_request(monkeypatch)
|
|
monkeypatch.setattr(
|
|
telegram_module,
|
|
"_send_telegram_message",
|
|
lambda chat_id, text: (_ for _ in ()).throw(AssertionError("sendMessage should not be called")),
|
|
)
|
|
|
|
telegram_client = TestClient(telegram_app)
|
|
created = telegram_client.post(
|
|
"/integrations/telegram/webhook",
|
|
json={"chat_id": "chat_close_reply", "text": "Need closure"},
|
|
)
|
|
thread_id = created.json()["thread_id"]
|
|
|
|
claimed = telegram_client.post(
|
|
f"/integrations/telegram/threads/{thread_id}/claim",
|
|
headers=operator_headers("operator_close"),
|
|
)
|
|
assert claimed.status_code == 200
|
|
|
|
closed = telegram_client.post(
|
|
f"/integrations/telegram/threads/{thread_id}/close",
|
|
headers=operator_headers("operator_close"),
|
|
)
|
|
assert closed.status_code == 200
|
|
assert closed.json()["status"] == "closed"
|
|
assert closed.json()["claimed_by_user"] is None
|
|
assert closed.json()["claimed_at"] is None
|
|
|
|
reply = telegram_client.post(
|
|
f"/integrations/telegram/threads/{thread_id}/messages",
|
|
headers=operator_headers("operator_close"),
|
|
json={"text": "Should fail"},
|
|
)
|
|
assert reply.status_code == 409
|
|
|
|
session = get_session()
|
|
try:
|
|
thread = session.execute(select(TelegramThreadRow).where(TelegramThreadRow.thread_id == thread_id)).scalar_one()
|
|
messages = session.execute(
|
|
select(TelegramMessageRow).where(TelegramMessageRow.thread_id == thread_id)
|
|
).scalars().all()
|
|
assert thread.status == "closed"
|
|
assert thread.claimed_by_user is None
|
|
assert thread.claimed_at is None
|
|
assert len(messages) == 1
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def test_ai_handoff_is_rejected_for_human_owned_thread(monkeypatch):
|
|
patch_interaction_request(monkeypatch)
|
|
telegram_client = TestClient(telegram_app)
|
|
|
|
created = telegram_client.post(
|
|
"/integrations/telegram/webhook",
|
|
json={"chat_id": "chat_ai_handoff_locked", "text": "Need operator"},
|
|
)
|
|
assert created.status_code == 200
|
|
thread_id = created.json()["thread_id"]
|
|
|
|
claimed = telegram_client.post(
|
|
f"/integrations/telegram/threads/{thread_id}/claim",
|
|
headers=operator_headers("operator_handoff_lock"),
|
|
)
|
|
assert claimed.status_code == 200
|
|
assert claimed.json()["ai_state"] == "human_owned"
|
|
|
|
handoff = telegram_client.post(
|
|
f"/integrations/telegram/threads/{thread_id}/ai/handoff",
|
|
headers=admin_headers(),
|
|
json={"reason": "Late AI handoff"},
|
|
)
|
|
assert handoff.status_code == 409
|
|
assert handoff.json()["detail"] == "Telegram thread is owned by a human operator"
|
|
|
|
session = get_session()
|
|
try:
|
|
thread = session.execute(select(TelegramThreadRow).where(TelegramThreadRow.thread_id == thread_id)).scalar_one()
|
|
assert thread.ai_state == "human_owned"
|
|
assert thread.claimed_by_user == "operator_handoff_lock"
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def test_claim_race_returns_single_winner(monkeypatch):
|
|
entered_assign = threading.Event()
|
|
release_assign = threading.Event()
|
|
assign_calls: list[str] = []
|
|
assign_lock = threading.Lock()
|
|
|
|
def fake_request(method: str, path: str, *, payload: dict | None = None) -> dict:
|
|
if path.endswith("/assign"):
|
|
with assign_lock:
|
|
assign_calls.append(payload["assignee"])
|
|
if len(assign_calls) == 1:
|
|
entered_assign.set()
|
|
assert release_assign.wait(5)
|
|
|
|
session = get_session()
|
|
try:
|
|
interaction_id = path.split("/")[2]
|
|
interaction = session.execute(
|
|
select(Interaction).where(Interaction.interaction_id == interaction_id)
|
|
).scalar_one()
|
|
if path.endswith("/assign"):
|
|
interaction.assigned_to = payload["assignee"]
|
|
interaction.status = "in_progress"
|
|
elif path.endswith("/status"):
|
|
interaction.status = payload["status"]
|
|
elif path.endswith("/escalate"):
|
|
interaction.status = "escalated"
|
|
interaction.queue_id = payload["target_queue_id"]
|
|
interaction.updated_at = telegram_module.utc_now_iso()
|
|
session.commit()
|
|
return {
|
|
"interaction_id": interaction.interaction_id,
|
|
"status": interaction.status,
|
|
"assigned_to": interaction.assigned_to,
|
|
"queue_id": interaction.queue_id,
|
|
}
|
|
finally:
|
|
session.close()
|
|
|
|
monkeypatch.setattr(telegram_module, "_interaction_request", fake_request)
|
|
|
|
telegram_client = TestClient(telegram_app)
|
|
created = telegram_client.post(
|
|
"/integrations/telegram/webhook",
|
|
json={"chat_id": "chat_claim_race", "text": "Race me"},
|
|
)
|
|
thread_id = created.json()["thread_id"]
|
|
|
|
winner: dict[str, object] = {}
|
|
|
|
def run_first_claim() -> None:
|
|
first_client = TestClient(telegram_app)
|
|
try:
|
|
winner["response"] = first_client.post(
|
|
f"/integrations/telegram/threads/{thread_id}/claim",
|
|
headers=operator_headers("operator_a"),
|
|
)
|
|
finally:
|
|
first_client.close()
|
|
|
|
claim_thread = threading.Thread(target=run_first_claim)
|
|
claim_thread.start()
|
|
|
|
assert entered_assign.wait(5)
|
|
|
|
loser_client = TestClient(telegram_app)
|
|
try:
|
|
loser = loser_client.post(
|
|
f"/integrations/telegram/threads/{thread_id}/claim",
|
|
headers=operator_headers("operator_b"),
|
|
)
|
|
finally:
|
|
loser_client.close()
|
|
|
|
release_assign.set()
|
|
claim_thread.join(timeout=5)
|
|
|
|
first = winner["response"]
|
|
assert getattr(first, "status_code") == 200
|
|
assert loser.status_code == 409
|
|
assert assign_calls == ["operator_a"]
|
|
|
|
session = get_session()
|
|
try:
|
|
thread = session.execute(select(TelegramThreadRow).where(TelegramThreadRow.thread_id == thread_id)).scalar_one()
|
|
assert thread.claimed_by_user == "operator_a"
|
|
assert thread.status == "in_progress"
|
|
finally:
|
|
session.close()
|