atproto-devtool 0.1.1

//! Common test utilities shared across integration tests.
//!
//! This module uses the `tests/common/mod.rs` idiom because cargo treats each
//! `tests/*.rs` as a separate crate and requires this pattern to share code.
//! Some helpers may be unused in any given test binary, so we allow dead code at the module level.
//! The `#[expect]` attribute cannot be used here because different test binaries use different
//! subsets of helpers, causing the expect to fail in binaries where a helper happens to be unused.
#![allow(dead_code)]

use async_trait::async_trait;
use atproto_devtool::commands::test::labeler::create_report::{
    CreateReportStageError, CreateReportTee, PdsXrpcClient, RawCreateReportResponse,
    RawPdsXrpcResponse,
};
use atproto_devtool::commands::test::labeler::http::{HttpStageError, RawHttpTee, RawXrpcResponse};
use atproto_devtool::commands::test::labeler::subscription::{
    FrameStream, SubscriptionStageError, WebSocketClient,
};
use reqwest::StatusCode;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use url::Url;

/// Type alias for HTTP response map in tests.
pub type FakeHttpResponses = Arc<Mutex<HashMap<Option<String>, (reqwest::StatusCode, Vec<u8>)>>>;

/// Fake HTTP tee for testing, returns pre-defined responses.
pub struct FakeRawHttpTee {
    /// Map of cursor -> response bytes.
    responses: FakeHttpResponses,
    /// Whether to return a transport error.
    transport_error: Arc<Mutex<bool>>,
}

impl FakeRawHttpTee {
    /// Create a new FakeRawHttpTee.
    pub fn new() -> Self {
        Self {
            responses: Arc::new(Mutex::new(HashMap::new())),
            transport_error: Arc::new(Mutex::new(false)),
        }
    }

    /// Add a response for a given cursor.
    pub fn add_response(&self, cursor: Option<&str>, status: u16, body: Vec<u8>) {
        self.responses.lock().unwrap().insert(
            cursor.map(|s| s.to_string()),
            (reqwest::StatusCode::from_u16(status).unwrap(), body),
        );
    }

    /// Set the transport error flag to simulate network failures.
    pub fn set_transport_error(&self) {
        *self.transport_error.lock().unwrap() = true;
    }
}

impl Default for FakeRawHttpTee {
    fn default() -> Self {
        Self::new()
    }
}

#[async_trait]
impl RawHttpTee for FakeRawHttpTee {
    async fn query_labels(&self, cursor: Option<&str>) -> Result<RawXrpcResponse, HttpStageError> {
        if *self.transport_error.lock().unwrap() {
            // Return a realistic transport error simulating TCP connection refused.
            return Err(HttpStageError::Transport {
                message: "tcp connect: connection refused".into(),
                source: None,
            });
        }

        let cursor_key = cursor.map(|s| s.to_string());
        let responses = self.responses.lock().unwrap();

        match responses.get(&cursor_key) {
            Some((status, body)) => {
                let raw_body: Arc<[u8]> = Arc::from(body.as_slice());
                let decoded = serde_json::from_slice::<
                    atrium_api::com::atproto::label::query_labels::Output,
                >(body)
                .map_err(|source| HttpStageError::DecodeFailed {
                    raw_body: raw_body.clone(),
                    source,
                    source_url: "https://example.com/xrpc/com.atproto.label.queryLabels"
                        .to_string(),
                })?;
                Ok(RawXrpcResponse {
                    status: *status,
                    raw_body,
                    decoded,
                    source_url: "https://example.com/xrpc/com.atproto.label.queryLabels"
                        .to_string(),
                })
            }
            None => {
                // Return a realistic transport error for unconfigured cursor.
                Err(HttpStageError::Transport {
                    message: "tcp connect: connection refused".into(),
                    source: None,
                })
            }
        }
    }
}

/// Scripted response for a single `FakeCreateReportTee::post_create_report`
/// call. A `Transport` variant short-circuits with an error; a `Response`
/// variant returns a `RawCreateReportResponse` built from the supplied parts.
#[derive(Debug, Clone)]
pub enum FakeCreateReportResponse {
    /// Simulate a transport-level failure (no HTTP exchange took place).
    Transport {
        /// Error message the stage will surface.
        message: String,
    },
    /// Simulate a well-formed HTTP response.
    Response {
        /// HTTP status (200, 401, 400, 500, ...).
        status: u16,
        /// Optional content-type header. Fake normalizes to lowercase.
        content_type: Option<String>,
        /// Raw response body bytes.
        body: Vec<u8>,
    },
}

impl FakeCreateReportResponse {
    /// Convenience: a 200 OK with an empty atproto createReport#output body.
    pub fn ok_empty() -> Self {
        Self::Response {
            status: 200,
            content_type: Some("application/json".to_string()),
            body: br#"{"id":1,"reasonType":"com.atproto.moderation.defs#reasonOther","subject":{"$type":"com.atproto.admin.defs#repoRef","did":"did:plc:aaa22222222222222222bbbbbb"},"reportedBy":"did:web:127.0.0.1%3A0","createdAt":"2026-04-17T00:00:00.000Z"}"#.to_vec(),
        }
    }

    /// Convenience: a 401 Unauthorized with the atproto error envelope.
    pub fn unauthorized(error_name: &str, message: &str) -> Self {
        Self::Response {
            status: 401,
            content_type: Some("application/json".to_string()),
            body: serde_json::to_vec(&serde_json::json!({
                "error": error_name,
                "message": message,
            }))
            .unwrap(),
        }
    }

    /// Convenience: a 400 Bad Request with the given error and message.
    pub fn bad_request(error_name: &str, message: &str) -> Self {
        Self::Response {
            status: 400,
            content_type: Some("application/json".to_string()),
            body: serde_json::to_vec(&serde_json::json!({
                "error": error_name,
                "message": message,
            }))
            .unwrap(),
        }
    }
}

/// A recorded request observed by `FakeCreateReportTee`.
#[derive(Debug, Clone)]
pub struct RecordedCreateReportRequest {
    /// Authorization bearer token, if any (stripped of "Bearer " prefix).
    pub auth: Option<String>,
    /// JSON body as posted by the stage.
    pub body: serde_json::Value,
}

/// Fake `CreateReportTee` for integration tests.
///
/// Scripted per-call-index responses: first call gets `responses[0]`,
/// second gets `responses[1]`, etc. Panics if a call is made with no
/// script queued — tests must declare every `post_create_report` the
/// stage is expected to make.
pub struct FakeCreateReportTee {
    /// Queued responses.
    scripts: Arc<Mutex<Vec<FakeCreateReportResponse>>>,
    /// Every request observed (in order).
    recorded: Arc<Mutex<Vec<RecordedCreateReportRequest>>>,
}

impl FakeCreateReportTee {
    /// Create a fake with no scripted responses.
    pub fn new() -> Self {
        Self {
            scripts: Arc::new(Mutex::new(Vec::new())),
            recorded: Arc::new(Mutex::new(Vec::new())),
        }
    }

    /// Queue a scripted response for the next `post_create_report` call.
    pub fn enqueue(&self, response: FakeCreateReportResponse) {
        self.scripts.lock().unwrap().push(response);
    }

    /// Return the recorded request history (cloned).
    pub fn recorded_requests(&self) -> Vec<RecordedCreateReportRequest> {
        self.recorded.lock().unwrap().clone()
    }

    /// Get the last recorded request, panicking if none.
    pub fn last_request(&self) -> RecordedCreateReportRequest {
        self.recorded
            .lock()
            .unwrap()
            .last()
            .cloned()
            .expect("FakeCreateReportTee: no requests recorded yet")
    }
}

impl Default for FakeCreateReportTee {
    fn default() -> Self {
        Self::new()
    }
}

#[async_trait]
impl CreateReportTee for FakeCreateReportTee {
    async fn post_create_report(
        &self,
        auth: Option<&str>,
        body: &serde_json::Value,
    ) -> Result<RawCreateReportResponse, CreateReportStageError> {
        self.recorded
            .lock()
            .unwrap()
            .push(RecordedCreateReportRequest {
                auth: auth.map(|s| s.to_string()),
                body: body.clone(),
            });

        let mut scripts = self.scripts.lock().unwrap();
        if scripts.is_empty() {
            panic!(
                "FakeCreateReportTee: post_create_report called with no script queued. \
                 Each test must enqueue() exactly the responses it expects the stage to consume."
            );
        }
        let script = scripts.remove(0);

        match script {
            FakeCreateReportResponse::Transport { message } => {
                Err(CreateReportStageError::Transport {
                    source: Box::new(std::io::Error::other(message)),
                })
            }
            FakeCreateReportResponse::Response {
                status,
                content_type,
                body,
            } => {
                let raw_body: Arc<[u8]> = Arc::from(body.as_slice());
                Ok(RawCreateReportResponse {
                    status: StatusCode::from_u16(status).expect("test must use valid HTTP status"),
                    content_type: content_type.map(|s| s.to_ascii_lowercase()),
                    raw_body,
                    source_url: "https://labeler.test/xrpc/com.atproto.moderation.createReport"
                        .to_string(),
                })
            }
        }
    }
}

/// Scripted response for a single `FakePdsXrpcClient` call. A `Transport`
/// variant short-circuits with an error; a `Response` variant returns a
/// `RawPdsXrpcResponse` built from the supplied parts.
#[derive(Debug, Clone)]
pub enum FakePdsXrpcResponse {
    /// Simulate a transport-level failure (no HTTP exchange took place).
    Transport { message: String },
    /// Simulate a well-formed HTTP response.
    Response { status: u16, body: Vec<u8> },
}

/// A recorded request observed by `FakePdsXrpcClient`.
#[derive(Debug, Clone)]
pub struct RecordedPdsRequest {
    /// HTTP method: "POST" or "GET".
    pub method: &'static str,
    /// Request path (e.g., "xrpc/com.atproto.server.createSession").
    pub path: String,
    /// Bearer token, if any (stripped of "Bearer " prefix).
    pub bearer: Option<String>,
    /// atproto-proxy header, if any.
    pub atproto_proxy: Option<String>,
    /// JSON body for POST requests; None for GET.
    pub body: Option<serde_json::Value>,
    /// Query parameters for GET requests; empty for POST.
    pub query: Vec<(String, String)>,
}

/// Fake `PdsXrpcClient` for integration tests.
///
/// Scripted per-call-index responses: first call gets `responses[0]`,
/// second gets `responses[1]`, etc. Panics if a call is made with no
/// script queued — tests must declare every call the stage is expected to make.
pub struct FakePdsXrpcClient {
    /// Queued responses.
    scripts: Arc<Mutex<Vec<FakePdsXrpcResponse>>>,
    /// Every request observed (in order).
    recorded: Arc<Mutex<Vec<RecordedPdsRequest>>>,
}

impl FakePdsXrpcClient {
    /// Create a fake with no scripted responses.
    pub fn new() -> Self {
        Self {
            scripts: Arc::new(Mutex::new(Vec::new())),
            recorded: Arc::new(Mutex::new(Vec::new())),
        }
    }

    /// Queue a scripted response for the next call.
    pub fn enqueue(&self, response: FakePdsXrpcResponse) {
        self.scripts.lock().unwrap().push(response);
    }

    /// Return the recorded request history (cloned).
    pub fn recorded_requests(&self) -> Vec<RecordedPdsRequest> {
        self.recorded.lock().unwrap().clone()
    }

    /// Get the last recorded request, panicking if none.
    pub fn last_request(&self) -> RecordedPdsRequest {
        self.recorded
            .lock()
            .unwrap()
            .last()
            .cloned()
            .expect("FakePdsXrpcClient: no requests recorded yet")
    }

    /// Pop the next script and return the result. Panics if no script queued.
    fn dispatch_next(&self) -> Result<RawPdsXrpcResponse, CreateReportStageError> {
        let mut scripts = self.scripts.lock().unwrap();
        if scripts.is_empty() {
            panic!(
                "FakePdsXrpcClient: call made with no script queued. \
                 Each test must enqueue() exactly the responses it expects."
            );
        }
        let script = scripts.remove(0);

        match script {
            FakePdsXrpcResponse::Transport { message } => Err(CreateReportStageError::Transport {
                source: Box::new(std::io::Error::other(message)),
            }),
            FakePdsXrpcResponse::Response { status, body } => {
                let raw_body: Arc<[u8]> = Arc::from(body.as_slice());
                Ok(RawPdsXrpcResponse {
                    status: StatusCode::from_u16(status).expect("test must use valid HTTP status"),
                    raw_body,
                    content_type: Some("application/json".to_string()),
                    source_url: "https://pds.test/xrpc".to_string(),
                })
            }
        }
    }
}

impl Default for FakePdsXrpcClient {
    fn default() -> Self {
        Self::new()
    }
}

#[async_trait]
impl PdsXrpcClient for FakePdsXrpcClient {
    async fn post(
        &self,
        path: &str,
        bearer: Option<&str>,
        atproto_proxy: Option<&str>,
        body: &serde_json::Value,
    ) -> Result<RawPdsXrpcResponse, CreateReportStageError> {
        self.recorded.lock().unwrap().push(RecordedPdsRequest {
            method: "POST",
            path: path.to_string(),
            bearer: bearer.map(String::from),
            atproto_proxy: atproto_proxy.map(String::from),
            body: Some(body.clone()),
            query: Vec::new(),
        });
        self.dispatch_next()
    }

    async fn get(
        &self,
        path: &str,
        bearer: Option<&str>,
        query: &[(&str, &str)],
    ) -> Result<RawPdsXrpcResponse, CreateReportStageError> {
        self.recorded.lock().unwrap().push(RecordedPdsRequest {
            method: "GET",
            path: path.to_string(),
            bearer: bearer.map(String::from),
            atproto_proxy: None,
            body: None,
            query: query
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect(),
        });
        self.dispatch_next()
    }
}

/// Fake WebSocket client for testing subscription stage with scripted responses.
///
/// Two construction styles are supported:
///
/// * `new()` + `add_script()` — the subscription-test style. Each `connect()`
///   consumes exactly one script. Calling `connect()` when no script is queued
///   panics, so subscription tests must explicitly declare every connection
///   the stage is expected to make.
/// * `empty()` — the identity-test style. Every `connect()` returns an
///   empty-stream placeholder that closes immediately. Identity tests construct
///   one of these to satisfy the pipeline's subscription stage without asserting
///   on its output.
pub struct FakeWebSocketClient {
    /// Scripts queued for each connection.
    scripts: Arc<Mutex<Vec<FakeScript>>>,
    /// If true, `connect()` silently returns an empty stream when no script is queued.
    /// If false, `connect()` panics when no script is queued (forcing subscription tests
    /// to declare exactly the connections they expect).
    silent_default: bool,
}

/// A script for a single WebSocket connection.
pub struct FakeScript {
    /// Frames to return (if no transport error).
    pub frames: Vec<Vec<u8>>,
    /// Delay between frames.
    pub inter_frame_delay: Duration,
    /// Optional final wait after all frames (simulates idle gap or continued streaming).
    pub final_wait: Option<Duration>,
    /// Whether to return a transport error instead of returning frames.
    pub transport_error: bool,
    /// If set, after all `frames` are yielded the stream yields one `Transport`
    /// error, then closes. Used to test that a mid-stream transport error does
    /// not reset the idle-gap timer.
    pub mid_stream_error: bool,
}

impl FakeWebSocketClient {
    /// Create a FakeWebSocketClient for subscription tests.
    ///
    /// Every `connect()` consumes exactly one script added via `add_script()`.
    /// Calling `connect()` with no script queued panics.
    pub fn new() -> Self {
        Self {
            scripts: Arc::new(Mutex::new(Vec::new())),
            silent_default: false,
        }
    }

    /// Create a FakeWebSocketClient that silently returns an empty stream on every connect.
    ///
    /// Intended for identity tests that must satisfy the pipeline's subscription
    /// stage but do not assert on its output.
    pub fn empty() -> Self {
        Self {
            scripts: Arc::new(Mutex::new(Vec::new())),
            silent_default: true,
        }
    }

    /// Add a script to the queue for the next connection.
    pub fn add_script(&self, script: FakeScript) {
        self.scripts.lock().unwrap().push(script);
    }
}

impl Default for FakeWebSocketClient {
    fn default() -> Self {
        Self::new()
    }
}

/// A fake frame stream returned by FakeWebSocketClient.
struct FakeFrameStream {
    frames: Vec<Vec<u8>>,
    current_frame: usize,
    inter_frame_delay: Duration,
    final_wait: Option<Duration>,
    mid_stream_error: bool,
    mid_stream_error_yielded: bool,
}

#[async_trait]
impl FrameStream for FakeFrameStream {
    async fn next_frame(&mut self) -> Option<Result<Vec<u8>, SubscriptionStageError>> {
        // Return frames one by one with inter-frame delays.
        if self.current_frame < self.frames.len() {
            if self.current_frame > 0 {
                tokio::time::sleep(self.inter_frame_delay).await;
            }
            let frame = self.frames[self.current_frame].clone();
            self.current_frame += 1;
            return Some(Ok(frame));
        }

        // All frames consumed. If mid-stream error mode is on and we haven't yielded the
        // error yet, sleep one inter-frame delay and yield it once — this lets the stage
        // observe a transport error without the idle-gap timer being reset.
        if self.mid_stream_error && !self.mid_stream_error_yielded {
            self.mid_stream_error_yielded = true;
            tokio::time::sleep(self.inter_frame_delay).await;
            return Some(Err(SubscriptionStageError::Transport {
                message: "fake mid-stream transport error".to_string(),
                source: None,
            }));
        }

        // Apply final_wait once (long enough to let the stage's idle-gap or budget timer fire).
        if let Some(wait_duration) = self.final_wait.take() {
            tokio::time::sleep(wait_duration).await;
        }

        // Stream closed.
        None
    }

    async fn close(&mut self) {
        // Noop for fake.
    }
}

#[async_trait]
impl WebSocketClient for FakeWebSocketClient {
    async fn connect(&self, _url: &Url) -> Result<Box<dyn FrameStream>, SubscriptionStageError> {
        let mut scripts = self.scripts.lock().unwrap();

        if scripts.is_empty() {
            if self.silent_default {
                // Silent-default mode (for identity tests): return an empty stream that
                // closes immediately. Do not consume any script slot.
                return Ok(Box::new(FakeFrameStream {
                    frames: vec![],
                    current_frame: 0,
                    inter_frame_delay: Duration::from_millis(0),
                    final_wait: None,
                    mid_stream_error: false,
                    mid_stream_error_yielded: false,
                }));
            }
            // Script-driven mode (for subscription tests): panic loudly so the test
            // author notices they forgot to declare a connection.
            panic!(
                "FakeWebSocketClient: no script queued for this connect() call. \
                Each subscription test must declare exactly the scripts it expects \
                the stage to consume. Use fake_ws.add_script() for each connect() \
                the stage will make. (Identity tests should use FakeWebSocketClient::empty() instead.)"
            );
        }

        let script = scripts.remove(0);

        if script.transport_error {
            return Err(SubscriptionStageError::Transport {
                message: "fake transport error".to_string(),
                source: None,
            });
        }

        Ok(Box::new(FakeFrameStream {
            frames: script.frames,
            current_frame: 0,
            inter_frame_delay: script.inter_frame_delay,
            final_wait: script.final_wait,
            mid_stream_error: script.mid_stream_error,
            mid_stream_error_yielded: false,
        }))
    }
}

/// Helper to normalize elapsed time in snapshots.
///
/// Replaces `elapsed: <N>ms` with `elapsed: XXms`, advancing past each match
/// to avoid re-matching the replacement on the next iteration.
pub fn normalize_timing(rendered: String) -> String {
    let mut result = String::with_capacity(rendered.len());
    let mut rest = rendered.as_str();
    while let Some(pos) = rest.find("elapsed: ") {
        let after = pos + "elapsed: ".len();
        result.push_str(&rest[..after]);
        let tail = &rest[after..];
        if let Some(end) = tail.find("ms") {
            result.push_str("XXms");
            rest = &tail[end + 2..];
        } else {
            result.push_str(tail);
            return result;
        }
    }
    result.push_str(rest);
    result
}