fuchsia_hyper_test_support/
lib.rs

1// Copyright 2020 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5#![deny(missing_docs)]
6
7//! This module provides a test HTTP(S) server that can be instantiated simply by a unit test, for
8//! connecting components to where you need to vary the response(s) from the HTTP(S) server during
9//! the operation of the test.
10//!
11//! It handles the TCP setup, letting the user specify `Handler` implementations which return the
12//! responses from the server.  `Handler` implementations are meant to be composable to provide
13//! for fault injection and varying behavior in tests.
14
15// This is gratuitously borrowed from src/sys/pkg/lib/fuchsia-pkg-testing/src/serve.rs, and then
16// made generic across all requests by removing the repo-serving aspects of it.
17
18use anyhow::Error;
19use chrono::Utc;
20use fuchsia_async::{self as fasync, Task};
21use futures::future::BoxFuture;
22use futures::prelude::*;
23use hyper::service::service_fn;
24use hyper::{Request, Response, StatusCode};
25#[cfg(not(target_os = "fuchsia"))]
26use netext::TokioAsyncReadExt;
27use std::convert::Infallible;
28use std::net::{Ipv6Addr, SocketAddr};
29use std::pin::Pin;
30use std::sync::Arc;
31
32/// Body type implementation for hyper 1.0 test server support.
33pub mod body;
34pub use crate::body::Body;
35
36// Some provided Handler implementations.
37pub mod handler;
38
39// Some provided Handler implementations for injecting faults into the server's behavior.
40pub mod fault_injection;
41
42/// A "test" HTTP(S) server which is composed of `Handler` implementations, and holding the
43/// connection state.
44pub struct TestServer {
45    stop: futures::channel::oneshot::Sender<()>,
46    addr: SocketAddr,
47    use_https: bool,
48    task: Task<()>,
49}
50
51/// Base trait that all Handlers implement.
52pub trait Handler: 'static + Send + Sync {
53    /// A Handler impl signals that it wishes to handle a request by returning a response for it,
54    /// otherwise it returns None.
55    fn handles(
56        &self,
57        request: &Request<hyper::body::Incoming>,
58    ) -> Option<BoxFuture<'_, Response<Body>>>;
59}
60
61impl Handler for Arc<dyn Handler> {
62    fn handles(
63        &self,
64        request: &Request<hyper::body::Incoming>,
65    ) -> Option<BoxFuture<'_, Response<Body>>> {
66        (**self).handles(request)
67    }
68}
69
70impl TestServer {
71    /// return the scheme of the TestServer
72    fn scheme(&self) -> &'static str {
73        if self.use_https { "https" } else { "http" }
74    }
75
76    /// Returns the URL that can be used to connect to this repository from this device.
77    pub fn local_url(&self) -> String {
78        format!("{}://localhost:{}", self.scheme(), self.addr.port())
79    }
80
81    /// Returns the URL for the given path that can be used to connect to this repository from this
82    /// device.
83    pub fn local_url_for_path(&self, path: &str) -> String {
84        let path = path.trim_start_matches('/');
85        format!("{}://localhost:{}/{}", self.scheme(), self.addr.port(), path)
86    }
87
88    /// Gracefully signal the server to stop and returns a future that resolves when it terminates.
89    pub fn stop(self) -> impl Future<Output = ()> {
90        self.stop.send(()).expect("remote end to still be open");
91        self.task
92    }
93
94    /// Internal helper which iterates over all Handlers until it finds one that will respond to the
95    /// request.  It then returns that response.  If not response is found, it returns 404 NOT_FOUND.
96    async fn handle_request(
97        handlers: Arc<Vec<Arc<dyn Handler>>>,
98        req: Request<hyper::body::Incoming>,
99    ) -> Response<Body> {
100        let response = handlers.iter().find_map(|h| h.handles(&req));
101
102        match response {
103            Some(response) => response.await,
104            None => Response::builder().status(StatusCode::NOT_FOUND).body(Body::empty()).unwrap(),
105        }
106    }
107
108    /// Create a Builder
109    pub fn builder() -> TestServerBuilder {
110        TestServerBuilder::new()
111    }
112}
113
114/// A builder to construct a `TestServer`.
115#[derive(Default)]
116pub struct TestServerBuilder {
117    handlers: Vec<Arc<dyn Handler>>,
118    https_certs: Option<(
119        Vec<rustls::pki_types::CertificateDer<'static>>,
120        rustls::pki_types::PrivateKeyDer<'static>,
121    )>,
122}
123
124impl TestServerBuilder {
125    /// Create a new TestServerBuilder
126    pub fn new() -> Self {
127        Self::default()
128    }
129
130    /// Serve over TLS, using a server certificate rooted the provided certs
131    pub fn use_https(mut self, cert_chain: &[u8], private_key: &[u8]) -> Self {
132        let cert_chain = parse_cert_chain(cert_chain);
133        let private_key = parse_private_key(private_key);
134        self.https_certs = Some((cert_chain, private_key));
135        self
136    }
137
138    /// Add a Handler which implements the server's behavior.  These are given the ability to
139    /// handle a request in the order in which they are added to the `TestServerBuilder`.
140    pub fn handler(mut self, handler: impl Handler + 'static) -> Self {
141        self.handlers.push(Arc::new(handler));
142        self
143    }
144
145    /// Spawn the server on the current executor, returning a handle to manage the server.
146    pub async fn start(self) -> TestServer {
147        let (mut listener, addr) = {
148            let addr = SocketAddr::new(Ipv6Addr::UNSPECIFIED.into(), 0);
149            let listener = bind_listener(&addr).await;
150            let local_addr = listener.local_addr().unwrap();
151            (listener, local_addr)
152        };
153
154        let (stop, rx_stop) = futures::channel::oneshot::channel();
155
156        let (tls_acceptor, use_https) = if let Some((cert_chain, private_key)) = self.https_certs {
157            // build a server configuration using a test CA and cert chain
158            let tls_config = rustls::ServerConfig::builder()
159                .with_no_client_auth()
160                .with_single_cert(cert_chain, private_key)
161                .unwrap();
162            let tls_acceptor = tokio_rustls::TlsAcceptor::from(Arc::new(tls_config));
163
164            (Some(tls_acceptor), true)
165        } else {
166            (None, false)
167        };
168
169        let task = fasync::Task::spawn(async move {
170            let listener = accept_stream(&mut listener);
171            #[cfg(target_os = "fuchsia")]
172            let listener = listener
173                .map_err(Error::from)
174                .map_ok(|conn| fuchsia_hyper::TcpStream { stream: conn });
175            #[cfg(not(target_os = "fuchsia"))]
176            let listener = listener.map_err(Error::from).map_ok(|conn| fuchsia_hyper::TcpStream {
177                stream: conn.into_multithreaded_futures_stream(),
178            });
179
180            let connections = if let Some(tls_acceptor) = tls_acceptor {
181                // wrap incoming tcp streams
182                listener
183                    .and_then(move |conn| {
184                        tls_acceptor.accept(conn).map(|res| match res {
185                            Ok(conn) => {
186                                Ok(Pin::new(Box::new(conn)) as Pin<Box<dyn AsyncReadWrite>>)
187                            }
188                            Err(e) => Err(Error::from(e)),
189                        })
190                    })
191                    .boxed() // connections
192            } else {
193                listener
194                    .map_ok(|conn| Pin::new(Box::new(conn)) as Pin<Box<dyn AsyncReadWrite>>)
195                    .boxed() // connections
196            };
197
198            // This is the root Arc<Vec<Arc<dyn Handler>>>.
199            let handlers = Arc::new(self.handlers);
200            let builder = hyper_util::server::conn::auto::Builder::new(fuchsia_hyper::Executor);
201            let mut rx_stop = rx_stop.fuse();
202            let mut connections = connections.fuse();
203            let mut tasks = futures::stream::FuturesUnordered::new();
204            loop {
205                futures::select! {
206                    conn_res = connections.next() => {
207                        match conn_res {
208                            Some(Ok(conn)) => {
209                                let handlers = Arc::clone(&handlers);
210                                let builder = builder.clone();
211                                tasks.push(fasync::Task::spawn(async move {
212                                    let _ = builder.serve_connection(
213                                        hyper_util::rt::TokioIo::new(conn),
214                                        service_fn(move |req| {
215                                            let method = req.method().to_owned();
216                                            let path = req.uri().path().to_owned();
217                                            TestServer::handle_request(Arc::clone(&handlers), req)
218                                                .inspect(move |x| {
219                                                    println!(
220                                                        "{} [test http] {} {} => {}",
221                                                        Utc::now().format("%T.%6f"),
222                                                        method,
223                                                        path,
224                                                        x.status()
225                                                    )
226                                                })
227                                                .map(Ok::<_, Infallible>)
228                                        }),
229                                    ).await;
230                                }));
231                            }
232                            _ => break,
233                        }
234                    }
235                    _ = tasks.next() => {}
236                    _ = rx_stop => break,
237                }
238            }
239        });
240
241        TestServer { stop, addr, use_https, task }
242    }
243}
244
245#[cfg(target_os = "fuchsia")]
246async fn bind_listener(addr: &SocketAddr) -> fuchsia_async::net::TcpListener {
247    fuchsia_async::net::TcpListener::bind(addr).unwrap()
248}
249
250#[cfg(not(target_os = "fuchsia"))]
251async fn bind_listener(&addr: &SocketAddr) -> tokio::net::TcpListener {
252    tokio::net::TcpListener::bind(addr).await.unwrap()
253}
254
255#[cfg(target_os = "fuchsia")]
256fn accept_stream<'a>(
257    listener: &'a mut fuchsia_async::net::TcpListener,
258) -> impl Stream<Item = std::io::Result<fuchsia_async::net::TcpStream>> + 'a {
259    use std::task::{Context, Poll};
260
261    #[pin_project::pin_project]
262    struct AcceptStream<'a> {
263        #[pin]
264        listener: &'a mut fuchsia_async::net::TcpListener,
265    }
266
267    impl<'a> Stream for AcceptStream<'a> {
268        type Item = std::io::Result<fuchsia_async::net::TcpStream>;
269
270        fn poll_next(
271            self: Pin<&mut Self>,
272            cx: &mut Context<'_>,
273        ) -> Poll<Option<<Self as Stream>::Item>> {
274            let mut this = self.project();
275            match this.listener.async_accept(cx) {
276                Poll::Ready(Ok((conn, _addr))) => Poll::Ready(Some(Ok(conn))),
277                Poll::Ready(Err(err)) => Poll::Ready(Some(Err(err))),
278                Poll::Pending => Poll::Pending,
279            }
280        }
281    }
282
283    AcceptStream { listener }
284}
285
286#[cfg(not(target_os = "fuchsia"))]
287fn accept_stream<'a>(
288    listener: &'a mut tokio::net::TcpListener,
289) -> impl Stream<Item = std::io::Result<tokio::net::TcpStream>> + 'a {
290    netext::TcpListenerRefStream(listener)
291}
292
293fn parse_cert_chain(mut bytes: &[u8]) -> Vec<rustls::pki_types::CertificateDer<'static>> {
294    rustls_pemfile::certs(&mut bytes).collect::<Result<Vec<_>, _>>().expect("certs to parse")
295}
296
297fn parse_private_key(mut bytes: &[u8]) -> rustls::pki_types::PrivateKeyDer<'static> {
298    rustls_pemfile::private_key(&mut bytes)
299        .expect("private keys to parse")
300        .expect("one private key")
301}
302
303trait AsyncReadWrite: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send {}
304impl<T> AsyncReadWrite for T where T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send {}
305
306// These are a set of useful functions when writing tests.
307
308/// Create a GET request for a given url, which can be used with any hyper client.
309pub fn make_get(
310    url: impl AsRef<str>,
311) -> Result<Request<http_body_util::Full<hyper::body::Bytes>>, Error> {
312    Request::get(url.as_ref()).body(http_body_util::Full::default()).map_err(Error::from)
313}
314
315/// Perform an HTTP GET for the given url, returning the result.
316pub async fn get(url: impl AsRef<str>) -> Result<Response<hyper::body::Incoming>, Error> {
317    let request = make_get(url)?;
318    let client = fuchsia_hyper::new_client();
319    let response = client.request(request).await?;
320    Ok(response)
321}
322
323/// Collect a Response into a single Vec of bytes.
324pub async fn body_as_bytes<B>(response: Response<B>) -> Result<Vec<u8>, Error>
325where
326    B: hyper::body::Body,
327    B::Error: Into<Error>,
328{
329    use http_body_util::BodyExt as _;
330    let bytes = response.into_body().collect().await.map_err(Into::into)?.to_bytes();
331    Ok(bytes.to_vec())
332}
333
334/// Collect a Response's Body and convert the body to a string.
335pub async fn body_as_string<B>(response: Response<B>) -> Result<String, Error>
336where
337    B: hyper::body::Body,
338    B::Error: Into<Error>,
339{
340    let bytes = body_as_bytes(response).await?;
341    let string = String::from_utf8(bytes)?;
342    Ok(string)
343}
344
345/// Get a url and return the body of the response as a string.
346pub async fn get_body_as_string(url: impl AsRef<str>) -> Result<String, Error> {
347    let response = get(url).await?;
348    body_as_string(response).await
349}
350
351#[cfg(test)]
352mod tests {
353    use super::*;
354    use crate::fault_injection::*;
355    use crate::handler::*;
356    use anyhow::anyhow;
357    use fasync::TimeoutExt;
358
359    #[fuchsia_async::run_singlethreaded(test)]
360    async fn test_start_stop() {
361        let server = TestServer::builder().start().await;
362        server.stop().await;
363    }
364
365    #[fuchsia_async::run_singlethreaded(test)]
366    async fn test_empty_server_404s() {
367        let server = TestServer::builder().start().await;
368        let result = get(server.local_url()).await;
369        assert_eq!(result.unwrap().status(), StatusCode::NOT_FOUND);
370    }
371
372    #[fuchsia_async::run_singlethreaded(test)]
373    async fn test_shared_handler() {
374        let shared: Arc<dyn Handler> = Arc::new(StaticResponse::ok_body("shared"));
375
376        let server = TestServer::builder()
377            .handler(ForPath::new("/a", Arc::clone(&shared)))
378            .handler(shared)
379            .start()
380            .await;
381
382        assert_eq!(get_body_as_string(server.local_url_for_path("/a")).await.unwrap(), "shared");
383        assert_eq!(get_body_as_string(server.local_url_for_path("/foo")).await.unwrap(), "shared");
384    }
385
386    #[fuchsia_async::run_singlethreaded(test)]
387    async fn test_simple_responder() {
388        let server =
389            TestServer::builder().handler(StaticResponse::ok_body("some data")).start().await;
390        assert_eq!(
391            get_body_as_string(server.local_url_for_path("ignored")).await.unwrap(),
392            "some data"
393        );
394    }
395
396    #[fuchsia_async::run_singlethreaded(test)]
397    async fn test_simple_path() {
398        let server = TestServer::builder()
399            .handler(ForPath::new("/some/path", StaticResponse::ok_body("some data")))
400            .start()
401            .await;
402        assert_eq!(
403            get_body_as_string(server.local_url_for_path("/some/path")).await.unwrap(),
404            "some data"
405        );
406    }
407
408    #[fuchsia_async::run_singlethreaded(test)]
409    async fn test_simple_path_doesnt_respond_to_wrong_path() {
410        let server = TestServer::builder()
411            .handler(ForPath::new("/some/path", StaticResponse::ok_body("some data")))
412            .start()
413            .await;
414        // make sure a non-matching path fails
415        let result = get(server.local_url_for_path("/other/path")).await;
416        assert_eq!(result.unwrap().status(), StatusCode::NOT_FOUND);
417    }
418
419    #[fuchsia_async::run_singlethreaded(test)]
420    async fn test_hang() {
421        let server = TestServer::builder().handler(Hang).start().await;
422        let result = get(server.local_url_for_path("ignored"))
423            .on_timeout(std::time::Duration::from_secs(1), || Err(anyhow!("timed out")))
424            .await;
425        assert_eq!(result.unwrap_err().to_string(), Error::msg("timed out").to_string());
426    }
427
428    #[fuchsia_async::run_singlethreaded(test)]
429    async fn test_hang_body() {
430        let server = TestServer::builder().handler(HangBody::content_length(500)).start().await;
431        let result = get_body_as_string(server.local_url_for_path("ignored"))
432            .on_timeout(std::time::Duration::from_secs(1), || Err(anyhow!("timed out")))
433            .await;
434        assert_eq!(result.unwrap_err().to_string(), Error::msg("timed out").to_string());
435    }
436}
fuchsia_hyper_test_support/lib.rs

fuchsia_hyper_test_support/
lib.rs
