Tests

Author: =

README.md

@@ -187,43 +187,6 @@ engine.print()
 Serializer.dump(engine, "my_workflow", "my_workflow.json")
 ```
 
-## CapioCL Web API Documentation
-
-This section describes the REST-style Web API exposed by the CapioCL Web Server.
-The server provides HTTP endpoints for configuring and querying the CapioCL engine at runtime. 
-Within the `bruno_webapi_tests` you can find several tests and examples on how to perform
-requests to the API webserver using [bruno](https://www.usebruno.com).
-
-All endpoints communicate using JSON over HTTP. To enable the webserver, users needs to explicitly start it with:
-
-```cpp
-capiocl::engine::Engine engine();
-
-// start engine with default parameters
-engine.startApiServer();
-
-// or by specifying the address and port:
-engine.startApiServer("127.0.0.1", 5520);
-```
-
-
-or equivalently in python with:
-
-```python
-engine = py_capio_cl.Engine()
-
-#start engine with default parameters
-engine.startApiServer()
-
-# or by specifying the address and port:
-engine.startApiServer("127.0.0.1", 5520)
-```
-
-By default, the webserver listens only on local connection at the following address: ```127.0.0.1:5520```. No
-authentication
-services are currently available, and as such, users should put particular care when allowing connections from external
-endpoints.
-
 ## Notes
 
 - All GET endpoints expect a JSON body containing the targeted file path.

capiocl/api.h

@@ -1,5 +1,6 @@
 #ifndef CAPIO_CL_WEBAPI_H
 #define CAPIO_CL_WEBAPI_H
+#include <atomic>
 #include <thread>
 
 #include "capiocl.hpp"
@@ -14,6 +15,9 @@ class capiocl::api::CapioClApiServer {
     /// @brief port on which the current server runs
     const configuration::CapioClConfiguration &capiocl_configuration;
 
+    /// @brief variable to tell the thread to terminate
+    std::atomic<bool> _terminate = false;
+
   public:
     /// @brief default constructor.
     CapioClApiServer(engine::Engine *engine, configuration::CapioClConfiguration &config);

capiocl/engine.h

@@ -1,8 +1,8 @@
 #ifndef CAPIO_CL_ENGINE_H
 #define CAPIO_CL_ENGINE_H
 #include <jsoncons/basic_json.hpp>
-#include <vector>
 #include <shared_mutex>
+#include <vector>
 
 #include "capiocl.hpp"
 #include "capiocl/api.h"
@@ -26,8 +26,10 @@ struct CapioCLEntry final {
     std::string commit_rule            = commitRules::ON_TERMINATION;
     ///@brief Fire rule
     std::string fire_rule              = fireRules::UPDATE;
-    long directory_children_count      = 0; ///@brief Expected number of files in directory
-    long commit_on_close_count         = 0; ///@brief Expected close count
+    ///@brief Expected number of files in directory
+    long directory_children_count      = 0;
+    ///@brief Expected close count
+    long commit_on_close_count         = 0;
     /// @brief whether to update or not directory item count
     bool enable_directory_count_update = true;
     /// @brief Store in memory or on the file system
@@ -342,7 +344,7 @@ class Engine final {
     std::vector<std::string> getConsumers(const std::filesystem::path &path) const;
 
     /// @brief Get the commit-on-close counter for a file.
-    long getCommitCloseCount(const std::filesystem::path::iterator::reference &path) const;
+    long getCommitCloseCount(const std::filesystem::path &path) const;
 
     /// @brief Get file dependencies.
     std::vector<std::filesystem::path>

src/Engine.cpp

@@ -9,16 +9,19 @@
 #include "capiocl/monitor.h"
 #include "capiocl/printer.h"
 
+/// @brief Class to implement a shared mutex lock guard
 template <typename SharedMutex> class shared_lock_guard {
   public:
+    /// @brief Constructor: acquire semaphore shared
     explicit shared_lock_guard(SharedMutex &m) : mutex_(m) { mutex_.lock_shared(); }
-
+    /// @brief Destructor: release resources
     ~shared_lock_guard() { mutex_.unlock_shared(); }
 
     shared_lock_guard(const shared_lock_guard &)            = delete;
     shared_lock_guard &operator=(const shared_lock_guard &) = delete;
 
   private:
+    /// @brief Reference to mutex
     SharedMutex &mutex_;
 };
 
@@ -734,8 +737,7 @@ void capiocl::engine::Engine::setFileDeps(const std::filesystem::path &path,
     setFileDeps(path, dependencies);
 }
 
-long capiocl::engine::Engine::getCommitCloseCount(
-    std::filesystem::path::iterator::reference &path) const {
+long capiocl::engine::Engine::getCommitCloseCount(const std::filesystem::path &path) const {
     if (path.empty()) {
         return 0;
     }

src/api.cpp

@@ -1,86 +1,99 @@
 #include <arpa/inet.h>
+#include <condition_variable>
 #include <iostream>
 #include <jsoncons/json.hpp>
+#include <mutex>
 #include <netinet/in.h>
+#include <signal.h>
 #include <sys/socket.h>
 #include <unistd.h>
 
 #include "capiocl/api.h"
 #include "capiocl/engine.h"
 #include "capiocl/printer.h"
 
+std::mutex _setupMtx;
+std::condition_variable _setupCv;
+bool thread_ready = false;
+
 /// @brief Main WebServer thread function
-void server(const std::string &address, const int port, capiocl::engine::Engine *engine) {
-    pthread_setcancelstate(PTHREAD_CANCEL_ASYNCHRONOUS, nullptr);
+void server(const std::string &address, const int port, capiocl::engine::Engine *engine,
+            std::atomic<bool> *terminate) {
+
+    constexpr int RECV_BUF_SIZE = 65535;
 
-    capiocl::printer::print(capiocl::printer::CLI_LEVEL_INFO,
-                            "Starting API server @ " + address + ":" + std::to_string(port));
+    const auto &wf_name = engine->getWorkflowName();
 
     int fd = socket(AF_INET, SOCK_DGRAM, 0);
-    if (fd < 0) {
-        perror("Socket creation failed");
-        return;
-    }
 
     int reuse = 1;
-    if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)) < 0) {
-        perror("Setting SO_REUSEADDR failed");
-        close(fd);
-        return;
-    }
+    setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
 
     sockaddr_in localAddr{};
     localAddr.sin_family      = AF_INET;
     localAddr.sin_port        = htons(port);
     localAddr.sin_addr.s_addr = INADDR_ANY;
 
-    if (bind(fd, reinterpret_cast<sockaddr *>(&localAddr), sizeof(localAddr)) < 0) {
-        perror("Bind failed");
-        close(fd);
-        return;
-    }
+    bind(fd, reinterpret_cast<sockaddr *>(&localAddr), sizeof(localAddr));
 
     ip_mreq group{};
     group.imr_multiaddr.s_addr = inet_addr(address.c_str());
     group.imr_interface.s_addr = INADDR_ANY;
-    if (setsockopt(fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &group, sizeof(group)) < 0) {
-        perror("Adding multicast group failed");
-        close(fd);
-        return;
-    }
+    setsockopt(fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &group, sizeof(group));
 
-    char buffer[65535] = {0};
+    char buffer[RECV_BUF_SIZE] = {0};
     sockaddr_in srcAddr{};
     socklen_t addrlen = sizeof(srcAddr);
 
     // timeout of 1 second for termination
     timeval tv{};
-    tv.tv_sec  = 1;
-    tv.tv_usec = 0;
+    tv.tv_sec  = 0;
+    tv.tv_usec = 500;
     setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
 
-    while (true) {
-        ssize_t nbytes = recvfrom(fd, buffer, sizeof(buffer) - 1, 0,
-                                  reinterpret_cast<struct sockaddr *>(&srcAddr), &addrlen);
-        if (nbytes < 0) {
+    thread_ready = true;
+    _setupCv.notify_all();
+
+    while (!*terminate) {
+        // GCOVR_EXCL_START
+        ssize_t n = recvfrom(fd, buffer, RECV_BUF_SIZE - 1, 0,
+                             reinterpret_cast<sockaddr *>(&srcAddr), &addrlen);
+        // GCOVR_EXCL_STOP
+
+        if (n <= 0) {
             continue;
         }
 
-        buffer[nbytes] = '\0';
+        buffer[n] = '\0';
 
         try {
-            auto data          = jsoncons::json::parse(buffer);
-            auto path          = data["path"].as_string();
-            auto workflow_name = data["workflow_name"].as_string();
-            auto rule = capiocl::engine::CapioCLEntry::fromJson(data["CapioClEntry"].as_string());
+            auto data = jsoncons::json::parse(buffer); // GCOVR_EXCL_LINE
+            const auto path =
+                data.get_value_or<std::string, std::string>("path", ""); // GCOVR_EXCL_LINE
+            if (path.empty()) {
+                continue;
+            }
+            const auto workflow_name =
+                data.get_value_or<std::string, std::string>("workflow_name", ""); // GCOVR_EXCL_LINE
+            if (workflow_name.empty()) {
+                continue;
+            }
+            const auto jsonEntry =
+                data.get_value_or<std::string, std::string>("CapioClEntry", ""); // GCOVR_EXCL_LINE
+            if (jsonEntry.empty()) {
+                continue;
+            }
+            auto rule = capiocl::engine::CapioCLEntry::fromJson(jsonEntry);
 
-            if (workflow_name == engine->getWorkflowName()) {
+            if (workflow_name == wf_name) {
                 engine->add(path, rule);
+            } else {
+                continue;
             }
 
         } catch (const jsoncons::json_exception &e) {
             capiocl::printer::print(capiocl::printer::CLI_LEVEL_ERROR,
-                                    "JSON Parse Error: " + std::string(e.what()));
+                                    "APIServer: Received invalid json: " + std::string(e.what()));
         }
     }
 
@@ -92,15 +105,28 @@ capiocl::api::CapioClApiServer::CapioClApiServer(engine::Engine *engine,
     : capiocl_configuration(config) {
 
     std::string address;
-    config.getParameter("dynamic_api.ip", &address);
-
     int port;
-    config.getParameter("dynamic_api.port", &port);
+    try {
+        config.getParameter("dynamic_api.ip", &address); // GCOVR_EXCL_LINE
+    } catch (...) {
+        address = configuration::defaults::DEFAULT_API_MULTICAST_IP.v;
+    }
+
+    try {
+        config.getParameter("dynamic_api.port", &port); // GCOVR_EXCL_LINE
+    } catch (...) {
+        port = std::stoi(configuration::defaults::DEFAULT_API_MULTICAST_PORT.v);
+    }
+
+    _webApiThread = std::thread(server, address, port, engine, &_terminate);
+
+    std::unique_lock lock(_setupMtx);
+    _setupCv.wait(lock, [] { return thread_ready; });
 
-    _webApiThread = std::thread(server, address, port, engine);
+    printer::print(printer::CLI_LEVEL_INFO, "API server @ " + address + ":" + std::to_string(port));
 }
 
 capiocl::api::CapioClApiServer::~CapioClApiServer() {
-    pthread_cancel(_webApiThread.native_handle());
+    _terminate = true;
     _webApiThread.join();
 }

tests/cpp/test_apis.hpp

@@ -15,36 +15,34 @@
 
 #include <string>
 
-bool sendMulticast(const std::string &message, const std::string &multicast_ip, int port) {
-    int fd = socket(AF_INET, SOCK_DGRAM, 0);
+inline bool sendMulticast(const std::string &message, const std::string &multicast_ip, int port) {
+    const int fd = socket(AF_INET, SOCK_DGRAM, 0);
     if (fd < 0) {
         perror("socket");
         return false;
     }
 
-    u_char ttl = 1;
+    const u_char ttl = 3;
     if (setsockopt(fd, IPPROTO_IP, IP_MULTICAST_TTL, &ttl, sizeof(ttl)) < 0) {
         perror("setsockopt (TTL)");
         close(fd);
         return false;
     }
 
-    sockaddr_in addr;
+    sockaddr_in addr{};
     addr.sin_family      = AF_INET;
     addr.sin_addr.s_addr = inet_addr(multicast_ip.c_str());
     addr.sin_port        = htons(port);
 
-    ssize_t nbytes =
-        sendto(fd, message.c_str(), message.size(), 0, (struct sockaddr *) &addr, sizeof(addr));
+    ssize_t nbytes = sendto(fd, message.c_str(), message.size(), 0,
+                            reinterpret_cast<struct sockaddr *>(&addr), sizeof(addr));
 
     if (nbytes < 0) {
         perror("sendto");
         close(fd);
         return false;
     }
 
-    std::cout << "Sent " << nbytes << " bytes to " << multicast_ip << ":" << port << std::endl;
-
     close(fd);
     return true;
 }
@@ -93,11 +91,41 @@ TEST(WEBSERVER_SUITE_NAME, TestWebServerAPIS) {
     engine.startApiServer();
     EXPECT_FALSE(engine.contains("file.txt"));
 
+    EXPECT_TRUE(
+        sendMulticast(R"({"path" : "file.txt", "workflow_name" : "notMyWorkflow"})",
+                      capiocl::configuration::defaults::DEFAULT_API_MULTICAST_IP.v,
+                      stoi(capiocl::configuration::defaults::DEFAULT_API_MULTICAST_PORT.v)));
+
+    EXPECT_TRUE(
+        sendMulticast("notAValidJson", capiocl::configuration::defaults::DEFAULT_API_MULTICAST_IP.v,
+                      stoi(capiocl::configuration::defaults::DEFAULT_API_MULTICAST_PORT.v)));
+
+    EXPECT_TRUE(
+        sendMulticast(R"({"workflow_name" : "notMyWorkflow", "CapioClEntry":{}})",
+                      capiocl::configuration::defaults::DEFAULT_API_MULTICAST_IP.v,
+                      stoi(capiocl::configuration::defaults::DEFAULT_API_MULTICAST_PORT.v)));
+
+    EXPECT_TRUE(
+        sendMulticast(R"({"path" : "file.txt", "CapioClEntry":{}})",
+                      capiocl::configuration::defaults::DEFAULT_API_MULTICAST_IP.v,
+                      stoi(capiocl::configuration::defaults::DEFAULT_API_MULTICAST_PORT.v)));
+
+    EXPECT_TRUE(
+        sendMulticast(R"({"path" : "file.txt", "workflow_name" : "notMyWorkflow"})",
+                      capiocl::configuration::defaults::DEFAULT_API_MULTICAST_IP.v,
+                      stoi(capiocl::configuration::defaults::DEFAULT_API_MULTICAST_PORT.v)));
+
     capiocl::engine::CapioCLEntry entry;
     entry.commit_rule           = "on_file";
     entry.commit_on_close_count = 10;
     entry.fire_rule             = "no_update";
 
+    EXPECT_TRUE(sendMulticast(
+        R"({ "path" : "file.txt","workflow_name" : "notMyWorkflow", "CapioClEntry":)" +
+            entry.toJson() + "}",
+        capiocl::configuration::defaults::DEFAULT_API_MULTICAST_IP.v,
+        stoi(capiocl::configuration::defaults::DEFAULT_API_MULTICAST_PORT.v)));
+
     std::string request = R"({ "path" : "file.txt","workflow_name" : ")";
     request += capiocl::CAPIO_CL_DEFAULT_WF_NAME;
     request += R"(", "CapioClEntry":)" + entry.toJson() + "}";