ntex/web/

test.rs

//! Various helpers for ntex applications to use during testing.
use std::sync::{Arc, Mutex, mpsc};
use std::{convert::Infallible, fmt, future::Future, io, net, net::SocketAddr};
use std::{pin::Pin, rc::Rc, thread, time};

#[cfg(feature = "cookie")]
use coo_kie::Cookie;
use serde::{Serialize, de::DeserializeOwned};
use urly::Url;
use uuid::Uuid;

use crate::client::error::ClientPayloadError;
use crate::client::{Client, ClientConfig, ClientRequest, ClientResponse};
use crate::error::Error;
use crate::http::error::{HttpError, ResponseError};
use crate::http::header::{CONTENT_TYPE, HeaderName, HeaderValue};
use crate::http::test::TestRequest as HttpTestRequest;
use crate::http::{self, HttpService, Method, Payload, Request, Response, StatusCode, Version};
#[cfg(feature = "ws")]
use crate::io::Sealed;
use crate::router::{Path, ResourceDef};
use crate::service::boxed::{self, BoxServiceFactory};
use crate::service::{IntoServiceFactory, Pipeline, fn_service};
use crate::time::{Millis, Seconds};
use crate::util::{Bytes, BytesMut, Stream, stream_recv};
#[cfg(feature = "ws")]
use crate::ws::{WsClient, WsClientConfig, WsConnection, error::WsClientError};
use crate::{Service, ServiceFactory, SharedCfg, io::IoConfig, rt::System, server::Server};

use crate::web::config::WebAppConfig;
use crate::web::httprequest::HttpRequest;
use crate::web::rmap::ResourceMap;
use crate::web::{FromRequest, HttpResponse, Responder, State, WebRequest, WebResponse};

fn set_peer_addr(head: &mut http::Message<http::RequestHead>, addr: Option<SocketAddr>) {
    #[derive(Debug)]
    struct TestIo {
        io: crate::io::Io,
        _peer: crate::io::testing::IoTest,
    }

    impl http::IoAccess for TestIo {
        fn get(&self) -> Option<&crate::io::IoRef> {
            Some(self.io.as_ref())
        }

        fn take(&self) -> Option<(crate::io::IoBoxed, http::h1::Codec)> {
            None
        }
    }

    if let Some(addr) = addr {
        let (client, server) = crate::io::testing::IoTest::create();
        let io = crate::io::Io::new(server.set_peer_addr(addr), SharedCfg::default());
        head.io = http::CurrentIo::new(Rc::new(TestIo { io, _peer: client }));
    }
}

/// Create service that always responds with `HttpResponse::Ok()`
pub fn ok_service<St: State, In>()
-> impl Service<St, WebRequest<In>, Res = WebResponse, Error = std::convert::Infallible> {
    default_service::<St, In>(StatusCode::OK)
}

/// Create service that responds with response with specified status code
pub fn default_service<St: State, In>(
    status_code: StatusCode,
) -> impl Service<St, WebRequest<In>, Res = WebResponse, Error = Infallible> {
    fn_service(async move |req: WebRequest<In>| {
        Ok::<_, Infallible>(req.into_response(HttpResponse::builder(status_code).build()))
    })
}

/// This method accepts application builder instance, and constructs
/// service.
///
/// ```rust
/// use ntex::service::Service;
/// use ntex::http::StatusCode;
/// use ntex::web::{self, test, App, HttpResponse};
///
/// #[ntex::test]
/// async fn test_init_service() {
///     let mut app = test::init_service(
///         App::new()
///             .service(web::resource("/test").to(async || { HttpResponse::Ok() }))
///     ).await;
///
///     // Create request object
///     let req = test::TestRequest::with_uri("/test").to_request();
///
///     // Execute application
///     let resp = app.call(req).await.unwrap();
///     assert_eq!(resp.status(), StatusCode::OK);
/// }
/// ```
pub async fn init_service<R, S, E>(app: R) -> Pipeline<Request, Response, E>
where
    R: IntoServiceFactory<S, (), Request>,
    S: ServiceFactory<(), Request, Res = Response, Error = E> + 'static,
    S::InitError: fmt::Debug,
{
    let srv = app.into_factory().map_init_err(|e| log::error!("{e:?}"));
    srv.pipeline(()).await.unwrap()
}

/// This method accepts application builder instance, and constructs
/// service with specified state.
///
/// ```rust
/// use ntex::service::Service;
/// use ntex::http::StatusCode;
/// use ntex::web::{self, test, App, HttpResponse};
///
/// #[ntex::test]
/// async fn test_init_service() {
///     let mut app = test::init_service_st(
///         (),
///         App::new()
///             .service(web::resource("/test").to(async || { HttpResponse::Ok() }))
///     ).await;
///
///     // Create request object
///     let req = test::TestRequest::with_uri("/test").to_request();
///
///     // Execute application
///     let resp = app.call(req).await.unwrap();
///     assert_eq!(resp.status(), StatusCode::OK);
/// }
/// ```
pub async fn init_service_st<St, R, S, E>(st: St, app: R) -> Pipeline<Request, Response, E>
where
    St: 'static,
    R: IntoServiceFactory<S, St, Request>,
    S: ServiceFactory<St, Request, Res = Response, Error = E> + 'static,
    S::InitError: fmt::Debug,
{
    let srv = app.into_factory().map_init_err(|e| log::error!("{e:?}"));
    srv.pipeline(st).await.unwrap()
}

/// Calls service and waits for response future completion.
///
/// ```rust
/// use ntex::http::StatusCode;
/// use ntex::web::{self, test, App, HttpResponse};
///
/// #[ntex::test]
/// async fn test_response() {
///     let mut app = test::init_service(
///         App::new()
///             .service(web::resource("/test").to(async || {
///                 HttpResponse::Ok()
///             }))
///     ).await;
///
///     // Create request object
///     let req = test::TestRequest::with_uri("/test").to_request();
///
///     // Call application
///     let resp = test::call_service(&mut app, req).await;
///     assert_eq!(resp.status(), StatusCode::OK);
/// }
/// ```
pub async fn call_service<R, E>(app: &Pipeline<R, Response, E>, req: R) -> Response
where
    R: 'static,
    E: fmt::Debug + 'static,
{
    app.call(req).await.unwrap()
}

/// Helper function that returns a response body of a `TestRequest`
///
/// ```rust
/// use ntex::{http::header, util::Bytes};
/// use ntex::web::{self, test, App, HttpResponse};
///
/// #[ntex::test]
/// async fn test_index() {
///     let mut app = test::init_service(
///         App::new().service(
///             web::resource("/index.html")
///                 .route(web::post().to(async || {
///                     HttpResponse::Ok().body("welcome!")
///                 })))
///     ).await;
///
///     let req = test::TestRequest::post()
///         .uri("/index.html")
///         .header(header::CONTENT_TYPE, "application/json")
///         .to_request();
///
///     let result = test::read_response(&mut app, req).await;
///     assert_eq!(result, Bytes::from_static(b"welcome!"));
/// }
/// ```
pub async fn read_response<E>(app: &Pipeline<Request, Response, E>, req: Request) -> Bytes
where
    E: 'static,
{
    let mut resp = app
        .call(req)
        .await
        .unwrap_or_else(|_| panic!("read_response failed at application call"));

    let mut body = resp.take_body();
    let mut bytes = BytesMut::new();
    while let Some(item) = stream_recv(&mut body).await {
        bytes.extend_from_slice(&item.unwrap());
    }
    bytes.freeze()
}

/// Helper function that returns a response body of a `WebResponse`
///
/// ```rust
/// use ntex::{util::Bytes, http::header};
/// use ntex::web::{self, test, App, HttpResponse};
///
/// #[ntex::test]
/// async fn test_index() {
///     let mut app = test::init_service(
///         App::new().service(
///             web::resource("/index.html")
///                 .route(web::post().to(async || {
///                     HttpResponse::Ok().body("welcome!")
///                 })))
///     ).await;
///
///     let req = test::TestRequest::post()
///         .uri("/index.html")
///         .header(header::CONTENT_TYPE, "application/json")
///         .to_request();
///
///     let resp = test::call_service(&mut app, req).await;
///     let result = test::read_body(resp).await;
///     assert_eq!(result, Bytes::from_static(b"welcome!"));
/// }
/// ```
pub async fn read_body(res: impl Into<Response>) -> Bytes {
    let mut body = res.into().take_body();
    let mut bytes = BytesMut::new();
    while let Some(item) = stream_recv(&mut body).await {
        bytes.extend_from_slice(&item.unwrap());
    }
    bytes.freeze()
}

/// Reads response's body and combines it to a Bytes objects
pub async fn load_stream<S, E>(mut stream: S) -> Result<Bytes, E>
where
    S: Stream<Item = Result<Bytes, E>> + Unpin,
{
    let mut data = BytesMut::new();
    while let Some(item) = stream_recv(&mut stream).await {
        data.extend_from_slice(&item?);
    }
    Ok(data.freeze())
}

/// Helper function that returns a deserialized response body of a `TestRequest`
///
/// ```rust
/// use ntex::http::header;
/// use ntex::web::{self, test, App, HttpResponse};
/// use serde::{Serialize, Deserialize};
///
/// #[derive(Serialize, Deserialize)]
/// pub struct Person {
///     id: String,
///     name: String
/// }
///
/// #[ntex::test]
/// async fn test_add_person() {
///     let mut app = test::init_service(
///         App::new().service(
///             web::resource("/people")
///                 .route(web::post().to(async |person: web::types::Json<Person>| {
///                     HttpResponse::Ok()
///                         .json(&person.into_inner())})
///                     ))
///     ).await;
///
///     let payload = r#"{"id":"12345","name":"User name"}"#.as_bytes();
///
///     let req = test::TestRequest::post()
///         .uri("/people")
///         .header(header::CONTENT_TYPE, "application/json")
///         .payload(payload)
///         .to_request();
///
///     let result: Person = test::read_response_json(&mut app, req).await;
/// }
/// ```
pub async fn read_response_json<T, E>(app: &Pipeline<Request, Response, E>, req: Request) -> T
where
    T: DeserializeOwned,
    E: 'static,
{
    let body = read_response(app, req).await;

    serde_json::from_slice(&body)
        .unwrap_or_else(|e| panic!("read_response_json failed during deserialization, {e:?}"))
}

/// Helper method for extractors testing
pub async fn from_request<St, T: FromRequest<St>>(
    st: &St,
    req: &HttpRequest,
    payload: &mut Payload,
) -> Result<T, T::Error> {
    T::from_request(st, req, payload).await
}

/// Helper method for responders testing
pub async fn respond_to<T: Responder>(slf: T, req: &HttpRequest) -> HttpResponse {
    T::respond_to(slf, &(), req).await
}

/// Test `Request` builder.
///
/// For unit testing, ntex provides a request builder type and a simple handler runner. `TestRequest` implements a builder-like pattern.
/// You can generate various types of request via `TestRequest`'s methods:
///  * `TestRequest::to_request` creates `ntex::http::Request` instance.
///  * `TestRequest::to_srv_request` creates `WebRequest` instance, which is used for testing middlewares and chain adapters.
///  * `TestRequest::to_srv_response` creates `WebResponse` instance.
///  * `TestRequest::to_http_request` creates `HttpRequest` instance, which is used for testing handlers.
///
/// ```rust
/// use ntex::http::{header, StatusCode, HttpMessage};
/// use ntex::web::{self, test, HttpRequest, HttpResponse};
///
/// async fn index(req: HttpRequest) -> HttpResponse {
///     if req.headers().contains_key(header::CONTENT_TYPE) {
///         HttpResponse::Ok().into()
///     } else {
///         HttpResponse::BadRequest().into()
///     }
/// }
///
/// #[ntex::test]
/// async fn test_index() {
///     let req = test::TestRequest::with_header("content-type", "text/plain")
///         .to_http_request();
///
///     let resp = index(req).await;
///     assert_eq!(resp.status(), StatusCode::OK);
///
///     let req = test::TestRequest::default().to_http_request();
///     let resp = index(req).await;
///     assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
/// }
/// ```
#[derive(Debug)]
pub struct TestRequest<St = ()> {
    req: HttpTestRequest,
    rmap: ResourceMap,
    path: Path<Url>,
    peer_addr: Option<SocketAddr>,
    state: St,
    config: WebAppConfig,
}

impl Default for TestRequest {
    fn default() -> TestRequest {
        TestRequest {
            req: HttpTestRequest::default(),
            rmap: ResourceMap::new(ResourceDef::new("")),
            path: Path::new(Url::new()),
            peer_addr: None,
            state: (),
            config: WebAppConfig::new(),
        }
    }
}

impl TestRequest {
    #[must_use]
    /// Create `TestRequest` and set request uri.
    pub fn with_uri(path: &str) -> TestRequest {
        TestRequest::default().uri(path)
    }

    #[must_use]
    /// Create `TestRequest` and set header.
    pub fn with_header<K, V>(key: K, value: V) -> TestRequest
    where
        HeaderName: TryFrom<K>,
        HeaderValue: TryFrom<V>,
        <HeaderName as TryFrom<K>>::Error: Into<HttpError>,
    {
        TestRequest::default().header(key, value)
    }

    #[must_use]
    /// Create `TestRequest` and set method to `Method::GET`.
    pub fn get() -> TestRequest {
        TestRequest::default().method(Method::GET)
    }

    #[must_use]
    /// Create `TestRequest` and set method to `Method::POST`.
    pub fn post() -> TestRequest {
        TestRequest::default().method(Method::POST)
    }

    #[must_use]
    /// Create `TestRequest` and set method to `Method::PUT`.
    pub fn put() -> TestRequest {
        TestRequest::default().method(Method::PUT)
    }

    #[must_use]
    /// Create `TestRequest` and set method to `Method::PATCH`.
    pub fn patch() -> TestRequest {
        TestRequest::default().method(Method::PATCH)
    }

    #[must_use]
    /// Create `TestRequest` and set method to `Method::DELETE`.
    pub fn delete() -> TestRequest {
        TestRequest::default().method(Method::DELETE)
    }
}

impl<St> TestRequest<St> {
    #[must_use]
    /// Set HTTP version of this request.
    pub fn version(mut self, ver: Version) -> Self {
        self.req.version(ver);
        self
    }

    #[must_use]
    /// Set HTTP method of this request.
    pub fn method(mut self, meth: Method) -> Self {
        self.req.method(meth);
        self
    }

    #[must_use]
    /// Set HTTP Uri of this request.
    pub fn uri(mut self, path: &str) -> Self {
        self.req.uri(path);
        self
    }

    #[must_use]
    /// Set a header.
    pub fn header<K, V>(mut self, key: K, value: V) -> Self
    where
        HeaderName: TryFrom<K>,
        HeaderValue: TryFrom<V>,
        <HeaderName as TryFrom<K>>::Error: Into<HttpError>,
    {
        self.req.header(key, value);
        self
    }

    #[must_use]
    #[cfg(feature = "cookie")]
    /// Set cookie for this request.
    pub fn cookie<C>(mut self, cookie: C) -> Self
    where
        C: Into<Cookie<'static>>,
    {
        self.req.cookie(cookie);
        self
    }

    #[must_use]
    /// Set request path pattern parameter.
    pub fn param(mut self, name: &'static str, value: &'static str) -> Self {
        self.path.add_static(name, value);
        self
    }

    #[must_use]
    /// Set peer addr.
    ///
    /// The address is returned by `peer_addr()` of the generated request.
    /// Requests with a peer address must be created inside the ntex runtime.
    pub fn peer_addr(mut self, addr: SocketAddr) -> Self {
        self.peer_addr = Some(addr);
        self
    }

    #[must_use]
    /// Set request payload.
    pub fn payload<B: Into<Bytes>>(mut self, data: B) -> Self {
        self.req.set_payload(data);
        self
    }

    #[must_use]
    /// Serialize `data` to a URL encoded form and set it as the request payload.
    ///
    /// The `Content-Type` header is set to `application/x-www-form-urlencoded`.
    pub fn form<T: Serialize>(mut self, data: &T) -> Self {
        let bytes = serde_urlencoded::to_string(data)
            .expect("Failed to serialize test data as a urlencoded form");
        self.req.set_payload(bytes);
        self.req
            .header(CONTENT_TYPE, "application/x-www-form-urlencoded");
        self
    }

    #[must_use]
    /// Serialize `data` to JSON and set it as the request payload.
    ///
    /// The `Content-Type` header is set to `application/json`.
    pub fn json<T: Serialize>(mut self, data: &T) -> Self {
        let bytes = serde_json::to_string(data).expect("Failed to serialize test data to json");
        self.req.set_payload(bytes);
        self.req.header(CONTENT_TYPE, "application/json");
        self
    }

    #[must_use]
    /// Set request state.
    pub fn state<NewSt: 'static>(self, state: NewSt) -> TestRequest<NewSt> {
        TestRequest {
            state,
            req: self.req,
            rmap: self.rmap,
            path: self.path,
            peer_addr: self.peer_addr,
            config: self.config,
        }
    }

    #[must_use]
    /// Set application state.
    ///
    /// This is equivalent of [`WebAppConfig::set_state()`] method for testing
    /// purpose. The value is available via `HttpRequest::app_state()`.
    pub fn app_state<T: Send + Sync + 'static>(mut self, data: T) -> Self {
        self.config = self.config.set_state(data);
        self
    }

    #[must_use]
    #[cfg(test)]
    /// Set request config
    pub(crate) fn rmap(mut self, rmap: ResourceMap) -> Self {
        self.rmap = rmap;
        self
    }

    #[must_use]
    /// Complete request creation and generate `Request` instance.
    pub fn to_request(mut self) -> Request {
        self.req.build()
    }

    #[must_use]
    /// Complete request creation and generate `WebRequest` instance.
    pub fn to_srv_request(mut self) -> WebRequest<St> {
        let (mut head, payload) = self.req.build().into_parts();
        set_peer_addr(&mut head, self.peer_addr);
        *self.path.get_mut() = head.uri.clone();
        let cfg = SharedCfg::new("TEST").add(self.config).build();

        WebRequest::new(
            HttpRequest::new(self.path, head, Rc::new(self.rmap), cfg.get()),
            payload,
            self.state,
        )
    }

    #[must_use]
    /// Complete request creation and generate `WebResponse` instance.
    pub fn to_srv_response(self, res: HttpResponse) -> WebResponse {
        self.to_srv_request().into_response(res)
    }

    #[must_use]
    /// Complete request creation and generate `HttpRequest` instance.
    pub fn to_http_request(mut self) -> HttpRequest {
        let (mut head, _) = self.req.build().into_parts();
        set_peer_addr(&mut head, self.peer_addr);
        *self.path.get_mut() = head.uri.clone();
        let cfg = SharedCfg::new("TEST").add(self.config).build();

        HttpRequest::new(self.path, head, Rc::new(self.rmap), cfg.get())
    }

    #[must_use]
    /// Complete request creation and generate `HttpRequest` and `Payload` instances.
    pub fn to_http_parts(mut self) -> (HttpRequest, Payload, St) {
        let (mut head, payload) = self.req.build().into_parts();
        set_peer_addr(&mut head, self.peer_addr);
        *self.path.get_mut() = head.uri.clone();
        let cfg = SharedCfg::new("TEST").add(self.config).build();

        let req = HttpRequest::new(self.path, head, Rc::new(self.rmap), cfg.get());

        (req, payload, self.state)
    }
}

/// Start test server with default configuration
///
/// Test server is very simple server that simplify process of writing
/// integration tests cases for ntex web applications.
///
/// # Examples
///
/// ```rust
/// use ntex::web::{self, test, App, HttpResponse};
///
/// async fn my_handler() -> Result<HttpResponse, std::io::Error> {
///     Ok(HttpResponse::Ok().into())
/// }
///
/// #[ntex::test]
/// async fn test_example() {
///     let srv = test::server(
///         async |_| App::new().service(
///                 web::resource("/").to(my_handler))
///     );
///
///     let req = srv.get("/");
///     let response = req.send().await.unwrap();
///     assert!(response.status().is_success());
/// }
/// ```
pub fn server<F, I, Sf>(factory: F) -> TestServer
where
    F: AsyncFn(&()) -> I + Send + Clone + 'static,
    I: IntoServiceFactory<Sf, (), Request>,
    Sf: ServiceFactory<(), Request> + 'static,
    Sf::Res: Into<Response>,
    Sf::Error: ResponseError,
    Sf::InitError: fmt::Debug,
{
    server_with(TestServerConfig::default(), factory)
}

/// Start test server with custom configuration
///
/// Test server could be configured in different ways, for details check
/// `TestServerConfig` docs.
///
/// # Examples
///
/// ```rust
/// use ntex::web::{self, test, App, HttpResponse};
///
/// async fn my_handler() -> HttpResponse {
///     HttpResponse::Ok().into()
/// }
///
/// #[ntex::test]
/// async fn test_example() {
///     let srv = test::server_with(test::config().h1(), async |_|
///         App::new().service(web::resource("/").to(my_handler))
///     );
///
///     let req = srv.get("/");
///     let response = req.send().await.unwrap();
///     assert!(response.status().is_success());
/// }
/// ```
pub fn server_with<F, I, Sf>(cfg: TestServerConfig, factory: F) -> TestServer
where
    F: AsyncFn(&()) -> I + Send + Clone + 'static,
    I: IntoServiceFactory<Sf, (), Request>,
    Sf: ServiceFactory<(), Request> + 'static,
    Sf::Res: Into<Response>,
    Sf::Error: ResponseError + 'static,
    Sf::InitError: fmt::Debug,
{
    // Type-erase the application factory, so the HTTP server stacks below
    // are compiled once instead of once per test application.
    let factory = Mutex::new(factory);
    let factory: AppFactory = Arc::new(move || {
        let factory = factory.lock().unwrap().clone();
        Box::pin(async move {
            boxed::factory(
                factory(&())
                    .await
                    .into_factory()
                    .map(Into::into)
                    .map_err(AppError::new)
                    .map_init_err(|e| io::Error::other(format!("{e:?}"))),
            )
        })
    });
    start_server(cfg, factory)
}

type AppFactory = Arc<
    dyn Fn() -> Pin<
            Box<dyn Future<Output = BoxServiceFactory<(), Request, Response, AppError, io::Error>>>,
        > + Send
        + Sync,
>;

/// Type-erased application error
struct AppError(Box<dyn ResponseError>);

impl AppError {
    fn new<E: ResponseError + 'static>(err: E) -> Self {
        AppError(Box::new(err))
    }
}

impl fmt::Debug for AppError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt::Debug::fmt(&self.0, f)
    }
}

impl fmt::Display for AppError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt::Display::fmt(&self.0, f)
    }
}

impl std::error::Error for AppError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        self.0.source()
    }
}

impl ResponseError for AppError {
    fn error_response(&self) -> Response {
        self.0.error_response()
    }
}

fn start_server(cfg: TestServerConfig, factory: AppFactory) -> TestServer {
    let sys = System::current().config();
    let name = System::current().name().to_string();

    let id = Uuid::now_v7();
    let (tx, rx) = mpsc::channel();
    log::debug!("Starting {name:?} web server {id:?}");

    let ssl = match cfg.stream {
        StreamType::Tcp => false,
        #[cfg(feature = "openssl")]
        StreamType::Openssl(_) => true,
        #[cfg(feature = "rustls")]
        StreamType::Rustls(_) => true,
    };

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

        let ctimeout = cfg.client_timeout;
        let port = cfg.port;
        let tcp = cfg
            .listener
            .unwrap_or_else(|| net::TcpListener::bind(format!("127.0.0.1:{port}")).unwrap());
        let local_addr = tcp.local_addr().unwrap();

        sys.run(move || {
            let builder = crate::server::build().workers(1).disable_signals();
            let secure = match cfg.stream {
                StreamType::Tcp => false,
                #[cfg(feature = "openssl")]
                StreamType::Openssl(_) => true,
                #[cfg(feature = "rustls")]
                StreamType::Rustls(_) => true,
            };
            let c = cfg.srv_cfg.clone().unwrap_or_else(|| {
                SharedCfg::new("WEB-SRV")
                    .add(IoConfig::new())
                    .add(http::HttpServiceConfig::new().set_headers_read_rate(
                        ctimeout,
                        Seconds::ZERO,
                        256,
                    ))
                    .add(WebAppConfig::with(
                        &name,
                        secure,
                        local_addr,
                        format!("{local_addr}"),
                    ))
                    .into()
            });

            let srv = match cfg.stream {
                StreamType::Tcp => match cfg.tp {
                    HttpVer::Http1 => builder.listen("test", tcp, c, async move |st| {
                        HttpService::h1(factory(st).await)
                    }),
                    HttpVer::Http2 => builder.listen("test", tcp, c, async move |st| {
                        HttpService::h2(factory(st).await)
                    }),
                    HttpVer::Both => builder.listen("test", tcp, c, async move |st| {
                        HttpService::new(factory(st).await)
                    }),
                },
                #[cfg(feature = "openssl")]
                StreamType::Openssl(acceptor) => match cfg.tp {
                    HttpVer::Http1 => builder.listen("test", tcp, c, async move |st| {
                        http::openssl(acceptor.clone(), HttpService::h1(factory(st).await))
                    }),
                    HttpVer::Http2 => builder.listen("test", tcp, c, async move |st| {
                        http::openssl(acceptor.clone(), HttpService::h2(factory(st).await))
                    }),
                    HttpVer::Both => builder.listen("test", tcp, c, async move |st| {
                        http::openssl(acceptor.clone(), HttpService::new(factory(st).await))
                    }),
                },
                #[cfg(feature = "rustls")]
                StreamType::Rustls(config) => match cfg.tp {
                    HttpVer::Http1 => builder.listen("test", tcp, c, async move |st| {
                        http::rustls(
                            config.clone(),
                            http::ALPN_PROTO_H1,
                            HttpService::h1(factory(st).await),
                        )
                    }),
                    HttpVer::Http2 => builder.listen("test", tcp, c, async move |st| {
                        http::rustls(
                            config.clone(),
                            http::ALPN_PROTO_H2,
                            HttpService::h2(factory(st).await),
                        )
                    }),
                    HttpVer::Both => builder.listen("test", tcp, c, async move |st| {
                        http::rustls(
                            config.clone(),
                            http::ALPN_PROTOS,
                            HttpService::new(factory(st).await),
                        )
                    }),
                },
            }
            .unwrap()
            .run();

            tx.send((System::current(), srv, local_addr)).unwrap();
            Ok(())
        })
    });
    let (system, server, addr) = rx.recv().unwrap();
    thread::sleep(Millis(25).into());

    let cfg = cfg.client_cfg.clone().unwrap_or_else(|| {
        SharedCfg::new("TEST-CLIENT")
            .add(IoConfig::new().set_connect_timeout(Millis(90_000)))
            .add(ntex_tls::TlsConfig::new().set_handshake_timeout(Seconds(5)))
            .add(
                ntex_h2::ServiceConfig::new()
                    .set_max_header_list_size(256 * 1024)
                    .set_max_header_continuation_frames(96),
            )
            .add(
                ClientConfig::new()
                    .set_response_timeout(Seconds(30))
                    .set_response_payload_timeout(Seconds(30)),
            )
            .add(
                WsClientConfig::new()
                    .set_address(addr)
                    .set_handshake_timeout(Seconds(60)),
            )
            .build()
    });

    let client = {
        #[cfg(feature = "openssl")]
        {
            use tls_openssl::ssl::{SslConnector, SslMethod, SslVerifyMode};

            let mut builder = SslConnector::builder(SslMethod::tls()).unwrap();
            builder.set_verify(SslVerifyMode::NONE);
            let _ = builder
                .set_alpn_protos(b"\x02h2\x08http/1.1")
                .map_err(|e| log::error!("Cannot set alpn protocol: {e:?}"));
            Client::builder()
                .openssl(builder.build())
                .build(cfg.clone())
        }
        #[cfg(not(feature = "openssl"))]
        {
            Client::builder().build(cfg.clone())
        }
    };

    TestServer {
        id,
        cfg,
        addr,
        client,
        system,
        ssl,
        server,
    }
}

#[derive(Debug)]
/// Test server configuration
pub struct TestServerConfig {
    tp: HttpVer,
    stream: StreamType,
    client_timeout: Seconds,
    port: u16,
    listener: Option<net::TcpListener>,
    srv_cfg: Option<SharedCfg>,
    client_cfg: Option<SharedCfg>,
}

#[derive(Clone, Debug)]
enum HttpVer {
    Http1,
    Http2,
    Both,
}

#[derive(Clone)]
#[allow(clippy::large_enum_variant)]
enum StreamType {
    Tcp,
    #[cfg(feature = "openssl")]
    Openssl(tls_openssl::ssl::SslAcceptor),
    #[cfg(feature = "rustls")]
    Rustls(tls_rustls::ServerConfig),
}

impl fmt::Debug for StreamType {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            StreamType::Tcp => write!(f, "StreamType::Tcp"),
            #[cfg(feature = "openssl")]
            StreamType::Openssl(_) => write!(f, "StreamType::Openssl"),
            #[cfg(feature = "rustls")]
            StreamType::Rustls(_) => write!(f, "StreamType::Rustls"),
        }
    }
}

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

#[must_use]
/// Create default test server config
pub fn config() -> TestServerConfig {
    TestServerConfig::new()
}

impl TestServerConfig {
    #[must_use]
    /// Create default server configuration
    pub(crate) fn new() -> TestServerConfig {
        TestServerConfig {
            tp: HttpVer::Both,
            stream: StreamType::Tcp,
            client_timeout: Seconds(5),
            port: 0,
            listener: None,
            srv_cfg: None,
            client_cfg: None,
        }
    }

    #[must_use]
    /// Start http/1.1 server only.
    pub fn h1(mut self) -> Self {
        self.tp = HttpVer::Http1;
        self
    }

    #[must_use]
    /// Start http/2 server only.
    pub fn h2(mut self) -> Self {
        self.tp = HttpVer::Http2;
        self
    }

    #[must_use]
    /// Start openssl server.
    #[cfg(feature = "openssl")]
    pub fn openssl(mut self, acceptor: tls_openssl::ssl::SslAcceptor) -> Self {
        self.stream = StreamType::Openssl(acceptor);
        self
    }

    #[must_use]
    /// Start rustls server.
    #[cfg(feature = "rustls")]
    pub fn rustls(mut self, config: tls_rustls::ServerConfig) -> Self {
        self.stream = StreamType::Rustls(config);
        self
    }

    #[must_use]
    /// Set server client timeout in seconds for first request.
    pub fn client_timeout(mut self, val: Seconds) -> Self {
        self.client_timeout = val;
        self
    }

    #[must_use]
    /// Set server port. By default, test server binds to a random port assigned by OS.
    pub fn port(mut self, port: u16) -> Self {
        self.port = port;
        self
    }

    #[must_use]
    /// Use an already bound TCP listener for the test server.
    ///
    /// When set, the configured port is ignored.
    pub fn listener(mut self, listener: net::TcpListener) -> Self {
        self.listener = Some(listener);
        self
    }

    #[must_use]
    /// Use custom `SharedCfg` for test server.
    pub fn server_cfg(mut self, cfg: impl Into<SharedCfg>) -> Self {
        self.srv_cfg = Some(cfg.into());
        self
    }

    #[must_use]
    /// Use custom `SharedCfg` for client.
    pub fn client_cfg(mut self, cfg: impl Into<SharedCfg>) -> Self {
        self.client_cfg = Some(cfg.into());
        self
    }
}

#[derive(Debug)]
/// Test server controller
pub struct TestServer {
    id: Uuid,
    cfg: SharedCfg,
    addr: net::SocketAddr,
    client: Client,
    system: crate::rt::System,
    ssl: bool,
    server: Server,
}

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

    /// Construct test server url
    pub fn url(&self, uri: &str) -> String {
        let scheme = if self.ssl { "https" } else { "http" };

        if uri.starts_with('/') {
            format!("{}://localhost:{}{}", scheme, self.addr.port(), uri)
        } else {
            format!("{}://localhost:{}/{}", scheme, self.addr.port(), uri)
        }
    }

    /// Create `GET` request
    pub fn get<S: AsRef<str>>(&self, path: S) -> ClientRequest {
        self.client.get(self.url(path.as_ref()).as_str())
    }

    /// Create `POST` request
    pub fn post<S: AsRef<str>>(&self, path: S) -> ClientRequest {
        self.client.post(self.url(path.as_ref()).as_str())
    }

    /// Create `HEAD` request
    pub fn head<S: AsRef<str>>(&self, path: S) -> ClientRequest {
        self.client.head(self.url(path.as_ref()).as_str())
    }

    /// Create `PUT` request
    pub fn put<S: AsRef<str>>(&self, path: S) -> ClientRequest {
        self.client.put(self.url(path.as_ref()).as_str())
    }

    /// Create `PATCH` request
    pub fn patch<S: AsRef<str>>(&self, path: S) -> ClientRequest {
        self.client.patch(self.url(path.as_ref()).as_str())
    }

    /// Create `DELETE` request
    pub fn delete<S: AsRef<str>>(&self, path: S) -> ClientRequest {
        self.client.delete(self.url(path.as_ref()).as_str())
    }

    /// Create `OPTIONS` request
    pub fn options<S: AsRef<str>>(&self, path: S) -> ClientRequest {
        self.client.options(self.url(path.as_ref()).as_str())
    }

    /// Create `QUERY` request
    pub fn query<S: AsRef<str>>(&self, path: S) -> ClientRequest {
        self.client.query(self.url(path.as_ref()).as_str())
    }

    /// Create client request with the specified method
    ///
    /// A relative `path` is resolved against the test server url, an absolute
    /// url (containing `://`) is used as is.
    pub fn request<S: AsRef<str>>(&self, method: Method, path: S) -> ClientRequest {
        let path = path.as_ref();
        if path.contains("://") {
            self.client.request(method, path)
        } else {
            self.client.request(method, self.url(path).as_str())
        }
    }

    /// Load response's body
    pub async fn load_body(
        &self,
        response: ClientResponse,
    ) -> Result<Bytes, Error<ClientPayloadError>> {
        response.body().limit(10_485_760).await
    }

    #[cfg(feature = "ws")]
    /// Connect to websocket server at a given path
    pub async fn ws_at(&self, path: &str) -> Result<WsConnection<Sealed>, Error<WsClientError>> {
        if self.ssl {
            #[cfg(feature = "openssl")]
            {
                use tls_openssl::ssl::{SslConnector, SslMethod, SslVerifyMode};

                let mut builder = SslConnector::builder(SslMethod::tls()).unwrap();
                builder.set_verify(SslVerifyMode::NONE);
                let _ = builder
                    .set_alpn_protos(b"\x08http/1.1")
                    .map_err(|e| log::error!("Cannot set alpn protocol: {e:?}"));

                WsClient::new(self.url(path), &self.cfg)
                    .openssl(builder.build())
                    .connect()
                    .await
                    .map(WsConnection::seal)
            }
            #[cfg(not(feature = "openssl"))]
            {
                panic!("openssl feature is required")
            }
        } else {
            WsClient::new(self.url(path), &self.cfg)
                .connect()
                .await
                .map(WsConnection::seal)
        }
    }

    #[cfg(feature = "ws")]
    /// Connect to a websocket server
    pub async fn ws(&self) -> Result<WsConnection<Sealed>, Error<WsClientError>> {
        self.ws_at("/").await
    }

    /// Gracefully stop http server
    pub async fn stop(&self) {
        self.server.stop(true).await;
    }
}

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

#[cfg(test)]
mod tests {
    use serde::{Deserialize, Serialize};

    use super::*;
    use crate::http::{HttpMessage, header};
    use crate::web::{self, App};

    #[crate::rt_test]
    async fn test_basics() {
        let req = TestRequest::with_header(header::CONTENT_TYPE, "application/json")
            .version(Version::HTTP_2)
            .header(header::DATE, "some date")
            .param("test", "123")
            .app_state(20u64)
            .peer_addr("127.0.0.1:8081".parse().unwrap())
            .to_http_request();
        assert!(req.headers().contains_key(header::CONTENT_TYPE));
        assert!(req.headers().contains_key(header::DATE));
        assert_eq!(req.peer_addr(), Some("127.0.0.1:8081".parse().unwrap()));
        assert_eq!(&req.match_info()["test"], "123");
        assert_eq!(req.version(), Version::HTTP_2);
        let data = req.app_state::<u64>().unwrap();
        assert_eq!(*data, 20);
        assert_eq!(format!("{:?}", StreamType::Tcp), "StreamType::Tcp");

        let (_, pl, ()) = TestRequest::with_header(header::CONTENT_TYPE, "application/json")
            .to_srv_request()
            .into_parts();
        let res = load_stream(pl).await.unwrap();
        assert_eq!(res, &b""[..]);
    }

    #[crate::rt_test]
    async fn test_request_methods() {
        let app = init_service(
            App::new().service(
                web::resource("/index.html")
                    .route(web::put().to(async || HttpResponse::Ok().body("put!")))
                    .route(web::patch().to(async || HttpResponse::Ok().body("patch!")))
                    .route(web::delete().to(async || HttpResponse::Ok().body("delete!"))),
            ),
        )
        .await;

        let put_req = TestRequest::put()
            .uri("/index.html")
            .header(header::CONTENT_TYPE, "application/json")
            .to_request();

        let result = read_response(&app, put_req).await;
        assert_eq!(result, Bytes::from_static(b"put!"));

        let patch_req = TestRequest::patch()
            .uri("/index.html")
            .header(header::CONTENT_TYPE, "application/json")
            .to_request();

        let result = read_response(&app, patch_req).await;
        assert_eq!(result, Bytes::from_static(b"patch!"));

        let delete_req = TestRequest::delete().uri("/index.html").to_request();
        let result = read_response(&app, delete_req).await;
        assert_eq!(result, Bytes::from_static(b"delete!"));
    }

    #[crate::rt_test]
    async fn test_response() {
        let app = init_service(
            App::new().service(
                web::resource("/index.html")
                    .route(web::post().to(async || HttpResponse::Ok().body("welcome!"))),
            ),
        )
        .await;

        let req = TestRequest::post()
            .uri("/index.html")
            .header(header::CONTENT_TYPE, "application/json")
            .to_request();

        let result = read_response(&app, req).await;
        assert_eq!(result, Bytes::from_static(b"welcome!"));
    }

    #[derive(Serialize, Deserialize)]
    struct Person {
        id: String,
        name: String,
    }

    #[crate::rt_test]
    async fn test_response_json() {
        let app = init_service(
            App::new().service(web::resource("/people").route(web::post().to(
                async |person: web::types::Json<Person>| {
                    HttpResponse::Ok().json(&person.into_inner())
                },
            ))),
        )
        .await;

        let payload = r#"{"id":"12345","name":"User name"}"#.as_bytes();

        let req = TestRequest::post()
            .uri("/people")
            .header(header::CONTENT_TYPE, "application/json")
            .payload(payload)
            .to_request();

        let result: Person = read_response_json(&app, req).await;
        assert_eq!(&result.id, "12345");
    }

    #[crate::rt_test]
    async fn test_request_response_form() {
        let app = init_service(
            App::new().service(web::resource("/people").route(web::post().to(
                async |person: web::types::Form<Person>| {
                    HttpResponse::Ok().json(&person.into_inner())
                },
            ))),
        )
        .await;

        let payload = Person {
            id: "12345".to_string(),
            name: "User name".to_string(),
        };

        let req = TestRequest::post()
            .uri("/people")
            .form(&payload)
            .to_request();

        assert_eq!(req.content_type(), "application/x-www-form-urlencoded");

        let result: Person = read_response_json(&app, req).await;
        assert_eq!(&result.id, "12345");
        assert_eq!(&result.name, "User name");
    }

    #[crate::rt_test]
    async fn test_request_response_json() {
        let app = init_service(
            App::new().service(web::resource("/people").route(web::post().to(
                async |person: web::types::Json<Person>| {
                    HttpResponse::Ok().json(&person.into_inner())
                },
            ))),
        )
        .await;

        let payload = Person {
            id: "12345".to_string(),
            name: "User name".to_string(),
        };

        let req = TestRequest::post()
            .uri("/people")
            .json(&payload)
            .to_request();

        assert_eq!(req.content_type(), "application/json");

        let result: Person = read_response_json(&app, req).await;
        assert_eq!(&result.id, "12345");
        assert_eq!(&result.name, "User name");
    }

    #[crate::rt_test]
    async fn test_async_with_block() {
        async fn async_with_block() -> Result<HttpResponse, Infallible> {
            let res = web::block(move || Some(4usize).ok_or("wrong")).await;

            #[allow(clippy::match_wild_err_arm)]
            match res {
                Ok(value) => Ok(HttpResponse::Ok()
                    .content_type("text/plain")
                    .body(format!("Async with block value: {value}"))),
                Err(_) => panic!("Unexpected"),
            }
        }

        let app =
            init_service(App::new().service(web::resource("/index.html").to(async_with_block)))
                .await;

        let req = TestRequest::post().uri("/index.html").to_request();
        let res = app.call(req).await.unwrap();
        assert!(res.status().is_success());
    }

    #[crate::rt_test]
    async fn test_test_methods() {
        let srv = server(async |()| {
            App::new().service(
                web::resource("/").route((
                    web::route()
                        .method(Method::PUT)
                        .to(async || HttpResponse::Ok()),
                    web::route()
                        .method(Method::PATCH)
                        .to(async || HttpResponse::Ok()),
                    web::route()
                        .method(Method::DELETE)
                        .to(async || HttpResponse::Ok()),
                    web::route()
                        .method(Method::OPTIONS)
                        .to(async || HttpResponse::Ok()),
                )),
            )
        });

        assert_eq!(srv.put("/").send().await.unwrap().status(), StatusCode::OK);
        assert_eq!(
            srv.patch("/").send().await.unwrap().status(),
            StatusCode::OK
        );
        assert_eq!(
            srv.delete("/").send().await.unwrap().status(),
            StatusCode::OK
        );
        assert_eq!(
            srv.options("/").send().await.unwrap().status(),
            StatusCode::OK
        );

        let res = srv.put("").send().await.unwrap();
        assert_eq!(srv.load_body(res).await.unwrap(), Bytes::new());
    }

    #[crate::rt_test]
    async fn test_request_state() {
        let (_, _, st) = TestRequest::default().state(10usize).to_http_parts();
        assert_eq!(st, 10);

        let req = TestRequest::default()
            .state(web::AppState::new(5usize))
            .to_srv_request();
        assert_eq!(**req.st(), 5);
    }

    #[crate::rt_test]
    async fn test_server_config() {
        let cfg = config()
            .h1()
            .port(0)
            .client_cfg(SharedCfg::new("CUSTOM-CLIENT"));
        let dbg = format!("{cfg:?}");
        assert!(dbg.contains("StreamType::Tcp"), "{dbg}");

        let srv = server_with(cfg, async |()| {
            App::new().service(web::resource("/").to(async || HttpResponse::Ok()))
        });
        let res = srv.get("/").send().await.unwrap();
        assert_eq!(res.status(), StatusCode::OK);
        assert_eq!(res.version(), http::Version::HTTP_11);
        srv.stop().await;
    }

    #[crate::rt_test]
    async fn test_server_listener_ignores_port() {
        // regression: the configured port was bound even if a listener is set
        let lst = net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = lst.local_addr().unwrap();

        let srv = server_with(config().port(addr.port()).listener(lst), async |()| {
            App::new().service(web::resource("/").to(async || HttpResponse::Ok()))
        });
        assert_eq!(srv.addr(), addr);
        let res = srv.get("/").send().await.unwrap();
        assert_eq!(res.status(), StatusCode::OK);
    }

    #[cfg(feature = "openssl")]
    fn ssl_acceptor(h2: bool) -> tls_openssl::ssl::SslAcceptor {
        use tls_openssl::ssl::{SslAcceptor, SslFiletype, SslMethod};

        let mut builder = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap();
        builder
            .set_private_key_file("./tests/key.pem", SslFiletype::PEM)
            .unwrap();
        builder
            .set_certificate_chain_file("./tests/cert.pem")
            .unwrap();
        builder.set_alpn_select_callback(move |_, protos| {
            if h2 && protos.windows(3).any(|w| w == b"\x02h2") {
                Ok(b"h2")
            } else {
                Ok(b"http/1.1")
            }
        });
        builder.build()
    }

    #[cfg(feature = "openssl")]
    #[crate::rt_test]
    async fn test_server_openssl() {
        for (cfg, h2, version) in [
            (config().h1(), false, http::Version::HTTP_11),
            (config().h2(), true, http::Version::HTTP_2),
            (config(), true, http::Version::HTTP_2),
            (config(), false, http::Version::HTTP_11),
        ] {
            let cfg = cfg.openssl(ssl_acceptor(h2));
            assert!(format!("{cfg:?}").contains("StreamType::Openssl"));

            let srv = server_with(cfg, async |()| {
                App::new().service(web::resource("/").to(async || HttpResponse::Ok()))
            });
            assert!(srv.url("/").starts_with("https://"));
            let res = srv.get("/").send().await.unwrap();
            assert_eq!(res.status(), StatusCode::OK);
            assert_eq!(res.version(), version);
        }
    }

    #[cfg(all(feature = "openssl", feature = "ws"))]
    #[crate::rt_test]
    async fn test_server_openssl_ws() {
        let srv = server_with(config().h1().openssl(ssl_acceptor(false)), async |()| {
            App::new().service(
                web::resource("/").route(web::to(async |req: web::HttpRequest| {
                    let _ = web::ws::start(&req, None, async |_: web::ws::Frame| {
                        Ok::<_, io::Error>(None)
                    })
                    .await;
                })),
            )
        });
        assert!(srv.ws().await.is_ok());
    }

    #[cfg(feature = "rustls")]
    #[test]
    fn test_server_config_rustls_debug() {
        use std::{fs::File, io::BufReader};

        let cert_file = &mut BufReader::new(File::open("tests/cert.pem").unwrap());
        let key_file = &mut BufReader::new(File::open("tests/key.pem").unwrap());
        let cert_chain = rustls_pemfile::certs(cert_file)
            .map(|r| r.unwrap())
            .collect();
        let key = rustls_pemfile::private_key(key_file).unwrap().unwrap();
        let cfg = tls_rustls::ServerConfig::builder()
            .with_no_client_auth()
            .with_single_cert(cert_chain, key)
            .unwrap();
        assert!(format!("{:?}", config().rustls(cfg)).contains("StreamType::Rustls"));
    }

    #[cfg(feature = "cookie")]
    #[test]
    fn test_response_cookies() {
        let req = TestRequest::default()
            .cookie(
                coo_kie::Cookie::build(("name", "value"))
                    .domain("www.rust-lang.org")
                    .path("/test")
                    .http_only(true)
                    .max_age(::time::Duration::days(1)),
            )
            .to_http_request();

        let cookies = req.cookies().unwrap();
        assert_eq!(cookies.len(), 1);
        assert_eq!(cookies[0].name(), "name");
    }
}