ntex_server/net/
test.rs

1//! Test server utilities.
2#![allow(clippy::missing_panics_doc)]
3use std::{fmt, io, marker::PhantomData, net, rc::Rc, thread, time};
4
5use ntex_io::{Io, IoConfig};
6use ntex_net::tcp_connect;
7use ntex_rt::System;
8use ntex_service::{IntoService, Service, cfg::SharedCfg};
9use socket2::{Domain, SockAddr, Socket, Type};
10use uuid::Uuid;
11
12use super::{NoConfig, Server, ServerAppConfig, ServerBuilder};
13
14/// Builder for a single-worker test server.
15///
16/// The server listens on an ephemeral `127.0.0.1` port and runs in a separate
17/// thread with its own runtime.
18pub struct TestServerBuilder<Cfg, F, Sf, I> {
19    id: Uuid,
20    cfg: Cfg,
21    factory: F,
22    config: SharedCfg,
23    client_config: SharedCfg,
24    _t: PhantomData<(Sf, I)>,
25}
26
27impl<Cfg, F, Sf, I> fmt::Debug for TestServerBuilder<Cfg, F, Sf, I> {
28    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
29        f.debug_struct("TestServerBuilder")
30            .field("id", &self.id)
31            .field("config", &self.config)
32            .field("client_config", &self.client_config)
33            .finish()
34    }
35}
36
37impl<F, S, I> TestServerBuilder<NoConfig, F, S, I>
38where
39    F: AsyncFn() -> I + Send + Clone + 'static,
40    I: IntoService<S, (), Io> + 'static,
41    S: Service<(), Io> + 'static,
42{
43    #[must_use]
44    /// Creates a test server builder for the specified service factory.
45    pub fn new(factory: F) -> Self {
46        Self {
47            factory,
48            id: Uuid::now_v7(),
49            cfg: NoConfig,
50            config: SharedCfg::new("TEST-SERVER").into(),
51            client_config: SharedCfg::new("TEST-CLIENT").into(),
52            _t: PhantomData,
53        }
54    }
55}
56
57impl<Cfg, F, S, I> TestServerBuilder<Cfg, F, S, I>
58where
59    F: AsyncFn() -> I + Send + Clone + 'static,
60    I: IntoService<S, Cfg::State, Io> + 'static,
61    S: Service<Cfg::State, Io> + 'static,
62    Cfg: ServerAppConfig + 'static,
63{
64    #[must_use]
65    /// Creates a test server builder with application configuration.
66    pub fn with(cfg: Cfg, factory: F) -> Self {
67        Self {
68            cfg,
69            factory,
70            id: Uuid::now_v7(),
71            config: SharedCfg::new("TEST-SERVER").into(),
72            client_config: SharedCfg::new("TEST-CLIENT").into(),
73            _t: PhantomData,
74        }
75    }
76
77    #[must_use]
78    /// Sets the I/O configuration for accepted server connections.
79    pub fn config<T: Into<SharedCfg>>(mut self, cfg: T) -> Self {
80        self.config = cfg.into();
81        self
82    }
83
84    #[must_use]
85    /// Sets the I/O configuration used by [`TestServer::connect`].
86    pub fn client_config<T: Into<SharedCfg>>(mut self, cfg: T) -> Self {
87        self.client_config = cfg.into();
88        self
89    }
90
91    /// Starts the test server and returns its controller.
92    pub fn start(self) -> TestServer {
93        log::debug!("Starting test server {:?}", self.id);
94        let cfg = self.cfg;
95        let config = self.config;
96        let factory = self.factory;
97        let sys_cfg = System::current().config();
98        let name = System::current().name().to_string();
99
100        let (tx, rx) = oneshot::channel();
101        // run server in separate thread
102        thread::spawn(move || {
103            let sys = System::with_config(&name, sys_cfg);
104            let tcp = net::TcpListener::bind("127.0.0.1:0").unwrap();
105            let local_addr = tcp.local_addr().unwrap();
106
107            sys.run(move || {
108                let server = ServerBuilder::new(cfg)
109                    .listen("test", tcp, config, async move |_| factory().await)?
110                    .workers(1)
111                    .disable_signals()
112                    .enable_affinity()
113                    .run();
114
115                ntex_rt::spawn(async move {
116                    tx.send((System::current(), local_addr, server))
117                        .expect("Failed to send Server to TestServer");
118                });
119
120                Ok(())
121            })
122        });
123        let (system, addr, server) = rx.recv().unwrap();
124        thread::sleep(time::Duration::from_millis(25));
125
126        TestServer {
127            addr,
128            inner: Rc::new(TestServerInner {
129                server,
130                system,
131                id: self.id,
132                cfg: self.client_config,
133            }),
134        }
135    }
136}
137
138/// Starts a test server for the specified service factory.
139///
140/// [`TestServer`] is a simple server that makes it easier to write
141/// integration tests for ntex applications.
142///
143/// # Examples
144///
145/// ```rust
146/// use ntex::{server, http, client::Client};
147/// use ntex::web::{self, App, HttpResponse};
148///
149/// async fn my_handler() -> Result<HttpResponse, std::io::Error> {
150///     Ok(HttpResponse::Ok().into())
151/// }
152///
153/// #[ntex::test]
154/// async fn test_example() {
155///     let srv = server::test_server(
156///         async || http::HttpService::new(
157///             App::new().service(
158///                 web::resource("/").to(my_handler))
159///         )
160///     );
161///
162///     let req = Client::new().get(format!("http://127.0.0.1:{}", srv.addr().port()).as_str());
163///     let response = req.send().await.unwrap();
164///     assert!(response.status().is_success());
165/// }
166/// ```
167pub fn test_server<F, S>(factory: F) -> TestServer
168where
169    F: AsyncFn() -> S + Send + Clone + 'static,
170    S: Service<(), Io> + 'static,
171{
172    TestServerBuilder::new(factory).start()
173}
174
175/// Starts a test server configured by a server builder.
176///
177/// `factory` receives a single-worker builder with signals disabled and must
178/// register the listeners. The returned server's address is `127.0.0.1:0`;
179/// call [`TestServer::set_addr`] before using [`TestServer::connect`].
180pub fn build_test_server<Cfg, F>(cfg: Cfg, factory: F) -> TestServer
181where
182    Cfg: ServerAppConfig,
183    F: AsyncFnOnce(ServerBuilder<Cfg>) -> ServerBuilder<Cfg> + Send + 'static,
184{
185    let sys = System::current().config();
186    let name = System::current().name().to_string();
187
188    let id = Uuid::now_v7();
189    log::debug!("Starting {name:?} server {id:?}");
190
191    let (tx, rx) = oneshot::channel();
192
193    // run server in separate thread
194    thread::spawn(move || {
195        let sys = System::with_config(&name, sys);
196
197        sys.block_on(async move {
198            let server = factory(ServerBuilder::new(cfg))
199                .await
200                .workers(1)
201                .disable_signals()
202                .run();
203            tx.send((System::current(), server.clone()))
204                .expect("Failed to send Server to TestServer");
205            let _ = server.await;
206        });
207    });
208    let (system, server) = rx.recv().unwrap();
209    thread::sleep(time::Duration::from_millis(25));
210
211    TestServer {
212        addr: "127.0.0.1:0".parse().unwrap(),
213        inner: Rc::new(TestServerInner {
214            id,
215            system,
216            server,
217            cfg: SharedCfg::new("TEST-CLIENT").add(IoConfig::new()).into(),
218        }),
219    }
220}
221
222#[derive(Clone, Debug)]
223/// Test server controller.
224///
225/// Dropping the last clone stops the server and its runtime, blocking the
226/// current thread for about 100 ms.
227pub struct TestServer {
228    addr: net::SocketAddr,
229    inner: Rc<TestServerInner>,
230}
231
232#[derive(Debug)]
233struct TestServerInner {
234    id: Uuid,
235    system: System,
236    server: Server,
237    cfg: SharedCfg,
238}
239
240impl TestServer {
241    /// Returns the server address.
242    pub fn addr(&self) -> net::SocketAddr {
243        self.addr
244    }
245
246    #[must_use]
247    /// Sets the address used by [`connect`](Self::connect).
248    pub fn set_addr(mut self, addr: net::SocketAddr) -> Self {
249        self.addr = addr;
250        self
251    }
252
253    /// Returns the client I/O configuration.
254    pub fn config(&self) -> SharedCfg {
255        self.inner.cfg.clone()
256    }
257
258    /// Connects to the server address.
259    pub async fn connect(&self) -> io::Result<Io> {
260        tcp_connect(self.addr, self.inner.cfg.clone()).await
261    }
262
263    /// Requests a graceful server stop.
264    ///
265    /// Returns without waiting for the stop to complete.
266    pub fn stop(&self) {
267        drop(self.inner.server.stop(true));
268    }
269
270    /// Returns a currently unused local address.
271    ///
272    /// The port is released before this returns, so another process may take
273    /// it first.
274    pub fn unused_addr() -> net::SocketAddr {
275        let addr: net::SocketAddr = "127.0.0.1:0".parse().unwrap();
276        let socket = Socket::new(Domain::IPV4, Type::STREAM, None).unwrap();
277        socket.set_reuse_address(true).unwrap();
278        socket.bind(&SockAddr::from(addr)).unwrap();
279        let tcp = net::TcpListener::from(socket);
280        tcp.local_addr().unwrap()
281    }
282
283    /// Returns the running server controller.
284    pub fn server(&self) -> Server {
285        self.inner.server.clone()
286    }
287}
288
289impl Drop for TestServerInner {
290    fn drop(&mut self) {
291        log::debug!("Stopping test server (dropped) {:?}", self.id);
292        drop(self.server.stop(false));
293        thread::sleep(time::Duration::from_millis(75));
294        self.system.stop();
295        thread::sleep(time::Duration::from_millis(25));
296    }
297}
ntex_server/net/test.rs

ntex_server/net/
test.rs
