Probar tu bot¶

bostie.testing ejecuta un Discord falso en localhost: un gateway, una API REST con las rutas que más usa un bot y, de forma opcional, un servidor de voz. Tu bot real se conecta a él, y la prueba maneja el lado de Discord: envía eventos e interacciones y luego comprueba lo que el bot envió de vuelta. Necesita el extra testing (Instalación), y la propia suite de pruebas de bostie corre sobre él.
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) inicia un FakeDiscord, ejecuta async def scenario(fake) contra él en un máximo de 10 segundos y lo apaga; es una función común, así que pytest no necesita ningún plugin. async with running(bot, fake): inicia el bot sobre el Discord falso, espera a que esté listo y lo cierra al final. Apunta el bot al Discord falso con Bot(api_url=fake.api_url), o define BOSTIE_API_URL cuando el bot se construye en otro lugar.
Manejar el lado de Discord¶
| Llamada | Qué hace |
|---|---|
await fake.interaction(type, data) |
envía un INTERACTION_CREATE y devuelve la primera respuesta del bot como una solicitud Recorded |
await fake.dispatch(name, d, shard=0) |
envía cualquier evento del gateway y devuelve su número de secuencia |
fake.expect_callback(payload) |
para una interacción que despachas tú mismo: un future para su primera respuesta |
await fake.send_op(op, d) |
envía un opcode que no es dispatch, como 7 (reconectar) o 9 (sesión inválida) |
await fake.close_gateway(code), fake.drop_gateway() |
cierra la conexión con un código de cierre, o la corta sin ninguno |
fake.hold_acks() |
deja de responder a los heartbeats, como haría una conexión muerta |
await fake.next_identify(), await fake.next_resume() |
espera el siguiente IDENTIFY o RESUME del bot |
Los tipos de interacción son los números de Discord: 2 para un comando, 3 para un componente, 4 para el autocompletado, 5 para el envío de un modal.
bostie.testing.payloads construye los payloads de Discord a partir de los ejemplos de su documentación, con ids nuevos:
command(name, *options),option(name, type, value, focused=False),subcommand(...),group(...)yresolved()para los comandos y el autocompletado;context_menu(type, name, target_id, resolved)para los comandos de usuario y de mensaje;button(custom_id),select(custom_id, *values)ymodal(custom_id)para los componentes y los modales;message(...),guild(...),member(...),user(...),role(...),channel(...),voice_channel(...),stage_channel(...)yvoice_state(...)para los eventos, cada uno con argumentos por palabra clave que reemplazan campos;fixture(name)para cualquier payload documentado que incluye bostie, ysnowflake()para un id nuevo.
Comprobar lo que envió el bot¶
fake.requests enumera cada solicitud REST como un Recorded, y fake.expect_request(method, route_prefix) devuelve un future para la siguiente que coincida, para una solicitud que el bot hace por su cuenta, por ejemplo desde una tarea. Un Recorded tiene method, path (relativo a la base de la API), query, headers, json y, para las subidas, parts. Un Recorded que responde a una interacción tiene el cuerpo del callback en json, como arriba. fake.gateway_received es una cola de los payloads que el bot envió por el gateway, y fake.connections son las conexiones al gateway, objetos Connection con sus payloads received, su shard y su close_code.
fake.cdn_file(path, data) sirve data en una URL de la CDN que devuelve, para un archivo adjunto o un recurso que descarga el bot, y fake.cdn_requests registra esas descargas.
El Discord falso mantiene el estado que necesitan esas rutas, y las pruebas lo pueden leer o precargar: fake.guilds, fake.channels, fake.users, fake.members, fake.messages, fake.commands y fake.guild_commands, con fake.put_count para las sincronizaciones de comandos. fake.ready_guilds define los ids de servidor que enumera READY.
Fallos y tiempos¶
fake.fail(method, route_prefix, status, json)responde con un error a la siguiente solicitud que coincida, y sirve como stub de una ruta que el Discord falso no atiende.fake.rate_limit(method, route_prefix, scope="user", retry_after=0.1)la responde con un 429.fake.answer(method, route_prefix, body, content_type)la responde con un cuerpo que no es JSON.fake.hold(method, route_prefix)la mantiene en espera hasta que actives elasyncio.Eventque devuelve;fake.hold_identify(shard)yfake.hold_hello()hacen lo mismo con el handshake del gateway.await settle()le cede el control al event loop unas cuantas veces, para que se ejecuten los callbacks programados por lo que acabas de hacer.
Para esperar algo que el bot hace en un listener, haz que despache un evento propio y espera ese evento, como hace la segunda prueba, en lugar de dormir.
FakeDiscord(compress=True) comprime el gateway con zstd, lo que necesita el compression.zstd de Python 3.14; heartbeat_interval define el intervalo que pide, los 41 250 ms de Discord por defecto.
Voz¶
FakeDiscord(voice=True) agrega un canal de voz y un canal escenario, responde a las actualizaciones de estado de voz y ejecuta un gateway de voz y un servidor UDP como fake.voice, así que ctx.voice.connect() y vc.play(...) funcionan contra él. No habla DAVE; eso lo cubren las pruebas de Rust de bostie. La prueba puede:
- esperar lo que envió el bot con
await fake.next_voice_state_update(),await fake.voice.next_identify()yawait fake.voice.wait_for_packets(count), que devuelve los paquetes RTP descifrados como objetosPacket; cada conexión de voz es unaVoiceSessionenfake.voice.sessions, sobre unVoiceSocket, yfake.voice.op(op, d)le envía a la más reciente cualquier opcode de voz; - actuar como otros usuarios con
fake.voice.clients_connect(user_ids),fake.voice.client_disconnect(user_id),fake.voice.speaking(user_id, ssrc, flags),fake.voice.send_audio(...), que le envía paquetes Opus al bot,fake.voice.send_rtcp(packet), que le envía un paquete RTCP, yfake.voice.send_datagram(data), que envía bytes sin sellar, como haría un servidor roto; - mover o desconectar al bot como lo haría un moderador, con
fake.move_bot(channel_id)yfake.kick_bot(); - dejar sin respuesta la siguiente actualización de estado de voz, como hace Discord cuando el canal está lleno, con
fake.hold_voice_state(), hasta que se active elasyncio.Eventque devuelve; - definir
fake.mute,fake.deafyfake.suppresspara los estados de voz que envía sobre el bot, yvoice_orderpara el orden de los dos eventos que Discord envía al unirse.