234 lines
8.8 KiB
Python
234 lines
8.8 KiB
Python
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
|