//! The RFC 1928 core's differential RFC edge-case suite (the wire
//! form is the point; expectations are pinned as explicit bytes).
//!
//! Posture (docs/architecture/rfc1928.md §Maintenance Posture,
//! ADR-013): `/workspace/fast-socks5` (v1.0.0) is the behavioral
//! reference, consulted while writing these tests; the assertions
//! pin the expected bytes, never the reference checkout itself
//! (build-time dependency-free). Where the fork deliberately
//! diverges (the pruned error variants, the published `new_reply`),
//! the test pins the in-tree contract and the divergence is recorded
//! in rfc1928.md.
//!
//! Differential convention: every non-obvious expectation names the
//! upstream behavior it pins (file + line in the v1.0.0 checkout).
//!
//! Infrastructure: tokio duplex streams only (no sockets, no
//! `local` feature; the suite runs in the default wasm-clean crate's
//! native test build.

use alksocks::rfc1928::client::{Config, Socks5Stream};
use alksocks::rfc1928::consts;
use alksocks::rfc1928::server::{
    new_reply, transfer, wait_on_tcp, Socks5ServerProtocol, SocksServerError,
};
use alksocks::rfc1928::util::target_addr::{AddrError, TargetAddr};
use alksocks::rfc1928::{new_udp_header, parse_udp_request, ReplyError, Socks5Command, SocksError};
use std::net::SocketAddr;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

type ServerTask<T> = tokio::task::JoinHandle<Result<T, SocksServerError>>;

/// Drive the forked server through `accept_no_auth` (negotiation +
/// the auth-choice write) on a spawned task, mapping the success
/// value to `()` so the handle unwraps without a `Debug` bound.
fn spawn_no_auth(server: tokio::io::DuplexStream) -> ServerTask<()> {
    tokio::spawn(async move {
        Socks5ServerProtocol::accept_no_auth(server).await?;
        Ok(())
    })
}

// ---------------------------------------------------------------------------
// Group 1: negotiation edge cases
// ---------------------------------------------------------------------------

/// An empty method list (methods_len = 0) matches nothing → 0xff.
/// Upstream: `negotiate_auth` loops `server_methods` × client methods
/// (`src/server.rs:903-914`); zero client methods means the inner
/// loop never matches, so the not-acceptable write at
/// `src/server.rs:917-923` fires with `AuthMethodUnacceptable([])`.
#[tokio::test]
async fn negotiation_empty_method_list_replies_0xff() {
    let (mut client, server) = tokio::io::duplex(64);
    let handle = spawn_no_auth(server);
    client.write_all(&[0x05, 0x00]).await.unwrap();
    let mut method_reply = [0u8; 2];
    client.read_exact(&mut method_reply).await.unwrap();
    assert_eq!(method_reply, [0x05, 0xff]);
    let err = handle.await.unwrap().unwrap_err();
    assert!(matches!(
        err,
        SocksServerError::AuthMethodUnacceptable(ref m) if m.is_empty()
    ));
}

/// Methods offered that the server doesn't support → 0xff, and the
/// error carries the *client's* full method list verbatim (upstream
/// `src/server.rs:918-924` returns the read-back `methods` buffer).
#[tokio::test]
async fn negotiation_zero_methods_offered_replies_0xff() {
    let (mut client, server) = tokio::io::duplex(64);
    let handle = spawn_no_auth(server);
    client.write_all(&[0x05, 0x01, 0x02]).await.unwrap();
    let mut method_reply = [0u8; 2];
    client.read_exact(&mut method_reply).await.unwrap();
    assert_eq!(method_reply, [0x05, 0xff]);
    let err = handle.await.unwrap().unwrap_err();
    assert!(matches!(
        err,
        SocksServerError::AuthMethodUnacceptable(ref m) if m == &vec![0x02]
    ));
}

/// Method-order preference: the server picks *its* first match in
/// `server_methods` order, not the client's first offer. Upstream
/// pins this as the outer-loop-in-server-order contract
/// (`src/server.rs:901-902`, "server_methods order matter!");
/// in-tree, `accept_no_auth` offers only `NoAuthentication`, so a
/// client offering password-first-then-none still gets 0x00.
#[tokio::test]
async fn negotiation_server_picks_its_own_first_match() {
    let (mut client, server) = tokio::io::duplex(64);
    let handle = spawn_no_auth(server);
    client.write_all(&[0x05, 0x02, 0x02, 0x00]).await.unwrap();
    let mut method_reply = [0u8; 2];
    client.read_exact(&mut method_reply).await.unwrap();
    assert_eq!(method_reply, [0x05, 0x00]);
    handle.await.unwrap().unwrap();
}

/// The same contract under `accept_password_auth`: a client offering
/// none-then-password is answered with the *server's* preferred
/// method (0x02; `negotiate_auth` walks `server_methods` in server
/// order; RFC 1928 §3 lets the server pick by its own preference, so
/// the method list is ordered more-secure/preferred first).
#[tokio::test]
async fn negotiation_password_server_prefers_password_over_no_auth() {
    let (mut client, server) = tokio::io::duplex(64);
    let handle: ServerTask<()> = tokio::spawn(async move {
        let (_proto, _check) =
            Socks5ServerProtocol::accept_password_auth(server, |_u, _p| true).await?;
        Ok(())
    });
    client.write_all(&[0x05, 0x02, 0x00, 0x02]).await.unwrap();
    let mut method_reply = [0u8; 2];
    client.read_exact(&mut method_reply).await.unwrap();
    assert_eq!(method_reply, [0x05, 0x02]);
    client
        .write_all(&[0x01, 0x01, b'u', 0x01, b'p'])
        .await
        .unwrap();
    let mut status = [0u8; 2];
    client.read_exact(&mut status).await.unwrap();
    assert_eq!(status, [0x01, 0x00]);
    handle.await.unwrap().unwrap();
}

/// Version mismatch at the greeting (0x04) → typed refusal, no hang.
/// Upstream: `negotiate_auth` checks `version` immediately after
/// reading the 2-byte header, before any method bytes
/// (`src/server.rs:890-892`); so a truncated greeting (no methods
/// byte) still refuses on the version alone.
#[tokio::test]
async fn negotiation_version_mismatch_refuses_before_methods_byte() {
    let (mut client, server) = tokio::io::duplex(64);
    let handle = spawn_no_auth(server);
    client.write_all(&[0x04, 0x01]).await.unwrap();
    let err = handle.await.unwrap().unwrap_err();
    assert!(matches!(
        err,
        SocksServerError::UnsupportedSocksVersion(0x04)
    ));
}

/// Truncated greeting (EOF mid-handshake) → typed error, no hang.
/// Upstream: `read_exact!` propagates an
/// `UnexpectedEof`-flavored `io::Error` via `err_when("reading
/// methods")` (`src/server.rs:882-883`); in-tree the same shape
/// surfaces as `SocksServerError::Io` with that context.
#[tokio::test]
async fn negotiation_truncated_greeting_is_a_typed_io_error() {
    let (mut client, server) = tokio::io::duplex(64);
    let handle = spawn_no_auth(server);
    client.write_all(&[0x05]).await.unwrap();
    drop(client);
    let err = handle.await.unwrap().unwrap_err();
    match err {
        SocksServerError::Io { context, .. } => assert_eq!(context, "reading methods"),
        other => panic!("expected Io with context, got {other:?}"),
    }
}

/// Password sub-protocol: empty username → `EmptyUsername` before any
/// credential byte is read. Upstream `src/server.rs:572-582`.
#[tokio::test]
async fn password_auth_empty_username_is_typed_error() {
    let (mut client, server) = tokio::io::duplex(64);
    let handle: ServerTask<()> = tokio::spawn(async move {
        Socks5ServerProtocol::accept_password_auth(server, |_u, _p| true).await?;
        Ok(())
    });
    client.write_all(&[0x05, 0x01, 0x02]).await.unwrap();
    let mut method_reply = [0u8; 2];
    client.read_exact(&mut method_reply).await.unwrap();
    assert_eq!(method_reply, [0x05, 0x02]);
    client.write_all(&[0x01, 0x00]).await.unwrap();
    let err = handle.await.unwrap().unwrap_err();
    assert!(matches!(err, SocksServerError::EmptyUsername));
}

/// Password sub-protocol: empty password → `EmptyPassword`. Upstream
/// `src/server.rs:587-596`.
#[tokio::test]
async fn password_auth_empty_password_is_typed_error() {
    let (mut client, server) = tokio::io::duplex(64);
    let handle: ServerTask<()> = tokio::spawn(async move {
        Socks5ServerProtocol::accept_password_auth(server, |_u, _p| true).await?;
        Ok(())
    });
    client.write_all(&[0x05, 0x01, 0x02]).await.unwrap();
    let mut method_reply = [0u8; 2];
    client.read_exact(&mut method_reply).await.unwrap();
    assert_eq!(method_reply, [0x05, 0x02]);
    client.write_all(&[0x01, 0x01, b'u', 0x00]).await.unwrap();
    let err = handle.await.unwrap().unwrap_err();
    assert!(matches!(err, SocksServerError::EmptyPassword));
}

/// Password sub-protocol reject path: the 0x01/0xff status frame is
/// written *before* the error surfaces. Upstream `accept_password_auth`
/// awaits `auth.reject()` (the `[1, 0xff]` write) before returning
/// `AuthenticationRejected` (`src/server.rs:464-475`; the reject
/// write itself is `src/server.rs:620-627`).
#[tokio::test]
async fn password_auth_reject_writes_0x01_0xff_then_errors() {
    let (mut client, server) = tokio::io::duplex(64);
    let handle: ServerTask<()> = tokio::spawn(async move {
        Socks5ServerProtocol::accept_password_auth(server, |_u, _p| false).await?;
        Ok(())
    });
    client.write_all(&[0x05, 0x01, 0x02]).await.unwrap();
    let mut method_reply = [0u8; 2];
    client.read_exact(&mut method_reply).await.unwrap();
    assert_eq!(method_reply, [0x05, 0x02]);
    client
        .write_all(&[0x01, 0x01, b'u', 0x01, b'p'])
        .await
        .unwrap();
    let mut status = [0u8; 2];
    client.read_exact(&mut status).await.unwrap();
    assert_eq!(status, [0x01, 0xff]);
    let err = handle.await.unwrap().unwrap_err();
    assert!(matches!(err, SocksServerError::AuthenticationRejected));
}

// ---------------------------------------------------------------------------
// Group 2: command-read edge cases
// ---------------------------------------------------------------------------

/// Unknown command byte (0x04+) → `UnknownCommand` → 0x07 reply. The
/// address is parsed *first* (upstream `src/server.rs:1018`: the
/// try_notify order), so the request must carry a valid target for
/// the reply to fire; the reply uses the error form (ATYP=1, zeros).
#[tokio::test]
async fn command_unknown_byte_replies_0x07() {
    let (mut client, server) = tokio::io::duplex(64);
    let handle: ServerTask<()> = tokio::spawn(async move {
        let proto = Socks5ServerProtocol::accept_no_auth(server).await?;
        proto.read_command().await?;
        Ok(())
    });
    client.write_all(&[0x05, 0x01, 0x00]).await.unwrap();
    let mut method_reply = [0u8; 2];
    client.read_exact(&mut method_reply).await.unwrap();
    assert_eq!(method_reply, [0x05, 0x00]);
    client
        .write_all(&[0x05, 0x7f, 0x00, 0x01, 10, 0, 0, 1, 0, 80])
        .await
        .unwrap();
    let mut reply = [0u8; 10];
    client.read_exact(&mut reply).await.unwrap();
    assert_eq!(reply, [0x05, 0x07, 0x00, 0x01, 0, 0, 0, 0, 0, 0]);
    let err = handle.await.unwrap().unwrap_err();
    assert!(matches!(err, SocksServerError::UnknownCommand(0x7f)));
}

/// Reserved-byte tolerance: RSV ≠ 0 is read but never checked by
/// upstream (`src/server.rs:1001-1002`, destructured as `rsv`, only
/// debug-logged; no validation). Pin the tolerance: a garbage RSV
/// byte still parses to a successful command read.
#[tokio::test]
async fn command_garbage_reserved_byte_is_tolerated() {
    let (mut client, server) = tokio::io::duplex(64);
    let handle: ServerTask<(Socks5Command, TargetAddr)> = tokio::spawn(async move {
        let proto = Socks5ServerProtocol::accept_no_auth(server).await?;
        let (_proto, cmd, target) = proto.read_command().await?;
        Ok((cmd, target))
    });
    client.write_all(&[0x05, 0x01, 0x00]).await.unwrap();
    let mut method_reply = [0u8; 2];
    client.read_exact(&mut method_reply).await.unwrap();
    assert_eq!(method_reply, [0x05, 0x00]);
    client
        .write_all(&[0x05, 0x01, 0xa5, 0x03, 4, b'h', b'o', b's', b't', 0, 80])
        .await
        .unwrap();
    let (cmd, target) = handle.await.unwrap().unwrap();
    assert_eq!(cmd, Socks5Command::TCPConnect);
    assert_eq!(target, TargetAddr::Domain("host".to_owned(), 80));
}

/// ATYP 0x02 (unsupported both by RFC and by the codec's 0x05+
/// fallthrough) → `IncorrectAddressType` → 0x08 reply, via
/// `try_notify!`'s secondary-error write. Upstream: `read_address`'s
/// `_ =>` arm (`src/util/target_addr.rs:257`) maps through
/// `AddrError::to_reply_error` (`src/util/target_addr.rs:40-48`) to
/// reply 0x08 (`src/server.rs:1018` try_notify site).
#[tokio::test]
async fn command_atyp_0x02_replies_0x08() {
    let (mut client, server) = tokio::io::duplex(64);
    let handle: ServerTask<()> = tokio::spawn(async move {
        let proto = Socks5ServerProtocol::accept_no_auth(server).await?;
        proto.read_command().await?;
        Ok(())
    });
    client.write_all(&[0x05, 0x01, 0x00]).await.unwrap();
    let mut method_reply = [0u8; 2];
    client.read_exact(&mut method_reply).await.unwrap();
    assert_eq!(method_reply, [0x05, 0x00]);
    client
        .write_all(&[0x05, 0x01, 0x00, 0x02, 1, 2, 3, 4, 0, 80])
        .await
        .unwrap();
    let mut reply = [0u8; 10];
    client.read_exact(&mut reply).await.unwrap();
    assert_eq!(reply, [0x05, 0x08, 0x00, 0x01, 0, 0, 0, 0, 0, 0]);
    let err = handle.await.unwrap().unwrap_err();
    match err {
        SocksServerError::AddrError(AddrError::IncorrectAddressType) => {}
        other => panic!("expected AddrError::IncorrectAddressType, got {other:?}"),
    }
}

/// ATYP 0x05+ (same `_ =>` arm as 0x02) → 0x08 reply.
#[tokio::test]
async fn command_atyp_0x05_replies_0x08() {
    let (mut client, server) = tokio::io::duplex(64);
    let handle: ServerTask<()> = tokio::spawn(async move {
        let proto = Socks5ServerProtocol::accept_no_auth(server).await?;
        proto.read_command().await?;
        Ok(())
    });
    client.write_all(&[0x05, 0x01, 0x00]).await.unwrap();
    let mut method_reply = [0u8; 2];
    client.read_exact(&mut method_reply).await.unwrap();
    assert_eq!(method_reply, [0x05, 0x00]);
    client
        .write_all(&[0x05, 0x01, 0x00, 0x05, 1, 2, 3, 4, 0, 80])
        .await
        .unwrap();
    let mut reply = [0u8; 10];
    client.read_exact(&mut reply).await.unwrap();
    assert_eq!(reply, [0x05, 0x08, 0x00, 0x01, 0, 0, 0, 0, 0, 0]);
    let err = handle.await.unwrap().unwrap_err();
    match err {
        SocksServerError::AddrError(AddrError::IncorrectAddressType) => {}
        other => panic!("expected AddrError::IncorrectAddressType, got {other:?}"),
    }
}

/// Domain-length boundary: a 255-byte domain is the wire maximum and
/// parses fine (`read_address` reads a 1-byte len then that many
/// bytes (`src/util/target_addr.rs:236-245`; nothing else bounds
/// it). 256+ is rejected at *encode* time by `to_be_bytes`
/// (`src/util/target_addr.rs:113-115`), so parsing never sees it.
#[tokio::test]
async fn command_domain_255_bytes_is_the_wire_maximum() {
    let (mut client, server) = tokio::io::duplex(64);
    let handle: ServerTask<TargetAddr> = tokio::spawn(async move {
        let proto = Socks5ServerProtocol::accept_no_auth(server).await?;
        let (_proto, _cmd, target) = proto.read_command().await?;
        Ok(target)
    });
    client.write_all(&[0x05, 0x01, 0x00]).await.unwrap();
    let mut method_reply = [0u8; 2];
    client.read_exact(&mut method_reply).await.unwrap();
    assert_eq!(method_reply, [0x05, 0x00]);
    let mut request = vec![0x05, 0x01, 0x00, 0x03, 0xff];
    request.extend_from_slice(&vec![b'a'; 255]);
    request.extend_from_slice(&[0x01, 0xbb]);
    client.write_all(&request).await.unwrap();
    let target = handle.await.unwrap().unwrap();
    assert_eq!(target, TargetAddr::Domain("a".repeat(255), 443));
}

/// The decode-side bound is the wire's own 1-byte length field,
/// stated explicitly (review 001 S-1): `read_address`'s domain arm
/// allocates from an attacker-controlled byte, and the allocation is
/// safe *only* because the length field is one byte (≤ 255). This
/// test pins both faces of that invariant: (a) the 255-byte maximum
/// decodes through `read_address` (the attacker-side path), and (b)
/// the encoder rejects a 256-byte claim (a 256-byte domain cannot
/// enter the wire at all, so the decode arm's `len as usize` can
/// never ask for more than 255. Widening the length field (a
/// hypothetical FRAG-style extension) requires an explicit cap
/// before the allocation (the `read_address` doc note).
#[tokio::test]
async fn decode_side_domain_allocation_is_bounded_by_the_1_byte_wire_field() {
    let maxed = {
        let (mut client, mut server) = tokio::io::duplex(512);
        client.write_all(&[0xff]).await.unwrap();
        client.write_all(&vec![b'a'; 255]).await.unwrap();
        client.write_all(&[0x00, 0x50]).await.unwrap();
        drop(client);
        alksocks::rfc1928::util::target_addr::read_address(&mut server, 0x03)
            .await
            .unwrap()
    };
    assert_eq!(maxed, TargetAddr::Domain("a".repeat(255), 80));
    // The 256-byte claim is impossible by construction: the encoder
    // refuses it before any byte is emitted.
    let over = TargetAddr::Domain("a".repeat(256), 80);
    assert!(matches!(
        over.to_be_bytes(),
        Err(AddrError::DomainLenTooLong(256))
    ));
}

/// Truncated address fields (EOF mid-address) → typed
/// `AddrError`-wrapped error, surfaced through `try_notify!`'s
/// 0x01-coded secondary reply (AddrError non-ATYP variants map to
/// `ConnectionRefused`; `src/util/target_addr.rs:40-48`).
#[tokio::test]
async fn command_truncated_ipv4_address_is_typed_error_with_0x05_reply() {
    let (mut client, server) = tokio::io::duplex(64);
    let handle: ServerTask<()> = tokio::spawn(async move {
        let proto = Socks5ServerProtocol::accept_no_auth(server).await?;
        proto.read_command().await?;
        Ok(())
    });
    client.write_all(&[0x05, 0x01, 0x00]).await.unwrap();
    let mut method_reply = [0u8; 2];
    client.read_exact(&mut method_reply).await.unwrap();
    assert_eq!(method_reply, [0x05, 0x00]);
    client
        .write_all(&[0x05, 0x01, 0x00, 0x01, 10, 0])
        .await
        .unwrap();
    drop(client);
    let err = handle.await.unwrap().unwrap_err();
    match err {
        SocksServerError::AddrError(AddrError::IPv4Unreadable(_)) => {}
        other => panic!("expected AddrError::IPv4Unreadable, got {other:?}"),
    }
}

/// Truncated domain content (EOF after the length byte) →
/// `DomainContentUnreadable`.
#[tokio::test]
async fn command_truncated_domain_content_is_typed_error() {
    let (mut client, server) = tokio::io::duplex(64);
    let handle: ServerTask<()> = tokio::spawn(async move {
        let proto = Socks5ServerProtocol::accept_no_auth(server).await?;
        proto.read_command().await?;
        Ok(())
    });
    client.write_all(&[0x05, 0x01, 0x00]).await.unwrap();
    let mut method_reply = [0u8; 2];
    client.read_exact(&mut method_reply).await.unwrap();
    assert_eq!(method_reply, [0x05, 0x00]);
    client
        .write_all(&[0x05, 0x01, 0x00, 0x03, 10, b'a'])
        .await
        .unwrap();
    drop(client);
    let err = handle.await.unwrap().unwrap_err();
    match err {
        SocksServerError::AddrError(AddrError::DomainContentUnreadable(_)) => {}
        other => panic!("expected AddrError::DomainContentUnreadable, got {other:?}"),
    }
}

/// Version mismatch at the command phase (post-negotiation, the
/// 4-byte request header read) → typed refusal, no reply. Upstream
/// `read_command` checks `version` before parsing the address
/// (`src/server.rs:1010-1014`), a distinct check from the greeting's
/// (`negotiate_auth`, `src/server.rs:889-892`), so this exercises the
/// command-phase arm in isolation.
#[tokio::test]
async fn command_version_mismatch_refuses_before_address_parse() {
    let (mut client, server) = tokio::io::duplex(64);
    let handle: ServerTask<()> = tokio::spawn(async move {
        let proto = Socks5ServerProtocol::accept_no_auth(server).await?;
        proto.read_command().await?;
        Ok(())
    });
    client.write_all(&[0x05, 0x01, 0x00]).await.unwrap();
    let mut method_reply = [0u8; 2];
    client.read_exact(&mut method_reply).await.unwrap();
    assert_eq!(method_reply, [0x05, 0x00]);
    client.write_all(&[0x06, 0x01, 0x00, 0x01]).await.unwrap();
    drop(client);
    let err = handle.await.unwrap().unwrap_err();
    assert!(matches!(
        err,
        SocksServerError::UnsupportedSocksVersion(0x06)
    ));
}

// ---------------------------------------------------------------------------
// Group 3: reply-form edge cases (the `new_reply` carried change's
// regression guard)
// ---------------------------------------------------------------------------

/// v4 success reply is exactly 10 bytes with VER/RSV/ATYP pinned and
/// the caller's `sock_addr` carried verbatim. Upstream `new_reply`
/// (`src/server.rs:1380-1404`) + `reply_success`'s pass-through of
/// the caller-supplied address (`src/server.rs:931-941`).
#[tokio::test]
async fn reply_v4_form_is_10_bytes_with_caller_addr_verbatim() {
    let addr: SocketAddr = "203.0.113.9:65535".parse().unwrap();
    let reply = new_reply(&ReplyError::Succeeded, addr);
    assert_eq!(
        reply,
        vec![0x05, 0x00, 0x00, 0x01, 203, 0, 113, 9, 0xff, 0xff]
    );
    assert_eq!(reply.len(), 10);
}

/// v6 reply is exactly 22 bytes with ATYP=0x04 and the full 16-byte
/// address. Upstream `src/server.rs:1387-1392`.
#[tokio::test]
async fn reply_v6_form_is_22_bytes() {
    let addr: SocketAddr = "[2001:db8::ffff]:12345".parse().unwrap();
    let reply = new_reply(&ReplyError::Succeeded, addr);
    assert_eq!(reply.len(), 22);
    assert_eq!(&reply[..4], &[0x05, 0x00, 0x00, 0x04]);
    assert_eq!(
        &reply[4..20],
        &0x2001_0db8_0000_0000_0000_0000_0000_ffffu128.to_be_bytes()
    );
    assert_eq!(&reply[20..22], &12345u16.to_be_bytes());
}

/// VER=0x05 / RSV=0x00 hold in every reply regardless of the code.
#[tokio::test]
async fn reply_ver_and_rsv_are_constant_across_codes() {
    let addr: SocketAddr = "10.0.0.1:1".parse().unwrap();
    let codes = [
        ReplyError::Succeeded,
        ReplyError::GeneralFailure,
        ReplyError::ConnectionNotAllowed,
        ReplyError::NetworkUnreachable,
        ReplyError::HostUnreachable,
        ReplyError::ConnectionRefused,
        ReplyError::ConnectionTimeout,
        ReplyError::TtlExpired,
        ReplyError::CommandNotSupported,
        ReplyError::AddressTypeNotSupported,
    ];
    for code in codes {
        let reply = new_reply(&code, addr);
        assert_eq!(reply[0], 0x05);
        assert_eq!(reply[2], 0x00);
        assert_eq!(reply[3], 0x01);
        assert_eq!(reply[1], code.as_u8());
    }
}

/// Error replies carry the `0.0.0.0:0`-equivalent BND.ADDR: ATYP=1
/// with a zero address + zero port. Upstream `reply_error` passes
/// `"0.0.0.0:0".parse().unwrap()` (`src/server.rs:944-956`); in-tree
/// the same form is the compile-time `V4_UNSPECIFIED_ZERO` const.
#[tokio::test]
async fn reply_error_form_carries_zero_addr() {
    let (mut client, server) = tokio::io::duplex(64);
    let handle: ServerTask<()> = tokio::spawn(async move {
        let proto = Socks5ServerProtocol::accept_no_auth(server).await?;
        let (proto, _cmd, _target) = proto.read_command().await?;
        proto.reply_error(&ReplyError::HostUnreachable).await
    });
    client.write_all(&[0x05, 0x01, 0x00]).await.unwrap();
    let mut method_reply = [0u8; 2];
    client.read_exact(&mut method_reply).await.unwrap();
    assert_eq!(method_reply, [0x05, 0x00]);
    let mut request = vec![0x05, 0x01, 0x00, 0x04];
    request.extend_from_slice(&[0x20; 16]);
    request.extend_from_slice(&[0, 80]);
    client.write_all(&request).await.unwrap();
    let mut reply = [0u8; 10];
    client.read_exact(&mut reply).await.unwrap();
    assert_eq!(reply, [0x05, 0x04, 0x00, 0x01, 0, 0, 0, 0, 0, 0]);
    handle.await.unwrap().unwrap();
}

/// Success replies carry the caller's `sock_addr` verbatim *on the
/// wire*; `reply_success` does not substitute its own address
/// (upstream `src/server.rs:931-941`; the 127.0.0.1 substitution is
/// the `run_tcp_proxy` caller's choice, `src/server.rs:1087`, not
/// the typestate's).
#[tokio::test]
async fn reply_success_carries_caller_sock_addr_on_the_wire() {
    let (mut client, server) = tokio::io::duplex(64);
    let handle: ServerTask<()> = tokio::spawn(async move {
        let proto = Socks5ServerProtocol::accept_no_auth(server).await?;
        let (proto, _cmd, _target) = proto.read_command().await?;
        let bound: SocketAddr = "192.0.2.1:443".parse().unwrap();
        proto.reply_success(bound).await?;
        Ok(())
    });
    client.write_all(&[0x05, 0x01, 0x00]).await.unwrap();
    let mut method_reply = [0u8; 2];
    client.read_exact(&mut method_reply).await.unwrap();
    assert_eq!(method_reply, [0x05, 0x00]);
    client
        .write_all(&[0x05, 0x01, 0x00, 0x01, 10, 0, 0, 1, 0, 80])
        .await
        .unwrap();
    let mut reply = [0u8; 10];
    client.read_exact(&mut reply).await.unwrap();
    assert_eq!(reply, [0x05, 0x00, 0x00, 0x01, 192, 0, 2, 1, 0x01, 0xbb]);
    handle.await.unwrap().unwrap();
}

// ---------------------------------------------------------------------------
// Group 4: client-session edge cases
// ---------------------------------------------------------------------------

/// Reply#1 with a non-zero reply code surfaces as the mapped
/// `ReplyError` (the `from_u8` table; upstream
/// `src/client.rs:363`). Covered per-code including the 0x06 double
/// mapping (ConnectionTimeout and TtlExpired both carry 0x06).
#[tokio::test]
async fn client_reply_code_maps_to_reply_error() {
    for (code, expect) in [
        (
            consts::SOCKS5_REPLY_GENERAL_FAILURE,
            ReplyError::GeneralFailure,
        ),
        (
            consts::SOCKS5_REPLY_CONNECTION_NOT_ALLOWED,
            ReplyError::ConnectionNotAllowed,
        ),
        (
            consts::SOCKS5_REPLY_NETWORK_UNREACHABLE,
            ReplyError::NetworkUnreachable,
        ),
        (
            consts::SOCKS5_REPLY_HOST_UNREACHABLE,
            ReplyError::HostUnreachable,
        ),
        (
            consts::SOCKS5_REPLY_CONNECTION_REFUSED,
            ReplyError::ConnectionRefused,
        ),
        (
            consts::SOCKS5_REPLY_TTL_EXPIRED,
            ReplyError::ConnectionTimeout,
        ),
        (
            consts::SOCKS5_REPLY_COMMAND_NOT_SUPPORTED,
            ReplyError::CommandNotSupported,
        ),
        (
            consts::SOCKS5_REPLY_ADDRESS_TYPE_NOT_SUPPORTED,
            ReplyError::AddressTypeNotSupported,
        ),
    ] {
        let (mut mock, server) = tokio::io::duplex(64);
        let handle: tokio::task::JoinHandle<Result<(), SocksError>> = tokio::spawn(async move {
            let mut session = Socks5Stream::use_stream(server, None, Config::default()).await?;
            session
                .request(
                    Socks5Command::TCPConnect,
                    TargetAddr::Ip("10.0.0.1:80".parse().unwrap()),
                )
                .await?;
            Ok(())
        });
        let mut greeting = [0u8; 3];
        mock.read_exact(&mut greeting).await.unwrap();
        assert_eq!(greeting, [0x05, 0x01, 0x00]);
        mock.write_all(&[0x05, 0x00]).await.unwrap();
        let mut request = [0u8; 10];
        mock.read_exact(&mut request).await.unwrap();
        assert_eq!(&request[..4], &[0x05, 0x01, 0x00, 0x01]);
        mock.write_all(&[0x05, code, 0x00, 0x01, 0, 0, 0, 0, 0, 0])
            .await
            .unwrap();
        let err = handle.await.unwrap().unwrap_err();
        match err {
            SocksError::ReplyError(reply) => assert_eq!(reply.as_u8(), expect.as_u8()),
            other => panic!("expected ReplyError for code {code:#x}, got {other:?}"),
        }
    }
}

/// The client's domain-target request encodes ATYP=3 len-prefixed,
/// truncated to exactly the header length (no zero padding). Upstream
/// `request_header`'s domain arm (`src/client.rs:310-317`) and the
/// write limited to `padding` (`src/client.rs:331`).
#[tokio::test]
async fn client_domain_request_encoding_is_len_prefixed() {
    let (mut mock, server) = tokio::io::duplex(64);
    let handle: tokio::task::JoinHandle<Result<TargetAddr, SocksError>> =
        tokio::spawn(async move {
            let mut session = Socks5Stream::use_stream(server, None, Config::default()).await?;
            session
                .request(
                    Socks5Command::TCPConnect,
                    TargetAddr::Domain("a.example".to_owned(), 8080),
                )
                .await
        });
    let mut greeting = [0u8; 3];
    mock.read_exact(&mut greeting).await.unwrap();
    assert_eq!(greeting, [0x05, 0x01, 0x00]);
    mock.write_all(&[0x05, 0x00]).await.unwrap();
    let mut request = [0u8; 16];
    mock.read_exact(&mut request).await.unwrap();
    assert_eq!(
        request,
        [
            0x05, 0x01, 0x00, 0x03, 9, b'a', b'.', b'e', b'x', b'a', b'm', b'p', b'l', b'e', 0x1f,
            0x90
        ]
    );
    let mut bind_reply = vec![0x05, 0x00, 0x00, 0x03, 9];
    bind_reply.extend_from_slice(b"a.example");
    bind_reply.extend_from_slice(&[0x1f, 0x90]);
    mock.write_all(&bind_reply).await.unwrap();
    let bind = handle.await.unwrap().unwrap();
    assert_eq!(bind, TargetAddr::Domain("a.example".to_owned(), 8080));
}

/// UDP-ASSOCIATE with no target address falls back to the 10-byte
/// zero form (ATYP=1, zero IP + zero port). Upstream
/// `request_header`'s `target_addr: None` branch
/// (`src/client.rs:277-285`): the request reads
/// `[05, 03, 00, 01, 0, 0, 0, 0, 0, 0]`. Driven through the public
/// surface (an in-tree client speaking the wire form against the
/// mock); the in-tree divergence: upstream's missing-target error
/// for non-ASSOCIATE commands was
/// `anyhow::Error::msg("target addr should be present")`
/// (`src/client.rs:285`); in-tree it is the typed
/// `SocksError::MissingTargetAddr` (rfc1928.md's adjustment list),
/// pinned by the wire-vocabulary module's own test
/// (`mod.rs` tests: `socks_error_from_variants` keeps the variant
/// reachable; the private-API form of this test lives in
/// `src/rfc1928/client.rs`).
#[tokio::test]
async fn client_udp_associate_zero_target_fallback_is_10_zero_bytes() {
    // The forked client's `target_addr: None` + UDPAssociate arm
    // emits exactly this 10-byte form; reproduced byte-for-byte here
    // (integration surface) and covered against the real code path in
    // the client module's in-module test (privacy-scoped).
    let expected: &[u8] = &[0x05, 0x03, 0x00, 0x01, 0, 0, 0, 0, 0, 0];
    let mut addr_field = &expected[4..10];
    let parsed = alksocks::rfc1928::util::target_addr::read_address(&mut addr_field, 0x01)
        .await
        .unwrap();
    // The zero form decodes to the unspecified IPv4 addr, port 0;
    // the wire meaning of the fallback.
    assert_eq!(parsed.to_string(), "0.0.0.0:0");
    assert_eq!(expected.len(), 10);
}

/// The client's v6 request encoding: ATYP=4, the full 16-byte
/// address, the 2-byte port, 22 bytes total: the `packet[4..20]`/
/// `packet[20..22]` slice boundaries pinned byte-for-byte. Upstream
/// `request_header`'s v6 arm (`src/client.rs:293-300`); the in-tree
/// arm is byte-identical (`client.rs`, `TargetAddr::Ip
/// (SocketAddr::V6)`). The v6 *reply* form is pinned elsewhere
/// (`reply_v6_form_is_22_bytes`); this pins the request direction.
#[tokio::test]
async fn client_v6_request_encoding_is_22_bytes_with_atyp_4() {
    let (mut mock, server) = tokio::io::duplex(64);
    let handle: tokio::task::JoinHandle<Result<TargetAddr, SocksError>> =
        tokio::spawn(async move {
            let mut session = Socks5Stream::use_stream(server, None, Config::default()).await?;
            session
                .request(
                    Socks5Command::TCPConnect,
                    TargetAddr::Ip("[2001:db8::1]:443".parse().unwrap()),
                )
                .await
        });
    let mut greeting = [0u8; 3];
    mock.read_exact(&mut greeting).await.unwrap();
    assert_eq!(greeting, [0x05, 0x01, 0x00]);
    mock.write_all(&[0x05, 0x00]).await.unwrap();
    let mut request = [0u8; 22];
    mock.read_exact(&mut request).await.unwrap();
    assert_eq!(&request[..4], &[0x05, 0x01, 0x00, 0x04]);
    assert_eq!(
        &request[4..20],
        &0x2001_0db8_0000_0000_0000_0000_0000_0001u128.to_be_bytes()
    );
    assert_eq!(&request[20..22], &443u16.to_be_bytes());
    // The reply form doubles as the request's echo target: v6 replies
    // ride the same 22-byte shape the request just used.
    mock.write_all(&[0x05, 0x00, 0x00, 0x04]).await.unwrap();
    mock.write_all(&0x2001_0db8_0000_0000_0000_0000_0000_0001u128.to_be_bytes())
        .await
        .unwrap();
    mock.write_all(&443u16.to_be_bytes()).await.unwrap();
    let bind = handle.await.unwrap().unwrap();
    assert_eq!(bind, TargetAddr::Ip("[2001:db8::1]:443".parse().unwrap()));
}

/// `Socks5Stream`'s `AsyncWrite::poll_shutdown` passthrough reaches
/// the backing stream: `shutdown()` closes the write side and the
/// mock observes EOF, the half-close propagation shape the consumer
/// half's two-pump teardown relies on (the EOF-sentinel convention).
/// The passthrough is a 3-line delegate (`client.rs` impl block);
/// without this test a regression in the `poll_shutdown` arm (e.g.
/// silently returning `Ready(Ok(()))` without forwarding) would pass
/// the whole suite. Duplex note: `poll_shutdown` on a `DuplexStream`
/// closes the whole pipe's write side for the pair (tokio
/// `io/util/mem.rs` `SimplexStream::poll_shutdown` → `close_write`),
/// so the mock's read side drains one final EOF after the byte.
#[tokio::test]
async fn client_session_shutdown_propagates_to_the_backing_stream() {
    let (mut mock, server) = tokio::io::duplex(64);
    let handle: tokio::task::JoinHandle<Result<(), SocksError>> = tokio::spawn(async move {
        let mut session = Socks5Stream::use_stream(server, None, Config::default()).await?;
        session
            .request(
                Socks5Command::TCPConnect,
                TargetAddr::Ip("10.0.0.1:80".parse().unwrap()),
            )
            .await?;
        session.flush().await?;
        session.shutdown().await?;
        // After the shutdown the write side is closed; a further
        // write surfaces `BrokenPipe` (the passthrough's close
        // semantics, tokio `DuplexStream`'s closed-writer shape).
        match session.write_all(b"late").await {
            Ok(()) => panic!("write after shutdown should not succeed"),
            Err(err) => {
                assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe)
            }
        }
        Ok(())
    });
    let mut greeting = [0u8; 3];
    mock.read_exact(&mut greeting).await.unwrap();
    assert_eq!(greeting, [0x05, 0x01, 0x00]);
    mock.write_all(&[0x05, 0x00]).await.unwrap();
    let mut request = [0u8; 10];
    mock.read_exact(&mut request).await.unwrap();
    assert_eq!(&request[..4], &[0x05, 0x01, 0x00, 0x01]);
    mock.write_all(&[0x05, 0x00, 0x00, 0x01, 0, 0, 0, 0, 0, 0])
        .await
        .unwrap();
    // The handshake's flush landed before the shutdown; the EOF read
    // proves the write side is closed for the peer.
    let mut eof = [0u8; 1];
    match mock.read_exact(&mut eof).await {
        Ok(_) => panic!("expected EOF after shutdown"),
        Err(err) if err.kind() == std::io::ErrorKind::UnexpectedEof => {}
        Err(err) => panic!("expected UnexpectedEof, got {err:?}"),
    }
    handle.await.unwrap().unwrap();
}

/// `skip_auth` writes no method-negotiation bytes: the first wire
/// output is the request itself. Upstream: the `skip_auth` branch
/// skips `send_version_and_methods`/`which_method_accepted`
/// entirely (`src/client.rs:86-93`).
#[tokio::test]
async fn client_skip_auth_emits_no_greeting() {
    let (mut mock, server) = tokio::io::duplex(64);
    let handle: tokio::task::JoinHandle<Result<(), SocksError>> = tokio::spawn(async move {
        let mut config = Config::default();
        config.set_skip_auth(true);
        let mut session = Socks5Stream::use_stream(server, None, config).await?;
        session
            .request(
                Socks5Command::TCPConnect,
                TargetAddr::Ip("10.0.0.1:80".parse().unwrap()),
            )
            .await?;
        Ok(())
    });
    let mut request = [0u8; 10];
    mock.read_exact(&mut request).await.unwrap();
    assert_eq!(request, [0x05, 0x01, 0x00, 0x01, 10, 0, 0, 1, 0, 80]);
    mock.write_all(&[0x05, 0x00, 0x00, 0x01, 0, 0, 0, 0, 0, 0])
        .await
        .unwrap();
    handle.await.unwrap().unwrap();
}

// ---------------------------------------------------------------------------
// Group 5: UDP codec edge cases
// ---------------------------------------------------------------------------

/// `parse_udp_request` with garbage RSV → `GarbageInReserved`. The
/// parser checks the 2-byte RSV before anything else
/// (`src/lib.rs:332-336`).
#[tokio::test]
async fn udp_codec_garbage_reserved_is_rejected() {
    let mut frame = new_udp_header(TargetAddr::Ip("10.0.0.1:53".parse().unwrap())).unwrap();
    frame[1] = 0xa5;
    let err = parse_udp_request(&frame).await.unwrap_err();
    assert!(matches!(
        err,
        alksocks::rfc1928::UdpHeaderError::GarbageInReserved
    ));
}

/// FRAG ≠ 0 passes *through* the parser; the parser's contract is
/// to return the frag byte; refusing fragmentation is the *caller's*
/// job (upstream discards frag≠0 packets in `handle_udp_request`,
/// `src/server.rs:1236-1242`, not in `parse_udp_request`). Pin the
/// parser half of that split.
#[tokio::test]
async fn udp_codec_frag_byte_passes_through_the_parser() {
    let mut frame = new_udp_header(TargetAddr::Domain("example.com".to_owned(), 53)).unwrap();
    frame[2] = 0x07;
    frame.extend_from_slice(b"payload");
    let (frag, target, payload) = parse_udp_request(&frame).await.unwrap();
    assert_eq!(frag, 0x07);
    assert_eq!(target, TargetAddr::Domain("example.com".to_owned(), 53));
    assert_eq!(payload, b"payload");
}

/// `new_udp_header`/`parse_udp_request` round-trip for v6 (the v4
/// and domain forms have in-module coverage; v6 completes the
/// atyp-set). Upstream `new_udp_header` (`src/lib.rs:313-327`).
#[tokio::test]
async fn udp_codec_round_trip_ipv6() {
    let target = TargetAddr::Ip("[2001:db8::1]:53".parse().unwrap());
    let header = new_udp_header(target.clone()).unwrap();
    assert_eq!(header.len(), 4 + 16 + 2);
    assert_eq!(&header[..3], &[0, 0, 0]);
    assert_eq!(header[3], consts::SOCKS5_ADDR_TYPE_IPV6);
    let (frag, parsed, payload) = parse_udp_request(&header).await.unwrap();
    assert_eq!(frag, 0);
    assert_eq!(parsed, target);
    assert!(payload.is_empty());
}

/// The UDP header round-trip preserves a non-empty payload split
/// (header + data): the wire frame is RSV+FRAG+target-bytes and the
/// parser returns the remainder as payload. Upstream's payload split
/// is the returned `req` slice (`src/lib.rs:342`).
#[tokio::test]
async fn udp_codec_round_trip_carries_payload() {
    let target = TargetAddr::Ip("192.0.2.1:5353".parse().unwrap());
    let mut frame = new_udp_header(target.clone()).unwrap();
    frame.extend_from_slice(b"dns query bytes");
    let (frag, parsed, payload) = parse_udp_request(&frame).await.unwrap();
    assert_eq!(frag, 0);
    assert_eq!(parsed, target);
    assert_eq!(payload, b"dns query bytes");
}

// ---------------------------------------------------------------------------
// Group 6: `transfer`/`wait_on_tcp` semantics
// ---------------------------------------------------------------------------

/// `transfer` pumps both directions to completion. Upstream is a
/// `copy_bidirectional` wrapper with the result discarded
/// (`src/server.rs:1209-1218`). Duplex half-close note: a
/// `DuplexStream`'s shutdown closes the whole pipe's write side for
/// both directions of its simplex pair, so the half-close
/// propagation shape is driven with drops (full EOF) here.
#[tokio::test]
async fn transfer_pumps_both_directions_to_completion() {
    let (mut left, right) = tokio::io::duplex(64);
    let (a, mut b) = tokio::io::duplex(64);
    let handle = tokio::spawn(transfer(right, a));
    left.write_all(b"request->").await.unwrap();
    b.write_all(b"<-response").await.unwrap();
    // Drain left→a and b→left via the pump.
    let mut got = [0u8; 9];
    b.read_exact(&mut got).await.unwrap();
    assert_eq!(&got, b"request->");
    let mut back = [0u8; 10];
    left.read_exact(&mut back).await.unwrap();
    assert_eq!(&back, b"<-response");
    // EOF propagates: dropping the writers ends the pump.
    drop(left);
    drop(b);
    handle.await.unwrap();
}

/// `wait_on_tcp` returns `EOF` on a clean close (read → 0 bytes).
/// Upstream `src/server.rs:1195-1206`.
#[tokio::test]
async fn wait_on_tcp_returns_eof_on_clean_close() {
    let (mut client, mut server) = tokio::io::duplex(64);
    client.write_all(b"x").await.unwrap();
    drop(client);
    // Drain the byte, then the next read is the clean EOF.
    let mut buf = [0u8; 1];
    server.read_exact(&mut buf).await.unwrap();
    let err = wait_on_tcp(&mut server).await.unwrap_err();
    assert!(matches!(err, SocksServerError::EOF));
}

/// `wait_on_tcp` returns `UnexpectedUdpControlGarbage` on a control
/// byte; the received byte is carried in the error. Upstream
/// `src/server.rs:1202`.
#[tokio::test]
async fn wait_on_tcp_carries_control_garbage_byte() {
    let (mut client, mut server) = tokio::io::duplex(64);
    client.write_all(&[0xde]).await.unwrap();
    let err = wait_on_tcp(&mut server).await.unwrap_err();
    assert!(matches!(
        err,
        SocksServerError::UnexpectedUdpControlGarbage(0xde)
    ));
}

/// `wait_on_tcp`'s io-error arm: the read *fails* (not EOF, not
/// data) → the io::Error is wrapped with the "waiting on UDP control
/// stream" context. Upstream `src/server.rs:1203`
/// (`Err(err) => Err(err).err_when("waiting on UDP control stream")`).
/// Driven by a real error producer: writing to a duplex whose peer is
/// gone surfaces `BrokenPipe` on the *write* side; for the read arm
/// the harness swaps in a failing reader so the `Err(err)` shape is
/// exercised deterministically (a dropped-peer duplex read is a clean
/// EOF, `Ok(0)`, which is `SocksServerError::EOF` (the other arm).
#[tokio::test]
async fn wait_on_tcp_wraps_io_errors_with_context() {
    struct FailingReader;
    impl tokio::io::AsyncRead for FailingReader {
        fn poll_read(
            self: std::pin::Pin<&mut Self>,
            _cx: &mut std::task::Context<'_>,
            _buf: &mut tokio::io::ReadBuf<'_>,
        ) -> std::task::Poll<std::io::Result<()>> {
            std::task::Poll::Ready(Err(std::io::Error::other("read side failure")))
        }
    }
    let mut stream = FailingReader;
    let err = wait_on_tcp(&mut stream).await.unwrap_err();
    match err {
        SocksServerError::Io { context, .. } => {
            assert_eq!(context, "waiting on UDP control stream")
        }
        other => panic!("expected Io with context, got {other:?}"),
    }
}