TelegramBotAdmin/tests/test_channel_management.py

from __future__ import annotations

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

import pytest
from pyrogram.enums import ChatMembersFilter, ChatMemberStatus, ChatType
from pyrogram.errors import MessageEmpty


def prepare_channel(app_modules, *, chat_id: int = -100123) -> None:
    app = app_modules.wbb.app
    app.chats[chat_id] = SimpleNamespace(
        id=chat_id,
        type=ChatType.CHANNEL,
        title="测试频道",
        username="test_channel",
        description="简介",
        permissions=None,
        photo=None,
    )
    app.members[(chat_id, 999)] = SimpleNamespace(
        status=ChatMemberStatus.ADMINISTRATOR,
        privileges=SimpleNamespace(
            can_manage_chat=True,
            can_post_messages=True,
            can_edit_messages=True,
            can_delete_messages=True,
            can_pin_messages=True,
            can_change_info=True,
            can_promote_members=True,
            can_invite_users=True,
        ),
    )


async def test_channel_connect_rejects_group_and_non_admin(app_modules):
    service = app_modules.load("wbb.services.channel_management")
    prepare_channel(app_modules)
    result = await service.ensure_channel(-100123)
    assert result["type"] == "channel"

    app_modules.wbb.app.members[(-100123, 999)].status = ChatMemberStatus.MEMBER
    with pytest.raises(service.ChatManagementError) as caught:
        await service.ensure_channel(-100123)
    assert caught.value.code == "bot_not_admin"

    app_modules.wbb.app.chats[-100123].type = ChatType.SUPERGROUP
    with pytest.raises(service.ChatManagementError) as caught:
        await service.ensure_channel(-100123)
    assert caught.value.code == "not_channel"


async def test_channel_admin_candidates_use_telegram_recent_and_search(app_modules, monkeypatch):
    service = app_modules.load("wbb.services.channel_management")
    prepare_channel(app_modules)
    calls = []

    def member(user_id: int, name: str, *, status=ChatMemberStatus.MEMBER, is_bot=False):
        return SimpleNamespace(
            status=status, privileges=None, custom_title=None, until_date=None,
            user=SimpleNamespace(
                id=user_id, username=None, first_name=name, last_name=None,
                is_bot=is_bot, is_deleted=False,
            ),
        )

    async def get_chat_members(chat_id, *, query, limit, filter):
        calls.append((chat_id, query, limit, filter))
        for item in [
            member(42, "Keven"),
            member(999, "Bot", status=ChatMemberStatus.ADMINISTRATOR, is_bot=True),
            member(43, "Owner", status=ChatMemberStatus.OWNER),
        ]:
            yield item

    monkeypatch.setattr(app_modules.app, "get_chat_members", get_chat_members, raising=False)
    recent = await service.list_channel_admin_candidates(-100123)
    searched = await service.list_channel_admin_candidates(-100123, query="@keven", limit=80)
    assert [item["user"]["id"] for item in recent] == ["42"]
    assert [item["user"]["id"] for item in searched] == ["42"]
    assert calls == [
        (-100123, "", 30, ChatMembersFilter.RECENT),
        (-100123, "keven", 30, ChatMembersFilter.SEARCH),
    ]

    app_modules.app.members[(-100123, 999)].privileges.can_promote_members = False
    with pytest.raises(service.ChatManagementError) as caught:
        await service.list_channel_admin_candidates(-100123)
    assert caught.value.code == "bot_permission_missing"


async def test_channel_admin_candidates_lookup_numeric_member(app_modules, monkeypatch):
    service = app_modules.load("wbb.services.channel_management")
    prepare_channel(app_modules)
    calls = []

    async def bot_api(method, **params):
        calls.append((method, params))
        return {
            "status": "member",
            "user": {"id": 8960210159, "first_name": "Keven", "username": "uesd1515", "is_bot": False},
        }

    monkeypatch.setattr(service, "channel_bot_api", bot_api)
    result = await service.list_channel_admin_candidates(-100123, query="8960210159")
    assert result[0]["user"]["id"] == "8960210159"
    assert calls == [("getChatMember", {"chat_id": -100123, "user_id": 8960210159})]


async def test_manual_channel_connect_resolves_uncached_public_id(app_modules, monkeypatch):
    service = app_modules.load("wbb.services.channel_management")
    prepare_channel(app_modules)
    app_modules.wbb.BOT_TOKEN = "test-token"
    original_get_chat = app_modules.app.get_chat
    warmed = False

    async def get_chat(peer):
        nonlocal warmed
        if peer == -100123 and not warmed:
            raise ValueError("Peer id invalid: -100123")
        if peer == "@test_channel":
            warmed = True
            return await original_get_chat(-100123)
        return await original_get_chat(peer)

    class Response:
        async def __aenter__(self):
            return self

        async def __aexit__(self, *_args):
            return None

        async def json(self):
            return {"ok": True, "result": {"id": -100123, "type": "channel", "username": "test_channel"}}

    class Session:
        async def __aenter__(self):
            return self

        async def __aexit__(self, *_args):
            return None

        def post(self, *_args, **_kwargs):
            return Response()

    monkeypatch.setattr(app_modules.app, "get_chat", get_chat)
    monkeypatch.setattr(service, "ClientSession", lambda **_kwargs: Session())
    result = await service.ensure_channel(-100123)
    assert result["chat_id"] == -100123 and warmed


async def test_channel_list_and_group_list_are_isolated(app_modules):
    service = app_modules.load("wbb.services.channel_management")
    chat_service = app_modules.load("wbb.services.chat_management")
    prepare_channel(app_modules)
    await service.ensure_channel(-100123)
    await app_modules.wbb.db.managed_chats.insert_one({
        "bot_id": "primary", "chat_id": -100456, "type": "supergroup",
        "title": "群", "accessible": True,
    })
    channels, channel_total = await service.list_channels(query="", page=1, page_size=20)
    groups, group_total = await chat_service.list_accessible_chats(page=1, page_size=20)
    assert channel_total == 1 and channels[0]["chat_id"] == -100123
    assert group_total == 1 and groups[0]["chat_id"] == -100456


async def test_generic_channel_observation_keeps_unavailable_status(app_modules):
    dbadmin = app_modules.load("wbb.utils.dbadmin")
    await dbadmin.upsert_managed_chat(
        chat_id=-100123, title="频道", username=None, chat_type="channel",
        accessible=False, bot_status="administrator",
    )
    await dbadmin.upsert_managed_chat(
        chat_id=-100123, title="频道新名", username=None, chat_type="channel",
        accessible=None,
    )
    record = await app_modules.wbb.db.managed_chats.find_one({"chat_id": -100123})
    assert record["accessible"] is False
    assert record["title"] == "频道新名"


async def test_channel_observation_deduplicates_updates_and_scopes_bot(app_modules):
    service = app_modules.load("wbb.services.channel_management")
    prepare_channel(app_modules)
    message = SimpleNamespace(
        chat=app_modules.wbb.app.chats[-100123], id=8,
        text="原文", caption=None, media=None, date=datetime.now(UTC),
    )
    await service.observe_channel_post(message)
    message.text = "改文"
    await service.observe_channel_post(message)
    assert await service.postsdb.count_documents({}) == 1
    post = await service.postsdb.find_one({"bot_id": "primary"})
    assert post["text"] == "改文" and post["source"] == "telegram"
    await service.observe_channel_deletion(-100123, 8)
    assert (await service.postsdb.find_one({"post_id": post["post_id"]}))["status"] == "deleted"

    app_modules.wbb.BOT_PROFILE_ID = "second"
    service.BOT_PROFILE_ID = "second"
    dbmodule = app_modules.load("wbb.utils.dbchannel")
    dbmodule.BOT_PROFILE_ID = "second"
    await service.observe_channel_post(message)
    assert await service.postsdb.count_documents({}) == 2


async def test_scheduled_post_publishes_once_and_can_be_canceled(app_modules):
    service = app_modules.load("wbb.services.channel_management")
    prepare_channel(app_modules)
    scheduled = await service.create_channel_post(-100123, {
        "text": "计划帖", "publish_at": (datetime.now(UTC) + timedelta(hours=1)).isoformat(),
    })
    assert scheduled["status"] == "scheduled"
    changed = await service.edit_channel_post(-100123, scheduled["post_id"], {"text": "更新计划"})
    assert changed["text"] == "更新计划"
    await service.postsdb.update_one(
        {"post_id": scheduled["post_id"]},
        {"$set": {"publish_at": datetime.now(UTC) - timedelta(seconds=1)}},
    )
    await service.sweep_channel_posts()
    await service.sweep_channel_posts()
    assert app_modules.wbb.app.sent_messages == [(-100123, "更新计划")]
    assert (await service.get_post(-100123, scheduled["post_id"]))["status"] == "published"

    canceled = await service.create_channel_post(-100123, {
        "text": "不发", "publish_at": (datetime.now(UTC) + timedelta(hours=1)).isoformat(),
    })
    assert (await service.delete_channel_post(-100123, canceled["post_id"]))["status"] == "canceled"


async def test_scheduled_post_can_be_edited_near_publish_time(app_modules):
    service = app_modules.load("wbb.services.channel_management")
    prepare_channel(app_modules)
    scheduled = await service.create_channel_post(-100123, {
        "text": "临近发布", "publish_at": (datetime.now(UTC) + timedelta(hours=1)).isoformat(),
    })
    await service.postsdb.update_one(
        {"post_id": scheduled["post_id"]},
        {"$set": {"publish_at": datetime.now(UTC) + timedelta(seconds=2)}},
    )
    current = await service.get_post(-100123, scheduled["post_id"])
    updated = await service.edit_channel_post(-100123, scheduled["post_id"], {
        "text": "已修改", "publish_at": current["publish_at"].replace(tzinfo=UTC).isoformat(),
    })
    assert updated["text"] == "已修改"
    assert updated["status"] == "scheduled"


async def test_uncertain_send_is_not_retried(app_modules, monkeypatch):
    service = app_modules.load("wbb.services.channel_management")
    prepare_channel(app_modules)
    send = AsyncMock(side_effect=TimeoutError("timeout after send"))
    monkeypatch.setattr(service.app, "send_message", send)
    with pytest.raises(service.ChatManagementError) as caught:
        await service.create_channel_post(-100123, {"text": "结果不明"})
    assert caught.value.code == "post_delivery_uncertain"
    await service.sweep_channel_posts()
    assert send.await_count == 1
    post = await service.postsdb.find_one({"text": "结果不明"})
    assert post["status"] == "uncertain"


async def test_telegram_rejection_is_failed_not_uncertain(app_modules, monkeypatch):
    service = app_modules.load("wbb.services.channel_management")
    prepare_channel(app_modules)
    monkeypatch.setattr(
        service.app, "send_message", AsyncMock(side_effect=MessageEmpty()),
    )
    with pytest.raises(service.ChatManagementError) as caught:
        await service.create_channel_post(-100123, {"text": "Telegram 拒绝"})
    assert caught.value.code == "post_send_failed"
    post = await service.postsdb.find_one({"text": "Telegram 拒绝"})
    assert post["status"] == "failed"


async def test_published_post_edit_and_delete_require_telegram_permission(app_modules, monkeypatch):
    service = app_modules.load("wbb.services.channel_management")
    prepare_channel(app_modules)
    post = await service.create_channel_post(-100123, {"text": "原文"})
    edited = await service.edit_channel_post(-100123, post["post_id"], {"text": "新文"})
    assert edited["text"] == "新文"
    assert app_modules.wbb.app.edited_messages == [(-100123, 1, "新文")]
    monkeypatch.setattr(service.app, "delete_messages", AsyncMock(return_value=1))
    deleted = await service.delete_channel_post(-100123, post["post_id"])
    assert deleted["status"] == "deleted"


async def test_media_post_can_replace_media_and_channel_admin_rights_are_checked(app_modules, monkeypatch):
    service = app_modules.load("wbb.services.channel_management")
    prepare_channel(app_modules)
    post = await service.create_channel_post(-100123, {
        "text": "第一张", "media_type": "photo", "file_id": "photo-file-1",
        "media_filename": "first.jpg", "media_size": 1234, "media_mime_type": "image/jpeg",
    })
    assert app_modules.app.sent_photos == [(-100123, "photo-file-1", "第一张")]
    assert post["media_filename"] == "first.jpg" and post["media_size"] == 1234
    edit_media = AsyncMock()
    monkeypatch.setattr(service.app, "edit_message_media", edit_media, raising=False)
    changed = await service.edit_channel_post(-100123, post["post_id"], {
        "text": "第二张", "media_type": "photo", "file_id": "photo-file-2",
    })
    assert changed["file_id"] == "photo-file-2"
    assert edit_media.await_count == 1

    member = {"status": "member", "user": {"id": 42, "first_name": "用户"}}
    admin = {"status": "administrator", "user": member["user"], "can_post_messages": True}
    bot_api = AsyncMock(side_effect=[member, True, admin])
    monkeypatch.setattr(service, "channel_bot_api", bot_api)
    promoted = await service.set_channel_admin(-100123, 42, ["can_post_messages"])
    assert promoted["status"] == "administrator"
    assert promoted["privileges"] == ["can_post_messages"]
    assert [item.args[0] for item in bot_api.await_args_list] == [
        "getChatMember", "promoteChatMember", "getChatMember",
    ]
    assert bot_api.await_args_list[1].kwargs["can_post_messages"] is True
    assert bot_api.await_args_list[1].kwargs["can_edit_messages"] is False
    bot_api.reset_mock(side_effect=True)
    bot_api.side_effect = [admin, True]
    removed = await service.remove_channel_admin(-100123, 42)
    assert removed["removed"] is True
    assert [item.args[0] for item in bot_api.await_args_list] == [
        "getChatMember", "promoteChatMember",
    ]
    assert not any(
        bot_api.await_args_list[1].kwargs[key] for key in service.CHANNEL_ADMIN_PRIVILEGES
    )
    with pytest.raises(service.ChatManagementError) as caught:
        await service.set_channel_admin(-100123, 42, ["can_restrict_members"])
    assert caught.value.code == "invalid_privileges"


@pytest.mark.parametrize("media_type, method, argument", [
    ("animation", "send_animation", "animation"),
    ("video", "send_video", "video"),
    ("document", "send_document", "document"),
])
async def test_media_post_dispatches_supported_types(app_modules, monkeypatch, media_type, method, argument):
    service = app_modules.load("wbb.services.channel_management")
    prepare_channel(app_modules)
    send = AsyncMock(return_value=SimpleNamespace(id=91))
    monkeypatch.setattr(service.app, method, send, raising=False)
    post = await service.create_channel_post(-100123, {
        "text": "媒体帖", "media_type": media_type, "file_id": "telegram-file-id",
    })
    assert post["status"] == "published" and post["message_id"] == 91
    assert send.await_args.args == (-100123,)
    assert send.await_args.kwargs[argument] == "telegram-file-id"


async def test_revoked_bot_post_role_prevents_scheduled_send(app_modules):
    service = app_modules.load("wbb.services.channel_management")
    prepare_channel(app_modules)
    scheduled = await service.create_channel_post(-100123, {
        "text": "原定发布", "publish_at": (datetime.now(UTC) + timedelta(hours=1)).isoformat(),
    })
    await service.postsdb.update_one(
        {"post_id": scheduled["post_id"]},
        {"$set": {"publish_at": datetime.now(UTC) - timedelta(seconds=1)}},
    )
    service.BOT_PERMISSIONS = {"channel.profile"}
    await service.sweep_channel_posts()
    assert app_modules.app.sent_messages == []
    assert (await service.get_post(-100123, scheduled["post_id"]))["status"] == "failed"


async def test_concurrent_claim_and_stale_restart_do_not_resend(app_modules):
    service = app_modules.load("wbb.services.channel_management")
    dbmodule = app_modules.load("wbb.utils.dbchannel")
    prepare_channel(app_modules)
    scheduled = await service.create_channel_post(-100123, {
        "text": "重启前任务", "publish_at": (datetime.now(UTC) + timedelta(hours=1)).isoformat(),
    })
    await service.postsdb.update_one(
        {"post_id": scheduled["post_id"]},
        {"$set": {"publish_at": datetime.now(UTC) - timedelta(seconds=1)}},
    )
    first, second = await asyncio.gather(dbmodule.claim_due_posts(), dbmodule.claim_due_posts())
    assert len(first) + len(second) == 1
    await service.postsdb.update_one(
        {"post_id": scheduled["post_id"]},
        {"$set": {"updated_at": datetime.now(UTC) - timedelta(minutes=6)}},
    )
    await service.sweep_channel_posts()
    assert app_modules.app.sent_messages == []
    assert (await service.get_post(-100123, scheduled["post_id"]))["status"] == "uncertain"


def test_channel_route_permissions(app_modules):
    permissions = app_modules.load("wbb.services.bot_permissions")
    assert permissions.api_permission("GET", "/api/admin/v1/channels") == "channel.any"
    assert permissions.api_permission("GET", "/api/admin/v1/channels/-100") == "channel.any"
    assert permissions.api_permission("POST", "/api/admin/v1/channels") == "channel.profile"
    assert permissions.api_permission("PUT", "/api/admin/v1/channels/-100/admins/42") == "channel.admins"
    assert permissions.api_permission("PATCH", "/api/admin/v1/channels/-100/posts/p1") == "channel.posts"