[12/13] mesos git commit: Added a test for request streaming and GZIP compression.

[an...@apache.org]

Added a test for request streaming and GZIP compression.

These tests validate that clients can stream requests and send
compressed GZIP requests using the connection abstraction. This
also test the implementation of the streaming decoder indirectly.

Review: https://reviews.apache.org/r/53490/


Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/5cd134fe
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/5cd134fe
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/5cd134fe

Branch: refs/heads/master
Commit: 5cd134fedf69900523f3088de32daae81f216437
Parents: a24cb49
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Nov 21 18:08:30 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Nov 21 19:27:08 2016 -0800

----------------------------------------------------------------------
 3rdparty/libprocess/src/tests/http_tests.cpp | 118 ++++++++++++++++++++++
 1 file changed, 118 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/mesos/blob/5cd134fe/3rdparty/libprocess/src/tests/http_tests.cpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/tests/http_tests.cpp b/3rdparty/libprocess/src/tests/http_tests.cpp
index 10d3fb0..d41929a 100644
--- a/3rdparty/libprocess/src/tests/http_tests.cpp
+++ b/3rdparty/libprocess/src/tests/http_tests.cpp
@@ -21,6 +21,7 @@
 #include <netinet/tcp.h>
 #endif // __WINDOWS__
 
+#include <algorithm>
 #include <string>
 #include <vector>
 
@@ -86,6 +87,7 @@ public:
   MOCK_METHOD1(requestDelete, Future<http::Response>(const http::Request&));
   MOCK_METHOD1(a, Future<http::Response>(const http::Request&));
   MOCK_METHOD1(abc, Future<http::Response>(const http::Request&));
+  MOCK_METHOD1(requestStreaming, Future<http::Response>(const http::Request&));
 
   MOCK_METHOD2(
       authenticated,
@@ -103,6 +105,12 @@ protected:
     route("/a", None(), &HttpProcess::a);
     route("/a/b/c", None(), &HttpProcess::abc);
     route("/authenticated", "realm", None(), &HttpProcess::authenticated);
+
+    // Route accepting a streaming request.
+    RouteOptions options;
+    options.requestStreaming = true;
+
+    route("/requeststreaming", None(), &HttpProcess::requestStreaming, options);
   }
 };
 
@@ -832,6 +840,51 @@ TEST(HTTPTest, Request)
 }
 
 
+// This test verifies that the server can correctly receive the
+// uncompressed data from the request.
+TEST(HTTPConnectionTest, GzipRequestBody)
+{
+  Http http;
+
+  http::URL url = http::URL(
+      "http",
+      http.process->self().address.ip,
+      http.process->self().address.port,
+      http.process->self().id + "/body");
+
+  Future<http::Connection> connect = http::connect(url);
+  AWAIT_READY(connect);
+
+  http::Connection connection = connect.get();
+
+  Promise<http::Response> promise;
+  Future<http::Request> expected;
+
+  EXPECT_CALL(*http.process, body(_))
+    .WillOnce(DoAll(FutureArg<0>(&expected), Return(promise.future())));
+
+  string uncompressed = "Hello World";
+
+  http::Request request;
+  request.method = "POST";
+  request.url = url;
+  request.body = gzip::compress(uncompressed).get();
+  request.keepAlive = true;
+
+  request.headers["Content-Encoding"] = "gzip";
+  request.headers["Content-Length"] = request.body.length();
+
+  Future<http::Response> response = connection.send(request);
+
+  AWAIT_READY(expected);
+  EXPECT_EQ(uncompressed, expected->body);
+
+  // Disconnect.
+  AWAIT_READY(connection.disconnect());
+  AWAIT_READY(connection.disconnected());
+}
+
+
 TEST(HTTPConnectionTest, Serial)
 {
   Http http;
@@ -1175,6 +1228,71 @@ TEST(HTTPConnectionTest, Equality)
 }
 
 
+// This test verifies that we can stream the request body using the
+// connection abstraction to a streaming enabled route.
+TEST(HTTPConnectionTest, RequestStreaming)
+{
+  Http http;
+
+  http::URL url = http::URL(
+      "http",
+      http.process->self().address.ip,
+      http.process->self().address.port,
+      http.process->self().id + "/requeststreaming");
+
+  Future<http::Connection> connect = http::connect(url);
+  AWAIT_READY(connect);
+
+  http::Connection connection = connect.get();
+
+  Promise<http::Response> promise;
+  Future<http::Request> expected;
+
+  EXPECT_CALL(*http.process, requestStreaming(_))
+    .WillOnce(DoAll(FutureArg<0>(&expected), Return(promise.future())));
+
+  http::Pipe pipe;
+
+  http::Request request;
+  request.method = "POST";
+  request.url = url;
+  request.type = http::Request::PIPE;
+  request.reader = pipe.reader();
+  request.keepAlive = true;
+
+  Future<http::Response> response = connection.send(request);
+
+  AWAIT_READY(expected);
+  ASSERT_EQ(http::Request::PIPE, expected->type);
+  ASSERT_SOME(expected->reader);
+
+  http::Pipe::Reader reader = expected->reader.get();
+
+  // Start streaming the request body.
+  pipe.writer().write("Hello");
+
+  string read;
+  while (read != "Hello") {
+    Future<string> future = reader.read();
+    AWAIT_READY(future);
+
+    read.append(future.get());
+    ASSERT_TRUE(std::equal(read.begin(), read.end(), "Hello"));
+  }
+
+  pipe.writer().close();
+
+  promise.set(http::OK("1"));
+
+  AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::OK().status, response);
+  EXPECT_EQ("1", response->body);
+
+  // Disconnect.
+  AWAIT_READY(connection.disconnect());
+  AWAIT_READY(connection.disconnected());
+}
+
+
 // TODO(hausdorff): This test seems to create inconsistent (though not
 // incorrect) results across platforms. Fix and enable the test on Windows. In
 // particular, the encoding in the 3rd example puts the first variable into the