Files
site_aegisone/max_bot/tests/test_handlers.py
T
angel 72b6879f4b v1.7.0: refactor max_bot to flat structure, add VCF+UserModel+NLP history context, portal pages and proxy fixes
- Refactored max_bot from nested packages to flat module structure
- Q2: Extended BotUser model (patronymic, email, org, address, vcf_raw, contact_hash, phone_verified, email_verified, last_interaction, total_conversations, total_tickets)
- Q2: VCF parser (FN, N, TEL, EMAIL, ORG, ADR), upsert on re-contact, NLP history context (_get_user_history_context -> YandexGPT)
- Q1: Broadcast preview modal with 10s confirmation timer
- Q3: CSS var(--white)->var(--bg-card), var(--text)->var(--text-primary)
- Q4: bot_settings showNotification(), editable max_bot_id
- Q5: Webhook secret passthrough via X-Max-Bot-Api-Secret
- Masking sensitive keys, dialog_cleared handler, migrate via _add_column_if_not_exists()
- Rate limit (asyncio.sleep 0.5 per 10), dead code removed, conv.intent context in contact.py
- Portal pages: bot_consent, bot_kb (edit), bot_settings, bot_test, bot_tickets, portal_settings
- Tests: 21/21 passing, added test_yandex_gpt.py, test_email_sender.py
- Deploy: deploy_full.sh, schema.sql, seed_knowledge_base.sql
2026-05-29 02:30:30 +03:00

175 lines
6.1 KiB
Python

import pytest
from unittest.mock import AsyncMock, MagicMock, patch, ANY
from app.handlers.greeting import analyze_and_respond
from app.handlers.consent import handle_consent_response
from app.handlers.contact import handle_manual_contact
from app.handlers.inquiry import handle_inquiry
from app.handlers.handoff import handle_callback
from app.models import BotConversation, BotUser, BotTicket, BotKnowledgeBase
def make_mock_db():
db = AsyncMock()
result = MagicMock()
result.scalar_one_or_none.return_value = None
result.scalars.return_value.all.return_value = []
db.execute.return_value = result
return db
@pytest.mark.asyncio
async def test_consent_response_yes():
mock_max_api = MagicMock()
mock_max_api.send_message = AsyncMock()
with patch("app.handlers.consent.max_api", mock_max_api), \
patch("app.handlers.consent.async_session") as mock_session:
mock_db = make_mock_db()
mock_session.return_value.__aenter__.return_value = mock_db
mock_user = MagicMock(spec=BotUser)
mock_user.consent_given = False
mock_user.consent_date = None
mock_conv = MagicMock(spec=BotConversation)
mock_conv.id = 1
mock_result = MagicMock()
mock_result.scalar_one_or_none.side_effect = [mock_user, mock_conv]
mock_db.execute.return_value = mock_result
await handle_consent_response(12345, 1, "даю согласие")
assert mock_max_api.send_message.called
@pytest.mark.asyncio
async def test_consent_response_no():
mock_max_api = MagicMock()
mock_max_api.send_message = AsyncMock()
with patch("app.handlers.consent.max_api", mock_max_api), \
patch("app.handlers.consent.async_session") as mock_session, \
patch("app.handlers.consent.config_reader") as mock_config:
mock_db = make_mock_db()
mock_session.return_value.__aenter__.return_value = mock_db
mock_config.get_phones.return_value = [("+7 (861) 203-33-30", "+78612033330")]
mock_config.get_site_mail.return_value = "mail@aegisone.ru"
mock_conv = MagicMock(spec=BotConversation)
mock_conv.id = 1
mock_result = MagicMock()
mock_result.scalar_one_or_none.return_value = mock_conv
mock_db.execute.return_value = mock_result
await handle_consent_response(12345, 1, "нет")
assert mock_max_api.send_message.called
@pytest.mark.asyncio
async def test_manual_contact_valid_phone():
mock_max_api = MagicMock()
mock_max_api.send_message = AsyncMock()
with patch("app.handlers.contact.max_api", mock_max_api), \
patch("app.handlers.contact.async_session") as mock_session, \
patch("app.handlers.contact.handle_contact_received",
new=AsyncMock()) as mock_received:
mock_db = make_mock_db()
mock_session.return_value.__aenter__.return_value = mock_db
await handle_manual_contact(12345, 1, "+78612033330")
assert mock_received.called
@pytest.mark.asyncio
async def test_manual_contact_invalid():
mock_max_api = MagicMock()
mock_max_api.send_message = AsyncMock()
with patch("app.handlers.contact.max_api", mock_max_api), \
patch("app.handlers.contact.async_session") as mock_session:
mock_db = make_mock_db()
mock_session.return_value.__aenter__.return_value = mock_db
await handle_manual_contact(12345, 1, "просто текст")
assert mock_max_api.send_message.called
@pytest.mark.asyncio
async def test_callback_consent_yes():
mock_max_api = MagicMock()
mock_max_api.send_message = AsyncMock()
with patch("app.handlers.handoff.max_api", mock_max_api), \
patch("app.handlers.consent.max_api", mock_max_api), \
patch("app.handlers.handoff.async_session") as mock_hoff_session, \
patch("app.handlers.consent.async_session") as mock_consent_session:
mock_db = make_mock_db()
mock_hoff_session.return_value.__aenter__.return_value = mock_db
mock_consent_session.return_value.__aenter__.return_value = mock_db
mock_conv = MagicMock(spec=BotConversation)
mock_conv.id = 1
mock_user = MagicMock(spec=BotUser)
mock_user.consent_given = False
mock_user.consent_date = None
mock_result = MagicMock()
mock_result.scalar_one_or_none.side_effect = [mock_conv, mock_user, mock_conv]
mock_db.execute.return_value = mock_result
await handle_callback(12345, "consent_yes", {})
assert mock_max_api.send_message.called
@pytest.mark.asyncio
async def test_inquiry_handling():
mock_max_api = MagicMock()
mock_max_api.send_message = AsyncMock()
mock_email = MagicMock()
mock_email.send_ticket_escalation = MagicMock(return_value=True)
with patch("app.handlers.inquiry.max_api", mock_max_api), \
patch("app.handlers.inquiry.email_sender", mock_email), \
patch("app.handlers.inquiry.async_session") as mock_session, \
patch("app.handlers.inquiry.settings_cache") as mock_cache:
mock_db = make_mock_db()
mock_session.return_value.__aenter__.return_value = mock_db
mock_conv = MagicMock(spec=BotConversation)
mock_conv.id = 1
mock_conv.intent = "ticket"
mock_conv.state = "awaiting_inquiry"
mock_conv.inquiry_text = None
mock_user = MagicMock(spec=BotUser)
mock_user.id = 12345
mock_user.first_name = "Иван"
mock_user.phone = "+78612033330"
mock_conv_result = MagicMock()
mock_conv_result.scalar_one_or_none.return_value = mock_conv
mock_user_result = MagicMock()
mock_user_result.scalar_one_or_none.return_value = mock_user
mock_db.execute.side_effect = [
mock_conv_result, mock_user_result,
mock_conv_result,
MagicMock(), MagicMock(), MagicMock(),
]
mock_cache.get = AsyncMock(side_effect=lambda k, d="": {
"support_email": "support@aegisone.ru",
"email_subject": "Test subject"
}.get(k, d))
await handle_inquiry(12345, 1, "Нужна консультация")
assert mock_max_api.send_message.called