82 lines
3.5 KiB
Python
82 lines
3.5 KiB
Python
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from services.collector import ScrapeResult
|
|
from services.admin_bot import AdminBotService
|
|
from db.models import Post, SourceChannel, SourceWebsite
|
|
|
|
def test_collector_summary_fa_with_title():
|
|
res = ScrapeResult(scanned=10, collected=5, already_stored=3, duplicates=2)
|
|
summary = res.summary_fa(channel_id=-100123456, channel_title="کانال خبری فناوری")
|
|
assert "کانال خبری فناوری" in summary
|
|
assert "<code>-100123456</code>" not in summary
|
|
|
|
def test_collector_summary_fa_fallback_to_id():
|
|
res = ScrapeResult(scanned=10, collected=0, already_stored=10, duplicates=0)
|
|
summary = res.summary_fa(channel_id=-100123456, channel_title=None)
|
|
assert "-100123456" in summary
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_bot_format_raw_post_caption_telegram_channel():
|
|
repo = MagicMock()
|
|
repo.get_source_by_channel_id = AsyncMock(return_value=SourceChannel(
|
|
id=1, channel_id=-100123456, username="tech_news", title="اخبار روز تکنولوژی"
|
|
))
|
|
bot = AdminBotService(repo=repo, ai_processor=MagicMock(), queue=MagicMock())
|
|
|
|
post = Post(id=1, source_channel_id=-100123456, source_message_id=10, raw_text="متن تست پست")
|
|
caption = await bot._format_raw_post_caption(post)
|
|
|
|
assert "اخبار روز تکنولوژی" in caption
|
|
assert "@tech_news" in caption
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_bot_format_raw_post_caption_website():
|
|
repo = MagicMock()
|
|
pseudo_id = -900000005
|
|
repo.get_source_website_by_id = AsyncMock(return_value=SourceWebsite(
|
|
id=5, name="دیجیاتو", url="https://digiato.com"
|
|
))
|
|
bot = AdminBotService(repo=repo, ai_processor=MagicMock(), queue=MagicMock())
|
|
|
|
post = Post(id=2, source_channel_id=pseudo_id, source_message_id=1, raw_text="خبر جدید وبسایت")
|
|
caption = await bot._format_raw_post_caption(post)
|
|
|
|
assert "دیجیاتو" in caption
|
|
assert "وبسایت" in caption
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_bot_format_raw_post_caption_real_telegram_supergroup_id():
|
|
repo = MagicMock()
|
|
real_tg_id = -1009199982627
|
|
repo.get_source_by_channel_id = AsyncMock(return_value=SourceChannel(
|
|
id=1, channel_id=real_tg_id, username="mychannel", title="کانال تست"
|
|
))
|
|
repo.get_source_website_by_id = AsyncMock()
|
|
bot = AdminBotService(repo=repo, ai_processor=MagicMock(), queue=MagicMock())
|
|
|
|
post = Post(id=3, source_channel_id=real_tg_id, source_message_id=100, raw_text="متن خبر کانال")
|
|
caption = await bot._format_raw_post_caption(post)
|
|
|
|
assert "کانال تست" in caption
|
|
# Verify get_source_website_by_id was NEVER called for telegram channel
|
|
repo.get_source_website_by_id.assert_not_called()
|
|
repo.get_source_by_channel_id.assert_called_once_with(real_tg_id)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_bot_get_effective_review_channel_id():
|
|
repo = MagicMock()
|
|
real_tg_id = -1009199982627
|
|
repo.get_source_by_channel_id = AsyncMock(return_value=SourceChannel(
|
|
id=1, channel_id=real_tg_id, username="mychannel", title="کانال تست", admin_channel_id=-100111222333
|
|
))
|
|
repo.get_source_website_by_id = AsyncMock()
|
|
bot = AdminBotService(repo=repo, ai_processor=MagicMock(), queue=MagicMock())
|
|
|
|
post = Post(id=3, source_channel_id=real_tg_id, source_message_id=100, raw_text="متن خبر")
|
|
rev_id = await bot.get_effective_review_channel_id(post)
|
|
assert rev_id == -100111222333
|
|
repo.get_source_website_by_id.assert_not_called()
|
|
|