142 lines
5.1 KiB
Python
142 lines
5.1 KiB
Python
import importlib
|
|
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"))
|
|
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))
|
|
|
|
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, "is_connected", lambda: True)
|
|
|
|
test_client = TestClient(main_module.app)
|
|
test_client.call_log = call_log # 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']}")
|
|
|
|
list_resp = client.get("/api/hooks")
|
|
assert list_resp.status_code == 200
|
|
hooks = list_resp.json()
|
|
assert len(hooks) == 1
|
|
|
|
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"])]
|
|
|
|
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}")
|
|
|
|
# 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"])]
|
|
|
|
|
|
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"
|