// Copyright (c) 2014-2026 Savoir-faire Linux Inc. // SPDX-License-Identifier: MIT #include "dhtproxytester.h" // std #include #include #include #include using namespace std::chrono_literals; namespace test { CPPUNIT_TEST_SUITE_REGISTRATION(DhtProxyTester); void DhtProxyTester::setUp() { clientConfig.dht_config.node_config.max_peer_req_per_sec = -1; clientConfig.dht_config.node_config.max_req_per_sec = -1; nodePeer.run(0, clientConfig); nodeProxy = std::make_shared(); nodeProxy->run(0, clientConfig); auto bound = nodePeer.getBound(); if (bound.isUnspecified()) bound.setLoopback(); nodeProxy->bootstrap(bound); // auto serverCAIdentity = dht::crypto::generateEcIdentity("DHT Node CA"); uint16_t port = 1024 + (std::rand() % (65535 - 1024)); dht::ProxyServerConfig serverConfig; // serverConfig.identity = dht::crypto::generateIdentity("DHT Node", serverCAIdentity); serverConfig.port = port; // serverConfig.pushServer = "127.0.0.1:8090"; serverProxy = std::make_unique(nodeProxy, serverConfig); /*clientConfig.server_ca = serverCAIdentity.second; clientConfig.client_identity = dht::crypto::generateIdentity("DhtProxyTester"); clientConfig.push_node_id = "dhtnode";*/ clientConfig.proxy_server = "http://127.0.0.1:" + std::to_string(port); //"https://127.0.0.1:8080"; } void DhtProxyTester::tearDown() { nodePeer.join(); nodeClient.join(); bool done = false; std::condition_variable cv; std::mutex cv_m; nodeProxy->shutdown([&] { std::lock_guard lk(cv_m); done = true; cv.notify_all(); }); std::unique_lock lk(cv_m); CPPUNIT_ASSERT(cv.wait_for(lk, 15s, [&] { return done; })); serverProxy.reset(); nodeProxy.reset(); } void DhtProxyTester::testGetPut() { nodeClient.run(0, clientConfig); bool done = false; std::condition_variable cv; std::mutex cv_m; auto key = dht::InfoHash::get("GLaDOs"); dht::Value val {"Hey! It's been a long time. How have you been?"}; auto val_data = val.data; { nodePeer.put(key, std::move(val), [&](bool) { std::lock_guard lk(cv_m); done = true; cv.notify_all(); }); std::unique_lock lk(cv_m); CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] { return done; })); } auto vals = nodeClient.get(key).get(); CPPUNIT_ASSERT(not vals.empty()); CPPUNIT_ASSERT(vals.front()->data == val_data); } void DhtProxyTester::testListen() { nodeClient.run(0, clientConfig); std::condition_variable cv; std::mutex cv_m; std::unique_lock lk(cv_m); auto key = dht::InfoHash::get("GLaDOs"); bool done = false; // If a peer send a value, the listen operation from the client // should retrieve this value dht::Value firstVal {"Hey! It's been a long time. How have you been?"}; auto firstVal_data = firstVal.data; nodePeer.put(key, std::move(firstVal), [&](bool ok) { CPPUNIT_ASSERT(ok); std::lock_guard lk(cv_m); done = true; cv.notify_all(); }); CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] { return done; })); done = false; std::vector values; auto token = nodeClient.listen(key, [&](const std::vector>& v, bool expired) { if (not expired) { std::lock_guard lk(cv_m); for (const auto& value : v) values.emplace_back(value->data); done = true; cv.notify_all(); } return true; }); CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] { return done; })); done = false; // Here values should contains 1 values CPPUNIT_ASSERT_EQUAL(static_cast(values.size()), 1); CPPUNIT_ASSERT(values.front() == firstVal_data); // And the listen should retrieve futures values // All values dht::Value secondVal {"You're a monster"}; auto secondVal_data = secondVal.data; nodePeer.put(key, std::move(secondVal)); CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] { return done; })); nodeClient.cancelListen(key, std::move(token)); // Here values should contains 2 values CPPUNIT_ASSERT_EQUAL(static_cast(values.size()), 2); CPPUNIT_ASSERT(values.back() == secondVal_data); } void DhtProxyTester::testResubscribeGetValues() { clientConfig.push_token = "atlas"; nodeClient.run(0, clientConfig); bool done = false; std::condition_variable cv; std::mutex cv_m; std::unique_lock lk(cv_m); auto key = dht::InfoHash::get("GLaDOs"); // If a peer sent a value, the listen operation from the client // should retrieve this value dht::Value firstVal {"Hey! It's been a long time. How have you been?"}; auto firstVal_data = firstVal.data; nodePeer.put(key, std::move(firstVal), [&](bool ok) { std::lock_guard lk(cv_m); CPPUNIT_ASSERT(ok); done = true; cv.notify_all(); }); CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] { return done; })); done = false; // Send a first subscribe, the value is sent via a push notification // So ignore values here. nodeClient.listen(key, [&](const std::vector>&, bool) { return true; }); cv.wait_for(lk, std::chrono::seconds(1)); // Reboot node (to avoid cache) nodeClient.join(); clientConfig.push_token = ""; nodeClient.run(0, clientConfig); // For the second subscribe, the proxy will return the value in the body std::vector> values; auto ftoken = nodeClient.listen(key, [&](const std::vector>& v, bool expired) { if (not expired) { std::lock_guard lk(cv_m); values.insert(values.end(), v.begin(), v.end()); done = true; cv.notify_all(); } return true; }); CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] { return done; })); auto token = ftoken.get(); CPPUNIT_ASSERT(token); nodeClient.cancelListen(key, token); // Here values should still contains 1 values CPPUNIT_ASSERT_EQUAL((size_t) 1u, values.size()); CPPUNIT_ASSERT(firstVal_data == values.front()->data); } void DhtProxyTester::testPutGet40KChars() { nodeClient.run(0, clientConfig); constexpr size_t N = 40000; // Arrange auto key = dht::InfoHash::get("testPutGet40KChars"); std::vector> values; std::vector mtu; mtu.reserve(N); for (size_t i = 0; i < N; i++) mtu.emplace_back((i % 2) ? 'T' : 'M'); std::condition_variable cv; std::mutex cv_m; std::unique_lock lk(cv_m); bool done_put = false; bool done_get = false; // Act dht::Value val {mtu}; nodePeer.put(key, std::move(val), [&](bool ok) { std::lock_guard lk(cv_m); done_put = ok; cv.notify_all(); }); CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] { return done_put; })); nodeClient.get( key, [&](const std::vector>& vals) { values.insert(values.end(), vals.begin(), vals.end()); return true; }, [&](bool ok) { std::lock_guard lk(cv_m); done_get = ok; cv.notify_all(); }); CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] { return done_get; })); // Assert CPPUNIT_ASSERT_EQUAL((size_t) 1u, values.size()); for (const auto& value : values) CPPUNIT_ASSERT(value->data == mtu); } void DhtProxyTester::testFuzzy() { constexpr size_t N = 40000; // Arrange auto key = dht::InfoHash::get("testFuzzy"); std::vector> values; std::vector mtu; mtu.reserve(N); for (size_t i = 0; i < N; i++) mtu.emplace_back((i % 2) ? 'T' : 'M'); // Act for (size_t i = 0; i < 100; i++) { auto nodeTest = std::make_shared(); nodeTest->run(0, clientConfig); nodeTest->put(key, dht::Value(mtu), [&](bool ok) { CPPUNIT_ASSERT(ok); }); nodeTest->get( key, [&](const std::vector>& vals) { values.insert(values.end(), vals.begin(), vals.end()); return true; }, [&](bool ok) { CPPUNIT_ASSERT(ok); }); std::this_thread::sleep_for(5ms); } // Assert for (const auto& value : values) CPPUNIT_ASSERT(value->data == mtu); } void DhtProxyTester::testShutdownStop() { constexpr size_t N = 40000; constexpr unsigned C = 100; // Arrange auto key = dht::InfoHash::get("testShutdownStop"); std::vector> values; std::vector mtu; mtu.reserve(N); for (size_t i = 0; i < N; i++) mtu.emplace_back((i % 2) ? 'T' : 'M'); std::atomic_uint callback_count {0}; // Act for (size_t i = 0; i < C; i++) { auto nodeTest = std::make_shared(); nodeTest->run(0, clientConfig); nodeTest->put(key, dht::Value(mtu), [&](bool /*ok*/) { callback_count++; }); nodeTest->get( key, [&](const std::vector>& vals) { values.insert(values.end(), vals.begin(), vals.end()); return true; }, [&](bool /*ok*/) { callback_count++; }); bool done = false; std::condition_variable cv; std::mutex cv_m; nodeTest->shutdown( [&] { std::lock_guard lk(cv_m); done = true; cv.notify_all(); }, true); std::unique_lock lk(cv_m); CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] { return done; })); } CPPUNIT_ASSERT_EQUAL(2 * C, callback_count.load()); } } // namespace test