ntex_server/net/

test.rs

//! Test server utilities.
#![allow(clippy::missing_panics_doc)]
use std::{fmt, io, marker::PhantomData, net, rc::Rc, thread, time};

use ntex_io::{Io, IoConfig};
use ntex_net::tcp_connect;
use ntex_rt::System;
use ntex_service::{IntoService, Service, cfg::SharedCfg};
use socket2::{Domain, SockAddr, Socket, Type};
use uuid::Uuid;

use super::{NoConfig, Server, ServerAppConfig, ServerBuilder};

/// Builder for a single-worker test server.
///
/// The server listens on an ephemeral `127.0.0.1` port and runs in a separate
/// thread with its own runtime. The service factory receives a reference to
/// the worker state, or `&()` when no application configuration is supplied.
pub struct TestServerBuilder<Cfg, F, Sf, I> {
    id: Uuid,
    cfg: Cfg,
    factory: F,
    config: SharedCfg,
    client_config: SharedCfg,
    _t: PhantomData<(Sf, I)>,
}

impl<Cfg, F, Sf, I> fmt::Debug for TestServerBuilder<Cfg, F, Sf, I> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("TestServerBuilder")
            .field("id", &self.id)
            .field("config", &self.config)
            .field("client_config", &self.client_config)
            .finish()
    }
}

impl<F, S, I> TestServerBuilder<NoConfig, F, S, I>
where
    F: AsyncFn(&()) -> I + Send + Clone + 'static,
    I: IntoService<S, (), Io> + 'static,
    S: Service<(), Io> + 'static,
{
    #[must_use]
    /// Creates a test server builder for the specified service factory.
    ///
    /// The factory receives `&()`: use `async |_| service`.
    pub fn new(factory: F) -> Self {
        Self {
            factory,
            id: Uuid::now_v7(),
            cfg: NoConfig,
            config: SharedCfg::new("TEST-SERVER").into(),
            client_config: SharedCfg::new("TEST-CLIENT").into(),
            _t: PhantomData,
        }
    }
}

impl<Cfg, F, S, I> TestServerBuilder<Cfg, F, S, I>
where
    F: AsyncFn(&Cfg::State) -> I + Send + Clone + 'static,
    I: IntoService<S, Cfg::State, Io> + 'static,
    S: Service<Cfg::State, Io> + 'static,
    Cfg: ServerAppConfig + 'static,
{
    #[must_use]
    /// Creates a test server builder with application configuration.
    ///
    /// `cfg` creates the worker state; `factory` receives a reference to that
    /// state when constructing the service.
    pub fn with(cfg: Cfg, factory: F) -> Self {
        Self {
            cfg,
            factory,
            id: Uuid::now_v7(),
            config: SharedCfg::new("TEST-SERVER").into(),
            client_config: SharedCfg::new("TEST-CLIENT").into(),
            _t: PhantomData,
        }
    }

    #[must_use]
    /// Sets the I/O configuration for accepted server connections.
    pub fn config<T: Into<SharedCfg>>(mut self, cfg: T) -> Self {
        self.config = cfg.into();
        self
    }

    #[must_use]
    /// Sets the I/O configuration used by [`TestServer::connect`].
    pub fn client_config<T: Into<SharedCfg>>(mut self, cfg: T) -> Self {
        self.client_config = cfg.into();
        self
    }

    /// Starts the test server and returns its controller.
    pub fn start(self) -> TestServer {
        log::debug!("Starting test server {:?}", self.id);
        let cfg = self.cfg;
        let config = self.config;
        let factory = self.factory;
        let sys_cfg = System::current().config();
        let name = System::current().name().to_string();

        let (tx, rx) = oneshot::channel();
        // run server in separate thread
        thread::spawn(move || {
            let sys = System::with_config(&name, sys_cfg);
            let tcp = net::TcpListener::bind("127.0.0.1:0").unwrap();
            let local_addr = tcp.local_addr().unwrap();

            sys.run(move || {
                let server = ServerBuilder::new(cfg)
                    .listen("test", tcp, config, factory)?
                    .workers(1)
                    .disable_signals()
                    .run();

                ntex_rt::spawn_detached(async move {
                    tx.send((System::current(), local_addr, server))
                        .expect("Failed to send Server to TestServer");
                });

                Ok(())
            })
        });
        let (system, addr, server) = rx.recv().unwrap();
        thread::sleep(time::Duration::from_millis(25));

        TestServer {
            addr,
            inner: Rc::new(TestServerInner {
                server,
                system,
                id: self.id,
                cfg: self.client_config,
            }),
        }
    }
}

/// Starts a test server for the specified service factory.
///
/// [`TestServer`] is a simple server that makes it easier to write
/// integration tests for ntex applications.
///
/// The factory receives `&()`: use `async |_| service`. For application state,
/// use [`TestServerBuilder::with`].
///
/// # Examples
///
/// ```rust
/// use ntex::{server, http, client::Client};
/// use ntex::web::{self, App, HttpResponse};
///
/// async fn my_handler() -> Result<HttpResponse, std::io::Error> {
///     Ok(HttpResponse::Ok().into())
/// }
///
/// /// STATEMENT: A test server serves a successful HTTP response.
/// ///
/// /// HOW:
/// /// - Start an HTTP service with a factory accepting unit state.
/// /// - Send a request to its assigned port and check the response status.
/// #[ntex::test]
/// async fn test_example() {
///     let srv = server::test_server(
///         async |_| http::HttpService::new(
///             App::new().service(
///                 web::resource("/").to(my_handler))
///         )
///     );
///
///     let req = Client::new().get(format!("http://127.0.0.1:{}", srv.addr().port()).as_str());
///     let response = req.send().await.unwrap();
///     assert!(response.status().is_success());
/// }
/// ```
pub fn test_server<F, S>(factory: F) -> TestServer
where
    F: AsyncFn(&()) -> S + Send + Clone + 'static,
    S: Service<(), Io> + 'static,
{
    TestServerBuilder::new(factory).start()
}

/// Starts a test server configured by a server builder.
///
/// `factory` receives a single-worker builder with signals disabled and must
/// register the listeners. The returned server's address is `127.0.0.1:0`;
/// call [`TestServer::set_addr`] before using [`TestServer::connect`].
pub fn build_test_server<Cfg, F>(cfg: Cfg, factory: F) -> TestServer
where
    Cfg: ServerAppConfig,
    F: AsyncFnOnce(ServerBuilder<Cfg>) -> ServerBuilder<Cfg> + Send + 'static,
{
    let sys = System::current().config();
    let name = System::current().name().to_string();

    let id = Uuid::now_v7();
    log::debug!("Starting {name:?} server {id:?}");

    let (tx, rx) = oneshot::channel();

    // run server in separate thread
    thread::spawn(move || {
        let sys = System::with_config(&name, sys);

        sys.block_on(async move {
            let server = factory(ServerBuilder::new(cfg))
                .await
                .workers(1)
                .disable_signals()
                .run();
            tx.send((System::current(), server.clone()))
                .expect("Failed to send Server to TestServer");
            let _ = server.await;
        });
    });
    let (system, server) = rx.recv().unwrap();
    thread::sleep(time::Duration::from_millis(25));

    TestServer {
        addr: "127.0.0.1:0".parse().unwrap(),
        inner: Rc::new(TestServerInner {
            id,
            system,
            server,
            cfg: SharedCfg::new("TEST-CLIENT").add(IoConfig::new()).into(),
        }),
    }
}

#[derive(Clone, Debug)]
/// Test server controller.
///
/// Dropping the last clone stops the server and its runtime, blocking the
/// current thread for about 100 ms.
pub struct TestServer {
    addr: net::SocketAddr,
    inner: Rc<TestServerInner>,
}

#[derive(Debug)]
struct TestServerInner {
    id: Uuid,
    system: System,
    server: Server,
    cfg: SharedCfg,
}

impl TestServer {
    /// Returns the server address.
    pub fn addr(&self) -> net::SocketAddr {
        self.addr
    }

    #[must_use]
    /// Sets the address used by [`connect`](Self::connect).
    pub fn set_addr(mut self, addr: net::SocketAddr) -> Self {
        self.addr = addr;
        self
    }

    /// Returns the client I/O configuration.
    pub fn config(&self) -> SharedCfg {
        self.inner.cfg.clone()
    }

    /// Connects to the server address.
    pub async fn connect(&self) -> io::Result<Io> {
        tcp_connect(self.addr, self.inner.cfg.clone()).await
    }

    /// Requests a graceful server stop.
    ///
    /// Returns without waiting for the stop to complete.
    pub fn stop(&self) {
        drop(self.inner.server.stop(true));
    }

    /// Returns a currently unused local address.
    ///
    /// The port is released before this returns, so another process may take
    /// it first.
    pub fn unused_addr() -> net::SocketAddr {
        let addr: net::SocketAddr = "127.0.0.1:0".parse().unwrap();
        let socket = Socket::new(Domain::IPV4, Type::STREAM, None).unwrap();
        socket.set_reuse_address(true).unwrap();
        socket.bind(&SockAddr::from(addr)).unwrap();
        let tcp = net::TcpListener::from(socket);
        tcp.local_addr().unwrap()
    }

    /// Returns the running server controller.
    pub fn server(&self) -> Server {
        self.inner.server.clone()
    }
}

impl Drop for TestServerInner {
    fn drop(&mut self) {
        log::debug!("Stopping test server (dropped) {:?}", self.id);
        drop(self.server.stop(false));
        thread::sleep(time::Duration::from_millis(75));
        self.system.stop();
        thread::sleep(time::Duration::from_millis(25));
    }
}