Tester votre bot¶

bostie.testing lance un Discord factice sur localhost : une gateway, une API REST avec les routes qu'un bot utilise le plus, et en option un serveur vocal. Votre vrai bot s'y connecte, et le test pilote le côté Discord : il envoie des événements et des interactions, puis vérifie ce que le bot a renvoyé. Il nécessite l'extra testing (Installation), et la suite de tests de bostie elle-même tourne dessus.

from bostie import Bot, ComponentContext, Context, ui
from bostie.events import BotEvent, MessageCreate
from bostie.testing import FakeDiscord, payloads, run, running

INTERACTION_COMMAND = 2
INTERACTION_COMPONENT = 3


class Counted(BotEvent, kw_only=True):
    count: int


def make_bot(api_url: str) -> Bot:
    bot = Bot(api_url=api_url)
    seen: list[str] = []

    @bot.command
    async def hello(ctx: Context) -> None:
        """Say hello."""
        wave = ui.button("Wave", custom_id="wave")
        await ctx.respond("Hello!", components=[ui.row(wave)])

    @bot.on_component("wave")
    async def wave(ctx: ComponentContext) -> None:
        await ctx.update(content="*waves*")

    @bot.listen
    async def count(event: MessageCreate) -> None:
        seen.append(event.message.content)
        await bot.rest.messages.create(event.message.channel_id, f"{len(seen)} seen")
        bot.dispatch(Counted(count=len(seen)))

    return bot


def test_hello_answers_and_the_button_edits_the_message() -> None:
    async def scenario(fake: FakeDiscord) -> None:
        async with running(make_bot(fake.api_url), fake):
            reply = await fake.interaction(
                INTERACTION_COMMAND, payloads.command("hello")
            )
            update = await fake.interaction(
                INTERACTION_COMPONENT, payloads.button("wave")
            )

        assert reply.json["data"]["content"] == "Hello!"
        assert update.json["data"]["content"] == "*waves*"

    run(scenario)


def test_each_message_is_counted_out_loud() -> None:
    async def scenario(fake: FakeDiscord) -> None:
        bot = make_bot(fake.api_url)
        async with running(bot, fake), bot.stream(Counted) as counted:
            await fake.dispatch("MESSAGE_CREATE", payloads.message(content="hi"))
            first = await anext(counted)

        sent = [r.json for r in fake.requests if r.path.endswith("/messages")]
        assert first.count == 1
        assert [message["content"] for message in sent] == ["1 seen"]

    run(scenario)

run(scenario) démarre un FakeDiscord, exécute async def scenario(fake) contre lui en 10 secondes maximum, puis l'arrête ; c'est une fonction ordinaire, donc pytest n'a besoin d'aucun plugin. async with running(bot, fake): démarre le bot sur le Discord factice, attend qu'il soit prêt et le ferme à la fin. Dirigez le bot vers le Discord factice avec Bot(api_url=fake.api_url), ou définissez BOSTIE_API_URL quand le bot est construit ailleurs.

Piloter le côté Discord¶

Les types d'interaction sont les numéros de Discord : 2 pour une commande, 3 pour un composant, 4 pour l'autocomplétion, 5 pour l'envoi d'une fenêtre modale.

bostie.testing.payloads construit les payloads de Discord à partir des exemples de sa documentation, avec de nouveaux id :

• command(name, *options), option(name, type, value, focused=False), subcommand(...), group(...) et resolved() pour les commandes et l'autocomplétion ;
• context_menu(type, name, target_id, resolved) pour les commandes utilisateur et les commandes de message ;
• button(custom_id), select(custom_id, *values) et modal(custom_id) pour les composants et les fenêtres modales ;
• message(...), guild(...), member(...), user(...), role(...), channel(...), voice_channel(...), stage_channel(...) et voice_state(...) pour les événements, chacun avec des arguments nommés qui remplacent des champs ;
• fixture(name) pour n'importe quel payload documenté que bostie embarque, et snowflake() pour un nouvel id.

Vérifier ce que le bot a envoyé¶

fake.requests liste chaque requête REST sous forme de Recorded, et fake.expect_request(method, route_prefix) renvoie un future pour la prochaine requête correspondante, pour une requête que le bot fait de lui-même, par exemple depuis une tâche. Un Recorded a method, path (relatif à la base de l'API), query, headers, json et, pour les envois de fichiers, parts. Un Recorded qui répond à une interaction contient le corps du callback dans json, comme ci-dessus. fake.gateway_received est une file d'attente des payloads que le bot a envoyés sur la gateway, et fake.connections les connexions à la gateway, des objets Connection avec leurs payloads received, leur shard et leur close_code.

fake.cdn_file(path, data) sert data à une URL de CDN qu'il renvoie, pour une pièce jointe ou une ressource que le bot télécharge, et fake.cdn_requests enregistre ces téléchargements.

Le Discord factice garde l'état dont ces routes ont besoin, et les tests peuvent le lire ou le pré-remplir : fake.guilds, fake.channels, fake.users, fake.members, fake.messages, fake.commands et fake.guild_commands, avec fake.put_count pour les synchronisations de commandes. fake.ready_guilds définit les id de serveurs que READY liste.

Échecs et timing¶

• fake.fail(method, route_prefix, status, json) répond à la prochaine requête correspondante par une erreur, et remplace une route que le Discord factice ne sert pas.
• fake.rate_limit(method, route_prefix, scope="user", retry_after=0.1) y répond par un 429.
• fake.answer(method, route_prefix, body, content_type) y répond par un corps qui n'est pas du JSON.
• fake.hold(method, route_prefix) la fait attendre jusqu'à ce que vous déclenchiez l'asyncio.Event renvoyé ; fake.hold_identify(shard) et fake.hold_hello() font de même pour la poignée de main de la gateway.
• await settle() rend la main à la boucle d'événements quelques fois, pour que les callbacks planifiés par ce que vous venez de faire s'exécutent.

Pour attendre quelque chose que le bot fait dans un écouteur, faites-lui émettre un événement à lui et attendez celui-ci, comme le fait le second test, plutôt que de faire un sleep.

FakeDiscord(compress=True) compresse la gateway avec zstd, ce qui nécessite le module compression.zstd de Python 3.14 ; heartbeat_interval définit l'intervalle qu'il demande, 41 250 ms par défaut comme chez Discord.

Voix¶

FakeDiscord(voice=True) ajoute un salon vocal et un salon de conférence, répond aux mises à jour d'état vocal et fait tourner une gateway vocale et un serveur UDP sous fake.voice, de sorte que ctx.voice.connect() et vc.play(...) fonctionnent contre lui. Il ne parle pas DAVE ; les tests Rust de bostie couvrent cette partie. Le test peut :

• attendre ce que le bot a envoyé avec await fake.next_voice_state_update(), await fake.voice.next_identify() et await fake.voice.wait_for_packets(count), qui renvoie les paquets RTP déchiffrés sous forme d'objets Packet ; chaque connexion vocale est une VoiceSession dans fake.voice.sessions, sur un VoiceSocket, et fake.voice.op(op, d) envoie n'importe quel opcode vocal à la plus récente ;
• jouer d'autres utilisateurs avec fake.voice.clients_connect(user_ids), fake.voice.client_disconnect(user_id), fake.voice.speaking(user_id, ssrc, flags), fake.voice.send_audio(...), qui envoie des paquets Opus au bot, fake.voice.send_rtcp(packet), qui lui envoie un paquet RTCP, et fake.voice.send_datagram(data), qui envoie des octets non chiffrés, comme le ferait un serveur défaillant ;
• déplacer ou déconnecter le bot comme le ferait un modérateur, avec fake.move_bot(channel_id) et fake.kick_bot() ;
• laisser la prochaine mise à jour d'état vocal sans réponse, comme le fait Discord quand le salon est plein, avec fake.hold_voice_state(), jusqu'à ce que l'asyncio.Event renvoyé soit déclenché ;
• définir fake.mute, fake.deaf et fake.suppress pour les états vocaux qu'il envoie à propos du bot, et voice_order pour l'ordre des deux événements que Discord envoie lors d'une connexion.