Testando o seu bot¶

bostie.testing roda um Discord falso no localhost: um gateway, uma API REST com as rotas que um bot mais usa e, opcionalmente, um servidor de voz. O seu bot de verdade se conecta a ele, e o teste conduz o lado do Discord: envia eventos e interações e depois confere o que o bot mandou de volta. Ele precisa do extra testing (Instalação), e a própria suíte de testes do bostie roda nele.
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 um FakeDiscord, roda async def scenario(fake) contra ele em até 10 segundos e o encerra; é uma função comum, então o pytest não precisa de plugin. async with running(bot, fake): inicia o bot no Discord falso, espera até ele ficar pronto e o fecha no final. Aponte o bot para o Discord falso com Bot(api_url=fake.api_url), ou defina BOSTIE_API_URL quando o bot é montado em outro lugar.
Conduzindo o lado do Discord¶
| Chamada | Faz |
|---|---|
await fake.interaction(type, data) |
envia um INTERACTION_CREATE e retorna a primeira resposta do bot como uma requisição Recorded |
await fake.dispatch(name, d, shard=0) |
envia qualquer evento do gateway e retorna o número de sequência dele |
fake.expect_callback(payload) |
para uma interação que você mesmo envia: um future para a primeira resposta dela |
await fake.send_op(op, d) |
envia um opcode que não é dispatch, como 7 (reconectar) ou 9 (sessão inválida) |
await fake.close_gateway(code), fake.drop_gateway() |
fecha a conexão com um código de fechamento, ou a derruba sem um |
fake.hold_acks() |
para de responder aos heartbeats, como uma conexão morta faria |
await fake.next_identify(), await fake.next_resume() |
espera o próximo IDENTIFY ou RESUME do bot |
Os tipos de interação são os números do Discord: 2 para um comando, 3 para um componente, 4 para preenchimento automático, 5 para o envio de um modal.
bostie.testing.payloads monta os payloads do Discord a partir dos exemplos da documentação dele, com IDs novos:
command(name, *options),option(name, type, value, focused=False),subcommand(...),group(...)eresolved()para comandos e preenchimento automático;context_menu(type, name, target_id, resolved)para comandos de usuário e de mensagem;button(custom_id),select(custom_id, *values)emodal(custom_id)para componentes e modais;message(...),guild(...),member(...),user(...),role(...),channel(...),voice_channel(...),stage_channel(...)evoice_state(...)para eventos, cada um com argumentos nomeados que sobrescrevem campos;fixture(name)para qualquer payload documentado que o bostie traz, esnowflake()para um ID novo.
Conferindo o que o bot enviou¶
fake.requests lista cada requisição REST como um Recorded, e fake.expect_request(method, route_prefix) retorna um future para a próxima que corresponder, para uma requisição que o bot faz por conta própria, como a partir de uma tarefa. Um Recorded tem method, path (relativo à base da API), query, headers, json e, para uploads, parts. Um Recorded que responde a uma interação tem o corpo do callback em json, como acima. fake.gateway_received é uma fila dos payloads que o bot enviou pelo gateway, e fake.connections as conexões do gateway, objetos Connection com os seus payloads received, shard e close_code.
fake.cdn_file(path, data) serve data em uma URL da CDN que ele retorna, para um anexo ou asset que o bot baixa, e fake.cdn_requests registra esses downloads.
O Discord falso guarda o estado de que essas rotas precisam, e os testes podem lê-lo ou populá-lo: fake.guilds, fake.channels, fake.users, fake.members, fake.messages, fake.commands e fake.guild_commands, com fake.put_count para as sincronizações de comandos. fake.ready_guilds define os IDs de servidor que o READY lista.
Falhas e tempo¶
fake.fail(method, route_prefix, status, json)responde à próxima requisição correspondente com um erro, e simula uma rota que o Discord falso não serve.fake.rate_limit(method, route_prefix, scope="user", retry_after=0.1)responde a ela com um 429.fake.answer(method, route_prefix, body, content_type)responde a ela com um corpo que não é JSON.fake.hold(method, route_prefix)a deixa esperando até você definir oasyncio.Eventretornado;fake.hold_identify(shard)efake.hold_hello()fazem o mesmo para o handshake do gateway.await settle()cede ao event loop algumas vezes, para que os callbacks agendados pelo que você acabou de fazer rodem.
Para esperar algo que o bot faz em um listener, faça-o enviar um evento próprio e espere por ele, como o segundo teste faz, em vez de usar sleep.
FakeDiscord(compress=True) comprime o gateway com zstd, o que precisa do compression.zstd do Python 3.14; heartbeat_interval define o intervalo que ele pede, os 41 250 ms do Discord por padrão.
Voz¶
FakeDiscord(voice=True) adiciona um canal de voz e um canal de palco, responde às atualizações de estado de voz e roda um gateway de voz e um servidor UDP como fake.voice, para que ctx.voice.connect() e vc.play(...) funcionem contra ele. Ele não fala DAVE; os testes em Rust do bostie cobrem isso. O teste pode:
- esperar o que o bot enviou com
await fake.next_voice_state_update(),await fake.voice.next_identify()eawait fake.voice.wait_for_packets(count), que retorna os pacotes RTP descriptografados como objetosPacket; cada conexão de voz é umaVoiceSessionemfake.voice.sessions, sobre umVoiceSocket, efake.voice.op(op, d)envia à mais recente qualquer opcode de voz; - agir como outros usuários com
fake.voice.clients_connect(user_ids),fake.voice.client_disconnect(user_id),fake.voice.speaking(user_id, ssrc, flags),fake.voice.send_audio(...), que envia pacotes Opus ao bot,fake.voice.send_rtcp(packet), que envia a ele um pacote RTCP, efake.voice.send_datagram(data), que envia bytes sem criptografia, como um servidor quebrado faria; - mover ou desconectar o bot como um moderador faria, com
fake.move_bot(channel_id)efake.kick_bot(); - deixar sem resposta a próxima atualização de estado de voz, como o Discord faz quando o canal está cheio, com
fake.hold_voice_state(), até oasyncio.Eventretornado ser definido; - definir
fake.mute,fake.deafefake.suppresspara os estados de voz que ele envia sobre o bot, evoice_orderpara a ordem dos dois eventos que o Discord envia ao entrar.