TgMessageHook/tests/test_app.py

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"))
    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']}")
    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)

    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


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"