Add test coverage for ConnectStream

test/CMakeLists.txt

@@ -28,6 +28,7 @@ if(BUILD_TESTING AND TARGET CapnProto::kj-test)
     mp/test/foo.h
     mp/test/listen_tests.cpp
     mp/test/spawn_tests.cpp
+    mp/test/connect_tests.cpp
     mp/test/test.cpp
   )
   include(${PROJECT_SOURCE_DIR}/cmake/TargetCapnpSources.cmake)

test/mp/test/connect_tests.cpp

@@ -0,0 +1,182 @@
+// Copyright (c) The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://opensource.org.
+#include "unixlistener.cpp"
+#include <kj/async.h>
+#include <kj/common.h>
+#include <kj/debug.h>
+#include <kj/memory.h>
+#include <kj/test.h>
+#include <mp/proxy-io.h>
+#include <mp/test/foo.capnp.h>
+#include <mp/test/foo.capnp.proxy.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <unistd.h>
+
+#include <cstring> // IWYU pragma: keep
+#include <functional>
+#include <future>
+#include <memory>
+#include <stdexcept>
+#include <string>
+#include <string_view>
+#include <thread>
+
+namespace mp {
+namespace test {
+namespace {
+
+class TestSetup
+{
+public:
+    int client_fd;
+    int server_fd;
+
+    mp::EventLoop* loop;
+    //! Thread variable should be after other struct members so the thread does
+    //! not start until the other members are initialized.
+    std::thread loop_thread;
+
+    TestSetup() : TestSetup([](int fds[2]) {
+                      KJ_REQUIRE(socketpair(AF_UNIX, SOCK_STREAM, 0, fds) != -1);
+                  }) {}
+
+    TestSetup(const std::function<void(int[2])>& init_sockets)
+    {
+        std::promise<mp::EventLoop*> loop_promise;
+        loop_thread = std::thread([&] {
+            mp::EventLoop loop("mptest-connect", [](mp::LogMessage log) {
+                if (log.level == mp::Log::Raise)
+                    throw std::runtime_error(log.message);
+            });
+            loop_promise.set_value(&loop);
+            loop.loop();
+        });
+        loop = loop_promise.get_future().get();
+
+        // Initalize and store sockets
+        int fds[2] = {-1, -1};
+        init_sockets(fds);
+
+        client_fd = fds[0];
+        server_fd = fds[1];
+    }
+
+    ~TestSetup()
+    {
+        loop_thread.join();
+    }
+};
+
+KJ_TEST("Successfully connects to a socket serving a valid init interface")
+{
+    TestSetup setup;
+
+    std::thread server_thread([&setup]() {
+        mp::EventLoop server_loop("mptest-valid-server", [](mp::LogMessage log) {
+            if (log.level == mp::Log::Raise)
+                throw std::runtime_error(log.message);
+        });
+        std::unique_ptr<FooImplementation> init = std::make_unique<FooImplementation>();
+        ServeStream<messages::FooInterface>(server_loop, setup.server_fd, *init);
+        server_loop.loop();
+    });
+
+    auto init = ConnectStream<messages::FooInterface>(*setup.loop, setup.client_fd);
+    init->initThreadMap();
+
+    init.reset();
+    server_thread.join();
+    KJ_EXPECT(true);
+}
+
+KJ_TEST("Passing a disconnected socket will throw")
+{
+    TestSetup setup;
+
+    close(setup.server_fd);
+
+    try {
+        auto init = ConnectStream<messages::FooInterface>(*setup.loop, setup.client_fd);
+        init->initThreadMap();
+        KJ_EXPECT(false);
+    } catch (const std::runtime_error& e) {
+        std::string_view reason = e.what();
+
+        KJ_EXPECT(reason == "IPC client method called after disconnect.");
+    }
+}
+
+KJ_TEST("Passing a connected socket that disconnects on recv will throw")
+{
+    TestSetup setup;
+
+    std::thread server_thread([&setup]() {
+        char buf[128];
+
+        ssize_t bytes_received =
+            recv(setup.server_fd, buf, sizeof(buf), 0);
+
+        if (bytes_received > 0) {
+            close(setup.server_fd);
+        }
+    });
+
+    try {
+        auto init = ConnectStream<messages::FooInterface>(*setup.loop, setup.client_fd);
+        init->initThreadMap();
+
+        if (server_thread.joinable()) server_thread.join();
+        KJ_EXPECT(false);
+    } catch (const std::runtime_error& e) {
+        if (server_thread.joinable()) server_thread.join();
+
+        std::string_view reason = e.what();
+        KJ_EXPECT(reason == "IPC client method called after disconnect." || reason == "IPC client method call interrupted by disconnect.");
+    }
+}
+
+
+KJ_TEST("Passing a custom listening socket that disconnects on recv will throw")
+{
+    UnixListener listener;
+
+    TestSetup setup([&listener](int fds[2]) {
+        fds[0] = listener.MakeConnectedSocket(); // client_fd
+        fds[1] = listener.release();             // server_fd
+    });
+
+    std::thread server_thread([&setup]() {
+        char buf[128];
+
+        int connection_fd = accept(setup.server_fd, nullptr, nullptr);
+
+        if (connection_fd >= 0) {
+            ssize_t bytes_received =
+                recv(connection_fd, buf, sizeof(buf), 0);
+
+            if (bytes_received > 0) {
+                close(connection_fd);
+            }
+        }
+        close(setup.server_fd);
+    });
+
+    try {
+        auto init = ConnectStream<messages::FooInterface>(*setup.loop, setup.client_fd);
+        init->initThreadMap();
+
+        if (server_thread.joinable()) server_thread.join();
+        KJ_EXPECT(false);
+    } catch (const std::runtime_error& e) {
+        if (server_thread.joinable()) server_thread.join();
+
+        std::string_view reason = e.what();
+        KJ_EXPECT(reason == "IPC client method called after disconnect." || reason == "IPC client method call interrupted by disconnect.");
+    }
+}
+
+} // namespace
+} // namespace test
+} // namespace mp

test/mp/test/listen_tests.cpp

@@ -2,99 +2,35 @@
 // Distributed under the MIT software license, see the accompanying
 // file COPYING or http://www.opensource.org/licenses/mit-license.php.
 
-#include <cassert>
-#include <filesystem>
+#include "unixlistener.cpp"
+#include <kj/async.h>
+#include <kj/common.h>
+#include <kj/debug.h>
+#include <kj/memory.h>
+#include <kj/test.h>
+#include <mp/proxy-io.h>
+#include <mp/proxy.h>
 #include <mp/test/foo.capnp.h>
 #include <mp/test/foo.capnp.proxy.h>
+#include <mp/util.h>
 
 #include <chrono>
 #include <condition_variable>
-#include <cstdlib>
 #include <cstring>
-#include <future>
 #include <functional>
-#include <kj/async.h>
-#include <kj/common.h>
-#include <kj/debug.h>
-#include <kj/memory.h>
-#include <kj/test.h>
+#include <future>
 #include <memory>
-#include <mp/proxy.h>
-#include <mp/proxy-io.h>
-#include <mp/util.h>
-#include <ratio> // IWYU pragma: keep
 #include <optional>
+#include <ratio> // IWYU pragma: keep
 #include <stdexcept>
-#include <string>
-#include <sys/socket.h>
-#include <sys/un.h>
 #include <thread>
-#include <unistd.h>
 
 namespace mp {
 namespace test {
 namespace {
 
 constexpr auto FAILURE_TIMEOUT = std::chrono::seconds{30};
 
-//! Owns a temporary Unix-domain listening socket used by ListenSetup. Tests call
-//! Connect() to create client socket FDs and release() to transfer the listening
-//! FD to ListenConnections().
-class UnixListener
-{
-public:
-    UnixListener()
-    {
-        std::string dir_template = (std::filesystem::temp_directory_path() / "mptest-listener-XXXXXX").string();
-        char* dir = mkdtemp(dir_template.data());
-        KJ_REQUIRE(dir != nullptr);
-        m_dir = dir;
-        m_path = m_dir + "/socket";
-
-        m_fd = socket(AF_UNIX, SOCK_STREAM, 0);
-        KJ_REQUIRE(m_fd >= 0);
-
-        sockaddr_un addr{};
-        addr.sun_family = AF_UNIX;
-        KJ_REQUIRE(m_path.size() < sizeof(addr.sun_path));
-        std::strncpy(addr.sun_path, m_path.c_str(), sizeof(addr.sun_path) - 1);
-        KJ_REQUIRE(bind(m_fd, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) == 0);
-        KJ_REQUIRE(listen(m_fd, SOMAXCONN) == 0);
-    }
-
-    ~UnixListener()
-    {
-        if (m_fd >= 0) close(m_fd);
-        if (!m_path.empty()) unlink(m_path.c_str());
-        if (!m_dir.empty()) rmdir(m_dir.c_str());
-    }
-
-    int release()
-    {
-        assert(m_fd >= 0);
-        int fd = m_fd;
-        m_fd = -1;
-        return fd;
-    }
-
-    int MakeConnectedSocket() const
-    {
-        int fd = socket(AF_UNIX, SOCK_STREAM, 0);
-        KJ_REQUIRE(fd >= 0);
-
-        sockaddr_un addr{};
-        addr.sun_family = AF_UNIX;
-        KJ_REQUIRE(m_path.size() < sizeof(addr.sun_path));
-        std::strncpy(addr.sun_path, m_path.c_str(), sizeof(addr.sun_path) - 1);
-        KJ_REQUIRE(connect(fd, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) == 0);
-        return fd;
-    }
-
-private:
-    int m_fd{-1};
-    std::string m_dir;
-    std::string m_path;
-};
 
 //! Runs a client EventLoop on its own thread and connects one socket FD to the
 //! server. The constructed ProxyClient can be used by the test thread to make

test/mp/test/unixlistener.cpp

@@ -0,0 +1,71 @@
+// Copyright (c) The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+#include <kj/test.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <unistd.h>
+
+#include <cassert>
+#include <cstring>
+#include <filesystem>
+
+//! Owns a temporary Unix-domain listening socket used by ListenSetup. Tests call
+//! Connect() to create client socket FDs and release() to transfer the listening
+//! FD to ListenConnections().
+class UnixListener
+{
+public:
+    UnixListener()
+    {
+        std::string dir_template = (std::filesystem::temp_directory_path() / "mptest-listener-XXXXXX").string();
+        char* dir = mkdtemp(dir_template.data());
+        KJ_REQUIRE(dir != nullptr);
+        m_dir = dir;
+        m_path = m_dir + "/socket";
+
+        m_fd = socket(AF_UNIX, SOCK_STREAM, 0);
+        KJ_REQUIRE(m_fd >= 0);
+
+        sockaddr_un addr{};
+        addr.sun_family = AF_UNIX;
+        KJ_REQUIRE(m_path.size() < sizeof(addr.sun_path));
+        std::strncpy(addr.sun_path, m_path.c_str(), sizeof(addr.sun_path) - 1);
+        KJ_REQUIRE(bind(m_fd, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) == 0);
+        KJ_REQUIRE(listen(m_fd, SOMAXCONN) == 0);
+    }
+
+    ~UnixListener()
+    {
+        if (m_fd >= 0) close(m_fd);
+        if (!m_path.empty()) unlink(m_path.c_str());
+        if (!m_dir.empty()) rmdir(m_dir.c_str());
+    }
+
+    int release()
+    {
+        assert(m_fd >= 0);
+        int fd = m_fd;
+        m_fd = -1;
+        return fd;
+    }
+
+    int MakeConnectedSocket() const
+    {
+        int fd = socket(AF_UNIX, SOCK_STREAM, 0);
+        KJ_REQUIRE(fd >= 0);
+
+        sockaddr_un addr{};
+        addr.sun_family = AF_UNIX;
+        KJ_REQUIRE(m_path.size() < sizeof(addr.sun_path));
+        std::strncpy(addr.sun_path, m_path.c_str(), sizeof(addr.sun_path) - 1);
+        KJ_REQUIRE(connect(fd, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) == 0);
+        return fd;
+    }
+
+private:
+    int m_fd{-1};
+    std::string m_dir;
+    std::string m_path;
+};