suppaftp 12.0.1

//! # Sync
//!
//! This module contains the definition for all Sync implementation of suppaftp

mod control;
mod data_stream;
mod tls;
mod transfer_stream;

use std::io::{BufRead, BufReader, Cursor, Read, copy};
#[cfg(not(feature = "secure"))]
use std::marker::PhantomData;
use std::net::{Ipv4Addr, SocketAddr, TcpListener, TcpStream, ToSocketAddrs};
use std::sync::{Arc, MutexGuard};
use std::time::{Duration, Instant};

use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
// export
pub use control::ControlSocket;
use control::{ControlChannel, SharedControl};
pub use data_stream::DataStream;
#[cfg(feature = "secure")]
pub use tls::TlsConnector;
#[cfg(feature = "native-tls")]
pub use tls::{NativeTlsConnector, NativeTlsStream};
pub use tls::{NoTlsStream, TlsStream};
#[cfg(any(feature = "rustls-aws-lc-rs", feature = "rustls-ring"))]
pub use tls::{RustlsConnector, RustlsStream};
pub use transfer_stream::TransferStream;

use super::Status;
use super::regex::{EPSV_PORT_RE, MDTM_RE, PASV_PORT_RE, SIZE_RE};
use super::types::{FileType, FtpError, FtpResult, Mode, Response};
use crate::command::Command;
#[cfg(feature = "secure")]
use crate::command::ProtectionLevel;
use crate::types::Features;

/// A function that creates a new stream for the data connection in passive mode.
///
/// It takes a [`SocketAddr`] and returns a [`TcpStream`].
pub type PassiveStreamBuilder = dyn Fn(SocketAddr) -> FtpResult<TcpStream> + Send + Sync;

/// Stream to interface with the FTP server. This interface is only for the command stream.
///
/// The control connection is shared with the [`TransferStream`]s handed out by the data
/// commands, so that each transfer can close its data connection and read the completion reply
/// on its own; see [`TransferStream`] for details.
pub struct ImplFtpStream<T>
where
    T: TlsStream,
{
    /// Control connection, locked for the duration of each command.
    control: SharedControl<T>,
    mode: Mode,
    nat_workaround: bool,
    welcome_msg: Option<String>,
    active_timeout: Duration,
    passive_stream_builder: Box<PassiveStreamBuilder>,
    #[cfg(not(feature = "secure"))]
    marker: PhantomData<T>,
    #[cfg(feature = "secure")]
    tls_ctx: Option<Box<dyn TlsConnector<Stream = T> + Send + Sync + 'static>>,
    #[cfg(feature = "secure")]
    domain: Option<String>,
}

impl<T> ImplFtpStream<T>
where
    T: TlsStream,
{
    /// Try to connect to the remote server
    pub fn connect<A: ToSocketAddrs>(addr: A) -> FtpResult<Self> {
        debug!("Connecting to server");
        TcpStream::connect(addr)
            .map_err(FtpError::ConnectionError)
            .and_then(|stream| Self::connect_with_stream(stream))
    }

    /// Try to connect to the remote server but with the specified timeout
    pub fn connect_timeout(addr: std::net::SocketAddr, timeout: Duration) -> FtpResult<Self> {
        debug!("Connecting to server {addr}");
        TcpStream::connect_timeout(&addr, timeout)
            .map_err(FtpError::ConnectionError)
            .and_then(|stream| Self::connect_with_stream(stream))
    }

    /// Connect using provided configured tcp stream
    pub fn connect_with_stream(stream: TcpStream) -> FtpResult<Self> {
        debug!("Established connection with server");
        let mut ftp_stream = Self {
            control: ControlChannel::shared(DataStream::Tcp(stream)),
            mode: Mode::Passive,
            nat_workaround: false,
            welcome_msg: None,
            active_timeout: Duration::from_secs(60),
            passive_stream_builder: Self::default_passive_stream_builder(),
            #[cfg(feature = "secure")]
            tls_ctx: None,
            #[cfg(feature = "secure")]
            domain: None,
            #[cfg(not(feature = "secure"))]
            marker: PhantomData {},
        };
        debug!("Reading server response...");
        let ready = ftp_stream.control().read_response(Status::Ready);
        match ready {
            Ok(response) => {
                let welcome_msg = response.as_string().ok();
                debug!("Server READY; response: {:?}", welcome_msg);
                ftp_stream.welcome_msg = welcome_msg;
                Ok(ftp_stream)
            }
            Err(err) => Err(err),
        }
    }

    /// Enable active mode for data channel
    pub fn active_mode(mut self, accept_timeout: Duration) -> Self {
        self.mode = Mode::Active;
        self.active_timeout = accept_timeout;
        self
    }

    /// Set a custom [`PassiveStreamBuilder`] for passive mode.
    ///
    /// The stream builder is a function that takes a `SocketAddr` and returns a `TcpStream` and it's used
    /// to create the [`TcpStream`] for the data connection in passive mode.
    pub fn passive_stream_builder<F>(mut self, stream_builder: F) -> Self
    where
        F: Fn(SocketAddr) -> FtpResult<TcpStream> + Send + Sync + 'static,
    {
        self.passive_stream_builder = Box::new(stream_builder);
        self
    }

    /// Set the data channel transfer mode
    pub fn set_mode(&mut self, mode: Mode) {
        debug!("Changed mode to {:?}", mode);
        self.mode = mode;
    }

    /// Set NAT workaround for passive mode
    pub fn set_passive_nat_workaround(&mut self, nat_workaround: bool) {
        self.nat_workaround = nat_workaround;
    }

    /// Switch to explicit secure mode if possible (FTPS), using a provided SSL configuration.
    /// Returns [`FtpError::DataConnectionAlreadyOpen`] before sending `AUTH` if a transfer is alive.
    /// This method does nothing if the connect is already secured.
    ///
    /// ## Example
    ///
    /// ```rust,ignore
    /// use suppaftp::{NativeTlsFtpStream, NativeTlsConnector};
    /// use suppaftp::native_tls::{TlsConnector, TlsStream};
    /// use std::path::Path;
    ///
    /// // Create a TlsConnector
    /// // NOTE: For custom options see <https://docs.rs/native-tls/0.2.6/native_tls/struct.TlsConnectorBuilder.html>
    /// let mut ctx = NativeTlsConnector::from(TlsConnector::new().unwrap());
    /// let mut ftp_stream = NativeTlsFtpStream::connect("127.0.0.1:21").unwrap();
    /// let mut ftp_stream = ftp_stream.into_secure(ctx, "localhost").unwrap();
    /// ```
    #[cfg(feature = "secure")]
    #[cfg_attr(docsrs, doc(cfg(feature = "secure")))]
    pub fn into_secure(
        self,
        tls_connector: impl TlsConnector<Stream = T> + Send + Sync + 'static,
        domain: &str,
    ) -> FtpResult<Self> {
        // Reject a live transfer before asking the server to change the control protocol.
        if Arc::strong_count(&self.control) != 1 {
            return Err(FtpError::DataConnectionAlreadyOpen);
        }
        {
            let mut cc = self.control();
            // Ask the server to start securing data.
            debug!("Initializing TLS auth");
            cc.perform(Command::Auth)?;
            cc.read_response(Status::AuthOk)?;
        }
        debug!("TLS OK; initializing ssl stream");
        let plain = control::into_exclusive(self.control)?
            .reader
            .into_inner()
            .into_tcp_stream()?;
        let stream = tls_connector
            .connect(domain, plain)
            .map_err(|e| FtpError::SecureError(format!("{e}")))?;
        debug!("TLS Steam OK");
        let secured_ftp_tream = Self {
            control: ControlChannel::shared(DataStream::Ssl(Box::new(stream))),
            mode: self.mode,
            nat_workaround: self.nat_workaround,
            passive_stream_builder: self.passive_stream_builder,
            tls_ctx: Some(Box::new(tls_connector)),
            domain: Some(String::from(domain)),
            welcome_msg: self.welcome_msg,
            active_timeout: self.active_timeout,
        };
        {
            let mut cc = secured_ftp_tream.control();
            // Set protection buffer size
            cc.perform(Command::Pbsz(0))?;
            cc.read_response(Status::CommandOk)?;
            // Change the level of data protectio to Private
            cc.perform(Command::Prot(ProtectionLevel::Private))?;
            cc.read_response(Status::CommandOk)?;
        }
        Ok(secured_ftp_tream)
    }

    /// Connect to remote ftps server using IMPLICIT secure connection.
    ///
    /// > Warning: mind that implicit ftps should be considered deprecated, if you can use explicit mode with [`ImplFtpStream::into_secure`]
    ///
    ///
    /// ## Example
    ///
    /// ```rust,ignore
    /// use suppaftp::FtpStream;
    /// use suppaftp::native_tls::{TlsConnector, TlsStream};
    /// use std::path::Path;
    ///
    /// //Create a TlsConnector
    /// //NOTE: For custom options see <https://docs.rs/native-tls/0.2.6/native_tls/struct.TlsConnectorBuilder.html>
    /// let mut ctx = TlsConnector::new().unwrap();
    /// let mut ftp_stream = FtpStream::connect_secure_implicit("127.0.0.1:990", ctx, "localhost").unwrap();
    /// ```
    #[cfg(all(feature = "secure", feature = "deprecated"))]
    #[cfg_attr(docsrs, doc(cfg(all(feature = "secure", feature = "deprecated"))))]
    pub fn connect_secure_implicit<A: ToSocketAddrs>(
        addr: A,
        tls_connector: impl TlsConnector<Stream = T> + Send + Sync + 'static,
        domain: &str,
    ) -> FtpResult<Self> {
        debug!("Connecting to server (secure)");
        let stream = TcpStream::connect(addr).map_err(FtpError::ConnectionError)?;
        debug!("Established connection with server");
        debug!("TLS OK; initializing ssl stream");
        let stream = tls_connector
            .connect(domain, stream)
            .map_err(|e| FtpError::SecureError(format!("{e}")))?;
        debug!("TLS Steam OK");
        let mut stream = Self {
            control: ControlChannel::shared(DataStream::Ssl(Box::new(stream))),
            mode: Mode::Passive,
            nat_workaround: false,
            tls_ctx: Some(Box::new(tls_connector)),
            passive_stream_builder: Self::default_passive_stream_builder(),
            domain: Some(String::from(domain)),
            welcome_msg: None,
            active_timeout: Duration::from_secs(60),
        };
        debug!("Reading server response...");
        let response = stream.control().read_response(Status::Ready)?;
        let welcome_msg = response.as_string().ok();
        debug!("Server READY; response: {:?}", welcome_msg);
        stream.welcome_msg = welcome_msg;

        Ok(stream)
    }

    /// Returns welcome message retrieved from server (if available)
    pub fn get_welcome_msg(&self) -> Option<&str> {
        self.welcome_msg.as_deref()
    }

    /// Returns a locked view of the underlying control [`TcpStream`].
    ///
    /// The returned [`ControlSocket`] dereferences to the socket and keeps the control connection
    /// locked while alive, so drop it before finishing a [`TransferStream`].
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use std::time::Duration;
    ///
    /// use suppaftp::FtpStream;
    ///
    /// let stream = FtpStream::connect("127.0.0.1:21").expect("Couldn't connect to the server...");
    /// stream
    ///     .get_ref()
    ///     .set_read_timeout(Some(Duration::from_secs(10)))
    ///     .expect("set_read_timeout call failed");
    /// ```
    pub fn get_ref(&self) -> ControlSocket<'_, T> {
        ControlSocket::lock(&self.control)
    }

    /// Log in to the FTP server.
    pub fn login<S: AsRef<str>>(&mut self, user: S, password: S) -> FtpResult<()> {
        let mut cc = self.control();
        debug!("Signin in with user '{}'", user.as_ref());
        cc.perform(Command::User(user.as_ref().to_string()))?;
        cc.read_response_in(&[Status::LoggedIn, Status::NeedPassword])
            .and_then(|Response { status, body: _ }| {
                if status == Status::NeedPassword {
                    debug!("Password is required");
                    cc.perform(Command::Pass(password.as_ref().to_string()))?;
                    cc.read_response(Status::LoggedIn)?;
                }
                debug!("Login OK");
                Ok(())
            })
    }

    /// Perform clear command channel (CCC).
    /// Once the command is performed, the command channel will be encrypted no more.
    /// The data stream will still be secure.
    /// Returns [`FtpError::DataConnectionAlreadyOpen`] before sending `CCC` if a transfer is alive.
    #[cfg(feature = "secure")]
    #[cfg_attr(docsrs, doc(cfg(feature = "secure")))]
    pub fn clear_command_channel(mut self) -> FtpResult<Self> {
        // Reject a live transfer before asking the server to change the control protocol.
        if Arc::strong_count(&self.control) != 1 {
            return Err(FtpError::DataConnectionAlreadyOpen);
        }
        {
            let mut cc = self.control();
            // Ask the server to stop securing data
            debug!("performing clear command channel");
            cc.perform(Command::ClearCommandChannel)?;
            cc.read_response(Status::CommandOk)?;
        }
        trace!("CCC OK");
        let plain = control::into_exclusive(self.control)?
            .reader
            .into_inner()
            .into_tcp_stream()?;
        self.control = ControlChannel::shared(DataStream::Tcp(plain));
        Ok(self)
    }

    /// Change the current directory to the path specified.
    pub fn cwd<S: AsRef<str>>(&mut self, path: S) -> FtpResult<()> {
        let mut cc = self.control();
        debug!("Changing working directory to {}", path.as_ref());
        cc.perform(Command::Cwd(path.as_ref().to_string()))?;
        cc.read_response_in(&[Status::CommandOk, Status::RequestedFileActionOk])
            .map(|_| ())
    }

    /// Move the current directory to the parent directory.
    pub fn cdup(&mut self) -> FtpResult<()> {
        let mut cc = self.control();
        debug!("Going to parent directory");
        cc.perform(Command::Cdup)?;
        cc.read_response_in(&[Status::CommandOk, Status::RequestedFileActionOk])
            .map(|_| ())
    }

    /// Gets the current directory
    pub fn pwd(&mut self) -> FtpResult<String> {
        let mut cc = self.control();
        debug!("Getting working directory");
        cc.perform(Command::Pwd)?;
        cc.read_response(Status::PathCreated).and_then(|response| {
            let body = response.as_string().map_err(|_| FtpError::BadResponse)?;
            let status = response.status;
            match (body.find('"'), body.rfind('"')) {
                (Some(begin), Some(end)) if begin < end => Ok(body[begin + 1..end].to_string()),
                _ => Err(FtpError::UnexpectedResponse(Response::new(
                    status,
                    response.body,
                ))),
            }
        })
    }

    /// This does nothing. This is usually just used to keep the connection open.
    pub fn noop(&mut self) -> FtpResult<()> {
        let mut cc = self.control();
        debug!("Pinging server");
        cc.perform(Command::Noop)?;
        cc.read_response(Status::CommandOk).map(|_| ())
    }

    /// The EPRT command allows for the specification of an extended address
    /// for the data connection. The extended address MUST consist of the
    /// network protocol as well as the network and transport addresses
    pub fn eprt(&mut self, address: SocketAddr) -> FtpResult<()> {
        let mut cc = self.control();
        debug!("EPRT with address {address}");
        cc.perform(Command::Eprt(address))?;
        cc.read_response(Status::CommandOk).map(|_| ())
    }

    /// This creates a new directory on the server.
    pub fn mkdir<S: AsRef<str>>(&mut self, pathname: S) -> FtpResult<()> {
        let mut cc = self.control();
        debug!("Creating directory at {}", pathname.as_ref());
        cc.perform(Command::Mkd(pathname.as_ref().to_string()))?;
        // Some non-compliant servers (e.g. bftpd) reply with 200 instead of 257.
        cc.read_response_in(&[Status::PathCreated, Status::CommandOk])
            .map(|_| ())
    }

    /// Sets the type of file to be transferred. That is the implementation
    /// of `TYPE` command.
    pub fn transfer_type(&mut self, file_type: FileType) -> FtpResult<()> {
        let mut cc = self.control();
        debug!("Setting transfer type {}", file_type);
        cc.perform(Command::Type(file_type))?;
        cc.read_response(Status::CommandOk).map(|_| ())
    }

    /// Quits the current FTP session.
    pub fn quit(&mut self) -> FtpResult<()> {
        let mut cc = self.control();
        debug!("Quitting stream");
        cc.perform(Command::Quit)?;
        cc.read_response(Status::Closing).map(|_| ())
    }

    /// Renames the file from_name to to_name
    pub fn rename<S: AsRef<str>>(&mut self, from_name: S, to_name: S) -> FtpResult<()> {
        let mut cc = self.control();
        debug!(
            "Renaming '{}' to '{}'",
            from_name.as_ref(),
            to_name.as_ref()
        );
        cc.perform(Command::RenameFrom(from_name.as_ref().to_string()))?;
        cc.read_response(Status::RequestFilePending).and_then(|_| {
            cc.perform(Command::RenameTo(to_name.as_ref().to_string()))?;
            // Some non-compliant servers (e.g. bftpd) reply with 200 instead of 250.
            cc.read_response_in(&[Status::RequestedFileActionOk, Status::CommandOk])
                .map(|_| ())
        })
    }

    /// The implementation of `RETR` command where `filename` is the name of the file
    /// to download from FTP and `reader` is the function which operates with the
    /// data stream opened.
    ///
    /// The data connection is finished once `reader` returns, whether it succeeded or not, so the
    /// control connection can be reused after successful cleanup.
    ///
    /// ```rust,ignore
    /// use suppaftp::{FtpStream, FtpError};
    /// use std::io::Cursor;
    /// let mut conn = FtpStream::connect("127.0.0.1:10021").unwrap();
    /// conn.login("test", "test").and_then(|_| {
    ///     let mut reader = Cursor::new("hello, world!".as_bytes());
    ///     conn.put_file("retr.txt", &mut reader)
    /// }).unwrap();
    /// assert!(conn.retr("retr.txt", |stream| {
    ///     let mut buf = Vec::new();
    ///     stream.read_to_end(&mut buf).map(|_|
    ///         assert_eq!(buf, "hello, world!".as_bytes())
    ///     ).map_err(|e| FtpError::ConnectionError(e))
    /// }).is_ok());
    /// assert!(conn.rm("retr.txt").is_ok());
    /// ```
    pub fn retr<F, D>(&mut self, file_name: &str, mut reader: F) -> FtpResult<D>
    where
        F: FnMut(&mut dyn Read) -> FtpResult<D>,
    {
        let mut stream = self.retr_as_stream(file_name)?;
        let result = reader(&mut stream);
        // Always finish, but report the reader's error first: it is the root cause.
        let finished = stream.finish();
        let value = result?;
        finished?;
        Ok(value)
    }

    /// Simple way to retr a file from the server. This stores the file in a buffer in memory.
    ///
    /// ```rust,ignore
    /// # use suppaftp::{FtpStream, FtpError};
    /// # use std::io::Cursor;
    /// # let mut conn = FtpStream::connect("127.0.0.1:10021").unwrap();
    /// # conn.login("test", "test").and_then(|_| {
    /// #     let mut reader = Cursor::new("hello, world!".as_bytes());
    /// #     conn.put_file("simple_retr.txt", &mut reader)
    /// # }).unwrap();
    /// let cursor = conn.retr_as_buffer("simple_retr.txt").unwrap();
    /// // do something with bytes
    /// assert_eq!(cursor.into_inner(), "hello, world!".as_bytes());
    /// # assert!(conn.rm("simple_retr.txt").is_ok());
    /// ```
    pub fn retr_as_buffer(&mut self, file_name: &str) -> FtpResult<Cursor<Vec<u8>>> {
        self.retr(file_name, |reader| {
            let mut buffer = Vec::new();
            reader
                .read_to_end(&mut buffer)
                .map(|_| buffer)
                .map_err(FtpError::ConnectionError)
        })
        .map(Cursor::new)
    }

    /// Retrieves the file `file_name` from the server as a readable [`TransferStream`].
    ///
    /// Read the payload from the returned stream, then call [`TransferStream::finish`] to close
    /// the data connection and read the server's completion reply. Until then any other data
    /// command fails with [`FtpError::DataConnectionAlreadyOpen`].
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use std::io::Read;
    ///
    /// use suppaftp::FtpStream;
    ///
    /// let mut ftp = FtpStream::connect("127.0.0.1:21").unwrap();
    /// ftp.login("test", "test").unwrap();
    /// let mut download = ftp.retr_as_stream("hello.txt").unwrap();
    /// let mut buf = Vec::new();
    /// download.read_to_end(&mut buf).unwrap();
    /// download.finish().unwrap();
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`FtpError::DataConnectionAlreadyOpen`] if a transfer is still in progress, and
    /// [`FtpError::UnexpectedResponse`] if the server refuses the `RETR` command.
    pub fn retr_as_stream<S: AsRef<str>>(&mut self, file_name: S) -> FtpResult<TransferStream<T>> {
        debug!("Retrieving '{}'", file_name.as_ref());
        let (_, stream) = self.open_transfer(
            Command::Retr(file_name.as_ref().to_string()),
            &[Status::AboutToSend, Status::AlreadyOpen],
        )?;
        Ok(stream)
    }

    /// Removes the remote pathname from the server.
    pub fn rmdir<S: AsRef<str>>(&mut self, pathname: S) -> FtpResult<()> {
        let mut cc = self.control();
        debug!("Removing directory {}", pathname.as_ref());
        cc.perform(Command::Rmd(pathname.as_ref().to_string()))?;
        // Some non-compliant servers (e.g. bftpd) reply with 200 instead of 250.
        cc.read_response_in(&[Status::RequestedFileActionOk, Status::CommandOk])
            .map(|_| ())
    }

    /// Remove the remote file from the server.
    pub fn rm<S: AsRef<str>>(&mut self, filename: S) -> FtpResult<()> {
        let mut cc = self.control();
        debug!("Removing file {}", filename.as_ref());
        cc.perform(Command::Dele(filename.as_ref().to_string()))?;
        // Some non-compliant servers (e.g. bftpd) reply with 200 instead of 250.
        cc.read_response_in(&[Status::RequestedFileActionOk, Status::CommandOk])
            .map(|_| ())
    }

    /// This stores a file on the server.
    /// r argument must be any struct which implemenents the [`Read`] trait.
    /// Returns amount of written bytes
    pub fn put_file<S: AsRef<str>, R: Read>(&mut self, filename: S, r: &mut R) -> FtpResult<u64> {
        // Get stream
        let mut data_stream = self.put_with_stream(filename.as_ref())?;
        let bytes = copy(r, &mut data_stream).map_err(FtpError::ConnectionError)?;
        data_stream.finish()?;
        Ok(bytes)
    }

    /// Sends `STOR` and returns a writable [`TransferStream`] for the file `filename`.
    ///
    /// Write the payload to the returned stream, then call [`TransferStream::finish`] to close
    /// the data connection and read the server's completion reply. Until then any other data
    /// command fails with [`FtpError::DataConnectionAlreadyOpen`].
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use std::io::Write;
    ///
    /// use suppaftp::FtpStream;
    ///
    /// let mut ftp = FtpStream::connect("127.0.0.1:21").unwrap();
    /// ftp.login("test", "test").unwrap();
    /// let mut upload = ftp.put_with_stream("hello.txt").unwrap();
    /// upload.write_all(b"hello, world!").unwrap();
    /// upload.finish().unwrap();
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`FtpError::DataConnectionAlreadyOpen`] if a transfer is still in progress, and
    /// [`FtpError::UnexpectedResponse`] if the server refuses the `STOR` command.
    pub fn put_with_stream<S: AsRef<str>>(&mut self, filename: S) -> FtpResult<TransferStream<T>> {
        debug!("Put file {}", filename.as_ref());
        let (_, stream) = self.open_transfer(
            Command::Store(filename.as_ref().to_string()),
            &[Status::AlreadyOpen, Status::AboutToSend],
        )?;
        Ok(stream)
    }

    /// Sends `APPE` and returns a writable [`TransferStream`] appending to the file `filename`.
    ///
    /// Behaves like [`ImplFtpStream::put_with_stream`], except that the data is appended.
    ///
    /// # Errors
    ///
    /// Returns [`FtpError::DataConnectionAlreadyOpen`] if a transfer is still in progress, and
    /// [`FtpError::UnexpectedResponse`] if the server refuses the `APPE` command.
    pub fn append_with_stream<S: AsRef<str>>(
        &mut self,
        filename: S,
    ) -> FtpResult<TransferStream<T>> {
        debug!("Appending to file {}", filename.as_ref());
        let (_, stream) = self.open_transfer(
            Command::Appe(filename.as_ref().to_string()),
            &[Status::AlreadyOpen, Status::AboutToSend],
        )?;
        Ok(stream)
    }

    /// Append data from reader to file at `filename`
    pub fn append_file<R: Read>(&mut self, filename: &str, r: &mut R) -> FtpResult<u64> {
        // Get stream
        let mut data_stream = self.append_with_stream(filename)?;
        let bytes = copy(r, &mut data_stream).map_err(FtpError::ConnectionError)?;
        data_stream.finish()?;
        Ok(bytes)
    }

    /// Aborts the transfer running on `transfer` with the `ABOR` command.
    ///
    /// The data connection is closed and the server's abort replies (`426` followed by `226`,
    /// or a single `226`) are consumed, so the control connection is ready for the next command.
    /// `transfer` must have been obtained from this client.
    ///
    /// # Errors
    ///
    /// Returns [`FtpError::UnexpectedResponse`] if the server does not acknowledge the abort.
    pub fn abort(&mut self, transfer: TransferStream<T>) -> FtpResult<()> {
        debug!("Aborting active file transfer");
        // Detach the socket first: the stream must not finalize itself while the lock is held.
        let data_stream = transfer.detach();
        let mut cc = self.control();
        cc.perform(Command::Abor)?;
        // Drop stream NOTE: must be done first, otherwise server won't return any response
        drop(data_stream);
        // mark data channel as closed
        cc.data_connection_open = false;
        trace!("dropped stream");
        let response =
            cc.read_response_in(&[Status::ClosingDataConnection, Status::TransferAborted])?;
        // If server sent 426 (TransferAborted), expect a follow-up 226
        if response.status == Status::TransferAborted {
            cc.read_response(Status::ClosingDataConnection)?;
        }
        debug!("Transfer aborted");
        Ok(())
    }

    /// Tell the server to resume the transfer from a certain offset. The offset indicates the amount of bytes to skip
    /// from the beginning of the file.
    /// the REST command does not actually initiate the transfer.
    /// After issuing a REST command, the client must send the appropriate FTP command to transfer the file
    ///
    /// It is possible to cancel the REST command, sending a REST command with offset 0
    pub fn resume_transfer(&mut self, offset: usize) -> FtpResult<()> {
        let mut cc = self.control();
        debug!("Requesting to resume transfer at offset {}", offset);
        cc.perform(Command::Rest(offset))?;
        cc.read_response(Status::RequestFilePending)?;
        debug!("Resume transfer accepted");
        Ok(())
    }

    /// Execute `LIST` command which returns the detailed file listing in human readable format.
    /// If `pathname` is omited then the list of files in the current directory will be
    /// returned otherwise it will the list of files on `pathname`.
    ///
    /// ### Parse result
    ///
    /// You can parse the output of this command with
    ///
    /// ```rust
    ///
    /// use std::str::FromStr;
    /// use suppaftp::list::File;
    ///
    /// let file: File = File::from_str("-rw-rw-r-- 1 0  1  8192 Nov 5 2018 omar.txt")
    ///
    ///     .unwrap();
    /// ```
    pub fn list(&mut self, pathname: Option<&str>) -> FtpResult<Vec<String>> {
        debug!(
            "Reading {} directory content",
            pathname.unwrap_or("working")
        );

        self.stream_lines(
            Command::List(pathname.map(|x| x.to_string())),
            Status::AboutToSend,
        )
    }

    /// Execute `NLST` command which returns the list of file names only.
    /// If `pathname` is omited then the list of files in the current directory will be
    /// returned otherwise it will the list of files on `pathname`.
    pub fn nlst(&mut self, pathname: Option<&str>) -> FtpResult<Vec<String>> {
        debug!(
            "Getting file names for {} directory",
            pathname.unwrap_or("working")
        );

        self.stream_lines(
            Command::Nlst(pathname.map(|x| x.to_string())),
            Status::AboutToSend,
        )
    }

    /// Execute `MLSD` command which returns the machine-processable listing of a directory.
    /// If `pathname` is omited then the list of files in the current directory will be
    pub fn mlsd(&mut self, pathname: Option<&str>) -> FtpResult<Vec<String>> {
        debug!(
            "Reading {} directory content",
            pathname.unwrap_or("working")
        );

        self.stream_lines(
            Command::Mlsd(pathname.map(|x| x.to_string())),
            Status::AboutToSend,
        )
    }

    /// Execute `MLST` command which returns the machine-processable listing of a file.
    /// If `pathname` is omited then the list of files in the current directory will be
    pub fn mlst(&mut self, pathname: Option<&str>) -> FtpResult<String> {
        let mut cc = self.control();
        debug!("Reading {} path information", pathname.unwrap_or("working"));

        cc.perform(Command::Mlst(pathname.map(|x| x.to_string())))?;
        let response = cc.read_response_in(&[Status::RequestedFileActionOk])?;
        // read body at line 1
        let response_str = String::from_utf8_lossy(&response.body).to_string();
        match response_str.lines().nth(1) {
            Some("") => Err(FtpError::BadResponse),
            Some(line) => Ok(line.trim().to_string()),
            None => Err(FtpError::BadResponse),
        }
    }

    /// Retrieves the modification time of the file at `pathname` if it exists.
    pub fn mdtm<S: AsRef<str>>(&mut self, pathname: S) -> FtpResult<NaiveDateTime> {
        let mut cc = self.control();
        debug!("Getting modification time for {}", pathname.as_ref());
        cc.perform(Command::Mdtm(pathname.as_ref().to_string()))?;
        let response: Response = cc.read_response(Status::File)?;
        let body = response.as_string().map_err(|_| FtpError::BadResponse)?;

        match MDTM_RE.captures(&body) {
            Some(caps) => {
                let (year, month, day) = (
                    caps[1].parse::<i32>().map_err(|_| FtpError::BadResponse)?,
                    caps[2].parse::<u32>().map_err(|_| FtpError::BadResponse)?,
                    caps[3].parse::<u32>().map_err(|_| FtpError::BadResponse)?,
                );
                let (hour, minute, second) = (
                    caps[4].parse::<u32>().map_err(|_| FtpError::BadResponse)?,
                    caps[5].parse::<u32>().map_err(|_| FtpError::BadResponse)?,
                    caps[6].parse::<u32>().map_err(|_| FtpError::BadResponse)?,
                );

                let date = match NaiveDate::from_ymd_opt(year, month, day) {
                    Some(d) => d,
                    None => return Err(FtpError::BadResponse),
                };

                let time = match NaiveTime::from_hms_opt(hour, minute, second) {
                    Some(t) => t,
                    None => return Err(FtpError::BadResponse),
                };

                Ok(NaiveDateTime::new(date, time))
            }
            None => Err(FtpError::BadResponse),
        }
    }

    /// Retrieves the size of the file in bytes at `pathname` if it exists.
    pub fn size<S: AsRef<str>>(&mut self, pathname: S) -> FtpResult<usize> {
        let mut cc = self.control();
        debug!("Getting file size for {}", pathname.as_ref());
        cc.perform(Command::Size(pathname.as_ref().to_string()))?;
        let response: Response = cc.read_response(Status::File)?;
        let body = response.as_string().map_err(|_| FtpError::BadResponse)?;

        match SIZE_RE.captures(&body) {
            Some(caps) => caps[1].parse().map_err(|_| FtpError::BadResponse),
            None => Err(FtpError::BadResponse),
        }
    }

    /// Retrieves the features supported by the server, through the FEAT command.
    pub fn feat(&mut self) -> FtpResult<Features> {
        let mut cc = self.control();
        debug!("Getting server supported features");
        // Send FEAT command
        cc.perform(Command::Feat)?;

        // Read the response
        let response = cc.read_response(Status::System)?;

        let first_line = String::from_utf8_lossy(&response.body);
        let mut feat_lines = vec![first_line.to_string()];
        loop {
            let mut line = Vec::new();
            let line_sz = cc.read_line(&mut line)?;
            if line_sz == 0 {
                // EOF reached
                break;
            }
            let line = String::from_utf8_lossy(&line);
            trace!("FEAT IN: {:?}", line);
            feat_lines.push(line.to_string());
            if crate::command::feat::is_last_line(&line) {
                break;
            }
        }

        crate::command::feat::parse_features(&feat_lines)
    }

    /// Set option `option` with an optional value
    pub fn opts(&mut self, option: impl ToString, value: Option<impl ToString>) -> FtpResult<()> {
        let mut cc = self.control();
        debug!("Getting server supported features");
        cc.perform(Command::Opts(
            option.to_string(),
            value.map(|x| x.to_string()),
        ))?;
        cc.read_response(Status::CommandOk)?;

        Ok(())
    }

    /// Execute a command on the server and return the response
    pub fn site(&mut self, command: impl ToString) -> FtpResult<Response> {
        let mut cc = self.control();
        debug!("Sending SITE command: {}", command.to_string());
        cc.perform(Command::Site(command.to_string()))?;
        cc.read_response(Status::CommandOk)
    }

    /// Perform custom command.
    ///
    /// The command is sent as a single control-channel line, so it must not
    /// contain CR or LF: embedding a line break would let a second command be
    /// smuggled to the server.
    ///
    /// # Errors
    ///
    /// Returns [`FtpError::ConnectionError`] with [`std::io::ErrorKind::InvalidInput`]
    /// if `command` contains CR or LF.
    pub fn custom_command(
        &mut self,
        command: impl ToString,
        expected_code: &[Status],
    ) -> FtpResult<Response> {
        let mut cc = self.control();
        let command = command.to_string();
        debug!("Sending custom command: {}", command);
        cc.perform(Command::Custom(command))?;
        cc.read_response_in(expected_code)
    }

    /// Perform a custom command using the data connection.
    ///
    /// It returns both the [`Response`] and a [`TransferStream`], which implements both
    /// [`std::io::Write`] and [`Read`] and so it can be written or read to interact with the data
    /// channel.
    ///
    /// If you want you can easily parse lines from the stream using [`Self::get_lines_from_stream`].
    ///
    /// Once done, call [`TransferStream::finish`] to close the data connection and read the
    /// server's completion reply.
    ///
    /// # Errors
    ///
    /// Returns [`FtpError::DataConnectionAlreadyOpen`] if a transfer is still in progress, and
    /// [`FtpError::UnexpectedResponse`] if the server's reply is not in `expected_code`.
    pub fn custom_data_command(
        &mut self,
        command: impl ToString,
        expected_code: &[Status],
    ) -> FtpResult<(Response, TransferStream<T>)> {
        let command = command.to_string();
        debug!("Sending custom data command: {}", command);
        self.open_transfer(Command::Custom(command), expected_code)
    }

    /// Read a data stream line by line.
    pub fn get_lines_from_stream<R>(data_stream: &mut R) -> FtpResult<Vec<String>>
    where
        R: BufRead,
    {
        let mut lines: Vec<String> = Vec::new();

        loop {
            let mut line_buf = vec![];
            match data_stream.read_until(b'\n', &mut line_buf) {
                Ok(0) => break,
                Ok(len) => {
                    let mut line = String::from_utf8_lossy(&line_buf[..len]).to_string();
                    trace!("STREAM IN: {:?}", line);
                    if line.ends_with('\n') {
                        line.pop();
                    }
                    if line.ends_with('\r') {
                        line.pop();
                    }
                    if line.is_empty() {
                        continue;
                    }
                    lines.push(line);
                }
                Err(err) => {
                    error!("failed to get lines from stream: {err}");
                    return Err(FtpError::BadResponse);
                }
            }
        }
        trace!("Lines from stream {:?}", lines);

        Ok(lines)
    }

    /// Locks the control connection for the duration of one command.
    fn control(&self) -> MutexGuard<'_, ControlChannel<T>> {
        control::lock(&self.control)
    }

    /// Opens the data connection for `cmd` and wraps it into a self-finalizing [`TransferStream`].
    fn open_transfer(
        &self,
        cmd: Command,
        expected_code: &[Status],
    ) -> FtpResult<(Response, TransferStream<T>)> {
        let mut cc = self.control();
        let (response, data_stream) =
            self.data_command_with_response(&mut cc, cmd, expected_code)?;
        Ok((
            response,
            TransferStream::new(data_stream, Arc::clone(&self.control)),
        ))
    }

    /// Execute command which send data back in a separate stream
    fn data_command(&self, cc: &mut ControlChannel<T>, cmd: Command) -> FtpResult<DataStream<T>> {
        // guard data connection
        cc.guard_multiple_data_connections()?;

        let stream = match self.mode {
            Mode::Active => {
                let listener = self.active(cc)?;
                cc.perform(cmd)?;
                let start = Instant::now();
                loop {
                    match listener.accept() {
                        Ok((stream, _)) => break stream,
                        Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                            if start.elapsed() > self.active_timeout {
                                return Err(FtpError::ConnectionError(
                                    std::io::ErrorKind::WouldBlock.into(),
                                ));
                            }
                            std::thread::sleep(Duration::from_millis(100));
                        }
                        Err(e) => return Err(FtpError::ConnectionError(e)),
                    }
                }
            }
            Mode::ExtendedPassive => {
                let addr = self.epsv(cc)?;
                cc.perform(cmd)?;
                (self.passive_stream_builder)(addr)?
            }
            Mode::Passive => {
                let addr = self.pasv(cc)?;
                cc.perform(cmd)?;
                (self.passive_stream_builder)(addr)?
            }
        };

        #[cfg(not(feature = "secure"))]
        let result = Ok(DataStream::Tcp(stream));

        #[cfg(feature = "secure")]
        let result = match (self.tls_ctx.as_ref(), self.domain.as_ref()) {
            (Some(tls_ctx), Some(domain)) => tls_ctx
                .connect(domain, stream)
                .map(|x| DataStream::Ssl(Box::new(x)))
                .map_err(|e| FtpError::SecureError(format!("{e}"))),
            (Some(_), None) => Err(FtpError::SecureError(
                "TLS context is set but no domain is available for the secure connection"
                    .to_string(),
            )),
            (None, _) => Ok(DataStream::Tcp(stream)),
        };

        if result.is_ok() {
            cc.data_connection_open = true;
        }
        result
    }

    /// Open a data connection for `cmd` and read the preliminary response confirming the transfer.
    ///
    /// If the server rejects the command (the response is not one of `expected_code`), the data
    /// connection is dropped and `data_connection_open` is reset to `false`. This guarantees that a
    /// failed command never leaves the client believing a data connection is still open, which would
    /// otherwise make every subsequent data command fail with [`FtpError::DataConnectionAlreadyOpen`].
    fn data_command_with_response(
        &self,
        cc: &mut ControlChannel<T>,
        cmd: Command,
        expected_code: &[Status],
    ) -> FtpResult<(Response, DataStream<T>)> {
        let data_stream = self.data_command(cc, cmd)?;
        match cc.read_response_in(expected_code) {
            Ok(response) => Ok((response, data_stream)),
            Err(err) => {
                // server rejected the command: drop the data stream and reset the open flag
                drop(data_stream);
                cc.data_connection_open = false;
                Err(err)
            }
        }
    }

    /// Create a new tcp listener and send a PORT command for it
    fn active(&self, cc: &mut ControlChannel<T>) -> FtpResult<TcpListener> {
        debug!("Starting local tcp listener...");
        let conn = TcpListener::bind("0.0.0.0:0").map_err(FtpError::ConnectionError)?;
        conn.set_nonblocking(true)
            .map_err(FtpError::ConnectionError)?;

        let addr = conn.local_addr().map_err(FtpError::ConnectionError)?;
        trace!("Local address is {}", addr);

        let ip = cc
            .socket()
            .local_addr()
            .map_err(FtpError::ConnectionError)?
            .ip();

        debug!("Active mode, listening on {}:{}", ip, addr.port());

        match ip {
            std::net::IpAddr::V4(_) => {
                let msb = addr.port() / 256;
                let lsb = addr.port() % 256;
                let ip_port = format!("{},{},{}", ip.to_string().replace('.', ","), msb, lsb);
                debug!("Running PORT command");
                cc.perform(Command::Port(ip_port))?;
            }
            std::net::IpAddr::V6(_) => {
                debug!("Running EPRT command");
                cc.perform(Command::Eprt(SocketAddr::new(ip, addr.port())))?;
            }
        }
        cc.read_response(Status::CommandOk)?;

        Ok(conn)
    }

    /// Runs the EPSV to enter Extended passive mode.
    fn epsv(&self, cc: &mut ControlChannel<T>) -> FtpResult<SocketAddr> {
        debug!("EPSV command");
        cc.perform(Command::Epsv)?;
        // PASV response format : 229 Entering Extended Passive Mode (|||PORT|)
        let response: Response = cc.read_response(Status::ExtendedPassiveMode)?;
        let response_str = response.as_string().map_err(|_| FtpError::BadResponse)?;
        let caps = EPSV_PORT_RE
            .captures(&response_str)
            .ok_or_else(|| FtpError::UnexpectedResponse(response.clone()))?;
        let new_port = caps[1].parse::<u16>().map_err(|_| FtpError::BadResponse)?;
        trace!("Got port number from EPSV: {}", new_port);
        let mut remote = cc.socket().peer_addr().map_err(FtpError::ConnectionError)?;
        remote.set_port(new_port);
        trace!("Remote address for extended passive mode is {}", remote);
        Ok(remote)
    }

    /// Runs the PASV command  to enter passive mode.
    fn pasv(&self, cc: &mut ControlChannel<T>) -> FtpResult<SocketAddr> {
        debug!("PASV command");
        cc.perform(Command::Pasv)?;
        // PASV response format : 227 Entering Passive Mode (h1,h2,h3,h4,p1,p2).
        let response = cc.read_response(Status::PassiveMode)?;
        let addr = Self::parse_passive_address_from_response(response)?;
        trace!("Passive address: {addr}",);
        if self.nat_workaround {
            let mut remote = cc.socket().peer_addr().map_err(FtpError::ConnectionError)?;
            remote.set_port(addr.port());
            trace!("Replacing site local address {} with {}", addr, remote);
            Ok(remote)
        } else {
            Ok(addr)
        }
    }

    /// Parse passive address from response
    pub(crate) fn parse_passive_address_from_response(response: Response) -> FtpResult<SocketAddr> {
        let response_str = response.as_string().map_err(|_| FtpError::BadResponse)?;
        trace!("PASV response: {response_str}",);
        let caps = PASV_PORT_RE
            .captures(&response_str)
            .ok_or_else(|| FtpError::UnexpectedResponse(response.clone()))?;
        // The regex only guarantees the groups are digits; a malicious or buggy
        // server may still send an octet that overflows a `u8`, so parsing is fallible.
        let parse_octet = |index: usize| -> FtpResult<u8> {
            caps[index]
                .parse::<u8>()
                .map_err(|_| FtpError::UnexpectedResponse(response.clone()))
        };
        let (oct1, oct2, oct3, oct4) = (
            parse_octet(1)?,
            parse_octet(2)?,
            parse_octet(3)?,
            parse_octet(4)?,
        );
        let (msb, lsb) = (parse_octet(5)?, parse_octet(6)?);
        let ip = Ipv4Addr::new(oct1, oct2, oct3, oct4);
        let port = (u16::from(msb) << 8) | u16::from(lsb);
        let addr = SocketAddr::new(ip.into(), port);

        Ok(addr)
    }

    /// Execute a command which returns list of strings in a separate stream
    fn stream_lines(&self, cmd: Command, open_code: Status) -> FtpResult<Vec<String>> {
        let (_, stream) = self.open_transfer(cmd, &[open_code, Status::AlreadyOpen])?;
        let mut data_stream = BufReader::new(stream);
        let lines = Self::get_lines_from_stream(&mut data_stream);
        data_stream.into_inner().finish()?;
        lines
    }

    /// Default stream builder
    fn default_passive_stream_builder() -> Box<PassiveStreamBuilder> {
        Box::new(|addr| TcpStream::connect(addr).map_err(FtpError::ConnectionError))
    }
}

#[cfg(test)]
mod test {

    use std::io::Write;
    use std::net::IpAddr;
    use std::str::FromStr;
    use std::sync::Arc;

    #[cfg(feature = "secure")]
    use pretty_assertions::assert_eq;
    use rand::distr::Alphanumeric;
    use rand::{RngExt, rng};

    use super::*;
    use crate::FtpStream;
    use crate::test_container::SyncPureFtpRunner;
    use crate::types::FormatControl;

    #[test]
    fn connect() {
        crate::log_init();
        with_test_ftp_stream(|_stream| {});
    }

    #[test]
    fn test_should_parse_passive_address_from_response() {
        let response = vec![
            50, 50, 55, 32, 69, 110, 116, 101, 114, 105, 110, 103, 32, 80, 97, 115, 115, 105, 118,
            101, 32, 77, 111, 100, 101, 32, 40, 49, 50, 55, 44, 48, 44, 48, 44, 49, 44, 49, 49, 55,
            44, 53, 54, 41, 13, 10,
        ];
        let response = Response::new(Status::PassiveMode, response);

        let address = FtpStream::parse_passive_address_from_response(response)
            .expect("Failed to parse passive address");
        assert_eq!(
            address.ip(),
            IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)),
            "IP address is not correct"
        );
        assert_eq!(address.port(), 30008, "Port is not correct");

        let response = vec![
            50, 50, 55, 32, 69, 110, 116, 101, 114, 105, 110, 103, 32, 80, 97, 115, 115, 105, 118,
            101, 32, 77, 111, 100, 101, 32, 40, 53, 56, 44, 50, 52, 55, 44, 57, 50, 44, 49, 50, 50,
            44, 49, 52, 54, 44, 50, 51, 57, 41, 46, 13, 10,
        ];
        let response = Response::new(Status::PassiveMode, response);

        let address = FtpStream::parse_passive_address_from_response(response)
            .expect("Failed to parse passive address");
        assert_eq!(
            address.ip(),
            IpAddr::V4(Ipv4Addr::new(58, 247, 92, 122)),
            "IP address is not correct"
        );
        assert_eq!(address.port(), 37615, "Port is not correct");
    }

    #[test]
    fn should_change_mode() {
        with_test_ftp_stream(|stream| {
            assert_eq!(stream.mode, Mode::Passive);
            stream.set_mode(Mode::Active);
            assert_eq!(stream.mode, Mode::Active);
        })
    }

    #[test]
    fn should_connect_with_timeout() {
        crate::log_init();
        let container = SyncPureFtpRunner::start();
        let port = container.get_ftp_port();
        let url = format!("127.0.0.1:{port}");
        let addr: SocketAddr = url.parse().expect("invalid hostname");

        let mut stream = FtpStream::connect_timeout(addr, Duration::from_secs(15)).unwrap();
        assert!(stream.login("test", "test").is_ok());
        assert!(
            stream
                .get_welcome_msg()
                .unwrap()
                .contains("220 Welcome Alpine ftp server")
        );
    }

    #[test]
    fn welcome_message() {
        crate::log_init();
        with_test_ftp_stream(|stream| {
            assert!(
                stream
                    .get_welcome_msg()
                    .unwrap()
                    .contains("220 Welcome Alpine ftp server")
            );
        });
    }

    #[test]
    fn should_set_passive_nat_workaround() {
        with_test_ftp_stream(|stream| {
            stream.set_passive_nat_workaround(true);
            assert!(stream.nat_workaround);
        });
    }

    #[test]
    fn get_ref() {
        use std::time::Duration;
        with_test_ftp_stream(|stream| {
            assert!(
                stream
                    .get_ref()
                    .set_read_timeout(Some(Duration::from_secs(10)))
                    .is_ok()
            );
        });
    }

    #[test]
    fn change_wrkdir() {
        with_test_ftp_stream(|stream| {
            let wrkdir: String = stream.pwd().unwrap();
            assert!(stream.cwd("/").is_ok());
            assert_eq!(stream.pwd().unwrap().as_str(), "/");
            assert!(stream.cwd(wrkdir.as_str()).is_ok());
        })
    }

    #[test]
    fn cd_up() {
        with_test_ftp_stream(|stream| {
            let wrkdir: String = stream.pwd().unwrap();
            assert!(stream.cdup().is_ok());
            assert_eq!(stream.pwd().unwrap().as_str(), "/home/test");
            assert!(stream.cwd(wrkdir.as_str()).is_ok());
        })
    }

    #[test]
    fn noop() {
        with_test_ftp_stream(|stream| {
            assert!(stream.noop().is_ok());
        })
    }

    #[test]
    fn make_and_remove_dir() {
        with_test_ftp_stream(|stream| {
            // Make directory
            assert!(stream.mkdir("omar").is_ok());
            // It shouldn't allow me to re-create the directory; should return error code 550
            match stream.mkdir("omar").err().unwrap() {
                FtpError::UnexpectedResponse(Response { status, body: _ }) => {
                    assert_eq!(status, Status::FileUnavailable)
                }
                err => panic!("Expected UnexpectedResponse, got {}", err),
            }
            // Remove directory
            assert!(stream.rmdir("omar").is_ok());
        })
    }

    #[test]
    fn set_transfer_type() {
        with_test_ftp_stream(|stream| {
            assert!(stream.transfer_type(FileType::Binary).is_ok());
            assert!(
                stream
                    .transfer_type(FileType::Ascii(FormatControl::Default))
                    .is_ok()
            );
        })
    }

    #[test]
    fn test_should_list_files_with_non_utf8_names() {
        with_test_ftp_stream(|stream| {
            let files = stream
                .nlst(Some("/home/test/invalid-utf8/"))
                .expect("Failed to list files");
            assert_eq!(files.len(), 1);

            // list file and parse
            let files = stream
                .list(Some("/home/test/invalid-utf8/"))
                .expect("Failed to list files");
            assert_eq!(files.len(), 1);
            // parse
            crate::list::File::from_str(files[0].as_str()).expect("Failed to parse file");
        });
    }

    #[test]
    fn should_transfer_file() {
        with_test_ftp_stream(|stream| {
            // Set transfer type to Binary
            assert!(stream.transfer_type(FileType::Binary).is_ok());
            // Write file
            let file_data = "test data\n";
            let mut reader = Cursor::new(file_data.as_bytes());
            assert!(stream.put_file("test.txt", &mut reader).is_ok());
            // Read file
            assert_eq!(
                stream
                    .retr_as_buffer("test.txt")
                    .map(|bytes| bytes.into_inner())
                    .unwrap(),
                file_data.as_bytes()
            );
            // Get size
            assert_eq!(stream.size("test.txt").unwrap(), 10);
            // Size of non-existing file
            assert!(stream.size("omarone.txt").is_err());
            // List directory
            assert_eq!(stream.list(None).unwrap().len(), 1);
            // list names
            assert_eq!(stream.nlst(None).unwrap().as_slice(), &["test.txt"]);
            // modification time
            assert!(stream.mdtm("test.txt").is_ok());
            // Remove file
            assert!(stream.rm("test.txt").is_ok());
            assert!(stream.mdtm("test.txt").is_err());
            // Write file, rename and get
            let file_data = "test data\n";
            let mut reader = Cursor::new(file_data.as_bytes());
            assert!(stream.put_file("test.txt", &mut reader).is_ok());
            // Append file
            let mut reader = Cursor::new(file_data.as_bytes());
            assert!(stream.append_file("test.txt", &mut reader).is_ok());
            // Read file
            let mut reader = stream.retr_as_stream("test.txt").unwrap();
            let mut buffer = Vec::new();
            assert!(reader.read_to_end(&mut buffer).is_ok());
            // Finalize
            assert!(reader.finish().is_ok());
            // Verify file matches
            assert_eq!(buffer.as_slice(), "test data\ntest data\n".as_bytes());
            // Rename
            assert!(stream.rename("test.txt", "toast.txt").is_ok());
            assert!(stream.rm("toast.txt").is_ok());
            // List directory again
            assert_eq!(stream.list(None).unwrap().len(), 0);
        })
    }

    #[test]
    fn should_make_subsequent_retr_call_to_non_existing_files() {
        with_test_ftp_stream(|stream| {
            // First call to non-existing file should return error
            let e = stream.retr_as_stream("non-existing.txt").unwrap_err();
            let e2 = stream.retr_as_stream("non-existing.txt").unwrap_err();
            // Both errors should be the same and not be affected by each other
            assert_eq!(e.to_string(), e2.to_string());
        })
    }

    #[test]
    fn should_reset_data_connection_after_failed_data_command() {
        with_test_ftp_stream(|stream| {
            assert!(stream.transfer_type(FileType::Binary).is_ok());

            // A failed data command must leave the data connection free, otherwise the next data
            // command would wrongly fail with `DataConnectionAlreadyOpen` instead of the real error.
            for _ in 0..3 {
                assert!(stream.retr_as_stream("non-existing.txt").is_err());
                assert!(
                    stream
                        .custom_data_command("RETR non-existing.txt", &[Status::AboutToSend])
                        .is_err()
                );
            }

            // After the failures every kind of data command must still be usable.
            let mut reader = Cursor::new("test data\n".as_bytes());
            assert!(stream.put_file("test.txt", &mut reader).is_ok());
            let mut reader = Cursor::new("test data\n".as_bytes());
            assert!(stream.append_file("test.txt", &mut reader).is_ok());
            assert!(stream.retr_as_buffer("test.txt").is_ok());
            assert!(stream.list(None).is_ok());
            assert!(stream.nlst(None).is_ok());

            assert!(stream.rm("test.txt").is_ok());
        })
    }

    #[test]
    fn should_get_feat_and_set_opts() {
        with_test_ftp_stream(|stream| {
            let features = stream.feat().expect("Failed to get features");
            assert!(features.contains_key("UTF8"));
            assert!(stream.opts("UTF8", Some("ON")).is_ok());
        })
    }

    #[test]
    fn should_resume_transfer() {
        crate::log_init();
        let container = Arc::new(SyncPureFtpRunner::start());
        let port = container.get_ftp_port();

        let url = format!("localhost:{port}");

        // init stream with mapper
        let mut stream: FtpStream = setup_stream(&url, &container);

        // Set transfer type to Binary
        assert!(stream.transfer_type(FileType::Binary).is_ok());
        // get dir
        let wrkdir = stream.pwd().unwrap();
        // put as stream
        let mut transfer_stream = stream.put_with_stream("test.bin").unwrap();
        assert_eq!(
            transfer_stream
                .write(&[0x00, 0x01, 0x02, 0x03, 0x04])
                .unwrap(),
            5
        );
        // Drop stream on purpose to simulate a failed connection
        drop(stream);
        drop(transfer_stream);
        // Re-connect to server
        let mut stream = setup_stream(&url, &container);
        // Go back to previous dir
        assert!(stream.cwd(wrkdir).is_ok());
        // Set transfer type to Binary
        assert!(stream.transfer_type(FileType::Binary).is_ok());
        // Resume transfer
        assert!(stream.resume_transfer(5).is_ok());
        // Reopen stream
        let mut transfer_stream = stream.put_with_stream("test.bin").unwrap();
        assert_eq!(
            transfer_stream
                .write(&[0x05, 0x06, 0x07, 0x08, 0x09, 0x0a])
                .unwrap(),
            6
        );
        // Finalize
        assert!(transfer_stream.finish().is_ok());
        // Get size
        //assert_eq!(stream.size("test.bin").unwrap(), 11);
        // Remove file
        assert!(stream.rm("test.bin").is_ok());

        finalize_stream(stream);
    }

    #[test]
    fn should_transfer_with_extended_passive_mode() {
        with_test_ftp_stream(|stream| {
            // Set transfer type to Binary
            assert!(stream.transfer_type(FileType::Binary).is_ok());
            stream.set_mode(Mode::ExtendedPassive);
            // Write file
            let file_data = "test data\n";
            let mut reader = Cursor::new(file_data.as_bytes());
            assert!(stream.put_file("test.txt", &mut reader).is_ok());
            // Remove file
            assert!(stream.rm("test.txt").is_ok());
        })
    }

    #[test]
    fn test_should_perform_custom_command() {
        with_test_ftp_stream(|stream| {
            let command = "PWD";
            assert!(
                stream
                    .custom_command(command, &[Status::PathCreated])
                    .is_ok()
            );
        });
    }

    #[test]
    fn test_should_perform_custom_data_command() {
        with_test_ftp_stream(|stream| {
            let command = "LIST";
            let (response, data_stream) = stream
                .custom_data_command(command, &[Status::AboutToSend])
                .expect("Failed to perform custom data command");
            assert_eq!(response.status, Status::AboutToSend);
            let mut reader = BufReader::new(data_stream);
            FtpStream::get_lines_from_stream(&mut reader).expect("Failed to get lines from stream");
            // finalize
            assert!(reader.into_inner().finish().is_ok());
        });
    }

    #[test]
    fn test_abort_transfer() {
        crate::log_init();
        let container = Arc::new(SyncPureFtpRunner::start());
        let port = container.get_ftp_port();
        let url = format!("localhost:{port}");

        let mut stream: FtpStream = setup_stream(&url, &container);

        assert!(stream.transfer_type(FileType::Binary).is_ok());
        // Write a file first
        let file_data = "test data for abort\n";
        let mut reader = Cursor::new(file_data.as_bytes());
        assert!(stream.put_file("abort_test.txt", &mut reader).is_ok());
        // Open a retr stream
        let data_stream = stream.retr_as_stream("abort_test.txt").unwrap();
        // Abort the transfer - this is the method under test
        assert!(stream.abort(data_stream).is_ok());
        // NOTE: after abort, the server may leave extra responses in the buffer,
        // so we drop the stream without quit. The container cleanup handles the rest.
        drop(stream);
    }

    #[test]
    fn test_retr_with_callback() {
        with_test_ftp_stream(|stream| {
            assert!(stream.transfer_type(FileType::Binary).is_ok());
            let file_data = "hello callback";
            let mut reader = Cursor::new(file_data.as_bytes());
            assert!(stream.put_file("callback.txt", &mut reader).is_ok());

            let result = stream
                .retr("callback.txt", |reader| {
                    let mut buf = Vec::new();
                    reader
                        .read_to_end(&mut buf)
                        .map_err(FtpError::ConnectionError)?;
                    Ok(buf)
                })
                .unwrap();
            assert_eq!(result, file_data.as_bytes());

            assert!(stream.rm("callback.txt").is_ok());
        })
    }

    #[test]
    fn test_append_with_stream() {
        with_test_ftp_stream(|stream| {
            assert!(stream.transfer_type(FileType::Binary).is_ok());
            // Create file
            let mut reader = Cursor::new("part1".as_bytes());
            assert!(stream.put_file("append_stream.txt", &mut reader).is_ok());
            // Append via stream
            let mut data_stream = stream.append_with_stream("append_stream.txt").unwrap();
            data_stream.write_all(b"part2").unwrap();
            data_stream.finish().unwrap();
            // Verify content
            let content = stream
                .retr_as_buffer("append_stream.txt")
                .unwrap()
                .into_inner();
            assert_eq!(content, b"part1part2");
            assert!(stream.rm("append_stream.txt").is_ok());
        })
    }

    #[test]
    fn test_active_mode_builder() {
        crate::log_init();
        let stream = FtpStream::connect("test.rebex.net:21").unwrap();
        let stream = stream.active_mode(Duration::from_secs(30));
        assert_eq!(stream.mode, Mode::Active);
        assert_eq!(stream.active_timeout, Duration::from_secs(30));
    }

    #[test]
    fn test_passive_stream_builder() {
        crate::log_init();
        let stream = FtpStream::connect("test.rebex.net:21").unwrap();
        let stream = stream.passive_stream_builder(|addr| {
            TcpStream::connect(addr).map_err(FtpError::ConnectionError)
        });
        // Verify the builder was set (mode should still be passive)
        assert_eq!(stream.mode, Mode::Passive);
    }

    #[test]
    fn test_parse_passive_address_bad_response() {
        // Non-UTF8 body
        let response = Response::new(Status::PassiveMode, vec![0xff, 0xfe]);
        assert!(FtpStream::parse_passive_address_from_response(response).is_err());

        // No matching pattern
        let response = Response::new(
            Status::PassiveMode,
            "227 No passive mode info here".as_bytes().to_vec(),
        );
        assert!(FtpStream::parse_passive_address_from_response(response).is_err());

        // Octet out of range for a u8 must error instead of panicking
        let response = Response::new(
            Status::PassiveMode,
            "227 Entering Passive Mode (213,229,112,999,216,4)"
                .as_bytes()
                .to_vec(),
        );
        assert!(FtpStream::parse_passive_address_from_response(response).is_err());
    }

    #[test]
    fn test_should_prevent_multiple_data_connections() {
        with_test_ftp_stream(|stream| {
            let command = "LIST";

            let (_, data_stream) = stream
                .custom_data_command(command, &[Status::AboutToSend])
                .expect("Failed to perform custom data command");
            let reader = BufReader::new(data_stream);
            // Try to open another data connection without closing the previous one
            match stream.custom_data_command(command, &[Status::AboutToSend]) {
                Err(FtpError::DataConnectionAlreadyOpen) => {}
                _ => panic!("Expected DataConnectionAlreadyOpen error"),
            }

            // try with all other data commands
            match stream.retr_as_stream("somefile.txt") {
                Err(FtpError::DataConnectionAlreadyOpen) => {}
                _ => panic!("Expected DataConnectionAlreadyOpen error"),
            }

            match stream.put_with_stream("somefile.txt") {
                Err(FtpError::DataConnectionAlreadyOpen) => {}
                _ => panic!("Expected DataConnectionAlreadyOpen error"),
            }

            match stream.append_with_stream("somefile.txt") {
                Err(FtpError::DataConnectionAlreadyOpen) => {}
                _ => panic!("Expected DataConnectionAlreadyOpen error"),
            }

            assert!(reader.into_inner().finish().is_ok());
        });
    }

    #[test]
    fn test_should_free_data_connection_after_close() {
        with_test_ftp_stream(|stream| {
            let command = "LIST";

            let (response, data_stream) = stream
                .custom_data_command(command, &[Status::AboutToSend])
                .expect("Failed to perform custom data command");
            assert_eq!(response.status, Status::AboutToSend);
            let mut reader = BufReader::new(data_stream);
            FtpStream::get_lines_from_stream(&mut reader).expect("Failed to get lines from stream");
            // finalize
            assert!(reader.into_inner().finish().is_ok());

            // Now it should be possible to open another data connection
            let (response, data_stream) = stream
                .custom_data_command(command, &[Status::AboutToSend])
                .expect("Failed to perform custom data command");
            assert_eq!(response.status, Status::AboutToSend);
            let mut reader = BufReader::new(data_stream);
            FtpStream::get_lines_from_stream(&mut reader).expect("Failed to get lines from stream");
            // finalize
            assert!(reader.into_inner().finish().is_ok());
        });
    }

    #[test]
    fn test_site_command() {
        with_test_ftp_stream(|stream| {
            // SITE HELP is commonly supported
            let result = stream.site("HELP");
            // Some servers may not support SITE; just verify it doesn't panic
            // and returns either Ok or an expected error
            match result {
                Ok(response) => {
                    assert_eq!(response.status, Status::CommandOk);
                }
                Err(FtpError::UnexpectedResponse(_)) => {
                    // Server doesn't support SITE, that's ok
                }
                Err(err) => panic!("Unexpected error: {}", err),
            }
        })
    }

    #[test]
    fn test_eprt_command() {
        with_test_ftp_stream(|stream| {
            let addr: SocketAddr = "127.0.0.1:12345".parse().unwrap();
            // EPRT might succeed or fail depending on server config, but should not panic
            let _ = stream.eprt(addr);
        })
    }

    #[test]
    fn test_transfer_type_variants() {
        with_test_ftp_stream(|stream| {
            assert!(
                stream
                    .transfer_type(FileType::Ascii(FormatControl::NonPrint))
                    .is_ok()
            );
            assert!(stream.transfer_type(FileType::Image).is_ok());
            // Local type might not be supported, but should not panic
            let _ = stream.transfer_type(FileType::Local(8));
        })
    }

    #[test]
    fn test_cwd_error() {
        with_test_ftp_stream(|stream| match stream.cwd("/nonexistent/directory/path") {
            Err(FtpError::UnexpectedResponse(response)) => {
                assert_eq!(response.status, Status::FileUnavailable);
            }
            _ => panic!("Expected UnexpectedResponse for nonexistent directory"),
        })
    }

    #[test]
    fn test_rm_nonexistent_file() {
        with_test_ftp_stream(|stream| match stream.rm("nonexistent_file.txt") {
            Err(FtpError::UnexpectedResponse(_)) => {}
            _ => panic!("Expected error when removing nonexistent file"),
        })
    }

    #[test]
    fn test_rmdir_nonexistent() {
        with_test_ftp_stream(|stream| match stream.rmdir("nonexistent_dir") {
            Err(FtpError::UnexpectedResponse(_)) => {}
            _ => panic!("Expected error when removing nonexistent directory"),
        })
    }

    #[test]
    fn test_rename_nonexistent() {
        with_test_ftp_stream(
            |stream| match stream.rename("nonexistent.txt", "new_name.txt") {
                Err(FtpError::UnexpectedResponse(_)) => {}
                _ => panic!("Expected error when renaming nonexistent file"),
            },
        )
    }

    #[test]
    fn test_retr_nonexistent_file() {
        with_test_ftp_stream(
            |stream| match stream.retr_as_buffer("nonexistent_file.txt") {
                Err(FtpError::UnexpectedResponse(_)) => {}
                _ => panic!("Expected error when retrieving nonexistent file"),
            },
        )
    }

    #[test]
    fn test_list_nonexistent_path() {
        with_test_ftp_stream(|stream| {
            // Listing a nonexistent path - behavior varies per server
            let _ = stream.list(Some("/nonexistent/path"));
        })
    }

    #[test]
    fn test_nlst_with_path() {
        with_test_ftp_stream(|stream| {
            // Put a file first
            let mut reader = Cursor::new("data".as_bytes());
            assert!(stream.put_file("nlst_test.txt", &mut reader).is_ok());

            let files = stream.nlst(None).unwrap();
            assert!(files.contains(&"nlst_test.txt".to_string()));

            assert!(stream.rm("nlst_test.txt").is_ok());
        })
    }

    #[test]
    fn test_list_with_explicit_current_dir() {
        with_test_ftp_stream(|stream| {
            let mut reader = Cursor::new("data".as_bytes());
            assert!(stream.put_file("list_test.txt", &mut reader).is_ok());

            let files = stream.list(Some(".")).unwrap();
            assert!(!files.is_empty());

            assert!(stream.rm("list_test.txt").is_ok());
        })
    }

    #[test]
    fn should_upload_via_stream_and_finish() {
        with_test_ftp_stream(|stream| {
            assert!(stream.transfer_type(FileType::Binary).is_ok());
            let mut upload = stream.put_with_stream("upload.bin").unwrap();
            upload.write_all(b"0123456789").unwrap();
            upload
                .finish()
                .expect("finish should read the transfer reply");
            // The control connection is in sync: the next commands work.
            assert_eq!(stream.size("upload.bin").unwrap(), 10);
            assert_eq!(stream.nlst(None).unwrap().as_slice(), &["upload.bin"]);
            assert!(stream.rm("upload.bin").is_ok());
        })
    }

    #[test]
    fn should_download_via_stream_and_finish() {
        with_test_ftp_stream(|stream| {
            assert!(stream.transfer_type(FileType::Binary).is_ok());
            let mut reader = Cursor::new("download me".as_bytes());
            assert!(stream.put_file("download.txt", &mut reader).is_ok());
            let mut download = stream.retr_as_stream("download.txt").unwrap();
            let mut buffer = Vec::new();
            download.read_to_end(&mut buffer).unwrap();
            download
                .finish()
                .expect("finish should read the transfer reply");
            assert_eq!(buffer, b"download me");
            // The control connection is in sync: the next commands work.
            assert!(stream.pwd().is_ok());
            assert_eq!(stream.size("download.txt").unwrap(), 11);
            assert!(stream.rm("download.txt").is_ok());
        })
    }

    #[test]
    fn should_finalize_dropped_stream() {
        with_test_ftp_stream(|stream| {
            assert!(stream.transfer_type(FileType::Binary).is_ok());
            let mut upload = stream.put_with_stream("dropped.bin").unwrap();
            upload.write_all(b"dropped").unwrap();
            // Drop without finish(): the stream reads the reply on its own.
            drop(upload);
            assert_eq!(stream.size("dropped.bin").unwrap(), 7);
            let content = stream.retr_as_buffer("dropped.bin").unwrap().into_inner();
            assert_eq!(content, b"dropped");
            assert!(stream.rm("dropped.bin").is_ok());
        })
    }

    #[test]
    fn should_read_a_single_reply_on_finish_then_drop() {
        with_test_ftp_stream(|stream| {
            assert!(stream.transfer_type(FileType::Binary).is_ok());
            // A short read timeout makes a second (spurious) reply read fail slowly and visibly.
            let timeout = Duration::from_millis(500);
            stream.get_ref().set_read_timeout(Some(timeout)).unwrap();
            let mut upload = stream.put_with_stream("single.bin").unwrap();
            upload.write_all(b"single").unwrap();
            let started = Instant::now();
            upload
                .finish()
                .expect("finish should read exactly one reply");
            assert!(
                started.elapsed() < timeout,
                "finish() took {:?}: a second reply read blocked until the timeout",
                started.elapsed()
            );
            // The control connection is in sync: the next commands work.
            assert!(stream.noop().is_ok());
            assert_eq!(stream.size("single.bin").unwrap(), 6);
            stream.get_ref().set_read_timeout(None).unwrap();
            assert!(stream.rm("single.bin").is_ok());
        })
    }

    #[test]
    fn should_reject_data_commands_while_a_transfer_stream_is_alive() {
        with_test_ftp_stream(|stream| {
            assert!(stream.transfer_type(FileType::Binary).is_ok());
            let mut upload = stream.put_with_stream("alive.bin").unwrap();
            upload.write_all(b"alive").unwrap();
            assert!(matches!(
                stream.list(None),
                Err(FtpError::DataConnectionAlreadyOpen)
            ));
            assert!(matches!(
                stream.retr_as_stream("alive.bin"),
                Err(FtpError::DataConnectionAlreadyOpen)
            ));
            assert!(matches!(
                stream.put_with_stream("other.bin"),
                Err(FtpError::DataConnectionAlreadyOpen)
            ));
            upload.finish().unwrap();
            // Once finished, data commands work again.
            assert_eq!(stream.nlst(None).unwrap().as_slice(), &["alive.bin"]);
            assert!(stream.rm("alive.bin").is_ok());
        })
    }

    #[test]
    fn should_finish_transfer_stream_on_another_thread() {
        with_test_ftp_stream(|stream| {
            assert!(stream.transfer_type(FileType::Binary).is_ok());
            let upload = stream.put_with_stream("threaded.bin").unwrap();
            let handle = std::thread::spawn(move || {
                let mut upload = upload;
                upload.write_all(b"from another thread").unwrap();
                upload.finish()
            });
            handle
                .join()
                .expect("uploader thread panicked")
                .expect("finish should succeed on another thread");
            assert_eq!(stream.size("threaded.bin").unwrap(), 19);
            assert!(stream.rm("threaded.bin").is_ok());
        })
    }

    #[test]
    fn should_finalize_transfer_when_retr_callback_fails() {
        with_test_ftp_stream(|stream| {
            assert!(stream.transfer_type(FileType::Binary).is_ok());
            let mut reader = Cursor::new("callback".as_bytes());
            assert!(stream.put_file("callback_err.txt", &mut reader).is_ok());
            let result: FtpResult<()> =
                stream.retr("callback_err.txt", |_| Err(FtpError::BadResponse));
            assert!(matches!(result, Err(FtpError::BadResponse)));
            // The failed callback must not leave the data connection flagged as open.
            assert_eq!(stream.size("callback_err.txt").unwrap(), 8);
            assert!(stream.list(None).is_ok());
            assert!(stream.rm("callback_err.txt").is_ok());
        })
    }

    // -- test utils

    fn with_test_ftp_stream<F>(f: F)
    where
        F: FnOnce(&mut FtpStream),
    {
        crate::log_init();
        let container = Arc::new(SyncPureFtpRunner::start());
        let port = container.get_ftp_port();

        // init stream with mapper
        let mut stream: FtpStream = setup_stream(&format!("localhost:{port}"), &container);

        f(&mut stream);
        finalize_stream(stream);

        drop(container);
    }

    fn setup_stream(url: &str, container: &Arc<SyncPureFtpRunner>) -> FtpStream {
        let mut ftp_stream = FtpStream::connect(url).unwrap();
        assert!(ftp_stream.login("test", "test").is_ok());
        // Create wrkdir
        let tempdir: String = generate_tempdir();
        assert!(ftp_stream.mkdir(tempdir.as_str()).is_ok());
        // Change directory
        assert!(ftp_stream.cwd(tempdir.as_str()).is_ok());

        let container_t = container.clone();

        ftp_stream.passive_stream_builder(move |addr| {
            let mut addr = addr;
            let port = addr.port();
            let mapped = container_t.get_mapped_port(port);

            addr.set_port(mapped);

            info!("mapped port {port} to {mapped} for PASV");

            // open stream to this address instead
            TcpStream::connect(addr).map_err(FtpError::ConnectionError)
        })
    }

    fn finalize_stream(mut stream: FtpStream) {
        assert!(stream.quit().is_ok());
    }

    fn generate_tempdir() -> String {
        let mut rng = rng();
        let name: String = std::iter::repeat(())
            .map(|()| rng.sample(Alphanumeric))
            .map(char::from)
            .take(5)
            .collect();
        format!("temp_{}", name)
    }

    // M-MOCKABLE-SYSCALLS: use a local server for deterministic control-response tests.
    fn connect_with_control_response(
        expected_command: &'static str,
        response: &'static [u8],
    ) -> (FtpStream, std::thread::JoinHandle<()>) {
        use std::io::{BufRead as _, BufReader as IoBufReader, Write as _};

        let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind");
        let port = listener.local_addr().expect("missing local address").port();
        let handle = std::thread::spawn(move || {
            let (stream, _) = listener.accept().expect("no incoming connection");
            let mut writer = stream.try_clone().expect("failed to clone stream");
            let mut reader = IoBufReader::new(stream);

            writer.write_all(b"220 Welcome\r\n").unwrap();

            let mut line = String::new();
            reader.read_line(&mut line).unwrap();
            assert_eq!(line, format!("{expected_command}\r\n"));
            writer.write_all(response).unwrap();
        });

        let tcp = TcpStream::connect(("127.0.0.1", port)).expect("failed to connect");
        let stream = FtpStream::connect_with_stream(tcp).expect("failed handshake");
        (stream, handle)
    }

    /// Test if the stream is Send
    fn is_send<T: Send>(_send: T) {}

    fn is_sync<T: Sync>(_sync: T) {}

    #[test]
    #[ignore = "just needs to compile"]
    fn test_ftp_stream_should_be_send() {
        crate::log_init();

        let ftp_stream = FtpStream::connect("test.rebex.net:21")
            .unwrap()
            .passive_stream_builder(|addr| {
                println!("Connecting to {}", addr);
                TcpStream::connect(addr).map_err(FtpError::ConnectionError)
            });

        is_send::<FtpStream>(ftp_stream);
    }

    #[test]
    #[ignore = "just needs to compile"]
    fn test_ftp_stream_should_be_sync() {
        crate::log_init();
        let ftp_stream = FtpStream::connect("test.rebex.net:21")
            .unwrap()
            .passive_stream_builder(|addr| {
                println!("Connecting to {}", addr);
                TcpStream::connect(addr).map_err(FtpError::ConnectionError)
            });

        is_sync::<FtpStream>(ftp_stream);
    }

    #[test]
    fn should_accept_mismatched_multiline_terminal_code() {
        let response = b"553-\r\n\
553- Directory excluded (precheck).\r\n\
150 Opening BINARY mode data connection.\r\n";
        let (mut stream, handle) = connect_with_control_response("STOR upload.bin", response);

        let response = stream
            .custom_command("STOR upload.bin", &[Status::AboutToSend])
            .expect("mismatched terminal code should be accepted");

        assert_eq!(response.status, Status::AboutToSend);
        handle.join().expect("server thread panicked");
    }

    #[test]
    fn should_preserve_feat_multiline_handling() {
        let response = b"211-Features:\r\n UTF8\r\n211 END\r\n";
        let (mut stream, handle) = connect_with_control_response("FEAT", response);

        let features = stream.feat().expect("FEAT response should be parsed");

        assert!(features.contains_key("UTF8"));
        handle.join().expect("server thread panicked");
    }

    #[test]
    fn should_accept_200_for_file_operations() {
        use std::io::{BufRead, BufReader as IoBufReader, Write};
        use std::net::TcpListener;
        use std::thread;

        crate::log_init();

        let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind");
        let port = listener.local_addr().unwrap().port();

        // Fake FTP server that replies with 200 instead of the spec-mandated
        // 250/257 to file operations, mimicking non-compliant servers such as bftpd.
        let handle = thread::spawn(move || {
            let (stream, _) = listener.accept().expect("no incoming connection");
            let mut writer = stream.try_clone().expect("failed to clone stream");
            let mut reader = IoBufReader::new(stream);

            writer.write_all(b"220 Welcome\r\n").unwrap();

            let mut line = String::new();
            while reader.read_line(&mut line).unwrap() > 0 {
                let reply: &[u8] = match line.split_whitespace().next() {
                    // RNFR must be acknowledged with 350 before RNTO is sent.
                    Some("RNFR") => b"350 ready for destination name\r\n",
                    Some("QUIT") => b"221 goodbye\r\n",
                    // Everything else is answered with the non-compliant 200.
                    _ => b"200 ok\r\n",
                };
                writer.write_all(reply).unwrap();
                if line.starts_with("QUIT") {
                    break;
                }
                line.clear();
            }
        });

        let tcp = TcpStream::connect(("127.0.0.1", port)).expect("failed to connect");
        let mut stream = FtpStream::connect_with_stream(tcp).expect("failed handshake");

        assert!(stream.mkdir("dir").is_ok());
        assert!(stream.rename("a", "b").is_ok());
        assert!(stream.rmdir("dir").is_ok());
        assert!(stream.rm("file").is_ok());

        assert!(stream.quit().is_ok());
        handle.join().expect("server thread panicked");
    }

    #[test]
    fn should_reject_command_with_crlf_injection() {
        use std::io::{BufRead, BufReader as IoBufReader, Write};
        use std::thread;

        crate::log_init();

        let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind");
        let port = listener.local_addr().unwrap().port();

        // Fake FTP server that records every control-channel line it receives.
        let handle = thread::spawn(move || {
            let (stream, _) = listener.accept().expect("no incoming connection");
            let mut writer = stream.try_clone().expect("failed to clone stream");
            let mut reader = IoBufReader::new(stream);

            writer.write_all(b"220 Welcome\r\n").unwrap();

            let mut seen = Vec::new();
            let mut line = String::new();
            while reader.read_line(&mut line).unwrap() > 0 {
                let is_quit = line.starts_with("QUIT");
                seen.push(line.trim_end_matches(['\r', '\n']).to_string());
                let reply: &[u8] = if is_quit {
                    b"221 goodbye\r\n"
                } else {
                    b"200 ok\r\n"
                };
                writer.write_all(reply).unwrap();
                if is_quit {
                    break;
                }
                line.clear();
            }
            seen
        });

        let tcp = TcpStream::connect(("127.0.0.1", port)).expect("failed to connect");
        let mut stream = FtpStream::connect_with_stream(tcp).expect("failed handshake");

        let err = stream
            .cwd("dir\r\nDELE secret.txt")
            .expect_err("command with CRLF must be rejected");
        assert!(
            matches!(err, FtpError::ConnectionError(ref e) if e.kind() == std::io::ErrorKind::InvalidInput),
            "unexpected error: {err:?}"
        );
        assert!(stream.cwd("dir\nDELE secret.txt").is_err());
        assert!(stream.login("anon\r\nDELE secret.txt", "pw").is_err());
        assert!(
            stream
                .custom_command("NOOP\r\nDELE secret.txt", &[Status::CommandOk])
                .is_err()
        );

        assert!(stream.quit().is_ok());
        let seen = handle.join().expect("server thread panicked");
        assert_eq!(seen, vec!["QUIT".to_string()]);
    }
}