TelegramBotAdmin/tests/test_channel_api.py

from __future__ import annotations

import json
from datetime import UTC, datetime, timedelta
from types import SimpleNamespace
from unittest.mock import AsyncMock

from aiohttp import CookieJar
from aiohttp.test_utils import TestClient, TestServer
from pyrogram.enums import ChatMemberStatus, ChatType


async def test_gif_upload_uses_telegram_returned_media_type(app_modules, monkeypatch):
    api = app_modules.load("wbb.admin.api")
    monkeypatch.setattr(api, "MESSAGE_DUMP_CHAT", -100999)
    monkeypatch.setattr(
        api.telegram_app,
        "send_animation",
        AsyncMock(return_value=SimpleNamespace(
            animation=None, document=SimpleNamespace(file_id="gif-as-document")
        )),
        raising=False,
    )

    class Part:
        name = "file"
        filename = "sample.gif"
        headers = {"Content-Type": "image/gif"}

        def __init__(self):
            self.sent = False

        async def read_chunk(self, _size):
            if self.sent:
                return b""
            self.sent = True
            return b"GIF89a"

    class Reader:
        async def next(self):
            return Part()

    class Request(dict):
        async def multipart(self):
            return Reader()

    response = await api.AdminApi().upload_media(Request())
    result = json.loads(response.text)["data"]
    assert result["type"] == "document"
    assert result["file_id"] == "gif-as-document"
    assert result["mime_type"] == "image/gif"


async def test_channel_api_onboarding_schedule_and_role_gate(app_modules, monkeypatch):
    chat_id = -100123
    app_modules.app.chats[chat_id] = SimpleNamespace(
        id=chat_id, type=ChatType.CHANNEL, title="运营频道", username="ops_channel",
        description="简介", permissions=None, photo=None,
    )
    app_modules.app.members[(chat_id, 999)] = SimpleNamespace(
        status=ChatMemberStatus.ADMINISTRATOR,
        privileges=SimpleNamespace(
            can_manage_chat=True, can_post_messages=True, can_pin_messages=True,
            can_change_info=True, can_edit_messages=True, can_delete_messages=True,
            can_promote_members=True,
        ),
    )
    api = app_modules.load("wbb.admin.api")
    application = api.build_admin_application()
    await application["admin_api"].initialize()
    client = TestClient(TestServer(application), cookie_jar=CookieJar(unsafe=True))
    await client.start_server()
    try:
        login = await client.post(
            "/api/admin/v1/auth/login",
            json={"username": "admin", "password": "qwe0.123456"},
        )
        token = (await login.json())["data"]["csrf_token"]
        changed = await client.put(
            "/api/admin/v1/auth/password",
            headers={"X-CSRF-Token": token},
            json={"current_password": "qwe0.123456", "new_password": "changed-pass-123"},
        )
        token = (await changed.json())["data"]["csrf_token"]
        headers = {"X-CSRF-Token": token}

        connected = await client.post(
            "/api/admin/v1/channels", headers=headers,
            json={"chat_id": str(chat_id), "confirm": True},
        )
        assert connected.status == 201
        assert (await connected.json())["data"]["type"] == "channel"
        listed = await client.get("/api/admin/v1/channels")
        assert [item["chat_id"] for item in (await listed.json())["data"]["items"]] == [str(chat_id)]
        groups = await client.get("/api/admin/v1/chats")
        assert (await groups.json())["data"]["total"] == 0

        async def get_chat_members(_chat_id, *, query, limit, filter):
            assert query == "Keven" and limit == 20
            yield SimpleNamespace(
                status=ChatMemberStatus.MEMBER, privileges=None, custom_title=None, until_date=None,
                user=SimpleNamespace(
                    id=42, username="keven", first_name="Keven", last_name=None,
                    is_bot=False, is_deleted=False,
                ),
            )

        monkeypatch.setattr(app_modules.app, "get_chat_members", get_chat_members, raising=False)
        candidates = await client.get(
            f"/api/admin/v1/channels/{chat_id}/admins/candidates?query=Keven&limit=20"
        )
        assert candidates.status == 200
        assert (await candidates.json())["data"]["items"][0]["user"]["id"] == "42"

        scheduled = await client.post(
            f"/api/admin/v1/channels/{chat_id}/posts", headers=headers,
            json={
                "text": "稍后发布", "publish_at": (datetime.now(UTC) + timedelta(hours=1)).isoformat(),
                "confirm": True,
            },
        )
        assert scheduled.status == 201
        post = (await scheduled.json())["data"]
        assert post["status"] == "scheduled" and post["message_id"] is None
        updated = await client.patch(
            f"/api/admin/v1/channels/{chat_id}/posts/{post['post_id']}", headers=headers,
            json={"text": "更新的帖子", "confirm": True},
        )
        assert updated.status == 200
        assert (await updated.json())["data"]["text"] == "更新的帖子"
        canceled = await client.delete(
            f"/api/admin/v1/channels/{chat_id}/posts/{post['post_id']}", headers=headers,
            json={"confirm": True},
        )
        assert (await canceled.json())["data"]["status"] == "canceled"

        app_modules.wbb.BOT_PERMISSIONS = {"channel.profile"}
        denied = await client.get(f"/api/admin/v1/channels/{chat_id}/posts")
        assert denied.status == 403
        assert (await denied.json())["error"]["code"] == "bot_role_permission_denied"
        denied_candidates = await client.get(
            f"/api/admin/v1/channels/{chat_id}/admins/candidates"
        )
        assert denied_candidates.status == 403
    finally:
        await client.close()