added test client

meson.build

@@ -102,6 +102,8 @@ test_certs_dep = declare_dependency(
 alltests = executable('alltests',
 	'test/webfuse_provider/test_util/timeout_watcher.cc',
 	'test/webfuse_provider/test_util/ws_server.cc',
+	'test/webfuse_provider/test_util/webfuse_server.cc',
+	'test/webfuse_provider/test_util/client.cc',
 	'test/webfuse_provider/test_util/jansson_test_environment.cc',
 	'test/webfuse_provider/mocks/fake_invokation_context.cc',
 	'test/webfuse_provider/mocks/mock_request.cc',
@@ -126,6 +128,7 @@ alltests = executable('alltests',
 	'test/webfuse_provider/util/test_message.cc',
 	'test/webfuse_provider/util/test_message_queue.cc',
 	'test/webfuse_provider/util/test_url.cc',
+	'test/webfuse_provider/provider/test_client.cc',
 	'test/webfuse_provider/provider/test_client_protocol.cc',
 	'test/webfuse_provider/provider/test_dirbuffer.cc',
 	'test/webfuse_provider/provider/operation/test_close.cc',

test/webfuse_provider/provider/test_client.cc

@@ -0,0 +1,27 @@
+#include "webfuse_provider/webfuse_provider.h"
+#include "webfuse_provider/test_util/webfuse_server.hpp"
+#include "webfuse_provider/mocks/mock_provider_client.hpp"
+#include "webfuse_provider/test_util/client.hpp"
+
+#include <gtest/gtest.h>
+
+using webfuse_test::WebfuseServer;
+using webfuse_test::MockProviderClient;
+using webfuse_test::Client;
+
+TEST(Client, Connect)
+{
+    MockProviderClient provider;
+    EXPECT_CALL(provider, OnConnected()).Times(1);
+    EXPECT_CALL(provider, OnDisconnected()).Times(1);
+
+    wfp_client_config * config = wfp_client_config_create();
+    provider.AttachTo(config);
+
+    WebfuseServer server;
+    Client client(config, server.GetUrl());
+
+    server.AwaitConnection();
+
+    wfp_client_config_dispose(config);
+}

test/webfuse_provider/test_util/client.cc

@@ -0,0 +1,66 @@
+#include "webfuse_provider/test_util/client.hpp"
+#include "webfuse_provider/client.h"
+
+#include <thread>
+#include <mutex>
+
+namespace webfuse_test
+{
+
+class Client::Private
+{
+public:
+    Private(wfp_client_config * config, std::string const & url)
+    : client(wfp_client_create(config))
+    , is_shutdown_requested(false)
+    {
+        wfp_client_connect(client, url.c_str());
+        thread = std::thread(&Run, this);
+    }
+
+    ~Private()
+    {
+        {
+            std::unique_lock<std::mutex> lock(mutex);
+            is_shutdown_requested = true;
+        }
+
+        wfp_client_interrupt(client);
+        thread.join();
+        wfp_client_disconnect(client);
+        wfp_client_dispose(client);
+    }
+
+private:
+    static void Run(Private * self)
+    {
+        bool is_running = true;
+        while (is_running)
+        {
+            wfp_client_service(self->client);
+            {
+                std::unique_lock<std::mutex> lock(self->mutex);
+                is_running = !self->is_shutdown_requested;
+            }
+        }
+    }
+
+    wfp_client * client;
+    bool is_shutdown_requested;
+    std::thread thread;
+    std::mutex mutex;
+};
+
+Client::Client(wfp_client_config * config, std::string const & url)
+: d(new Private(config, url))
+{
+
+}
+
+Client::~Client()
+{
+    delete d;
+}
+
+
+}

test/webfuse_provider/test_util/client.hpp

@@ -0,0 +1,24 @@
+#ifndef WFP_TEST_UTIL_CLIENT_HPP
+#define WFP_TEST_UTIL_CLIENT_HPP
+
+#include "webfuse_provider/client_config.h"
+#include <string>
+
+namespace webfuse_test
+{
+
+class Client
+{
+    Client(Client const &) = delete;
+    Client & operator=(Client const &) = delete;
+public:
+    Client(wfp_client_config * config, std::string const & url);
+    ~Client();
+private:
+    class Private;
+    Private * d;
+};
+
+}
+
+#endif

test/webfuse_provider/test_util/webfuse_server.cc

@@ -0,0 +1,274 @@
+#include "webfuse_provider/test_util/webfuse_server.hpp"
+#include "webfuse_provider/impl/util/lws_log.h"
+#include "webfuse_provider/protocol_names.h"
+
+#include <libwebsockets.h>
+#include <stdexcept>
+#include <thread>
+#include <mutex>
+#include <sstream>
+#include <queue>
+
+namespace
+{
+
+class IServer
+{
+public:
+    virtual ~IServer() = default;
+    virtual void OnConnected(lws * wsi) = 0;
+    virtual void OnConnectionClosed(lws * wsi) = 0;
+    virtual void OnMessageReceived(lws * wsi, char const * data, size_t length) = 0;
+    virtual void OnWritable(lws * wsi) = 0;
+};
+
+}
+
+extern "C"
+{
+
+static int wfp_test_utils_webfuse_server_callback(
+    lws * wsi,
+    lws_callback_reasons reason,
+    void * user,
+    void * in,
+    size_t len)
+{
+    int result = 0;
+    lws_protocols const * protocol = lws_get_protocol(wsi);
+    auto * server = reinterpret_cast<IServer*>(nullptr != protocol ? protocol->user : nullptr);
+
+    if (nullptr != server)
+    {
+        switch (reason)
+        {
+            case LWS_CALLBACK_ESTABLISHED:
+                server->OnConnected(wsi);
+                break;
+            case LWS_CALLBACK_CLOSED:
+                server->OnConnectionClosed(wsi);
+                break;
+            case LWS_CALLBACK_RECEIVE:
+                {
+                    auto * data = reinterpret_cast<char const*>(in);
+                    server->OnMessageReceived(wsi, data, len);
+                }
+                break;
+            case LWS_CALLBACK_SERVER_WRITEABLE:
+                server->OnWritable(wsi);
+                break;
+            default:
+                break;
+        }
+    }
+
+    return result;
+}
+
+}
+
+namespace webfuse_test
+{
+
+class WebfuseServer::Private: public IServer
+{
+public:
+    Private()
+    : is_connected(false)
+    , is_shutdown_requested(false)
+    , client(nullptr)
+    {
+        wfp_impl_lwslog_disable();
+
+        IServer * server = this;
+        memset(protocols, 0, sizeof(struct lws_protocols) * 2 );
+
+        protocols[0].name = WFP_PROTOCOL_NAME_ADAPTER_SERVER;
+        protocols[0].callback = &wfp_test_utils_webfuse_server_callback;
+        protocols[0].per_session_data_size = 0;
+        protocols[0].user = reinterpret_cast<void*>(server);
+
+        memset(&info, 0, sizeof(struct lws_context_creation_info));
+        info.port = 0;
+        info.mounts = NULL;
+        info.protocols = protocols;
+        info.vhost_name = "localhost";
+        info.ws_ping_pong_interval = 10;
+        info.options = LWS_SERVER_OPTION_HTTP_HEADERS_SECURITY_BEST_PRACTICES_ENFORCE;
+        info.options |= LWS_SERVER_OPTION_EXPLICIT_VHOSTS;
+
+        context = lws_create_context(&info);
+
+        struct lws_vhost * vhost = lws_create_vhost(context, &info);
+        int port = lws_get_vhost_port(vhost);
+        std::ostringstream stream;
+        stream << "ws://localhost:" << port << "/";
+        url = stream.str();
+
+        thread = std::thread(&Run, this);
+
+    }
+
+    ~Private() override
+    {
+        {
+            std::unique_lock<std::mutex> lock(mutex);
+            is_shutdown_requested = true;
+        }
+
+        lws_cancel_service(context);
+        thread.join();
+        lws_context_destroy(context);
+    }
+
+    std::string const & GetUrl() const
+    {
+        return url;
+    }
+
+    void AwaitConnection()
+    {        
+        bool is_finished = false;
+        while (!is_finished)
+        {
+            std::this_thread::yield();
+            std::unique_lock<std::mutex> lock(mutex);
+            is_finished = is_connected;
+        }
+    }
+
+    json_t * Invoke(std::string const & method, json_t * params)
+    {
+        throw std::runtime_error("not implemented");
+    }
+
+    void OnConnected(lws * wsi) override
+    {
+        std::unique_lock<std::mutex> lock(mutex);
+        client = wsi;
+    }
+
+    void OnConnectionClosed(lws * wsi) override
+    {
+        std::unique_lock<std::mutex> lock(mutex);
+        client = nullptr;
+    }
+
+    void OnMessageReceived(lws * wsi, char const * data, size_t length) override
+    {
+        json_t * message = json_loadb(data, length, 0, nullptr);
+        if (message)
+        {
+            json_t * method = json_object_get(message, "method");
+            if (json_is_string(method))
+            {
+                if (0 == strcmp("add_filesystem", json_string_value(method)))
+                {
+                    json_t * id = json_object_get(message, "id");
+
+                    json_t * response = json_object();
+                    json_t * result = json_object();
+                    json_object_set_new(result, "id", json_string("fs")); 
+                    json_object_set_new(response, "result", result);
+                    json_object_set(response, "id", id);
+
+                    char * response_text = json_dumps(response, 0);
+                    {
+                        std::unique_lock<std::mutex> lock(mutex);
+                        write_queue.push(response_text);
+                        is_connected = true;
+                    }
+                    free(response_text);
+                    json_decref(response);
+
+                    lws_callback_on_writable(wsi);
+                }
+            }
+            json_decref(message);
+        }
+    }
+
+    void OnWritable(lws * wsi) override
+    {
+        bool notify = false;
+
+        {
+            std::unique_lock<std::mutex> lock(mutex);
+
+            if (!write_queue.empty())
+            {
+                std::string const & message = write_queue.front();
+
+                unsigned char * data = new unsigned char[LWS_PRE + message.size()];
+                memcpy(&data[LWS_PRE], message.c_str(), message.size());
+                lws_write(wsi, &data[LWS_PRE], message.size(), LWS_WRITE_TEXT);
+                delete[] data;
+
+                write_queue.pop();
+                notify = !write_queue.empty();
+            }
+        }
+
+        if (notify)
+        {
+            lws_callback_on_writable(wsi);
+        }
+    }
+
+
+
+private:
+    static void Run(Private * self)
+    {
+        bool is_running = true;
+        while (is_running)
+        {
+            lws_service(self->context, 0);
+            {
+                std::unique_lock<std::mutex> lock(self->mutex);
+                is_running = !self->is_shutdown_requested;
+            }
+        }
+    }
+
+    bool is_connected;
+    bool is_shutdown_requested;
+    lws * client;
+    std::string url;
+    lws_context * context;
+    lws_protocols protocols[2];
+    lws_context_creation_info info;
+    std::thread thread;
+    std::mutex mutex;
+    std::queue<std::string> write_queue;
+};
+
+
+WebfuseServer::WebfuseServer()
+: d(new Private())
+{
+
+}
+
+WebfuseServer::~WebfuseServer()
+{
+    delete d;
+}
+
+std::string const & WebfuseServer::GetUrl()
+{
+    return d->GetUrl();
+}
+
+void WebfuseServer::AwaitConnection()
+{
+    d->AwaitConnection();
+}
+
+json_t * WebfuseServer::Invoke(std::string method, json_t * params)
+{
+    return d->Invoke(method, params);
+}
+
+
+}

test/webfuse_provider/test_util/webfuse_server.hpp

@@ -0,0 +1,27 @@
+#ifndef WFP_TEST_UTIL_WEBFUSE_SERVER_HPP
+#define WFP_TEST_UTIL_WEBFUSE_SERVER_HPP
+
+#include <string>
+#include <jansson.h>
+
+namespace webfuse_test
+{
+
+class WebfuseServer
+{
+    WebfuseServer(WebfuseServer const &) = delete;
+    WebfuseServer& operator=(WebfuseServer const &) = delete;
+public:
+    WebfuseServer();
+    ~WebfuseServer();
+    std::string const & GetUrl();
+    void AwaitConnection();
+    json_t * Invoke(std::string method, json_t * params);
+private:
+    class Private;
+    Private * d;
+};
+
+}
+
+#endif