55 lines
2.1 KiB
Python
55 lines
2.1 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from property_bot.bot import TelegramPropertyBot
|
|
from property_bot.config import AppConfig, PropertyApiConfig, TelegramConfig
|
|
from property_bot.consumer import ListingEventConsumer
|
|
from property_bot.listings import Listing
|
|
from property_bot.repository import SubscriptionRepository
|
|
|
|
|
|
class FakeClient:
|
|
def __init__(self, fail_photo: bool = False) -> None:
|
|
self.fail_photo = fail_photo
|
|
self.photos: list[tuple[int, str, str]] = []
|
|
self.messages: list[tuple[int, str]] = []
|
|
|
|
async def send_photo(self, chat_id: int, photo: str, caption: str) -> None:
|
|
if self.fail_photo:
|
|
raise RuntimeError("media unavailable")
|
|
self.photos.append((chat_id, photo, caption))
|
|
|
|
async def send_message(self, chat_id: int, text: str, **_: object) -> None:
|
|
self.messages.append((chat_id, text))
|
|
|
|
|
|
def app_config(path: Path) -> AppConfig:
|
|
return AppConfig(
|
|
telegram=TelegramConfig(1, "hash", "token"),
|
|
admin_user_ids=frozenset(),
|
|
property_api=PropertyApiConfig("ws://localhost:3000/ws", 1, 1, 2),
|
|
database_path=path,
|
|
)
|
|
|
|
|
|
def listing() -> Listing:
|
|
return Listing("id", "https://example.test", "Title", 1, "PLN", None, None, None, None, None, None, ("https://example.test/image.jpg",))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_photo_failure_falls_back_to_text_and_keeps_other_recipients(tmp_path: Path) -> None:
|
|
repository = SubscriptionRepository(tmp_path / "db.sqlite3")
|
|
await repository.open()
|
|
try:
|
|
await repository.activate(1, 1)
|
|
await repository.activate(2, 2)
|
|
client = FakeClient(fail_photo=True)
|
|
consumer = ListingEventConsumer(app_config(tmp_path / "db.sqlite3").property_api, lambda _: None) # type: ignore[arg-type]
|
|
bot = TelegramPropertyBot(client, app_config(tmp_path / "db.sqlite3"), repository, consumer) # type: ignore[arg-type]
|
|
await bot.notify_listing(listing())
|
|
assert [chat_id for chat_id, _ in client.messages] == [1, 2]
|
|
assert client.photos == []
|
|
finally:
|
|
await repository.close()
|