First in-progress version of httptest

.gitignore

@@ -0,0 +1,3 @@
+/target
+**/*.rs.bk
+Cargo.lock

Cargo.toml

@@ -0,0 +1,23 @@
+[package]
+name = "httptest"
+version = "0.1.0"
+authors = ["Glenn Griffin <ggriffiniii@gmail.com>"]
+edition = "2018"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+hyper = {version = "=0.13.0-alpha.4", features = ["unstable-stream"]}
+futures-preview = {version = "=0.3.0-alpha.19", features = ["std", "async-await"]}
+tokio = "=0.2.0-alpha.6"
+crossbeam-channel = "0.4.0"
+http = "0.1.18"
+log = "0.4.8"
+bstr = "0.2.8"
+regex = "1.3.1"
+url = "2.1.0"
+serde_json = "1.0.44"
+serde = "1.0.103"
+
+[dev-dependencies]
+pretty_env_logger = "0.3.1"

src/lib.rs

@@ -0,0 +1,29 @@
+#[macro_export]
+macro_rules! all_of {
+    ($($x:expr),*) => ($crate::mappers::all_of($crate::vec_of_boxes![$($x),*]));
+    ($($x:expr,)*) => ($crate::all_of![$($x),*]);
+}
+
+#[macro_export]
+macro_rules! any_of {
+    ($($x:expr),*) => ($crate::mappers::any_of($crate::vec_of_boxes![$($x),*]));
+    ($($x:expr,)*) => ($crate::any_of![$($x),*]);
+}
+
+#[macro_export]
+macro_rules! vec_of_boxes {
+    ($($x:expr),*) => (std::vec![$(std::boxed::Box::new($x)),*]);
+    ($($x:expr,)*) => ($crate::vec_of_boxes![$($x),*]);
+}
+
+pub mod mappers;
+pub mod responders;
+pub mod server;
+
+pub type FullRequest = hyper::Request<Vec<u8>>;
+pub type FullResponse = hyper::Response<Vec<u8>>;
+pub use mappers::Matcher;
+
+pub use server::Expectation;
+pub use server::Server;
+pub use server::Times;

src/mappers.rs

@@ -0,0 +1,349 @@
+use std::borrow::Borrow;
+use std::fmt;
+use std::marker::PhantomData;
+
+// import the any_of and all_of macros from crate root so they are accessible if
+// people glob import this module.
+pub use crate::all_of;
+pub use crate::any_of;
+pub mod request;
+pub mod response;
+
+pub trait Mapper<IN>: Send + fmt::Debug
+where
+    IN: ?Sized,
+{
+    type Out;
+
+    fn map(&mut self, input: &IN) -> Self::Out;
+}
+
+// Matcher is just a special case of Mapper that returns a boolean. Simply
+// provides the `matches` method rather than `map` as that reads a little
+// better.
+pub trait Matcher<IN>: Send + fmt::Debug
+where
+    IN: ?Sized,
+{
+    fn matches(&mut self, input: &IN) -> bool;
+}
+impl<T, IN> Matcher<IN> for T
+where
+    T: Mapper<IN, Out = bool>,
+{
+    fn matches(&mut self, input: &IN) -> bool {
+        self.map(input)
+    }
+}
+
+pub fn any<IN>() -> impl Mapper<IN, Out = bool> {
+    Any
+}
+#[derive(Debug)]
+pub struct Any;
+impl<IN> Mapper<IN> for Any {
+    type Out = bool;
+
+    fn map(&mut self, _input: &IN) -> bool {
+        true
+    }
+}
+
+pub fn contains<T, IN>(value: T) -> impl Mapper<IN, Out = bool>
+where
+    T: AsRef<[u8]> + fmt::Debug + Send,
+    IN: AsRef<[u8]> + ?Sized,
+{
+    Contains(value)
+}
+#[derive(Debug)]
+pub struct Contains<T>(T);
+impl<IN, T> Mapper<IN> for Contains<T>
+where
+    T: AsRef<[u8]> + fmt::Debug + Send,
+    IN: AsRef<[u8]> + ?Sized,
+{
+    type Out = bool;
+
+    fn map(&mut self, input: &IN) -> bool {
+        use bstr::ByteSlice;
+        input.as_ref().contains_str(self.0.as_ref())
+    }
+}
+
+pub fn eq<T, IN>(value: T) -> impl Mapper<IN, Out = bool>
+where
+    T: Borrow<IN> + fmt::Debug + Send,
+    IN: PartialEq + ?Sized,
+{
+    Eq(value)
+}
+#[derive(Debug)]
+pub struct Eq<T>(T);
+impl<IN, T> Mapper<IN> for Eq<T>
+where
+    T: Borrow<IN> + fmt::Debug + Send,
+    IN: PartialEq + ?Sized,
+{
+    type Out = bool;
+
+    fn map(&mut self, input: &IN) -> bool {
+        self.0.borrow() == input
+    }
+}
+
+pub fn matches<IN>(value: &str) -> impl Mapper<IN, Out = bool>
+where
+    IN: AsRef<[u8]> + ?Sized,
+{
+    let regex = regex::bytes::Regex::new(value).expect("failed to create regex");
+    Matches(regex)
+}
+#[derive(Debug)]
+pub struct Matches(regex::bytes::Regex);
+impl<IN> Mapper<IN> for Matches
+where
+    IN: AsRef<[u8]> + ?Sized,
+{
+    type Out = bool;
+
+    fn map(&mut self, input: &IN) -> bool {
+        self.0.is_match(input.as_ref())
+    }
+}
+
+pub fn not<C, IN>(inner: C) -> impl Mapper<IN, Out = bool>
+where
+    C: Mapper<IN, Out = bool>,
+    IN: ?Sized,
+{
+    Not(inner, PhantomData)
+}
+pub struct Not<C, IN>(C, PhantomData<fn(IN)>)
+where
+    IN: ?Sized;
+impl<C, IN> Mapper<IN> for Not<C, IN>
+where
+    C: Mapper<IN, Out = bool>,
+    IN: ?Sized,
+{
+    type Out = bool;
+
+    fn map(&mut self, input: &IN) -> bool {
+        !self.0.map(input)
+    }
+}
+impl<C, IN> fmt::Debug for Not<C, IN>
+where
+    C: Mapper<IN, Out = bool>,
+    IN: ?Sized,
+{
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "Not({:?})", &self.0)
+    }
+}
+
+pub fn all_of<IN>(inner: Vec<Box<dyn Mapper<IN, Out = bool>>>) -> impl Mapper<IN, Out = bool>
+where
+    IN: fmt::Debug + ?Sized,
+{
+    AllOf(inner)
+}
+
+#[derive(Debug)]
+pub struct AllOf<IN>(Vec<Box<dyn Mapper<IN, Out = bool>>>)
+where
+    IN: ?Sized;
+impl<IN> Mapper<IN> for AllOf<IN>
+where
+    IN: fmt::Debug + ?Sized,
+{
+    type Out = bool;
+
+    fn map(&mut self, input: &IN) -> bool {
+        self.0.iter_mut().all(|maper| maper.map(input))
+    }
+}
+
+pub fn any_of<IN>(inner: Vec<Box<dyn Mapper<IN, Out = bool>>>) -> impl Mapper<IN, Out = bool>
+where
+    IN: fmt::Debug + ?Sized,
+{
+    AnyOf(inner)
+}
+#[derive(Debug)]
+pub struct AnyOf<IN>(Vec<Box<dyn Mapper<IN, Out = bool>>>)
+where
+    IN: ?Sized;
+impl<IN> Mapper<IN> for AnyOf<IN>
+where
+    IN: fmt::Debug + ?Sized,
+{
+    type Out = bool;
+
+    fn map(&mut self, input: &IN) -> bool {
+        self.0.iter_mut().any(|maper| maper.map(input))
+    }
+}
+
+pub fn uri_decoded<IN, C>(inner: C) -> impl Mapper<IN, Out = C::Out>
+where
+    IN: AsRef<[u8]> + ?Sized,
+    C: Mapper<[(String, String)]>,
+{
+    UriDecoded(inner)
+}
+#[derive(Debug)]
+pub struct UriDecoded<C>(C);
+impl<IN, C> Mapper<IN> for UriDecoded<C>
+where
+    IN: AsRef<[u8]> + ?Sized,
+    C: Mapper<[(String, String)]>,
+{
+    type Out = C::Out;
+
+    fn map(&mut self, input: &IN) -> C::Out {
+        let decoded: Vec<(String, String)> = url::form_urlencoded::parse(input.as_ref())
+            .into_owned()
+            .collect();
+        self.0.map(&decoded)
+    }
+}
+
+pub fn json_decoded<IN, C>(inner: C) -> impl Mapper<IN, Out = C::Out>
+where
+    IN: AsRef<[u8]> + ?Sized,
+    C: Mapper<serde_json::Value>,
+{
+    JsonDecoded(inner)
+}
+#[derive(Debug)]
+pub struct JsonDecoded<C>(C);
+impl<IN, C> Mapper<IN> for JsonDecoded<C>
+where
+    IN: AsRef<[u8]> + ?Sized,
+    C: Mapper<serde_json::Value>,
+{
+    type Out = C::Out;
+
+    fn map(&mut self, input: &IN) -> C::Out {
+        let json_value: serde_json::Value =
+            serde_json::from_slice(input.as_ref()).unwrap_or(serde_json::Value::Null);
+        self.0.map(&json_value)
+    }
+}
+
+pub fn lowercase<IN, C>(inner: C) -> impl Mapper<IN, Out = C::Out>
+where
+    IN: AsRef<[u8]> + ?Sized,
+    C: Mapper<[u8]>,
+{
+    Lowercase(inner)
+}
+#[derive(Debug)]
+pub struct Lowercase<C>(C);
+impl<IN, C> Mapper<IN> for Lowercase<C>
+where
+    IN: AsRef<[u8]> + ?Sized,
+    C: Mapper<[u8]>,
+{
+    type Out = C::Out;
+
+    fn map(&mut self, input: &IN) -> C::Out {
+        use bstr::ByteSlice;
+        self.0.map(&input.as_ref().to_lowercase())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_contains() {
+        let mut c = contains("foo");
+        assert_eq!(true, c.map("foobar"));
+        assert_eq!(true, c.map("bazfoobar"));
+        assert_eq!(false, c.map("bar"));
+    }
+
+    #[test]
+    fn test_eq() {
+        let mut c = eq("foo");
+        assert_eq!(false, c.map("foobar"));
+        assert_eq!(false, c.map("bazfoobar"));
+        assert_eq!(false, c.map("bar"));
+        assert_eq!(true, c.map("foo"));
+    }
+
+    #[test]
+    fn test_matches() {
+        let mut c = matches(r#"^foo\d*bar$"#);
+        assert_eq!(true, c.map("foobar"));
+        assert_eq!(true, c.map("foo99bar"));
+        assert_eq!(false, c.map("foo99barz"));
+        assert_eq!(false, c.map("bat"));
+    }
+
+    #[test]
+    fn test_not() {
+        let mut c = not(matches(r#"^foo\d*bar$"#));
+        assert_eq!(false, c.map("foobar"));
+        assert_eq!(false, c.map("foo99bar"));
+        assert_eq!(true, c.map("foo99barz"));
+        assert_eq!(true, c.map("bat"));
+    }
+
+    #[test]
+    fn test_all_of() {
+        let mut c = all_of![contains("foo"), contains("bar")];
+        assert_eq!(true, c.map("foobar"));
+        assert_eq!(true, c.map("barfoo"));
+        assert_eq!(false, c.map("foo"));
+        assert_eq!(false, c.map("bar"));
+    }
+
+    #[test]
+    fn test_any_of() {
+        let mut c = any_of![contains("foo"), contains("bar")];
+        assert_eq!(true, c.map("foobar"));
+        assert_eq!(true, c.map("barfoo"));
+        assert_eq!(true, c.map("foo"));
+        assert_eq!(true, c.map("bar"));
+        assert_eq!(false, c.map("baz"));
+    }
+
+    #[test]
+    fn test_uri_decoded() {
+        let expected = vec![
+            ("key 1".to_owned(), "value 1".to_owned()),
+            ("key2".to_owned(), "".to_owned()),
+        ];
+        let mut c = request::query(uri_decoded(eq(expected)));
+        let req = http::Request::get("https://example.com/path?key%201=value%201&key2")
+            .body(Vec::new())
+            .unwrap();
+
+        assert_eq!(true, c.map(&req));
+    }
+
+    #[test]
+    fn test_json_decoded() {
+        let mut c = json_decoded(eq(serde_json::json!({
+            "foo": 1,
+            "bar": 99,
+        })));
+        assert_eq!(true, c.map(r#"{"foo": 1, "bar": 99}"#));
+        assert_eq!(true, c.map(r#"{"bar": 99, "foo": 1}"#));
+        assert_eq!(false, c.map(r#"{"foo": 1, "bar": 100}"#));
+    }
+
+    #[test]
+    fn test_lowercase() {
+        let mut c = lowercase(contains("foo"));
+        assert_eq!(true, c.map("FOO"));
+        assert_eq!(true, c.map("FoOBar"));
+        assert_eq!(true, c.map("foobar"));
+        assert_eq!(false, c.map("bar"));
+    }
+}

src/mappers/request.rs

@@ -0,0 +1,177 @@
+use super::Mapper;
+use crate::FullRequest;
+
+pub fn method<C>(inner: C) -> impl Mapper<FullRequest, Out = C::Out>
+where
+    C: Mapper<str>,
+{
+    Method(inner)
+}
+#[derive(Debug)]
+pub struct Method<C>(C);
+impl<C> Mapper<FullRequest> for Method<C>
+where
+    C: Mapper<str>,
+{
+    type Out = C::Out;
+
+    fn map(&mut self, input: &FullRequest) -> C::Out {
+        self.0.map(input.method().as_str())
+    }
+}
+
+pub fn path<C>(inner: C) -> impl Mapper<FullRequest, Out = C::Out>
+where
+    C: Mapper<str>,
+{
+    Path(inner)
+}
+#[derive(Debug)]
+pub struct Path<C>(C);
+impl<C> Mapper<FullRequest> for Path<C>
+where
+    C: Mapper<str>,
+{
+    type Out = C::Out;
+
+    fn map(&mut self, input: &FullRequest) -> C::Out {
+        self.0.map(input.uri().path())
+    }
+}
+
+pub fn query<C>(inner: C) -> impl Mapper<FullRequest, Out = C::Out>
+where
+    C: Mapper<str>,
+{
+    Query(inner)
+}
+#[derive(Debug)]
+pub struct Query<C>(C);
+impl<C> Mapper<FullRequest> for Query<C>
+where
+    C: Mapper<str>,
+{
+    type Out = C::Out;
+
+    fn map(&mut self, input: &FullRequest) -> C::Out {
+        self.0.map(input.uri().query().unwrap_or(""))
+    }
+}
+
+pub fn headers<C>(inner: C) -> impl Mapper<FullRequest, Out = C::Out>
+where
+    C: Mapper<[(Vec<u8>, Vec<u8>)]>,
+{
+    Headers(inner)
+}
+#[derive(Debug)]
+pub struct Headers<C>(C);
+impl<C> Mapper<FullRequest> for Headers<C>
+where
+    C: Mapper<[(Vec<u8>, Vec<u8>)]>,
+{
+    type Out = C::Out;
+
+    fn map(&mut self, input: &FullRequest) -> C::Out {
+        let headers: Vec<(Vec<u8>, Vec<u8>)> = input
+            .headers()
+            .iter()
+            .map(|(k, v)| (k.as_str().into(), v.as_bytes().into()))
+            .collect();
+        self.0.map(&headers)
+    }
+}
+
+pub fn body<C>(inner: C) -> impl Mapper<FullRequest, Out = C::Out>
+where
+    C: Mapper<[u8]>,
+{
+    Body(inner)
+}
+#[derive(Debug)]
+pub struct Body<C>(C);
+impl<C> Mapper<FullRequest> for Body<C>
+where
+    C: Mapper<[u8]>,
+{
+    type Out = C::Out;
+
+    fn map(&mut self, input: &FullRequest) -> C::Out {
+        self.0.map(input.body())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mappers::*;
+
+    #[test]
+    fn test_path() {
+        let req = hyper::Request::get("https://example.com/foo")
+            .body(Vec::new())
+            .unwrap();
+        assert!(path(eq("/foo")).map(&req));
+
+        let req = hyper::Request::get("https://example.com/foobar")
+            .body(Vec::new())
+            .unwrap();
+        assert!(path(eq("/foobar")).map(&req))
+    }
+
+    #[test]
+    fn test_query() {
+        let req = hyper::Request::get("https://example.com/path?foo=bar&baz=bat")
+            .body(Vec::new())
+            .unwrap();
+        assert!(query(eq("foo=bar&baz=bat")).map(&req));
+        let req = hyper::Request::get("https://example.com/path?search=1")
+            .body(Vec::new())
+            .unwrap();
+        assert!(query(eq("search=1")).map(&req));
+    }
+
+    #[test]
+    fn test_method() {
+        let req = hyper::Request::get("https://example.com/foo")
+            .body(Vec::new())
+            .unwrap();
+        assert!(method(eq("GET")).map(&req));
+        let req = hyper::Request::post("https://example.com/foobar")
+            .body(Vec::new())
+            .unwrap();
+        assert!(method(eq("POST")).map(&req));
+    }
+
+    #[test]
+    fn test_headers() {
+        let expected = vec![
+            (Vec::from("host"), Vec::from("example.com")),
+            (Vec::from("content-length"), Vec::from("101")),
+        ];
+        let mut req = hyper::Request::get("https://example.com/path?key%201=value%201&key2")
+            .body(Vec::new())
+            .unwrap();
+        req.headers_mut().extend(vec![
+            (
+                hyper::header::HOST,
+                hyper::header::HeaderValue::from_static("example.com"),
+            ),
+            (
+                hyper::header::CONTENT_LENGTH,
+                hyper::header::HeaderValue::from_static("101"),
+            ),
+        ]);
+
+        assert!(headers(eq(expected)).map(&req));
+    }
+
+    #[test]
+    fn test_body() {
+        use bstr::{ByteVec, B};
+        let req = hyper::Request::get("https://example.com/foo")
+            .body(Vec::from_slice("my request body"))
+            .unwrap();
+        assert!(body(eq(B("my request body"))).map(&req));
+    }
+}

src/mappers/response.rs

@@ -0,0 +1,109 @@
+use super::Mapper;
+use crate::FullResponse;
+
+pub fn status_code<C>(inner: C) -> impl Mapper<FullResponse, Out = C::Out>
+where
+    C: Mapper<u16>,
+{
+    StatusCode(inner)
+}
+#[derive(Debug)]
+pub struct StatusCode<C>(C);
+impl<C> Mapper<FullResponse> for StatusCode<C>
+where
+    C: Mapper<u16>,
+{
+    type Out = C::Out;
+
+    fn map(&mut self, input: &FullResponse) -> C::Out {
+        self.0.map(&input.status().as_u16())
+    }
+}
+
+pub fn headers<C>(inner: C) -> impl Mapper<FullResponse, Out = C::Out>
+where
+    C: Mapper<[(Vec<u8>, Vec<u8>)]>,
+{
+    Headers(inner)
+}
+#[derive(Debug)]
+pub struct Headers<C>(C);
+impl<C> Mapper<FullResponse> for Headers<C>
+where
+    C: Mapper<[(Vec<u8>, Vec<u8>)]>,
+{
+    type Out = C::Out;
+
+    fn map(&mut self, input: &FullResponse) -> C::Out {
+        let headers: Vec<(Vec<u8>, Vec<u8>)> = input
+            .headers()
+            .iter()
+            .map(|(k, v)| (k.as_str().into(), v.as_bytes().into()))
+            .collect();
+        self.0.map(&headers)
+    }
+}
+
+pub fn body<C>(inner: C) -> impl Mapper<FullResponse, Out = C::Out>
+where
+    C: Mapper<[u8]>,
+{
+    Body(inner)
+}
+#[derive(Debug)]
+pub struct Body<C>(C);
+impl<C> Mapper<FullResponse> for Body<C>
+where
+    C: Mapper<[u8]>,
+{
+    type Out = C::Out;
+
+    fn map(&mut self, input: &FullResponse) -> C::Out {
+        self.0.map(input.body())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mappers::*;
+
+    #[test]
+    fn test_status_code() {
+        let resp = hyper::Response::builder()
+            .status(hyper::StatusCode::NOT_FOUND)
+            .body(Vec::new())
+            .unwrap();
+        assert!(status_code(eq(404)).map(&resp));
+
+        let resp = hyper::Response::builder()
+            .status(hyper::StatusCode::OK)
+            .body(Vec::new())
+            .unwrap();
+        assert!(status_code(eq(200)).map(&resp));
+    }
+
+    #[test]
+    fn test_headers() {
+        let expected = vec![
+            (Vec::from("host"), Vec::from("example.com")),
+            (Vec::from("content-length"), Vec::from("101")),
+        ];
+        let resp = hyper::Response::builder()
+            .header("host", "example.com")
+            .header("content-length", 101)
+            .body(Vec::new())
+            .unwrap();
+
+        assert!(headers(eq(expected)).map(&resp));
+    }
+
+    #[test]
+    fn test_body() {
+        use bstr::{ByteVec, B};
+        let resp = hyper::Response::builder()
+            .body(Vec::from_slice("my request body"))
+            .unwrap();
+        assert!(body(eq(B("my request body"))).map(&resp));
+    }
+}

src/responders.rs

@@ -0,0 +1,85 @@
+use std::fmt;
+use std::future::Future;
+use std::pin::Pin;
+
+pub trait Responder: Send + fmt::Debug {
+    fn respond(&mut self) -> Pin<Box<dyn Future<Output = http::Response<hyper::Body>> + Send>>;
+}
+
+pub fn status_code(code: u16) -> impl Responder {
+    StatusCode(code)
+}
+#[derive(Debug)]
+pub struct StatusCode(u16);
+impl Responder for StatusCode {
+    fn respond(&mut self) -> Pin<Box<dyn Future<Output = http::Response<hyper::Body>> + Send>> {
+        async fn _respond(status_code: u16) -> http::Response<hyper::Body> {
+            hyper::Response::builder()
+                .status(status_code)
+                .body(hyper::Body::empty())
+                .unwrap()
+        }
+        Box::pin(_respond(self.0))
+    }
+}
+
+pub fn json_encoded<T>(data: T) -> impl Responder
+where
+    T: serde::Serialize,
+{
+    JsonEncoded(serde_json::to_vec(&data).unwrap())
+}
+#[derive(Debug)]
+pub struct JsonEncoded(Vec<u8>);
+impl Responder for JsonEncoded {
+    fn respond(&mut self) -> Pin<Box<dyn Future<Output = http::Response<hyper::Body>> + Send>> {
+        async fn _respond(body: Vec<u8>) -> http::Response<hyper::Body> {
+            hyper::Response::builder()
+                .status(200)
+                .header("Content-Type", "application/json")
+                .body(body.into())
+                .unwrap()
+        }
+        Box::pin(_respond(self.0.clone()))
+    }
+}
+
+impl Responder for crate::FullResponse {
+    fn respond(&mut self) -> Pin<Box<dyn Future<Output = http::Response<hyper::Body>> + Send>> {
+        async fn _respond(resp: http::Response<hyper::Body>) -> http::Response<hyper::Body> {
+            resp
+        }
+        // Turn &hyper::Response<Vec<u8>> into a hyper::Response<hyper::Body>
+        let mut builder = hyper::Response::builder();
+        builder
+            .status(self.status().clone())
+            .version(self.version().clone());
+        *builder.headers_mut().unwrap() = self.headers().clone();
+        let resp = builder.body(self.body().clone().into()).unwrap();
+
+        Box::pin(_respond(resp))
+    }
+}
+
+// TODO: make a macro for this to avoid the vec![Box::new] dance.
+pub fn cycle(responders: Vec<Box<dyn Responder>>) -> impl Responder {
+    if responders.is_empty() {
+        panic!("empty vector provided to cycle");
+    }
+    Cycle { idx: 0, responders }
+}
+#[derive(Debug)]
+pub struct Cycle {
+    idx: usize,
+    responders: Vec<Box<dyn Responder>>,
+}
+impl Responder for Cycle {
+    fn respond(&mut self) -> Pin<Box<dyn Future<Output = http::Response<hyper::Body>> + Send>> {
+        let response = self.responders[self.idx].respond();
+        self.idx = (self.idx + 1) % self.responders.len();
+        response
+    }
+}
+
+#[cfg(test)]
+mod tests {}

src/server.rs

@@ -0,0 +1,286 @@
+use crate::responders::Responder;
+use crate::{FullRequest, Matcher};
+use std::future::Future;
+use std::net::SocketAddr;
+use std::pin::Pin;
+use std::sync::{Arc, Mutex};
+
+pub struct Server {
+    trigger_shutdown: Option<futures::channel::oneshot::Sender<()>>,
+    join_handle: Option<std::thread::JoinHandle<()>>,
+    addr: SocketAddr,
+    state: ServerState,
+}
+
+impl Server {
+    pub fn run() -> Self {
+        use futures::future::FutureExt;
+        use hyper::{
+            service::{make_service_fn, service_fn},
+            Error,
+        };
+        let bind_addr = ([127, 0, 0, 1], 0).into();
+        // And a MakeService to handle each connection...
+        let state = ServerState::default();
+        let make_service = make_service_fn({
+            let state = state.clone();
+            move |_| {
+                let state = state.clone();
+                async move {
+                    let state = state.clone();
+                    Ok::<_, Error>(service_fn({
+                        let state = state.clone();
+                        move |req: hyper::Request<hyper::Body>| {
+                            let state = state.clone();
+                            async move {
+                                // read the full body into memory prior to handing it to mappers.
+                                let (head, body) = req.into_parts();
+                                use futures::TryStreamExt;
+                                let full_body = body.try_concat().await?;
+                                let req = hyper::Request::from_parts(head, full_body.to_vec());
+                                log::debug!("Received Request: {:?}", req);
+                                let resp = on_req(state, req).await;
+                                log::debug!("Sending Response: {:?}", resp);
+                                hyper::Result::Ok(resp)
+                            }
+                        }
+                    }))
+                }
+            }
+        });
+        // Then bind and serve...
+        let server = hyper::Server::bind(&bind_addr).serve(make_service);
+        let addr = server.local_addr();
+        let (trigger_shutdown, shutdown_received) = futures::channel::oneshot::channel();
+        let join_handle = std::thread::spawn(move || {
+            let mut runtime = tokio::runtime::current_thread::Runtime::new().unwrap();
+            runtime.block_on(async move {
+                futures::select! {
+                    _ = server.fuse() => {},
+                    _ = shutdown_received.fuse() => {},
+                }
+            });
+        });
+
+        Server {
+            trigger_shutdown: Some(trigger_shutdown),
+            join_handle: Some(join_handle),
+            addr,
+            state,
+        }
+    }
+
+    pub fn addr(&self) -> SocketAddr {
+        self.addr
+    }
+
+    pub fn url<T>(&self, path_and_query: T) -> http::Uri
+    where
+        http::uri::PathAndQuery: http::HttpTryFrom<T>,
+    {
+        http::Uri::builder()
+            .scheme("http")
+            .authority(format!("{}", &self.addr).as_str())
+            .path_and_query(path_and_query)
+            .build()
+            .unwrap()
+    }
+
+    pub fn expect(&self, expectation: Expectation) {
+        self.state.push_expectation(expectation);
+    }
+
+    pub fn verify_and_clear(&mut self) {
+        let mut state = self.state.lock();
+        for expectation in state.expected.iter() {
+            let is_valid_cardinality = match &expectation.cardinality {
+                Times::AnyNumber => true,
+                Times::AtLeast(lower_bound) if expectation.hit_count >= *lower_bound => true,
+                Times::AtLeast(_) => false,
+                Times::AtMost(limit) if expectation.hit_count <= *limit => true,
+                Times::AtMost(_) => false,
+                Times::Between(range)
+                    if expectation.hit_count <= *range.end()
+                        && expectation.hit_count >= *range.start() =>
+                {
+                    true
+                }
+                Times::Between(_) => false,
+                Times::Exactly(limit) if expectation.hit_count == *limit => true,
+                Times::Exactly(_) => false,
+            };
+            if !is_valid_cardinality {
+                panic!(format!(
+                    "Unexpected number of requests for matcher '{:?}'; received {}; expected {:?}",
+                    &expectation.matcher, expectation.hit_count, &expectation.cardinality,
+                ));
+            }
+        }
+        state.expected.clear();
+        if !state.unexpected_requests.is_empty() {
+            // TODO: format and print the requests.
+            panic!("unexpected requests received");
+        }
+    }
+}
+
+impl Drop for Server {
+    fn drop(&mut self) {
+        // drop the trigger_shutdown channel to tell the server to shutdown.
+        // Then wait for the shutdown to complete.
+        self.trigger_shutdown = None;
+        let _ = self.join_handle.take().unwrap().join();
+        self.verify_and_clear();
+    }
+}
+
+async fn on_req(state: ServerState, req: FullRequest) -> http::Response<hyper::Body> {
+    let response_future = {
+        let mut state = state.lock();
+        let mut iter = state.expected.iter_mut();
+        let response_future = loop {
+            let expectation = match iter.next() {
+                None => break None,
+                Some(expectation) => expectation,
+            };
+            if expectation.matcher.matches(&req) {
+                log::debug!("found matcher: {:?}", &expectation.matcher);
+                expectation.hit_count += 1;
+                let is_valid_cardinality = match &expectation.cardinality {
+                    Times::AnyNumber => true,
+                    Times::AtLeast(_) => true,
+                    Times::AtMost(limit) if expectation.hit_count <= *limit => true,
+                    Times::AtMost(_) => false,
+                    Times::Between(range) if expectation.hit_count <= *range.end() => true,
+                    Times::Between(_) => false,
+                    Times::Exactly(limit) if expectation.hit_count <= *limit => true,
+                    Times::Exactly(_) => false,
+                };
+                if is_valid_cardinality {
+                    break Some(expectation.responder.respond());
+                } else {
+                    break Some(Box::pin(cardinality_error(
+                        &*expectation.matcher as &dyn Matcher<FullRequest>,
+                        &expectation.cardinality,
+                        expectation.hit_count,
+                    )));
+                }
+            }
+        };
+        if response_future.is_none() {
+            // TODO: provide real request id.
+            state.unexpected_requests.push(RequestID(1));
+        }
+        response_future
+    };
+    if let Some(f) = response_future {
+        f.await
+    } else {
+        http::Response::builder()
+            .status(http::StatusCode::INTERNAL_SERVER_ERROR)
+            .body(hyper::Body::empty())
+            .unwrap()
+    }
+}
+
+#[derive(Debug, Clone)]
+pub enum Times {
+    AnyNumber,
+    AtLeast(usize),
+    AtMost(usize),
+    Between(std::ops::RangeInclusive<usize>),
+    Exactly(usize),
+}
+
+pub struct Expectation {
+    matcher: Box<dyn Matcher<FullRequest>>,
+    cardinality: Times,
+    responder: Box<dyn Responder>,
+    hit_count: usize,
+}
+
+impl Expectation {
+    pub fn matching(matcher: impl Matcher<FullRequest> + 'static) -> ExpectationBuilder {
+        ExpectationBuilder {
+            matcher: Box::new(matcher),
+            cardinality: Times::Exactly(1),
+        }
+    }
+}
+
+pub struct ExpectationBuilder {
+    matcher: Box<dyn Matcher<FullRequest>>,
+    cardinality: Times,
+}
+
+impl ExpectationBuilder {
+    pub fn times(self, cardinality: Times) -> ExpectationBuilder {
+        ExpectationBuilder {
+            cardinality,
+            ..self
+        }
+    }
+
+    pub fn respond_with(self, responder: impl Responder + 'static) -> Expectation {
+        Expectation {
+            matcher: self.matcher,
+            cardinality: self.cardinality,
+            responder: Box::new(responder),
+            hit_count: 0,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+struct RequestID(u64);
+
+#[derive(Clone)]
+struct ServerState(Arc<Mutex<ServerStateInner>>);
+
+impl ServerState {
+    fn lock(&self) -> std::sync::MutexGuard<ServerStateInner> {
+        self.0.lock().expect("mutex poisoned")
+    }
+
+    fn push_expectation(&self, expectation: Expectation) {
+        let mut inner = self.lock();
+        inner.expected.push(expectation);
+    }
+}
+
+impl Default for ServerState {
+    fn default() -> Self {
+        ServerState(Default::default())
+    }
+}
+
+struct ServerStateInner {
+    unexpected_requests: Vec<RequestID>,
+    expected: Vec<Expectation>,
+}
+
+impl Default for ServerStateInner {
+    fn default() -> Self {
+        ServerStateInner {
+            unexpected_requests: Default::default(),
+            expected: Default::default(),
+        }
+    }
+}
+
+fn cardinality_error(
+    matcher: &dyn Matcher<FullRequest>,
+    cardinality: &Times,
+    hit_count: usize,
+) -> Pin<Box<dyn Future<Output = http::Response<hyper::Body>> + Send + 'static>> {
+    let body = hyper::Body::from(format!(
+        "Unexpected number of requests for matcher '{:?}'; received {}; expected {:?}",
+        matcher, hit_count, cardinality,
+    ));
+    Box::pin(async move {
+        http::Response::builder()
+            .status(http::StatusCode::INTERNAL_SERVER_ERROR)
+            .body(body)
+            .unwrap()
+    })
+}

tests/tests.rs

@@ -0,0 +1,145 @@
+use httptest::{mappers::*, responders::*, Expectation, Times};
+
+async fn read_response_body(resp: hyper::Response<hyper::Body>) -> hyper::Response<Vec<u8>> {
+    use futures::stream::TryStreamExt;
+    let (head, body) = resp.into_parts();
+    let body = body.try_concat().await.unwrap().to_vec();
+    hyper::Response::from_parts(head, body)
+}
+
+#[tokio::test]
+async fn test_server() {
+    let _ = pretty_env_logger::try_init();
+
+    // Setup a server to expect a single GET /foo request.
+    let server = httptest::Server::run();
+    server.expect(
+        Expectation::matching(all_of![
+            request::method(eq("GET")),
+            request::path(eq("/foo"))
+        ])
+        .times(Times::Exactly(1))
+        .respond_with(status_code(200)),
+    );
+
+    // Issue the GET /foo to the server and verify it returns a 200.
+    let client = hyper::Client::new();
+    let resp = read_response_body(client.get(server.url("/foo")).await.unwrap()).await;
+    assert!(response::status_code(eq(200)).matches(&resp));
+
+    // The Drop impl of the server will assert that all expectations were satisfied or else it will panic.
+}
+
+#[tokio::test]
+#[should_panic]
+async fn test_expectation_cardinality_not_reached() {
+    let _ = pretty_env_logger::try_init();
+
+    // Setup a server to expect a single GET /foo request.
+    let server = httptest::Server::run();
+    server.expect(
+        Expectation::matching(all_of![
+            request::method(eq("GET")),
+            request::path(eq("/foo"))
+        ])
+        .times(Times::Exactly(1))
+        .respond_with(status_code(200)),
+    );
+
+    // Don't send any requests. Should panic.
+}
+
+#[tokio::test]
+#[should_panic]
+async fn test_expectation_cardinality_exceeded() {
+    let _ = pretty_env_logger::try_init();
+
+    // Setup a server to expect a single GET /foo request.
+    let server = httptest::Server::run();
+    server.expect(
+        Expectation::matching(all_of![
+            request::method(eq("GET")),
+            request::path(eq("/foo"))
+        ])
+        .times(Times::Exactly(1))
+        .respond_with(
+            http::Response::builder()
+                .status(http::StatusCode::INTERNAL_SERVER_ERROR)
+                .body(Vec::new())
+                .unwrap(),
+        ),
+    );
+
+    // Issue the GET /foo to the server and verify it returns a 200.
+    let client = hyper::Client::new();
+    let resp = read_response_body(client.get(server.url("/foo")).await.unwrap()).await;
+    assert!(response::status_code(eq(200)).matches(&resp));
+
+    // Issue a second GET /foo and verify it returns a 500 because the cardinality of the expectation has been exceeded.
+    let resp = read_response_body(client.get(server.url("/foo")).await.unwrap()).await;
+    assert!(response::status_code(eq(500)).matches(&resp));
+
+    // Should panic on Server drop.
+}
+
+#[tokio::test]
+async fn test_json() {
+    let _ = pretty_env_logger::try_init();
+
+    let my_data = serde_json::json!({
+        "foo": "bar",
+        "baz": [1, 2, 3],
+    });
+
+    // Setup a server to expect a single GET /foo request and respond with a
+    // json encoding of my_data.
+    let server = httptest::Server::run();
+    server.expect(
+        Expectation::matching(all_of![
+            request::method(eq("GET")),
+            request::path(eq("/foo"))
+        ])
+        .times(Times::Exactly(1))
+        .respond_with(json_encoded(my_data.clone())),
+    );
+
+    // Issue the GET /foo to the server and verify it returns a 200.
+    let client = hyper::Client::new();
+    let resp = read_response_body(client.get(server.url("/foo")).await.unwrap()).await;
+    assert!(all_of![
+        response::status_code(eq(200)),
+        response::body(json_decoded(eq(my_data))),
+    ]
+    .matches(&resp));
+}
+
+#[tokio::test]
+async fn test_cycle() {
+    let _ = pretty_env_logger::try_init();
+
+    // Setup a server to expect a single GET /foo request and respond with a
+    // json encoding of my_data.
+    let server = httptest::Server::run();
+    server.expect(
+        Expectation::matching(all_of![
+            request::method(eq("GET")),
+            request::path(eq("/foo"))
+        ])
+        .times(Times::Exactly(4))
+        .respond_with(cycle(vec![
+            Box::new(status_code(200)),
+            Box::new(status_code(404)),
+        ])),
+    );
+
+    // Issue the GET /foo to the server and verify it returns a 200.
+    let client = hyper::Client::new();
+    let resp = read_response_body(client.get(server.url("/foo")).await.unwrap()).await;
+    assert!(response::status_code(eq(200)).matches(&resp));
+    let resp = read_response_body(client.get(server.url("/foo")).await.unwrap()).await;
+    assert!(response::status_code(eq(404)).matches(&resp));
+    let resp = read_response_body(client.get(server.url("/foo")).await.unwrap()).await;
+    assert!(response::status_code(eq(200)).matches(&resp));
+    let resp = read_response_body(client.get(server.url("/foo")).await.unwrap()).await;
+    assert!(response::status_code(eq(404)).matches(&resp));
+}