From 4e723e16572aa3374494444ad8ea3bc6a9608468 Mon Sep 17 00:00:00 2001 From: YoursFunny Date: Mon, 21 Sep 2026 03:02:43 +0800 Subject: [PATCH] test: drive a real Bot against a stand-in API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every test went through `MockSender`, so `media_sender`'s `Bot` implementation — the URL it builds, the multipart it sends, the per-chat limiter and the bot-wide budget it charges — was never exercised, and neither was any handler reached from a real update. The two things that made that hard are gone: - `media_sender::test_support::fake_api::FakeApi` is a stand-in for `api.telegram.org`: a `tokio` TCP listener that reads one HTTP/1.1 request (JSON or multipart), records it and answers the smallest result the method needs. No new dependency, and `Bot::new(token).set_api_url(api.url())` points a real `Bot` at it. Note for future tests: teloxide keys methods by payload type, so the path is `SendMediaGroup`, not `sendMediaGroup`. - `message_handler` built its own `AppContext::from_statics` internally, so no test could reach its branches; its body is now `handle_message(ctx, bot, message)` with `message_handler` as the thin `dptree` entry. Tests: a media group through the real `Bot` (asserting the multipart fields — chat, media URL, caption — and that the send charged the chat's limiter), the forward button through the real callback path (`CopyMessages`, `DeleteMessage`, `AnswerCallbackQuery` with the prompt's ids and the toast text), and `handle_message` twice (a prompt reply becoming an `EditMessageCaption`, and a supported link in a group producing the one explanatory `SendMessage`). Also closes the redirect-hop gap left open by the download guard: the live `a_redirect_into_the_hosts_network_is_refused` follows a public redirector to `169.254.169.254` and asserts the policy refuses the hop (verified against httpbin.org here, and by mutation — disabling the hop check fails it). Docs: AGENTS.md's testing conventions and untested-modules list (the Bot implementation and the handler branches are covered now; `main.rs`'s startup/shutdown and its `dptree` tree still are not). `cargo fmt`, `cargo clippy --workspace --all-targets --locked -- -D warnings`, `cargo test --workspace --locked` (201 passed, 16 ignored) clean. --- AGENTS.md | 6 +- crates/x-media/src/site/mod.rs | 15 ++ crates/xmedia-bot/src/handlers/callback.rs | 29 ++- crates/xmedia-bot/src/handlers/mod.rs | 86 +++++++-- crates/xmedia-bot/src/media_sender.rs | 194 +++++++++++++++++++++ crates/xmedia-bot/src/rate_limit.rs | 9 + crates/xmedia-bot/src/send/mod.rs | 44 +++++ 7 files changed, 369 insertions(+), 14 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 15e3a3b..5486f75 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -47,7 +47,7 @@ The `x-media` library: `site::fetch(url)` dispatches through the `SITES` registr | `crates/xmedia-bot/src/queue.rs` | `PersistentTaskQueue`: SQLite-backed queue (`tasks` table), `QUEUE_WORKERS = 4` concurrent workers (lease via `BEGIN IMMEDIATE` + `locked_until` TTL), retry→dead-letter, a `lease_token` fence: `lease_next` stamps a random token and every write-back (heartbeat, `delete`, `reschedule`, `mark_done`) is guarded by it, so a lease that expired and was re-leased cannot be written by its former holder — a lost lease stops the attempt instead; a finished row's `DELETE`/reschedule retried and a failed delete falling back to a `done` tombstone (the lease query and the sweep only look at `pending`/`in_progress`, so a task that already ran cannot be resurrected and re-run), `runnable_rows`/`replace_payload` (the startup repair's read/rewrite path: it runs before the workers exist, which is why it needs no lease token), `notify_one` worker wakeup plus a separate `Notify` for the 30 s lease-expiry sweep (a shared one let the sweep steal the workers' wakeup permit; the sweep does notify the workers after it actually recovered a row, since a recovered task is due immediately while every worker may be parked on `notify` with no pending row to sleep on), `busy_timeout` on all connections | | `crates/xmedia-bot/src/ctx.rs` | `AppContext`: the injected collaborators (`sender` + `ChatStore`/`PersistentTaskQueue`/`LinkCache`/`Config`), `from_statics` for production and the `CONTEXT` static the worker closures hold. `test_support::TestStores` backs handler tests with a tempdir store set, and the module also carries the fixtures those tests share — the canonical cached post (`cached_photo`), the edit-before-forward prompt (`seed_prompt` with its `PROMPT_ID`/`FORWARDED_ID`) and a scripted API error (`api_error`) — so no two test modules keep their own copies | | `crates/xmedia-bot/src/send/` | `send/mod.rs`: `Task`/`MediaItemPayload` payloads, `SendError`/`Classification`, `send_media_sequence`/`send_animation`/`forward_messages`; `send/input_media.rs`: payload → `InputFile`/`InputMedia` + `build_media_group` (caption on the first item only); `send/upload.rs`: the download-and-reupload fallback (`prepare_upload_item`/`send_batch_via_upload`, photo downscale handoff); `send/post_send.rs`: link-cache write, `KEEP_ALIVE` registry, `settle_task`, `post_send_actions`, `handle_task`/`dead_letter_notify` | -| `crates/xmedia-bot/src/media_sender.rs` | `MediaSender` trait: the user-flow surface (`send_media_group`/`send_animation`/`copy_messages`/`send_message`/`answer_callback_query`/`edit_message_text`/`edit_message_caption`/`delete_message`/`send_chat_action`) implemented by teloxide `Bot` (per-chat rate-limited) and by a recording `MockSender` in tests. Admin/setup APIs (`get_chat`, `set_my_commands`, …) stay on the concrete `Bot` | +| `crates/xmedia-bot/src/media_sender.rs` | `MediaSender` trait: the user-flow surface (`send_media_group`/`send_animation`/`copy_messages`/`send_message`/`answer_callback_query`/`edit_message_text`/`edit_message_caption`/`delete_message`/`send_chat_action`) implemented by teloxide `Bot` (per-chat rate-limited) and by a recording `MockSender` in tests. Admin/setup APIs (`get_chat`, `set_my_commands`, …) stay on the concrete `Bot`. `test_support` holds the scripted `MockSender` and `fake_api` (the stand-in API the real-`Bot` tests drive) | | `crates/xmedia-bot/src/rate_limit.rs` | Two token buckets paced before sends reach the API so batch forwards don't trip flood control: one per chat (`CAPACITY = 20`, ~20 msg/min refill) and one bot-wide (`acquire_global`, 30/s — Telegram's per-bot ceiling, invisible to any per-chat bucket and only binding when a batch fans out over many chats). `prune_idle` drops the per-chat buckets that refilled while unheld | ## Development Commands @@ -106,9 +106,9 @@ Docker: `docker build -t tgxmb .` then `docker run --rm -d --name tgxmb --env-fi ## Testing & QA - **~180 tests, all inline `#[cfg(test)] mod tests`** — no `tests/` integration directories. Framework: built-in Rust test + `#[tokio::test]` (dev-deps only in `x-media`: tokio macros/rt-multi-thread, dotenv). -- No mocking framework anywhere (no mockito/wiremock/mockall). Conventions: pure-function units (regex parsing, serde round-trips, chunking, retry math) tested synchronously; async tests use real dependencies — file-backed SQLite via `tempfile` (`queue.rs::new_queue()` helper), live network fetches. +- No mocking framework anywhere (no mockito/wiremock/mockall). Conventions: pure-function units (regex parsing, serde round-trips, chunking, retry math) tested synchronously; async tests use real dependencies — file-backed SQLite via `tempfile` (`queue.rs::new_queue()` helper), live network fetches. Tests that must go through a **real `Bot`** (its URL/multipart building, the per-chat limiter and the bot-wide budget) talk to a stand-in API instead (`media_sender::test_support::fake_api::FakeApi`, a `tokio` TCP listener that records every call and answers the smallest result each method needs — teloxide keys methods by payload type, so the recorded name is `SendMediaGroup`, not `sendMediaGroup`): a media group, the edit-before-forward prompt through the real callback path, and `handlers::handle_message` (the context-taking body of `message_handler`, split out for exactly this). - Live-network tests exist in `site/twitter/interface.rs` (5), `site/bsky/interface.rs` (1), `site/misskey/interface.rs` (1), `site/bilibili/interface.rs` (5), `site/pixiv/api.rs` (1); `photo.rs` adds one `#[ignore = "heavy: …"]` test. `site/mod.rs` also has a **token-gated but not `#[ignore]`d** pixiv download test (`download_media_pixiv_original_with_referer`): it hits `i.pximg.net` whenever `PIXIV_REFRESH_TOKEN` is set, so a local `cargo test --workspace` is not fully offline and can flake on a pixiv CDN body timeout. `disabled_site_is_reported_not_ignored` (same file) is gated the other way round: it asserts `fetch` answers `FetchError::Disabled { site: "pixiv" }` for a pixiv link and early-returns when `PIXIV_REFRESH_TOKEN` **is** set (the site is then enabled). Test gating convention (enforced by `.github/workflows/ci.yml`): pure unit tests always run; live-network tests carry `#[ignore = "live network: ..."]` (run via `cargo test --workspace -- --ignored live`); token-gated pixiv tests early-return when `PIXIV_REFRESH_TOKEN` is absent **or empty** (an unset GitHub secret arrives as `""` — `is_err()` alone would run them tokenless and fail), and the bilibili live tests early-return when the API answers risk control (`-352`, which bilibili applies per IP by request volume). Run the full offline suite with `cargo test --workspace`. - Fixtures are inline `serde_json::json!` builder fns (`fixture()`, `thread_json()`, `illust_json()`), not files. The shared `CLIENT` sets `pool_max_idle_per_host(0)` under `#[cfg(test)]` to avoid cross-runtime `DispatchGone`. - **CI** — `.github/workflows/ci.yml` (actions pinned to commit SHAs, `--locked` on every cargo invocation, `concurrency` cancels superseded runs, `RUST_BACKTRACE=1`) runs `cargo fmt --check` + `cargo clippy --workspace --all-targets --locked -- -D warnings` + `cargo test --workspace --locked` + a release-profile `cargo build --release --locked` + an `actions-rust-lang/audit` dependency-vulnerability gate (offline, no secrets, on every push/PR) and a `live` job (schedule/manual/tag only, `-p x-media` since every network/secret-gated test lives there, `continue-on-error`) for the `#[ignore]`d live + token tests. `.github/workflows/docker.yml` builds and pushes the image on master/tag and runs a **build-only check on pull requests touching the build inputs** (`Dockerfile`, entrypoint, manifests, `.dockerignore`); a release tag must match both crate versions or the build stops, and `FFMPEG_URL`/`FFMPEG_SHA256` are taken from repository variables when set (a release can pin an exact ffmpeg build). `.github/dependabot.yml` keeps crates, the pinned actions and the Docker base images current. -- Untested and hard to test without a mock seam: `config.rs`, `handlers/statics.rs`, `media_sender.rs` (holds the `MockSender` itself, so only the trait's mock side is exercised); `db.rs` is covered for the migration chain but not for pool behaviour under contention; `main.rs` is covered where it was split out (`periodic_sweep`, `sweep_temp_dir`) but not for startup/shutdown or the dispatcher tree; in `x-media`: `media.rs`, `lib.rs`, all `model.rs`. The `commands.rs` *executor* needs a real `Bot` (only its pure report builder is tested). Everything else — `handlers/{mod,callback,inline,urls}.rs`, `send/*`, `ctx.rs`, `state.rs`, `queue.rs`, `link_cache.rs`, `rate_limit.rs` — is driven through `TestStores`/`ctx::test_support` and the scripted `MockSender`. +- Untested and hard to test without a mock seam: `config.rs`, `handlers/statics.rs`; `db.rs` is covered for the migration chain but not for pool behaviour under contention; `main.rs` is covered where it was split out (`periodic_sweep`, `sweep_temp_dir`) but not for startup/shutdown or its `dptree` branch tree (the handlers themselves are, through the stand-in API); in `x-media`: `media.rs`, `lib.rs`, all `model.rs`. The `commands.rs` *executor* needs a real `Bot` (only its pure report builder is tested). Everything else — `handlers/{mod,callback,inline,urls}.rs`, `send/*`, `ctx.rs`, `state.rs`, `queue.rs`, `link_cache.rs`, `rate_limit.rs` — is driven through `TestStores`/`ctx::test_support` and the scripted `MockSender`. - No coverage tracking. diff --git a/crates/x-media/src/site/mod.rs b/crates/x-media/src/site/mod.rs index 37020aa..0c77551 100644 --- a/crates/x-media/src/site/mod.rs +++ b/crates/x-media/src/site/mod.rs @@ -956,6 +956,21 @@ mod tests { } } + /// The redirect-hop guard, against a public redirector: the initial URL is + /// checked by [`media_request`], but a redirect is the part of the path a + /// third-party response actually controls. + #[tokio::test] + #[ignore = "live network: requires outbound HTTPS to httpbin.org"] + async fn a_redirect_into_the_hosts_network_is_refused() { + let url = "https://httpbin.org/redirect-to?url=http://169.254.169.254/latest/meta-data/"; + match download_media(url).await.unwrap_err() { + // A policy refusal reaches the caller wrapped by reqwest. + FetchError::Http(e) => assert!(e.is_redirect(), "got {e}"), + FetchError::Blocked => {} + other => panic!("expected a refusal, got {other:?}"), + } + } + #[tokio::test] async fn a_download_into_the_hosts_network_is_refused() { // Refused on the URL alone: nothing has to be listening (or leaking) at diff --git a/crates/xmedia-bot/src/handlers/callback.rs b/crates/xmedia-bot/src/handlers/callback.rs index 3e23129..5a69bfe 100644 --- a/crates/xmedia-bot/src/handlers/callback.rs +++ b/crates/xmedia-bot/src/handlers/callback.rs @@ -216,7 +216,7 @@ async fn handle_callback( #[cfg(test)] mod tests { use super::*; - use crate::ctx::test_support::{PROMPT_ID, TestStores, api_error, seed_prompt}; + use crate::ctx::test_support::{FORWARDED_ID, PROMPT_ID, TestStores, api_error, seed_prompt}; use crate::media_sender::test_support::{MockSender, Outcome}; /// The Telegram wording the mocks answer with: a chat the bot cannot reach. @@ -315,6 +315,33 @@ mod tests { ); } + /// The whole callback path against a stand-in API through a real `Bot`: + /// copy, delete, toast, carrying the ids the prompt held. The scripted + /// mock records that a call happened; this records what the API received. + #[tokio::test] + async fn the_forward_button_talks_to_the_api_through_a_real_bot() { + use crate::media_sender::test_support::fake_api::FakeApi; + use teloxide::Bot; + + let api = FakeApi::start().await; + let bot = Bot::new("42:TEST").set_api_url(api.url()); + let stores = TestStores::new(); + let ctx = stores.ctx(&bot); + seed_prompt(&ctx, "", crate::db::unix_now()).await; + + handle_callback(&ctx, callback_id(), 1, PROMPT_ID, "forward").await; + + assert_eq!( + api.methods(), + vec!["CopyMessages", "DeleteMessage", "AnswerCallbackQuery"] + ); + let copy = api.body("CopyMessages"); + assert_eq!(copy["chat_id"], 2, "the prompt's channel"); + assert_eq!(copy["from_chat_id"], 1); + assert_eq!(copy["message_ids"], serde_json::json!([FORWARDED_ID])); + assert_eq!(api.body("AnswerCallbackQuery")["text"], "✅ Forwarded"); + } + #[tokio::test] async fn forward_without_a_channel_is_reported() { let sender = MockSender::scripted(vec![], || api_error(API_ERROR)); diff --git a/crates/xmedia-bot/src/handlers/mod.rs b/crates/xmedia-bot/src/handlers/mod.rs index c2052ba..c29a2fa 100644 --- a/crates/xmedia-bot/src/handlers/mod.rs +++ b/crates/xmedia-bot/src/handlers/mod.rs @@ -181,7 +181,21 @@ async fn edit_message_handler( true } +/// The `dptree` entry point: the process-wide context, plus the bot the +/// dispatcher handed us (used for the replies this module sends itself). pub async fn message_handler(bot: Bot, message: Message) -> Result<(), RequestError> { + handle_message(&AppContext::from_statics(&bot), &bot, message).await +} + +/// Body of [`message_handler`], taking its context. Every branch here — the +/// edit-reply interception, the command path, the private-chat link enqueue and +/// the group hint — is otherwise reachable only through the process-wide +/// statics, which is why none of them had a test. +pub(crate) async fn handle_message( + ctx: &AppContext<'_>, + bot: &Bot, + message: Message, +) -> Result<(), RequestError> { let is_private = matches!(message.chat.kind, ChatKind::Private(_)); let sender = message .from @@ -207,13 +221,7 @@ pub async fn message_handler(bot: Bot, message: Message) -> Result<(), RequestEr if is_private && let Some(reply) = message.reply_to_message() && let Some(text) = message.text() - && edit_message_handler( - &AppContext::from_statics(&bot), - message.chat.id.0, - reply.id.0 as i64, - text, - ) - .await + && edit_message_handler(ctx, message.chat.id.0, reply.id.0 as i64, text).await { return respond(()); } @@ -228,7 +236,7 @@ pub async fn message_handler(bot: Bot, message: Message) -> Result<(), RequestEr text.split_whitespace().next().unwrap_or("") ); log::trace!("command text: {text_preview}"); - execute_command(&bot, &message, command).await?; + execute_command(bot, &message, command).await?; return respond(()); } if is_private { @@ -262,7 +270,7 @@ pub async fn message_handler(bot: Bot, message: Message) -> Result<(), RequestEr // the expectation is there). Unsupported links stay ignored; the hint // names the two paths that do work. Channels are excluded — the reply // would be posted into the channel itself. - let _ = reply(&bot, message.chat.id.0, message.id, GROUP_LINK_HINT).await; + let _ = reply(ctx.sender, message.chat.id.0, message.id, GROUP_LINK_HINT).await; } respond(()) } @@ -287,7 +295,7 @@ fn is_group(kind: &ChatKind) -> bool { #[cfg(test)] mod tests { use super::*; - use crate::ctx::test_support::{PROMPT_ID, TestStores, api_error, seed_prompt}; + use crate::ctx::test_support::{FORWARDED_ID, PROMPT_ID, TestStores, api_error, seed_prompt}; use crate::media_sender::test_support::{MockSender, Outcome}; use teloxide::RequestError; @@ -393,6 +401,64 @@ mod tests { assert!(sender.calls().is_empty()); } + /// A reply driven through the real message entry point into a real `Bot`: + /// the routing (reply-to-prompt → caption swap, before the command and URL + /// branches) and the request teloxide builds. + #[tokio::test] + async fn a_prompt_reply_reaches_the_api_as_a_caption_edit() { + use crate::media_sender::test_support::fake_api::FakeApi; + use teloxide::Bot; + + let api = FakeApi::start().await; + let bot = Bot::new("42:TEST").set_api_url(api.url()); + let stores = TestStores::new(); + let ctx = stores.ctx(&bot); + seed_prompt(&ctx, "", crate::db::unix_now()).await; + let message: Message = serde_json::from_value(serde_json::json!({ + "message_id": PROMPT_ID + 1, + "date": 0, + "chat": { "id": 1, "type": "private" }, + "from": { "id": 5, "is_bot": false, "first_name": "u" }, + "reply_to_message": { + "message_id": PROMPT_ID, + "date": 0, + "chat": { "id": 1, "type": "private" }, + "text": "prompt", + }, + "text": "new caption", + })) + .expect("a minimal message deserializes"); + + handle_message(&ctx, &bot, message).await.unwrap(); + + assert_eq!(api.methods(), vec!["EditMessageCaption"]); + let body = api.body("EditMessageCaption"); + assert_eq!(body["chat_id"], 1); + assert_eq!(body["message_id"], FORWARDED_ID); + assert_eq!( + body["caption"], + "new caption" + ); + + // The other branch of the same entry point: a supported link in a group + // gets the one explanatory reply (the link pipeline is private-chat only, + // and dropping it in silence reads as a broken bot). + let group: Message = serde_json::from_value(serde_json::json!({ + "message_id": 2, + "date": 0, + "chat": { "id": -100, "type": "group", "title": "g" }, + "from": { "id": 5, "is_bot": false, "first_name": "u" }, + "text": "https://x.com/u/status/1", + "entities": [{ "type": "url", "offset": 0, "length": 24 }], + })) + .expect("a minimal group message deserializes"); + + handle_message(&ctx, &bot, group).await.unwrap(); + + assert_eq!(api.methods(), vec!["EditMessageCaption", "SendMessage"]); + assert_eq!(api.body("SendMessage")["text"], GROUP_LINK_HINT); + } + #[test] fn the_link_hint_is_for_groups_only() { use teloxide::types::{ChatPrivate, ChatPublic, PublicChatChannel, PublicChatSupergroup}; diff --git a/crates/xmedia-bot/src/media_sender.rs b/crates/xmedia-bot/src/media_sender.rs index 25bd65a..5904672 100644 --- a/crates/xmedia-bot/src/media_sender.rs +++ b/crates/xmedia-bot/src/media_sender.rs @@ -280,6 +280,200 @@ pub(crate) mod test_support { EditErr, } + /// A stand-in for `api.telegram.org` for the tests that must drive a real + /// `Bot` — its request building, the per-chat limiter, the bot-wide budget + /// — which the scripted mock bypasses entirely. Records every call and + /// answers the smallest result each method needs. + pub(crate) mod fake_api { + use parking_lot::Mutex; + use std::sync::Arc; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::{TcpListener, TcpStream}; + + pub(crate) struct FakeApi { + url: url::Url, + calls: Arc>>, + server: tokio::task::JoinHandle<()>, + } + + impl FakeApi { + /// Binds an ephemeral port and serves until dropped. + pub(crate) async fn start() -> FakeApi { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let calls = Arc::new(Mutex::new(Vec::new())); + let recorded = Arc::clone(&calls); + let server = tokio::spawn(async move { + while let Ok((mut socket, _)) = listener.accept().await { + let recorded = Arc::clone(&recorded); + tokio::spawn(async move { + let Some((method, body)) = read_request(&mut socket).await else { + return; + }; + recorded.lock().push((method.clone(), body)); + let payload = serde_json::json!({ + "ok": true, + "result": canned_result(&method), + }) + .to_string(); + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\n\ + content-length: {}\r\nconnection: close\r\n\r\n{}", + payload.len(), + payload + ); + let _ = socket.write_all(response.as_bytes()).await; + let _ = socket.flush().await; + }); + } + }); + FakeApi { + // Trailing slash: teloxide appends `bot/`. + url: url::Url::parse(&format!("http://{addr}/")).unwrap(), + calls, + server, + } + } + + /// Where to point a `Bot`: `Bot::new(token).set_api_url(api.url())`. + pub(crate) fn url(&self) -> url::Url { + self.url.clone() + } + + /// Method names in call order. + pub(crate) fn methods(&self) -> Vec { + self.calls.lock().iter().map(|(m, _)| m.clone()).collect() + } + + /// The JSON body of the first call to `method` (`Null` for a body + /// that is not JSON, i.e. a multipart upload). + pub(crate) fn body(&self, method: &str) -> serde_json::Value { + self.calls + .lock() + .iter() + .find(|(m, _)| m == method) + .map(|(_, body)| body.clone()) + .unwrap_or(serde_json::Value::Null) + } + } + + impl Drop for FakeApi { + fn drop(&mut self) { + self.server.abort(); + } + } + + /// The smallest result teloxide can deserialize for a method. The names + /// arrive as the payload type's own — `SendMediaGroup`, not + /// `sendMediaGroup`: teloxide builds the URL from that, and the Bot API + /// accepts the spelling. + fn canned_result(method: &str) -> serde_json::Value { + match method { + "CopyMessages" => serde_json::json!([{ "message_id": 11 }]), + "SendMediaGroup" => serde_json::json!([minimal_message()]), + "SendMessage" | "SendAnimation" | "EditMessageCaption" => minimal_message(), + _ => serde_json::Value::Bool(true), + } + } + + fn minimal_message() -> serde_json::Value { + serde_json::json!({ + "message_id": 1, + "date": 0, + "chat": { "id": 1, "type": "private" }, + }) + } + + /// One HTTP/1.1 request: the head up to the blank line, then + /// `content-length` bytes of body — JSON for most methods, multipart + /// for the media ones (teloxide sends `SendMediaGroup` that way). + async fn read_request(socket: &mut TcpStream) -> Option<(String, serde_json::Value)> { + let mut buf = Vec::new(); + let mut chunk = [0u8; 4096]; + loop { + let n = socket.read(&mut chunk).await.ok()?; + if n == 0 { + return None; + } + buf.extend_from_slice(&chunk[..n]); + let Some(headers_end) = find(&buf, b"\r\n\r\n") else { + continue; + }; + let head = String::from_utf8_lossy(&buf[..headers_end]).to_string(); + let length: usize = head + .lines() + .find_map(|line| { + line.to_ascii_lowercase() + .strip_prefix("content-length:") + .and_then(|v| v.trim().parse().ok()) + }) + .unwrap_or(0); + let body_start = headers_end + 4; + while buf.len() < body_start + length { + let n = socket.read(&mut chunk).await.ok()?; + if n == 0 { + break; + } + buf.extend_from_slice(&chunk[..n]); + } + let method = head + .lines() + .next() + // `POST /bot/` + .and_then(|line| line.split(' ').nth(1)) + .and_then(|path| path.rsplit('/').next()) + .unwrap_or_default() + .to_string(); + let body = parse_body(&buf[body_start..], &head); + return Some((method, body)); + } + } + + /// The request body as JSON: either the JSON body itself, or a + /// multipart form flattened into an object (each part's value parsed as + /// JSON when it is one, so `media` comes back as its array). + fn parse_body(body: &[u8], head: &str) -> serde_json::Value { + let content_type = head + .lines() + .find(|line| line.to_ascii_lowercase().starts_with("content-type:")) + .unwrap_or_default() + .to_ascii_lowercase(); + let Some(boundary) = content_type + .split("boundary=") + .nth(1) + .map(|b| b.trim().trim_matches('"').to_string()) + else { + return serde_json::from_slice(body).unwrap_or_default(); + }; + let text = String::from_utf8_lossy(body); + let mut fields = serde_json::Map::new(); + for part in text.split(&format!("--{boundary}")).skip(1) { + let Some((part_head, value)) = part.split_once("\r\n\r\n") else { + continue; + }; + let Some(name) = part_head + .split("name=\"") + .nth(1) + .and_then(|rest| rest.split('"').next()) + else { + continue; + }; + let value = value.trim_end_matches("\r\n"); + fields.insert( + name.to_string(), + serde_json::from_str(value).unwrap_or_else(|_| value.into()), + ); + } + serde_json::Value::Object(fields) + } + + fn find(haystack: &[u8], needle: &[u8]) -> Option { + haystack + .windows(needle.len()) + .position(|window| window == needle) + } + } + /// Replays a script and records what was sent, so tests can assert the /// user-visible text a path produced. pub(crate) struct MockSender { diff --git a/crates/xmedia-bot/src/rate_limit.rs b/crates/xmedia-bot/src/rate_limit.rs index e8188bc..1eb9d0f 100644 --- a/crates/xmedia-bot/src/rate_limit.rs +++ b/crates/xmedia-bot/src/rate_limit.rs @@ -91,6 +91,15 @@ impl TokenBucket { tokio::time::sleep(Duration::from_secs_f64(wait)).await; } + /// Current balance, for the tests that assert a call site charged the + /// bucket (a charge is otherwise only observable as a delay). + #[cfg(test)] + pub(crate) fn tokens(&self) -> f64 { + let mut state = self.state.lock(); + self.refill(&mut state); + state.tokens + } + /// True when the bucket has refilled to capacity: no debt outstanding, so /// the chat has not sent anything recently. fn is_idle(&self) -> bool { diff --git a/crates/xmedia-bot/src/send/mod.rs b/crates/xmedia-bot/src/send/mod.rs index 6282710..b055485 100644 --- a/crates/xmedia-bot/src/send/mod.rs +++ b/crates/xmedia-bot/src/send/mod.rs @@ -1396,6 +1396,50 @@ mod tests { assert_eq!(sender.calls(), vec!["send_animation", "send_animation"]); } + /// A media group through a **real** `Bot` — its request building, the + /// per-chat limiter, the bot-wide budget — against a stand-in API. The + /// scripted mock bypasses `media_sender`'s implementation entirely, so a + /// call site that stops charging the limiters (or a broken request shape) + /// is invisible to every other test. + #[tokio::test] + async fn a_media_group_reaches_the_api_through_a_real_bot() { + use crate::media_sender::test_support::fake_api::FakeApi; + use teloxide::Bot; + + let api = FakeApi::start().await; + let bot = Bot::new("42:TEST").set_api_url(api.url()); + let stores = TestStores::new(); + let ctx = stores.ctx(&bot); + // A chat of its own: the limiter buckets are process-wide. + let mut task = sequence_task("https://cdn.example/1.jpg"); + if let Task::SendMediaSequence { chat_id, .. } = &mut task { + *chat_id = 987_654; + } + let bucket = crate::rate_limit::limiter_for(987_654); + let before = bucket.tokens(); + + let outcome = send_media_sequence(&ctx, &task).await; + eprintln!( + "SCRATCH send methods={:?} outcome={outcome:?}", + api.methods() + ); + assert!(outcome.is_ok()); + + // The request teloxide built: one group, the URL, the caption on the + // first item. + assert_eq!(api.methods(), vec!["SendMediaGroup"]); + let body = api.body("SendMediaGroup"); + assert_eq!(body["chat_id"], 987_654); + assert_eq!(body["media"][0]["media"], "https://cdn.example/1.jpg"); + assert_eq!(body["media"][0]["caption"], "cap"); + // …and the send charged the pace limiter before it went out. + let after = bucket.tokens(); + assert!( + after < before, + "a send must charge the chat's budget ({before} -> {after})" + ); + } + #[tokio::test] async fn forward_classifies_retry_after_and_permanent() { use teloxide::types::Seconds;