Saltar a contenido

bostie.testing

bostie.testing

Order 

Order = Literal['state_first', 'server_first']

Scenario 

Scenario = Callable[[FakeDiscord], Coroutine[Any, Any, None]]

Connection dataclass 

Connection(request: Request, ws: WebSocketResponse, compress: Callable[[bytes], bytes] | None, live: bool = False, close_code: int | None = None, shard: int = 0, received: list[Any] = list[Any]())

A gateway connection, with the close code the client sent if it closed first.

request instance-attribute 

request: Request

ws instance-attribute 

ws: WebSocketResponse

compress instance-attribute 

compress: Callable[[bytes], bytes] | None

live class-attribute instance-attribute 

live: bool = False

close_code class-attribute instance-attribute 

close_code: int | None = None

shard class-attribute instance-attribute 

shard: int = 0

received class-attribute instance-attribute 

received: list[Any] = field(default_factory=list[Any])

path property 

path: str

/gateway or /gateway-resume.

send async 

send(frame: bytes) -> None

Send one gateway payload, compressed in compress=True mode.

A payload sent to a closing connection is dropped.

FakeDiscord 

FakeDiscord(*, token: str = _FAKE_TOKEN, heartbeat_interval: int = 41250, compress: bool = False, voice: bool = False, voice_heartbeat_interval: float = 13750.0, voice_order: Order = 'state_first')

A local Discord gateway and REST API for tests, served inside async with.

voice=True adds voice: op 4 answers, voice routes, and in voice the voice gateway and UDP server. mute, deaf and suppress go into every voice state the fake sends for the bot. ready_guilds are the guild ids READY lists, each on its own shard. cdn_requests records the downloads of cdn_file URLs.

requests instance-attribute 

requests: list[Recorded]

cdn_requests instance-attribute 

cdn_requests: list[Recorded]

voice instance-attribute 

voice = VoiceGateway(voice_heartbeat_interval, port)

api_url instance-attribute 

api_url = f'http://127.0.0.1:{port}{_API}'

gateway_url instance-attribute 

gateway_url = f'ws://127.0.0.1:{port}/gateway'

token instance-attribute 

token = token

application_id instance-attribute 

application_id = int(application['id'])

bot_user instance-attribute 

bot_user: Json = payloads.fixture('events/READY')['user']

guilds instance-attribute 

guilds: dict[str, Json] = {guild['id']: guild}

channels instance-attribute 

channels: dict[str, Json] = {channel['id']: channel for channel in channels}

ready_guilds instance-attribute 

ready_guilds: list[int] = []

users instance-attribute 

users: dict[str, Json] = {user['id']: user for user in (payloads.user(), member['user'])}

members instance-attribute 

members: dict[tuple[str, str], Json] = {(guild['id'], member['user']['id']): member}

messages instance-attribute 

messages: dict[str, dict[str, Json]] = {}

commands instance-attribute 

commands: list[Json] = []

guild_commands instance-attribute 

guild_commands: dict[str, list[Json]] = {}

put_count instance-attribute 

put_count = 0

gateway_received instance-attribute 

gateway_received: Queue[Any] = asyncio.Queue()

connections instance-attribute 

connections: list[Connection] = []

voice_order instance-attribute 

voice_order: Order = voice_order

mute instance-attribute 

mute = False

deaf instance-attribute 

deaf = False

suppress instance-attribute 

suppress = False

voice_states instance-attribute 

voice_states: dict[tuple[str, str], Json] = {}

dispatch async 

dispatch(name: str, d: Any, *, shard: int = 0) -> int

Send a dispatch event on a shard, kept for RESUME replay; return its seq.

send_op async 

send_op(op: int, d: Any = None) -> None

Send a non-dispatch opcode, such as 7 (reconnect) or 9 (invalid session).

close_gateway async 

close_gateway(code: int) -> None

Close the latest gateway connection with code.

drop_gateway 

drop_gateway() -> None

Abort the latest gateway connection's TCP stream without a close frame.

hold_acks 

hold_acks() -> None

Leave heartbeats unanswered from now on, like a zombie link.

next_voice_state_update async 

next_voice_state_update() -> Any

Wait for the next op 4 the bot sent and return its d.

hold_voice_state 

hold_voice_state() -> Event

Leave the next op 4 unanswered until the returned event is set.

move_bot async 

move_bot(channel_id: int) -> None

Move the bot to channel_id as a moderator would.

The voice socket of the old channel then closes with 4014.

kick_bot async 

kick_bot() -> None

Disconnect the bot from voice: a voice state without channel, then 4014.

next_identify async 

next_identify() -> Json

Wait for the next IDENTIFY and return its d.

hold_identify 

hold_identify(shard: int) -> Event

Leave shard's IDENTIFY unanswered until the returned event is set.

hold_hello 

hold_hello() -> Event

Leave the next gateway connection without HELLO until the event is set.

next_resume async 

next_resume() -> Json

Wait for the next RESUME and return its d.

interaction async 

interaction(type: int, data: Json, **fields: Any) -> Recorded

Dispatch payloads.interaction(...) and return its first callback.

expect_callback 

expect_callback(interaction: Json) -> Future[Recorded]

Accept callbacks for interaction; the future gets the first one.

rate_limit 

rate_limit(method: str, route_prefix: str, *, scope: Literal['user', 'shared', 'global'] = 'user', retry_after: float = 0.1) -> None

Answer the next matching request with a 429 of scope.

fail 

fail(method: str, route_prefix: str, status: int, json: Any) -> None

Answer the next matching request with status and json.

This also stubs routes the fake does not model, which otherwise get a 404.

answer 

answer(method: str, route_prefix: str, body: bytes, content_type: str) -> None

Answer the next matching request with a 200 that is not JSON.

cdn_file 

cdn_file(path: str, data: bytes) -> str

Serve data at a CDN path like attachments/1/2/a.png; return its URL.

expect_request 

expect_request(method: str, route_prefix: str) -> Future[Recorded]

A future for the next matching request the bot makes from now on.

hold 

hold(method: str, route_prefix: str) -> Event

Keep the next matching request waiting until the returned event is set.

Part dataclass 

Part(name: str | None, filename: str | None, content_type: str | None, data: bytes)

One part of a multipart request body.

name instance-attribute 

name: str | None

filename instance-attribute 

filename: str | None

content_type instance-attribute 

content_type: str | None

data instance-attribute 

data: bytes

Recorded dataclass 

Recorded(method: str, path: str, query: dict[str, str], headers: dict[str, str], json: Any, parts: list[Part])

A REST request, with path relative to api_url.

json holds the JSON body, the payload_json part of a multipart body, or the fields of a form-urlencoded body.

method instance-attribute 

method: str

path instance-attribute 

path: str

query instance-attribute 

query: dict[str, str]

headers instance-attribute 

headers: dict[str, str]

json instance-attribute 

json: Any

parts instance-attribute 

parts: list[Part]

Packet dataclass 

Packet(sequence: int, timestamp: int, ssrc: int, nonce: int, payload: bytes, arrived: float)

An RTP packet the fake received, opened with its session's AES-GCM key.

sequence instance-attribute 

sequence: int

timestamp instance-attribute 

timestamp: int

ssrc instance-attribute 

ssrc: int

nonce instance-attribute 

nonce: int

payload instance-attribute 

payload: bytes

arrived instance-attribute 

arrived: float

VoiceSession dataclass 

VoiceSession(identify: Json, ssrc: int, secret_key: bytes = (lambda: token_bytes(32))(), mode: str | None = None, socket: VoiceSocket | None = None, sent: list[Frame] = list[Frame](), address: Address | None = None, nonces: Iterator[int] = count())

A voice session from its Identify on.

sent keeps every sequenced frame, so a resume can replay what was missed; address is the bot's UDP address from discovery.

identify instance-attribute 

identify: Json

ssrc instance-attribute 

ssrc: int

secret_key class-attribute instance-attribute 

secret_key: bytes = field(default_factory=lambda: secrets.token_bytes(32))

mode class-attribute instance-attribute 

mode: str | None = None

socket class-attribute instance-attribute 

socket: VoiceSocket | None = None

sent class-attribute instance-attribute 

sent: list[Frame] = field(default_factory=list[Frame])

address class-attribute instance-attribute 

address: Address | None = None

nonces class-attribute instance-attribute 

nonces: Iterator[int] = field(default_factory=itertools.count)

VoiceSocket dataclass 

VoiceSocket(request: Request, ws: WebSocketResponse, session: VoiceSession | None = None, close_code: int | None = None, closing: bool = False)

A voice gateway connection, with the close code the client sent if any.

request instance-attribute 

request: Request

ws instance-attribute 

ws: WebSocketResponse

session class-attribute instance-attribute 

session: VoiceSession | None = None

close_code class-attribute instance-attribute 

close_code: int | None = None

closing class-attribute instance-attribute 

closing: bool = False

send async 

send(frame: Frame) -> None

Send one frame; nothing is sent once the socket is closing.

close async 

close(code: int) -> None

Send a close frame and keep reading until the client answers.

run 

run(scenario: Scenario, **options: Any) -> None

Run async def scenario(fake) against a new FakeDiscord, within 10 s.

settle async 

settle(rounds: int = 5) -> None

Yield to the event loop rounds times so ready callbacks and tasks run.

running async 

running(bot: Bot, fake: FakeDiscord) -> AsyncGenerator[None]

Start bot on fake, wait for READY, and close the bot on exit.

bostie.testing.payloads

Json 

Json = dict[str, Any]

fixture 

fixture(name: str) -> Json

A fresh copy of the documented payload fixtures/{name}.json.

snowflake 

snowflake(at: float | None = None) -> str

A unique snowflake created at at (Unix seconds, now by default).

user 

user(**fields: Any) -> Json

A user, as GET /users/{id} returns it.

member 

member(**fields: Any) -> Json

A guild member with its user.

role 

role(**fields: Any) -> Json

A guild role.

channel 

channel(**fields: Any) -> Json

A guild text channel.

guild 

guild(**fields: Any) -> Json

A GUILD_CREATE with channels (one obfuscated), a thread, roles and members.

message 

message(**fields: Any) -> Json

A MESSAGE_CREATE payload with a fresh id.

interaction 

interaction(type: int, data: Json, **fields: Any) -> Json

An INTERACTION_CREATE payload of type with a fresh id and token.

command 

command(name: str, *options: Json, resolved: Json | None = None) -> Json

The data of a slash command or autocomplete interaction.

context_menu 

context_menu(type: int, name: str, target_id: str, resolved: Json) -> Json

The data of a user (2) or message (3) command run on target_id.

group 

group(name: str, *subcommands: Json) -> Json

A subcommand group option.

subcommand 

subcommand(name: str, *options: Json) -> Json

A subcommand option.

option 

option(name: str, type: int, value: Any, *, focused: bool = False) -> Json

A command option value; focused marks the one being autocompleted.

resolved 

resolved() -> Json

Resolved users with their members, a role, a channel and an attachment.

button 

button(custom_id: str) -> Json

The data of a button interaction.

select 

select(custom_id: str, *values: str) -> Json

The data of a string select interaction.

modal 

modal(custom_id: str) -> Json

The data of a modal submit.

voice_channel 

voice_channel(**fields: Any) -> Json

The guild's voice channel, as GET /channels/{id} returns it.

stage_channel 

stage_channel(**fields: Any) -> Json

A stage channel in the same guild.

voice_state 

voice_state(**fields: Any) -> Json

A VOICE_STATE_UPDATE payload, by default the bot's own.