<commit_before>#![deny(warnings)]
extern crate hyper;
extern crate futures;
extern crate tokio_core;
extern crate pretty_env_logger;

use std::io::{self, Read, Write};
use std::net::TcpListener;
use std::thread;
use std::time::Duration;

use hyper::client::{Client, Request, HttpConnector};
use hyper::{Method, StatusCode};

use futures::{Future, Stream};
use futures::sync::oneshot;

use tokio_core::reactor::{Core, Handle};

fn client(handle: &Handle) -> Client<HttpConnector> {
    Client::new(handle)
}

fn s(buf: &[u8]) -> &str {
    ::std::str::from_utf8(buf).unwrap()
}

macro_rules! test {
    (
        name: $name:ident,
        server:
            expected: $server_expected:expr,
            reply: $server_reply:expr,
        client:
            request:
                method: $client_method:ident,
                url: $client_url:expr,
                headers: [ $($request_headers:expr,)* ],
                body: $request_body:expr,
                proxy: $request_proxy:expr,

            response:
                status: $client_status:ident,
                headers: [ $($response_headers:expr,)* ],
                body: $response_body:expr,
    ) => (
        #[test]
        fn $name() {
            #![allow(unused)]
            use hyper::header::*;
            let _ = pretty_env_logger::init();
            let server = TcpListener::bind("127.0.0.1:0").unwrap();
            let addr = server.local_addr().unwrap();
            let mut core = Core::new().unwrap();
            let client = client(&core.handle());
            let mut req = Request::new(Method::$client_method, format!($client_url, addr=addr).parse().unwrap());
            $(
                req.headers_mut().set($request_headers);
            )*

            if let Some(body) = $request_body {
                let body: &'static str = body;
                req.set_body(body);
            }
            req.set_proxy($request_proxy);

            let res = client.request(req);

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

            thread::spawn(move || {
                let mut inc = server.accept().unwrap().0;
                inc.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
                inc.set_write_timeout(Some(Duration::from_secs(5))).unwrap();
                let expected = format!($server_expected, addr=addr);
                let mut buf = [0; 4096];
                let mut n = 0;
                while n < buf.len() && n < expected.len() {
                    n += inc.read(&mut buf[n..]).unwrap();
                }
                assert_eq!(s(&buf[..n]), expected);

                inc.write_all($server_reply.as_ref()).unwrap();
                let _ = tx.send(());
            });

            let rx = rx.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked")));

            let work = res.join(rx).map(|r| r.0);

            let res = core.run(work).unwrap();
            assert_eq!(res.status(), StatusCode::$client_status);
            $(
                assert_eq!(res.headers().get(), Some(&$response_headers));
            )*

            let body = core.run(res.body().concat2()).unwrap();
<commit_msg>tests(client): improve panic message when client tests fail<commit_after>#![deny(warnings)]
extern crate hyper;
extern crate futures;
extern crate tokio_core;
extern crate pretty_env_logger;

use std::io::{self, Read, Write};
use std::net::TcpListener;
use std::thread;
use std::time::Duration;

use hyper::client::{Client, Request, HttpConnector};
use hyper::{Method, StatusCode};

use futures::{Future, Stream};
use futures::sync::oneshot;

use tokio_core::reactor::{Core, Handle};

fn client(handle: &Handle) -> Client<HttpConnector> {
    Client::new(handle)
}

fn s(buf: &[u8]) -> &str {
    ::std::str::from_utf8(buf).unwrap()
}

macro_rules! test {
    (
        name: $name:ident,
        server:
            expected: $server_expected:expr,
            reply: $server_reply:expr,
        client:
            request:
                method: $client_method:ident,
                url: $client_url:expr,
                headers: [ $($request_headers:expr,)* ],
                body: $request_body:expr,
                proxy: $request_proxy:expr,

            response:
                status: $client_status:ident,
                headers: [ $($response_headers:expr,)* ],
                body: $response_body:expr,
    ) => (
        #[test]
        fn $name() {
            #![allow(unused)]
            use hyper::header::*;
            let _ = pretty_env_logger::init();
            let server = TcpListener::bind("127.0.0.1:0").unwrap();
            let addr = server.local_addr().unwrap();
            let mut core = Core::new().unwrap();
            let client = client(&core.handle());
            let mut req = Request::new(Method::$client_method, format!($client_url, addr=addr).parse().unwrap());
            $(
                req.headers_mut().set($request_headers);
            )*

            if let Some(body) = $request_body {
                let body: &'static str = body;
                req.set_body(body);
            }
            req.set_proxy($request_proxy);

            let res = client.request(req);

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

            let thread = thread::Builder::new()
                .name(format!("tcp-server<{}>", stringify!($name)));
            thread.spawn(move || {
                let mut inc = server.accept().unwrap().0;
                inc.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
                inc.set_write_timeout(Some(Duration::from_secs(5))).unwrap();
                let expected = format!($server_expected, addr=addr);
                let mut buf = [0; 4096];
                let mut n = 0;
                while n < buf.len() && n < expected.len() {
                    n += match inc.read(&mut buf[n..]) {
                        Ok(n) => n,
                        Err(e) => panic!("failed to read request, partialy read = {:?}, error: {}", s(&buf[..n]), e),
                    };
                }
                assert_eq!(s(&buf[..n]), expected);

                inc.write_all($server_reply.as_ref()).unwrap();
                let _ = tx.send(());
            }).unwrap();

            let rx = rx.map_err(|_| hyper::Error::Io(io::Error::new(io::ErrorKind::Other, "thread panicked")));

            let work = res.join(rx).map(|r| r.0);

            let res = core.run(work).unwrap();
            assert_eq!(res.status(), StatusCode::$client_status);
            $(
                assert_eq!(res.headers().get(), Some(&$response_headers));
            )*

            let body = core.run(res.body().concat2()).unwrap();

{'fraction_non_alphanumeric': 0.16175156389633602, 'fraction_numerical': 0.005957700327673518, 'mean_word_length': 1.941305300043802, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 14, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4577002', 'n_tokens_mistral': 1999, 'n_tokens_neox': 1940, 'n_words': 376}