You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
100 lines
4.8 KiB
Python
100 lines
4.8 KiB
Python
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from app.integrations.telegram_satellite_service import TelegramSatelliteService
|
|
|
|
|
|
class _DummySession:
|
|
def close(self):
|
|
return None
|
|
|
|
|
|
class TelegramMultimodalTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_process_message_uses_extracted_image_message(self):
|
|
service = TelegramSatelliteService("token-teste")
|
|
tools_db = _DummySession()
|
|
mock_db = _DummySession()
|
|
|
|
with patch("app.integrations.telegram_satellite_service.SessionLocal", return_value=tools_db), patch(
|
|
"app.integrations.telegram_satellite_service.SessionMockLocal",
|
|
return_value=mock_db,
|
|
), patch("app.integrations.telegram_satellite_service.UserService") as user_service_cls, patch(
|
|
"app.integrations.telegram_satellite_service.OrquestradorService"
|
|
) as orchestrator_cls, patch.object(
|
|
service,
|
|
"_build_orchestration_message_from_image",
|
|
AsyncMock(return_value="[imagem recebida no telegram]\nDados extraidos da imagem: Registrar multa de aluguel: placa ABC1D23; valor 293,47; auto_infracao A123456."),
|
|
):
|
|
user_service_cls.return_value.get_or_create.return_value = SimpleNamespace(id=7)
|
|
orchestrator_cls.return_value.handle_message = AsyncMock(return_value="ok")
|
|
|
|
answer = await service._process_message(
|
|
text="segue a multa",
|
|
sender={"id": 99, "first_name": "Vitor"},
|
|
chat_id=99,
|
|
image_attachments=[{"mime_type": "image/jpeg", "data": b"123"}],
|
|
)
|
|
|
|
self.assertEqual(answer, "ok")
|
|
orchestrator_cls.return_value.handle_message.assert_awaited_once()
|
|
kwargs = orchestrator_cls.return_value.handle_message.await_args.kwargs
|
|
self.assertIn("Registrar multa de aluguel", kwargs["message"])
|
|
self.assertEqual(kwargs["user_id"], 7)
|
|
|
|
async def test_process_message_returns_direct_failure_for_unreadable_image(self):
|
|
service = TelegramSatelliteService("token-teste")
|
|
tools_db = _DummySession()
|
|
mock_db = _DummySession()
|
|
|
|
with patch("app.integrations.telegram_satellite_service.SessionLocal", return_value=tools_db), patch(
|
|
"app.integrations.telegram_satellite_service.SessionMockLocal",
|
|
return_value=mock_db,
|
|
), patch("app.integrations.telegram_satellite_service.UserService") as user_service_cls, patch(
|
|
"app.integrations.telegram_satellite_service.OrquestradorService"
|
|
) as orchestrator_cls, patch.object(
|
|
service,
|
|
"_build_orchestration_message_from_image",
|
|
AsyncMock(return_value="Nao consegui identificar os dados da imagem. Descreva o documento ou envie uma foto mais nitida."),
|
|
):
|
|
user_service_cls.return_value.get_or_create.return_value = SimpleNamespace(id=7)
|
|
orchestrator_cls.return_value.handle_message = AsyncMock()
|
|
|
|
answer = await service._process_message(
|
|
text="",
|
|
sender={"id": 99},
|
|
chat_id=99,
|
|
image_attachments=[{"mime_type": "image/jpeg", "data": b"123"}],
|
|
)
|
|
|
|
self.assertIn("Nao consegui identificar os dados da imagem", answer)
|
|
self.assertFalse(orchestrator_cls.return_value.handle_message.await_count)
|
|
|
|
async def test_process_message_returns_direct_failure_for_receipt_without_watermark(self):
|
|
service = TelegramSatelliteService("token-teste")
|
|
tools_db = _DummySession()
|
|
mock_db = _DummySession()
|
|
|
|
with patch("app.integrations.telegram_satellite_service.SessionLocal", return_value=tools_db), patch(
|
|
"app.integrations.telegram_satellite_service.SessionMockLocal",
|
|
return_value=mock_db,
|
|
), patch("app.integrations.telegram_satellite_service.UserService") as user_service_cls, patch(
|
|
"app.integrations.telegram_satellite_service.OrquestradorService"
|
|
) as orchestrator_cls, patch.object(
|
|
service,
|
|
"_build_orchestration_message_from_image",
|
|
AsyncMock(return_value="O comprovante enviado nao e valido. Envie um comprovante valido com a marca d'agua SysaltiIA visivel."),
|
|
):
|
|
user_service_cls.return_value.get_or_create.return_value = SimpleNamespace(id=7)
|
|
orchestrator_cls.return_value.handle_message = AsyncMock()
|
|
|
|
answer = await service._process_message(
|
|
text="segue o comprovante",
|
|
sender={"id": 99},
|
|
chat_id=99,
|
|
image_attachments=[{"mime_type": "image/jpeg", "data": b"123"}],
|
|
)
|
|
|
|
self.assertIn("marca d'agua SysaltiIA visivel", answer)
|
|
self.assertFalse(orchestrator_cls.return_value.handle_message.await_count)
|