test_bot.py

import asyncio
import logging
import os
from asyncio import AbstractEventLoop
from contextlib import suppress
from datetime import datetime, timedelta

import pytest
import pytest_asyncio

import interactions
from interactions import (
    GuildChannel,
    MessageableMixin,
    GuildNews,
    ThreadableMixin,
    ExplicitContentFilterLevel,
    VerificationLevel,
    Client,
    Guild,
    GuildText,
    Embed,
    PartialEmoji,
    BrandColors,
    Permissions,
    Status,
    EmbedField,
    EmbedAuthor,
    EmbedAttachment,
    EmbedFooter,
    StringSelectOption,
    Modal,
    ParagraphText,
    Message,
    GuildVoice,
    Poll,
    PollMedia,
)
from interactions.models.discord.asset import Asset
from interactions.models.discord.components import ActionRow, Button, StringSelectMenu
from interactions.models.discord.emoji import process_emoji_req_format
from interactions.api.gateway.websocket import WebsocketClient
from interactions.api.http.route import Route
from interactions.api.voice.audio import AudioVolume
from interactions.client.errors import NotFound
from interactions.models.discord.enums import ButtonStyle
from interactions.models.discord.role import Role

__all__ = ()


try:
    import dotenv

    dotenv.load_dotenv()
except ImportError:
    pass

TOKEN = os.environ.get("BOT_TOKEN")
if not TOKEN:
    pytest.skip(f"Skipping {os.path.basename(__file__)} - no token provided", allow_module_level=True)
if os.environ.get("GITHUB_ACTIONS") and not os.environ.get("RUN_TESTBOT"):
    pytest.skip(f"Skipping {os.path.basename(__file__)} - RUN_TESTBOT not set", allow_module_level=True)

log = logging.getLogger("Interations-Integration-Tests")


@pytest.fixture(scope="module")
def event_loop() -> AbstractEventLoop:
    return asyncio.get_event_loop()


@pytest_asyncio.fixture(scope="module")
async def bot(github_commit) -> Client:
    bot = interactions.Client(activity="Testing someones code")
    await bot.login(TOKEN)
    gw = asyncio.create_task(bot.start_gateway())

    await bot._ready.wait()
    bot.suffix = github_commit
    log.info(f"Logged in as {bot.user} ({bot.user.id}) -- {bot.suffix}")

    yield bot
    gw.cancel()
    with suppress(asyncio.CancelledError):
        await gw


@pytest_asyncio.fixture(scope="module")
async def guild(bot: Client) -> Guild:
    guild = next((g for g in bot.guilds if g.name in ["Interactions.py Test Suite", "NAFF Test Suite"]), None)
    if not guild:
        log.info("No guild found, creating one...")

        guild: interactions.Guild = await interactions.Guild.create("Interactions.py Test Suite", bot)
        community_channel = await guild.create_text_channel("community_channel")

        await guild.edit(
            features=["COMMUNITY"],
            rules_channel=community_channel,
            system_channel=community_channel,
            public_updates_channel=community_channel,
            explicit_content_filter=ExplicitContentFilterLevel.ALL_MEMBERS,
            verification_level=VerificationLevel.LOW,
        )

    yield guild


@pytest_asyncio.fixture(scope="module")
async def channel(bot, guild) -> GuildText:
    channel = await guild.create_text_channel(f"test_scene - {bot.suffix}")
    yield channel
    await channel.delete()


@pytest_asyncio.fixture(scope="module")
async def github_commit() -> str:
    import subprocess

    return subprocess.check_output(["git", "rev-parse", "--short", "HEAD"]).decode("ascii").strip()


def ensure_attributes(target_object) -> None:
    for attr in dir(target_object):
        # ensure all props and attributes load correctly
        getattr(target_object, attr)


@pytest.mark.asyncio
async def test_reaction_events(bot: Client, guild: Guild) -> None:
    """
    Tests reaction event handling on an uncached message.

    Requires manual setup:
    1. Set TARGET_CHANNEL_ID environment variable to a valid channel ID.
    2. A user must add a reaction to the test message within 60 seconds.
    """
    # Skip test if target channel not provided
    target_channel_id = os.environ.get("BOT_TEST_CHANNEL_ID")
    if not target_channel_id:
        pytest.skip("Set TARGET_CHANNEL_ID to run this test")

    # Get channel and post test message
    channel = await bot.fetch_channel(target_channel_id)
    test_msg = await channel.send("Reaction Event Test - React with ✅ within 60 seconds")

    try:
        # simulate uncached state
        bot.cache.delete_message(message_id=test_msg.id, channel_id=test_msg.channel.id)

        # wait for user to react with checkmark
        reaction_event = await bot.wait_for(
            "message_reaction_add", timeout=60, checks=lambda e: e.message.id == test_msg.id and str(e.emoji) == "✅"
        )

        assert reaction_event.message.id == test_msg.id
        assert reaction_event.emoji.name == "✅"
    finally:
        await test_msg.delete()


@pytest.mark.asyncio
async def test_channels(bot: Client, guild: Guild) -> None:
    channels = [
        guild_category := await guild.create_category("_test_category"),
        await guild.create_text_channel(f"_test_text-{bot.suffix}"),
        await guild.create_news_channel(f"_test_news-{bot.suffix}"),
        await guild.create_stage_channel(f"_test_stage-{bot.suffix}"),
        await guild.create_voice_channel(f"_test_voice-{bot.suffix}"),
    ]

    assert all(c in guild.channels for c in channels)

    channels.append(await bot.owner.fetch_dm())

    try:
        for channel in channels:
            ensure_attributes(channel)

            if isinstance(channel, GuildChannel) and channel != guild_category:
                await channel.edit(parent_id=guild_category.id)
                assert channel.category == guild_category

            if isinstance(channel, MessageableMixin) and not isinstance(channel, GuildVoice):
                _m = await channel.send("test")
                assert _m.channel == channel

                if isinstance(channel, GuildNews):
                    await _m.publish()

                await _m.delete()

            if isinstance(channel, ThreadableMixin):
                if isinstance(channel, GuildNews):
                    _tm = await channel.send("dummy message")
                    thread = await _tm.create_thread("new thread")
                else:
                    thread = await channel.create_thread("new thread")
                assert thread.parent_channel == channel
                _m = await thread.send("test")
                assert _m.channel == thread

                _m = await channel.send("start thread here")
                m_thread = await channel.create_thread("new message thread", message=_m)
                assert _m.id == m_thread.id

                assert m_thread in guild.threads
                assert thread in guild.threads
                await thread.delete()
                # We suppress bcu sometimes event fires too fast, before wait_for is called
                with suppress(asyncio.exceptions.TimeoutError):
                    await bot.wait_for("thread_delete", timeout=2)
                assert thread not in guild.threads
    finally:
        for channel in channels:
            with suppress(NotFound):
                await channel.delete()


@pytest.mark.asyncio
async def test_messages(bot: Client, guild: Guild, channel: GuildText) -> None:
    msg = await channel.send("Message Tests")
    thread = await msg.create_thread("Test Thread")

    try:
        _m = await thread.send("Test")
        ensure_attributes(_m)

        await _m.edit(content="Test Edit")
        assert _m.content == "Test Edit"
        await _m.add_reaction("❌")
        with suppress(asyncio.exceptions.TimeoutError):
            await bot.wait_for("message_reaction_add", timeout=2)

        assert len(_m.reactions) == 1

        assert len(await _m.fetch_reaction("❌")) != 0
        await _m.remove_reaction("❌")
        with suppress(asyncio.exceptions.TimeoutError):
            await bot.wait_for("message_reaction_remove", timeout=2)

        await _m.add_reaction("❌")
        await _m.clear_all_reactions()
        with suppress(asyncio.exceptions.TimeoutError):
            await bot.wait_for("message_reaction_remove_all", timeout=2)

        assert len(_m.reactions) == 0

        await _m.pin()
        assert _m.pinned is True
        await _m.suppress_embeds()
        await _m.unpin()

        _r = await _m.reply(f"test-reply {bot.user.mention} {channel.mention}")
        assert _r._referenced_message_id == _m.id

        mem_mentions = []
        async for member in _r.mention_users:
            mem_mentions.append(member)
        assert len(mem_mentions) == 1

        assert len(_r.mention_channels) == 1

        await thread.send(file=r"tests/LordOfPolls.png")

        assert _m.jump_url is not None
        assert _m.proto_url is not None

        await thread.send(embeds=Embed("Test"))

        await thread.delete()

        _m = await bot.owner.send("Test Message from TestSuite")
        await _m.delete()

    finally:
        with suppress(interactions.errors.NotFound):
            await thread.delete()


@pytest.mark.asyncio
async def test_roles(bot: Client, guild: Guild) -> None:
    roles: list[Role] = []

    try:
        with suppress(interactions.errors.Forbidden):
            roles.append(await guild.create_role("_test_role3"))
            # will fail if the test server has not enabled the feature
            roles.append(await guild.create_role("_test_role1", icon="💥"))
            roles.append(await guild.create_role("_test_role2", icon=r"tests/LordOfPolls.png"))

        assert roles[0].icon is None
        with suppress(IndexError):
            assert isinstance(roles[1].icon, PartialEmoji)
            assert isinstance(roles[2].icon, Asset)
        await guild.me.remove_role(roles[0])

        await roles[0].edit(name="_test_renamed", color=BrandColors.RED)

        assert roles[0].name == "_test_renamed"

        for role in roles:
            await role.delete()

    finally:
        for role in guild.roles:
            if role.name.startswith("_test"):
                with suppress(NotFound):
                    await role.delete()


@pytest.mark.asyncio
async def test_members(bot: Client, guild: Guild, channel: GuildText) -> None:
    member = guild.me
    ensure_attributes(member)

    await member.edit_nickname("Test Nickname")
    with suppress(asyncio.exceptions.TimeoutError):
        await bot.wait_for("member_update", timeout=2)
    assert member.display_name == "Test Nickname"

    await member.edit_nickname(None)
    with suppress(asyncio.exceptions.TimeoutError):
        await bot.wait_for("member_update", timeout=2)
    assert member.display_name == (bot.get_user(member.id)).username

    base_line = len(member.roles)

    assert len(member.roles) == base_line
    role = await guild.create_role(f"test-{bot.suffix}")
    try:
        await member.add_role(role)
        with suppress(asyncio.exceptions.TimeoutError):
            await bot.wait_for("member_update", timeout=2)
        assert len(member.roles) != base_line
        await member.remove_role(role)
        with suppress(asyncio.exceptions.TimeoutError):
            await bot.wait_for("member_update", timeout=2)
        assert len(member.roles) == base_line

        assert member.display_avatar is not None
        assert member.display_name is not None

        assert member.has_permission(Permissions.SEND_MESSAGES)
        assert member.channel_permissions(channel)

        assert member.guild_permissions is not None
    finally:
        await role.delete()


@pytest.mark.asyncio
async def test_gateway(bot: Client) -> None:
    try:
        gateway: WebsocketClient = bot._connection_state.gateway

        assert gateway._entered
        assert gateway._keep_alive is not None

        await bot.change_presence(Status.DO_NOT_DISTURB, activity="Testing")
        await bot.change_presence()

        await gateway.send_heartbeat()
        await gateway._acknowledged.wait()
    finally:
        await bot.change_presence(activity="Testing someones code")


@pytest.mark.asyncio
async def test_ratelimit(bot: Client, channel: GuildText) -> None:
    msg = await channel.send("- Abusing the api... please wait")
    await msg.add_reaction("🤔")
    await msg.remove_reaction("🤔")

    limit = bot.http.get_ratelimit(
        Route(
            "DELETE",
            "/channels/{channel_id}/messages/{message_id}/reactions/{emoji}/@me",
            channel_id=msg.channel.id,
            message_id=msg.id,
            emoji=process_emoji_req_format("🤔"),
        )
    )
    await msg.add_reaction("🤔")
    await msg.remove_reaction("🤔")
    assert limit.locked
    await msg.add_reaction("🤔")
    await msg.remove_reaction("🤔")
    assert limit.locked


@pytest.mark.asyncio
async def test_embeds(bot: Client, channel: GuildText) -> None:
    msg = await channel.send("Embed Tests")
    thread = await msg.create_thread("Test Thread")

    try:
        e = Embed("Test")
        await thread.send(embeds=e)

        e = Embed(
            "Test",
            "Test",
            BrandColors.RED,
            "https://github.com/",
            datetime.now(),
            [
                EmbedField("name", "value"),
                EmbedField("name2", "value2"),
                EmbedField("name3", "value3"),
            ],
            EmbedAuthor(bot.user.display_name, bot.user.avatar.url),
            EmbedAttachment(bot.user.avatar.url),
            EmbedAttachment(bot.owner.avatar.url),
            footer=EmbedFooter("Test", icon_url=bot.user.avatar.url),
        )
        await thread.send(embeds=e)

        e = Embed("Test")
        e.color = BrandColors.RED
        e.url = "https://github.com/"
        e.timestamp = datetime.now()
        e.set_image(bot.user.avatar.url)
        e.set_thumbnail(bot.user.avatar.url)
        e.set_author("Test", bot.owner.avatar.url)
        e.set_footer("Test")
        e.add_field("test", "test")
        e.add_field("test2", "test2")
        await thread.send(embeds=e)

        await thread.delete()
    finally:
        with suppress(interactions.errors.NotFound):
            await thread.delete()


@pytest.mark.asyncio
async def test_components(bot: Client, channel: GuildText) -> None:
    msg = await channel.send("Component Tests")
    thread = await msg.create_thread("Test Thread")

    try:
        await thread.send("Test - single", components=Button(style=ButtonStyle.PRIMARY, label="test"))
        await thread.send(
            "Test - list",
            components=[
                Button(style=ButtonStyle.PRIMARY, label="test"),
                Button(style=ButtonStyle.PRIMARY, label="test"),
            ],
        )
        await thread.send(
            "Test - ActionRow",
            components=ActionRow(
                *[Button(style=ButtonStyle.PRIMARY, label="test"), Button(style=ButtonStyle.PRIMARY, label="test")]
            ),
        )
        await thread.send(
            "Test - StringSelectMenu",
            components=StringSelectMenu(StringSelectOption(label="test", value="test")),
        )

        Modal(ParagraphText(label="test", value="test value, press send"), title="Test Modal")

    finally:
        with suppress(interactions.errors.NotFound):
            await thread.delete()


@pytest.mark.asyncio
async def test_polls(bot: Client, channel: GuildText) -> None:
    msg = await channel.send("Polls Tests")
    thread = await msg.create_thread("Test Thread")

    try:
        poll_1 = Poll.create("Test Poll", duration=1, answers=["Answer 1", "Answer 2"])
        test_data_1 = {
            "question": {"text": "Test Poll"},
            "layout_type": 1,
            "duration": 1,
            "allow_multiselect": False,
            "answers": [{"poll_media": {"text": "Answer 1"}}, {"poll_media": {"text": "Answer 2"}}],
        }
        poll_1_dict = poll_1.to_dict()
        for key in poll_1_dict.keys():
            assert poll_1_dict[key] == test_data_1[key]

        msg_1 = await thread.send(poll=poll_1)

        assert msg_1.poll is not None
        assert msg_1.poll.question.to_dict() == PollMedia(text="Test Poll").to_dict()
        assert msg_1.poll.expiry <= msg_1.created_at + timedelta(hours=1, minutes=1)
        poll_1_answer_medias = [poll_answer.poll_media.to_dict() for poll_answer in msg_1.poll.answers]
        assert poll_1_answer_medias == [
            PollMedia.create(text="Answer 1").to_dict(),
            PollMedia.create(text="Answer 2").to_dict(),
        ]

        poll_2 = Poll.create("Test Poll 2", duration=1, allow_multiselect=True)
        poll_2.add_answer("Answer 1")
        poll_2.add_answer("Answer 2")
        test_data_2 = {
            "question": {"text": "Test Poll 2"},
            "layout_type": 1,
            "duration": 1,
            "allow_multiselect": True,
            "answers": [{"poll_media": {"text": "Answer 1"}}, {"poll_media": {"text": "Answer 2"}}],
        }
        poll_2_dict = poll_2.to_dict()
        for key in poll_2_dict.keys():
            assert poll_2_dict[key] == test_data_2[key]
        msg_2 = await thread.send(poll=poll_2)

        assert msg_2.poll is not None
        assert msg_2.poll.question.to_dict() == PollMedia(text="Test Poll 2").to_dict()
        assert msg_2.poll.expiry <= msg_2.created_at + timedelta(hours=1, minutes=1)
        assert msg_2.poll.allow_multiselect
        poll_2_answer_medias = [poll_answer.poll_media.to_dict() for poll_answer in msg_2.poll.answers]
        assert poll_2_answer_medias == [
            PollMedia.create(text="Answer 1").to_dict(),
            PollMedia.create(text="Answer 2").to_dict(),
        ]

        poll_3 = Poll.create(
            "Test Poll 3",
            duration=1,
            answers=[PollMedia.create(text="One", emoji="1️⃣"), PollMedia.create(text="Two", emoji="2️⃣")],
        )
        test_data_3 = {
            "question": {"text": "Test Poll 3"},
            "layout_type": 1,
            "duration": 1,
            "allow_multiselect": False,
            "answers": [
                {"poll_media": {"text": "One", "emoji": {"name": "1️⃣", "animated": False}}},
                {"poll_media": {"text": "Two", "emoji": {"name": "2️⃣", "animated": False}}},
            ],
        }
        poll_3_dict = poll_3.to_dict()
        for key in poll_3_dict.keys():
            assert poll_3_dict[key] == test_data_3[key]

        msg_3 = await thread.send(poll=poll_3)

        assert msg_3.poll is not None
        assert msg_3.poll.question.to_dict() == PollMedia(text="Test Poll 3").to_dict()
        assert msg_3.poll.expiry <= msg_3.created_at + timedelta(hours=1, minutes=1)
        poll_3_answer_medias = [poll_answer.poll_media.to_dict() for poll_answer in msg_3.poll.answers]
        assert poll_3_answer_medias == [
            PollMedia.create(text="One", emoji="1️⃣").to_dict(),
            PollMedia.create(text="Two", emoji="2️⃣").to_dict(),
        ]

    finally:
        with suppress(interactions.errors.NotFound):
            await thread.delete()


@pytest.mark.asyncio
async def test_webhooks(bot: Client, guild: Guild, channel: GuildText) -> None:
    test_thread = await channel.create_thread("Test Thread")

    hook = await channel.create_webhook("Test")
    await hook.send("Test 123")
    await hook.delete()

    hook = await channel.create_webhook("Test-Avatar", r"tests/LordOfPolls.png")

    _m = await hook.send("Test", wait=True)
    assert isinstance(_m, Message)
    assert _m.webhook_id == hook.id
    await hook.send("Test", username="Different Name", wait=True)
    await hook.send("Test", avatar_url=bot.user.avatar.url, wait=True)
    _m = await hook.send("Test", thread=test_thread, wait=True)
    assert _m is not None
    assert _m.channel == test_thread

    await hook.delete()


@pytest.mark.asyncio
async def test_voice(bot: Client, guild: Guild) -> None:
    test_channel = await guild.create_voice_channel(f"_test_voice-{bot.suffix}")
    test_channel_two = await guild.create_voice_channel(f"_test_voice_two-{bot.suffix}")
    try:
        try:
            import nacl  # noqa
        except ImportError:
            # testing on a non-voice extra
            return

        vc = await test_channel.connect(deafened=True)
        assert vc == bot.get_bot_voice_state(guild.id)

        audio = AudioVolume("tests/test_audio.mp3")
        vc.play_no_wait(audio)
        await asyncio.sleep(5)

        assert len(vc.current_audio.buffer) != 0
        assert vc.player._sent_payloads != 0

        await vc.move(test_channel_two)
        await asyncio.sleep(2)

        _before = vc.player._sent_payloads

        await test_channel_two.connect(deafened=True)

        await asyncio.sleep(2)

        assert vc.player._sent_payloads != _before

        vc.volume = 1
        await asyncio.sleep(1)
        vc.volume = 0.5

        vc.pause()
        await asyncio.sleep(0.1)
        assert vc.player.paused
        vc.resume()
        await asyncio.sleep(0.1)
        assert not vc.player.paused

        await vc.disconnect()
        await vc._close_connection()
        await vc.ws._closed.wait()
    finally:
        await test_channel.delete()
        await test_channel_two.delete()


@pytest.mark.asyncio
async def test_emoji(bot: Client, guild: Guild) -> None:
    emoji = None
    try:
        emoji = await guild.create_custom_emoji("testEmoji", r"tests/LordOfPolls.png")
        assert emoji.animated is False

        fetched_emoji = await bot.fetch_custom_emoji(emoji.id, guild.id)

        assert emoji == fetched_emoji
        assert emoji.animated == fetched_emoji.animated
        ensure_attributes(emoji)
        ensure_attributes(fetched_emoji)

        await emoji.edit(name="testEditedName")
        await asyncio.sleep(1)
        fetched_emoji = await bot.fetch_custom_emoji(emoji.id, guild.id)
        assert fetched_emoji.name == "testEditedName"

    finally:
        if emoji:
            await emoji.delete()