import importlib from datetime import datetime from pathlib import Path from types import SimpleNamespace from typing import List, Tuple import pytest from fastapi.testclient import TestClient @pytest.fixture() def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> TestClient: monkeypatch.setenv("TELEGRAM_API_ID", "123456") monkeypatch.setenv("TELEGRAM_API_HASH", "testhash") monkeypatch.setenv("BASE_URL", "http://testserver") monkeypatch.setenv("DATABASE_PATH", str(tmp_path / "hooks.json")) monkeypatch.setenv("TELEGRAM_SESSION_PATH", str(tmp_path / "telegram.session")) class DummyTelegramClient: def __init__(self, *args, **kwargs): self._connected = False async def connect(self) -> None: self._connected = True async def disconnect(self) -> None: self._connected = False def is_connected(self) -> bool: return self._connected async def send_code_request(self, phone: str): # noqa: ANN001 return SimpleNamespace(phone_code_hash="dummy-hash") async def sign_in(self, *args, **kwargs): # noqa: ANN001, ANN002 return None def is_user_authorized(self) -> bool: return True async def get_me(self): # noqa: ANN201 return SimpleNamespace(first_name="Test", last_name="User", username="tester", id=1) async def log_out(self) -> None: self._connected = False async def send_message(self, *args, **kwargs) -> None: # noqa: ANN001, ANN002 return None monkeypatch.setattr("telethon.client.telegramclient.TelegramClient", DummyTelegramClient) monkeypatch.setattr("telethon.TelegramClient", DummyTelegramClient) # Clear cached settings and reload modules to pick up new paths from app import config config.get_settings.cache_clear() # type: ignore[attr-defined] importlib.reload(importlib.import_module("app.storage")) telegram_service_module = importlib.reload(importlib.import_module("app.telegram_service")) models_module = importlib.reload(importlib.import_module("app.models")) main_module = importlib.reload(importlib.import_module("app.main")) # Stub out Telegram interactions call_log: List[Tuple[str, str]] = [] async def fake_ensure_connected() -> None: return None async def fake_is_authorized() -> bool: return True async def fake_get_user() -> str: return "Test User" async def fake_send_message(chat_id: str, message: str) -> None: call_log.append((chat_id, message)) recent_chats_state: List[models_module.RecentChat] = [] async def fake_fetch_recent_chats(limit: int = 50) -> List[models_module.RecentChat]: return recent_chats_state[:limit] async def fake_disconnect() -> None: return None async def fake_verify_code(*args, **kwargs) -> None: # noqa: ANN002 return None monkeypatch.setattr(main_module.telegram_service, "ensure_connected", fake_ensure_connected) monkeypatch.setattr(main_module.telegram_service, "is_authorized", fake_is_authorized) monkeypatch.setattr(main_module.telegram_service, "get_user", fake_get_user) monkeypatch.setattr(main_module.telegram_service, "send_message", fake_send_message) monkeypatch.setattr(main_module.telegram_service, "disconnect", fake_disconnect) monkeypatch.setattr(main_module.telegram_service, "verify_code", fake_verify_code) monkeypatch.setattr(main_module.telegram_service, "fetch_recent_chats", fake_fetch_recent_chats) monkeypatch.setattr(main_module.telegram_service, "is_connected", lambda: True) test_client = TestClient(main_module.app) test_client.call_log = call_log # type: ignore[attr-defined] test_client.recent_chats_state = recent_chats_state # type: ignore[attr-defined] test_client.RecentChatModel = models_module.RecentChat # type: ignore[attr-defined] return test_client def test_create_list_and_trigger_hook(client: TestClient) -> None: payload = { "chat_id": "@example", "message": "Hello **Markdown**", } create_resp = client.post("/api/hooks", json=payload) assert create_resp.status_code == 201 data = create_resp.json() assert data["chat_id"] == payload["chat_id"] assert payload["message"] in data["message"] assert data["action_url"].endswith(f"/action/{data['hook_id']}") assert data["last_triggered_at"] is None list_resp = client.get("/api/hooks") assert list_resp.status_code == 200 hooks = list_resp.json() assert len(hooks) == 1 assert hooks[0]["last_triggered_at"] is None trigger_resp = client.get(f"/action/{data['hook_id']}") assert trigger_resp.status_code == 200 assert trigger_resp.json()["status"] == "sent" call_log = getattr(client, "call_log") assert call_log == [(payload["chat_id"], payload["message"])] refreshed_hooks = client.get("/api/hooks").json() last_triggered_value = refreshed_hooks[0]["last_triggered_at"] assert last_triggered_value is not None first_triggered_at = datetime.fromisoformat(last_triggered_value) client.recent_chats_state[:] = [ client.RecentChatModel( chat_id=payload["chat_id"], display_name=f"Chat {payload['chat_id']}", chat_type="user", username="exampleuser", phone_number="123456", last_used_at=datetime.now(), ) ] recent_resp = client.get("/api/recent-chats") assert recent_resp.status_code == 200 recent_data = recent_resp.json() assert len(recent_data) == 1 assert recent_data[0]["chat_id"] == payload["chat_id"] assert recent_data[0]["display_name"] == f"Chat {payload['chat_id']}" assert recent_data[0]["username"] == "exampleuser" assert recent_data[0]["phone_number"] == "123456" new_id = "customid123" patch_resp = client.patch(f"/api/hooks/{data['hook_id']}", json={"hook_id": new_id}) assert patch_resp.status_code == 200 patched = patch_resp.json() assert patched["hook_id"] == new_id assert patched["action_url"].endswith(f"/action/{new_id}") assert patched["last_triggered_at"] == last_triggered_value # Old ID should now be gone old_trigger = client.get(f"/action/{data['hook_id']}") assert old_trigger.status_code == 404 call_log.clear() new_trigger = client.get(f"/action/{new_id}") assert new_trigger.status_code == 200 assert getattr(client, "call_log") == [(payload["chat_id"], payload["message"])] update_payload = { "chat_id": "@updated", "message": "Updated message!", } update_resp = client.patch(f"/api/hooks/{new_id}", json=update_payload) assert update_resp.status_code == 200 updated_hook = update_resp.json() assert updated_hook["hook_id"] == new_id assert updated_hook["chat_id"] == update_payload["chat_id"] assert updated_hook["message"] == update_payload["message"] assert updated_hook["last_triggered_at"] == last_triggered_value call_log.clear() retrigger_resp = client.get(f"/action/{new_id}") assert retrigger_resp.status_code == 200 assert getattr(client, "call_log") == [(update_payload["chat_id"], update_payload["message"])] final_hooks = client.get("/api/hooks").json() second_triggered_value = final_hooks[0]["last_triggered_at"] assert second_triggered_value is not None second_triggered_at = datetime.fromisoformat(second_triggered_value) assert second_triggered_at >= first_triggered_at client.recent_chats_state[:] = [ client.RecentChatModel( chat_id=update_payload["chat_id"], display_name=f"Chat {update_payload['chat_id']}", chat_type="group", username="updateduser", phone_number=None, last_used_at=datetime.now(), ) ] history_resp = client.get("/api/recent-chats") assert history_resp.status_code == 200 history_data = history_resp.json() assert history_data[0]["chat_id"] == update_payload["chat_id"] assert history_data[0]["display_name"] == f"Chat {update_payload['chat_id']}" assert history_data[0]["username"] == "updateduser" assert history_data[0]["phone_number"] is None def test_login_verify_without_phone_number(client: TestClient) -> None: response = client.post("/api/login/verify", json={"code": "123456"}) assert response.status_code == 200 payload = response.json() assert payload["authorized"] is True assert payload["user"] == "Test User" def test_recent_chats_requires_authorization(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None: from app import main as main_module async def not_authorized() -> bool: return False monkeypatch.setattr(main_module.telegram_service, "is_authorized", not_authorized) response = client.get("/api/recent-chats") assert response.status_code == 401