diff --git a/packages/streamr-trackerless-network/CMakeLists.txt b/packages/streamr-trackerless-network/CMakeLists.txt index e006d2b7..9351d434 100644 --- a/packages/streamr-trackerless-network/CMakeLists.txt +++ b/packages/streamr-trackerless-network/CMakeLists.txt @@ -43,6 +43,8 @@ find_package(Protobuf REQUIRED) find_package(Boost CONFIG REQUIRED) find_package(magic_enum CONFIG REQUIRED) find_package(folly REQUIRED) +# streamPartIdToDataKey derives DHT data keys with SHA1 (OpenSSL EVP). +find_package(OpenSSL REQUIRED) if(NOT TARGET Boost::uuid) add_library(Boost::uuid INTERFACE IMPORTED) @@ -95,6 +97,7 @@ target_link_libraries(streamr-trackerless-network PUBLIC Boost::uuid PUBLIC magic_enum::magic_enum PUBLIC Folly::folly + PUBLIC OpenSSL::Crypto ) export(TARGETS streamr-trackerless-network @@ -113,6 +116,7 @@ file(WRITE "${CMAKE_BINARY_DIR}/streamr-trackerless-network-config.cmake" "find_package(Boost CONFIG REQUIRED)\n" "find_package(magic_enum CONFIG REQUIRED)\n" "find_package(folly REQUIRED)\n" + "find_package(OpenSSL REQUIRED)\n" "if(NOT TARGET Boost::uuid)\n" " add_library(Boost::uuid INTERFACE IMPORTED)\n" " set_target_properties(Boost::uuid PROPERTIES INTERFACE_INCLUDE_DIRECTORIES ${Boost_INCLUDE_DIRS})\n" @@ -162,6 +166,9 @@ if(NOT IOS AND STREAMR_MODULES_SUPPORTED) test/unit/ContentDeliveryRpcRemoteTest.cpp test/unit/ContentDeliveryRpcLocalTest.cpp test/unit/DuplicateMessageDetectorTest.cpp + test/unit/NumberPairTest.cpp + test/unit/StreamPartIdToDataKeyTest.cpp + test/unit/TemporaryConnectionRpcLocalTest.cpp ) target_include_directories(streamr-trackerless-network-test-unit @@ -177,6 +184,11 @@ if(NOT IOS AND STREAMR_MODULES_SUPPORTED) PUBLIC streamr-trackerless-network PUBLIC streamr-trackerless-network-test-utils PUBLIC streamr::streamr-logger + # The C1 tests import streamr.dht.* and streamr.protorpc.* + # modules directly; imported modules are usable only by DIRECT + # consumers, so the owning targets are linked here. + PUBLIC streamr::streamr-dht + PUBLIC streamr::streamr-proto-rpc PUBLIC GTest::gtest streamr-trackerless-network-test-main PUBLIC Folly::folly diff --git a/packages/streamr-trackerless-network/lint.sh b/packages/streamr-trackerless-network/lint.sh index 2f658722..f5ab9cfc 100755 --- a/packages/streamr-trackerless-network/lint.sh +++ b/packages/streamr-trackerless-network/lint.sh @@ -21,10 +21,22 @@ echo "Running clangd-tidy on $SRCFILES" ../../run-clang-format.py $SRCFILES fi +# Two phase-C1 test files are excluded from clangd-tidy (owner-approved +# selective disabling pattern, see packages/streamr-dht/lint.sh for the +# full catalog of both false-positive classes): +# ContentDeliveryRpcRemoteTest.cpp trips the global-namespace +# duplicate-declaration merge failure (NetworkRpc/ProtoRpc types arrive +# both textually and through imported BMIs; member calls like +# requestid() are then diagnosed ambiguous), and +# TemporaryConnectionRpcLocalTest.cpp trips the std-type unification +# false positive (std::string-vs-std::string mismatch on its own +# locals). The compiler builds and runs both (unit suite green on every +# platform); clang-format still checks them. TESTFILES=$(find test -type f \( -name "*.hpp" -o -name "*.cpp" \) -not -path '*/ts-integration/*' | sort | uniq | tr '\n' ' ') -echo "Running clangd-tidy on $TESTFILES" +TIDY_TESTFILES=$(echo "$TESTFILES" | tr ' ' '\n' | grep -v 'test/unit/ContentDeliveryRpcRemoteTest.cpp' | grep -v 'test/unit/TemporaryConnectionRpcLocalTest.cpp' | tr '\n' ' ') +echo "Running clangd-tidy on $TIDY_TESTFILES" -clangd-tidy -p "$COMPILE_DB" $TESTFILES +clangd-tidy -p "$COMPILE_DB" $TIDY_TESTFILES < /dev/null echo "Running clang-format --dry-run on $TESTFILES" ../../run-clang-format.py $TESTFILES diff --git a/packages/streamr-trackerless-network/modules/logic/NodeList.cppm b/packages/streamr-trackerless-network/modules/logic/NodeList.cppm index 6f6d661a..09db4717 100644 --- a/packages/streamr-trackerless-network/modules/logic/NodeList.cppm +++ b/packages/streamr-trackerless-network/modules/logic/NodeList.cppm @@ -118,9 +118,14 @@ public: std::ranges::to>(); } + // TS getNode() returns undefined for unknown ids; map::at would + // throw instead, so look up with find. [[nodiscard]] std::optional> get( const DhtAddress& id) const { - return this->nodes.at(id); + if (const auto it = this->nodes.find(id); it != this->nodes.end()) { + return it->second; + } + return std::nullopt; } [[nodiscard]] size_t size( diff --git a/packages/streamr-trackerless-network/modules/logic/streamPartIdToDataKey.cppm b/packages/streamr-trackerless-network/modules/logic/streamPartIdToDataKey.cppm new file mode 100644 index 00000000..c38756fd --- /dev/null +++ b/packages/streamr-trackerless-network/modules/logic/streamPartIdToDataKey.cppm @@ -0,0 +1,40 @@ +// Module streamr.trackerlessnetwork.streamPartIdToDataKey +// Ported from streamPartIdToDataKey() in packages/trackerless-network/ +// src/ContentDeliveryManager.ts (v103.8.0-rc.3): stream-part entry +// points are stored in the DHT under SHA1(streamPartId). The key must +// match the TS derivation byte for byte, or C++ and TS nodes would +// store/fetch the same stream part's entry points under different DHT +// keys (wire compatibility — TypeScript wins). +module; + +#include +#include + +export module streamr.trackerlessnetwork.streamPartIdToDataKey; + +import streamr.dht.Identifiers; +import streamr.utils.StreamPartID; + +using streamr::dht::DhtAddress; +using streamr::dht::DhtAddressRaw; +using streamr::dht::Identifiers; +using streamr::utils::StreamPartID; + +export namespace streamr::trackerlessnetwork { + +inline DhtAddress streamPartIdToDataKey(const StreamPartID& streamPartId) { + unsigned char digest[EVP_MAX_MD_SIZE]; // NOLINT + unsigned int digestLength = 0; + EVP_Digest( + streamPartId.data(), + streamPartId.size(), + digest, // NOLINT + &digestLength, + EVP_sha1(), + nullptr); + return Identifiers::getDhtAddressFromRaw( + DhtAddressRaw{std::string( + reinterpret_cast(digest), digestLength)}); // NOLINT +} + +} // namespace streamr::trackerlessnetwork diff --git a/packages/streamr-trackerless-network/modules/logic/temporary-connection/TemporaryConnectionRpcLocal.cppm b/packages/streamr-trackerless-network/modules/logic/temporary-connection/TemporaryConnectionRpcLocal.cppm new file mode 100644 index 00000000..a3850476 --- /dev/null +++ b/packages/streamr-trackerless-network/modules/logic/temporary-connection/TemporaryConnectionRpcLocal.cppm @@ -0,0 +1,115 @@ +// Module streamr.trackerlessnetwork.TemporaryConnectionRpcLocal +// Ported from packages/trackerless-network/src/content-delivery-layer/ +// temporary-connection/TemporaryConnectionRpcLocal.ts (v103.8.0-rc.3): +// the server side of temporary content-delivery connections (used by +// proxy/inspection flows), tracking the requesting nodes in a NodeList +// and weak-locking the underlying connection while it is in use. +module; + +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" +#include "packages/network/protos/NetworkRpc.pb.h" + +export module streamr.trackerlessnetwork.TemporaryConnectionRpcLocal; + +import streamr.trackerlessnetwork.NetworkRpcServer; +import streamr.trackerlessnetwork.NetworkRpcClient; +import streamr.trackerlessnetwork.ContentDeliveryRpcRemote; +import streamr.trackerlessnetwork.NodeList; +import streamr.dht.ConnectionLocker; +import streamr.dht.ConnectionLockStates; +import streamr.dht.DhtCallContext; +import streamr.dht.Identifiers; +import streamr.dht.ListeningRpcCommunicator; +import streamr.dht.protos; +import streamr.utils.StreamPartID; + +// Hoisted from the former header style (file scope, NOT exported); +// fully qualified because relative namespace names resolve +// differently at file scope than inside the package namespace. +using streamr::dht::DhtAddress; +using streamr::dht::Identifiers; +using streamr::dht::connection::ConnectionLocker; +using streamr::dht::connection::LockID; +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::dht::transport::ListeningRpcCommunicator; +using streamr::utils::StreamPartID; + +export namespace streamr::trackerlessnetwork { + +using ::dht::PeerDescriptor; +using TemporaryConnectionRpc = + streamr::protorpc::TemporaryConnectionRpc; +using ContentDeliveryRpcClient = + streamr::protorpc::ContentDeliveryRpcClient; + +struct TemporaryConnectionRpcLocalOptions { + PeerDescriptor localPeerDescriptor; + StreamPartID streamPartId; + ListeningRpcCommunicator& rpcCommunicator; + ConnectionLocker& connectionLocker; +}; + +constexpr auto temporaryConnectionLockIdBase = + "system/content-delivery/temporary-connection/"; + +class TemporaryConnectionRpcLocal : public TemporaryConnectionRpc { +private: + // TS TODO preserved: use an options option or a named constant? + static constexpr size_t temporaryNodesLimit = 10; + + TemporaryConnectionRpcLocalOptions options; + NodeList temporaryNodes; + LockID lockId; + +public: + explicit TemporaryConnectionRpcLocal( + TemporaryConnectionRpcLocalOptions options) + : options(std::move(options)), + temporaryNodes( + Identifiers::getNodeIdFromPeerDescriptor( + this->options.localPeerDescriptor), + temporaryNodesLimit), + lockId( + LockID{ + temporaryConnectionLockIdBase + this->options.streamPartId}) { + } + + [[nodiscard]] NodeList& getNodes() { return this->temporaryNodes; } + + [[nodiscard]] bool hasNode(const DhtAddress& node) const { + return this->temporaryNodes.has(node); + } + + void removeNode(const DhtAddress& nodeId) { + this->temporaryNodes.remove(nodeId); + this->options.connectionLocker.weakUnlockConnection( + nodeId, this->lockId); + } + + TemporaryConnectionResponse openConnection( + const TemporaryConnectionRequest& /*request*/, + const DhtCallContext& context) override { + const auto& sender = context.incomingSourceDescriptor.value(); + ContentDeliveryRpcClient client{this->options.rpcCommunicator}; + const auto remote = std::make_shared( + this->options.localPeerDescriptor, sender, client); + this->temporaryNodes.add(remote); + this->options.connectionLocker.weakLockConnection( + Identifiers::getNodeIdFromPeerDescriptor(sender), this->lockId); + TemporaryConnectionResponse response; + response.set_accepted(true); + return response; + } + + void closeConnection( + const CloseTemporaryConnection& /*request*/, + const DhtCallContext& context) override { + const auto remoteNodeId = Identifiers::getNodeIdFromPeerDescriptor( + context.incomingSourceDescriptor.value()); + this->removeNode(remoteNodeId); + } +}; + +} // namespace streamr::trackerlessnetwork diff --git a/packages/streamr-trackerless-network/modules/logic/temporary-connection/TemporaryConnectionRpcRemote.cppm b/packages/streamr-trackerless-network/modules/logic/temporary-connection/TemporaryConnectionRpcRemote.cppm new file mode 100644 index 00000000..63cf3506 --- /dev/null +++ b/packages/streamr-trackerless-network/modules/logic/temporary-connection/TemporaryConnectionRpcRemote.cppm @@ -0,0 +1,88 @@ +// Module streamr.trackerlessnetwork.TemporaryConnectionRpcRemote +// Ported from packages/trackerless-network/src/content-delivery-layer/ +// temporary-connection/TemporaryConnectionRpcRemote.ts (v103.8.0-rc.3): +// the client side of temporary content-delivery connections. Both RPCs +// swallow errors (open reports failure as `false`, close is +// fire-and-forget), matching the TS behavior. +module; + +// Coroutine definitions need std::coroutine_traits declared in THIS +// translation unit; it cannot arrive through an imported BMI. +#include // IWYU pragma: keep + +#include +#include "packages/dht/protos/DhtRpc.pb.h" +#include "packages/network/protos/NetworkRpc.pb.h" + +export module streamr.trackerlessnetwork.TemporaryConnectionRpcRemote; + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.NetworkRpcClient; +import streamr.dht.DhtCallContext; +import streamr.dht.Identifiers; +import streamr.dht.RpcRemote; +import streamr.dht.protos; +import streamr.logger.SLogger; + +// Hoisted (file scope, NOT exported); fully qualified because relative +// namespace names resolve differently at file scope than inside the +// package namespace. +using streamr::dht::Identifiers; +using streamr::dht::contact::RpcRemote; +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::logger::SLogger; + +export namespace streamr::trackerlessnetwork { + +using ::dht::PeerDescriptor; +using TemporaryConnectionRpcClient = + streamr::protorpc::TemporaryConnectionRpcClient; + +class TemporaryConnectionRpcRemote + : public RpcRemote { +public: + TemporaryConnectionRpcRemote( + PeerDescriptor localPeerDescriptor, // NOLINT + PeerDescriptor remotePeerDescriptor, + TemporaryConnectionRpcClient client, + std::optional timeout = std::nullopt) + : RpcRemote( + std::move(localPeerDescriptor), + std::move(remotePeerDescriptor), + client, + timeout) {} + + folly::coro::Task openConnection() { + TemporaryConnectionRequest request; + auto options = this->formDhtRpcOptions({}); + try { + const auto response = co_await this->getClient().openConnection( + std::move(request), std::move(options)); + co_return response.accepted(); + } catch (const std::exception& err) { + SLogger::debug( + "temporaryConnection to " + + Identifiers::getNodeIdFromPeerDescriptor( + this->getPeerDescriptor()) + + " failed: " + std::string(err.what())); + co_return false; + } + } + + folly::coro::Task closeConnection() { + CloseTemporaryConnection request; + auto options = this->formDhtRpcOptions({}); + try { + co_await this->getClient().closeConnection( + std::move(request), std::move(options)); + } catch (const std::exception& err) { + SLogger::trace( + "closeConnection to " + + Identifiers::getNodeIdFromPeerDescriptor( + this->getPeerDescriptor()) + + " failed: " + std::string(err.what())); + } + } +}; + +} // namespace streamr::trackerlessnetwork diff --git a/packages/streamr-trackerless-network/test/unit/ContentDeliveryRpcLocalTest.cpp b/packages/streamr-trackerless-network/test/unit/ContentDeliveryRpcLocalTest.cpp index 32540141..d14ebe53 100644 --- a/packages/streamr-trackerless-network/test/unit/ContentDeliveryRpcLocalTest.cpp +++ b/packages/streamr-trackerless-network/test/unit/ContentDeliveryRpcLocalTest.cpp @@ -1,9 +1,107 @@ +// Ported from packages/trackerless-network/test/unit/ +// ContentDeliveryRpcLocal.test.ts (v103.8.0-rc.3): the server side of +// content delivery invokes the duplicate check, broadcast and +// inspection-marking callbacks on sendStreamMessage, and the +// leave-notice callback on leaveStreamPartNotice. +// +// NB: NetworkRpc types are consumed ONLY through the +// streamr.trackerlessnetwork.protos module (no textual NetworkRpc.pb.h +// include) — see TestUtilsTest.cpp for the clangd rationale. +#include +#include #include +#include "packages/dht/protos/DhtRpc.pb.h" import streamr.trackerlessnetwork.ContentDeliveryRpcLocal; +import streamr.trackerlessnetwork.TestUtils; +import streamr.trackerlessnetwork.protos; +import streamr.dht.DhtCallContext; +import streamr.dht.FakeTransport; +import streamr.dht.Identifiers; +import streamr.dht.ListeningRpcCommunicator; +import streamr.dht.protos; +import streamr.utils.EthereumAddress; +import streamr.utils.StreamPartID; -using streamr::trackerlessnetwork::ContentDeliveryRpcLocal; // NOLINT +using ::dht::PeerDescriptor; +using streamr::dht::ServiceID; +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::dht::transport::FakeTransport; +using streamr::dht::transport::ListeningRpcCommunicator; +using streamr::trackerlessnetwork::ContentDeliveryRpcLocal; +using streamr::trackerlessnetwork::ContentDeliveryRpcLocalOptions; +using streamr::trackerlessnetwork::testutils::createMockPeerDescriptor; +using streamr::trackerlessnetwork::testutils::createStreamMessage; +using streamr::utils::StreamPartIDUtils; +using streamr::utils::toEthereumAddress; -TEST(ContentDeliveryRpcLocalTest, ItCanBeInstantiated) { - // ContentDeliveryRpcLocal contentDeliveryRpcLocal; +class ContentDeliveryRpcLocalTest : public ::testing::Test { +protected: + PeerDescriptor peerDescriptor = createMockPeerDescriptor(); + PeerDescriptor mockSender = createMockPeerDescriptor(); + FakeTransport transport{peerDescriptor, [](const auto& /*message*/) {}}; + ListeningRpcCommunicator rpcCommunicator{ + ServiceID{"random-graph-node"}, transport}; + + size_t duplicateCheckCalls = 0; + size_t broadcastCalls = 0; + size_t onLeaveNoticeCalls = 0; + size_t markForInspectionCalls = 0; + + std::optional rpcLocal; + + void SetUp() override { + this->rpcLocal.emplace( + ContentDeliveryRpcLocalOptions{ + .localPeerDescriptor = this->peerDescriptor, + .streamPartId = StreamPartIDUtils::parse("stream#0"), + .markAndCheckDuplicate = + [this]( + const auto& /*messageId*/, const auto& /*previous*/) { + this->duplicateCheckCalls++; + return true; + }, + .broadcast = + [this]( + const auto& /*message*/, const auto& /*previousNode*/) { + this->broadcastCalls++; + }, + .onLeaveNotice = + [this]( + const auto& /*remoteNodeId*/, bool /*isEntryPoint*/) { + this->onLeaveNoticeCalls++; + }, + .markForInspection = + [this]( + const auto& /*remoteNodeId*/, const auto& /*msgId*/) { + this->markForInspectionCalls++; + }, + .rpcCommunicator = this->rpcCommunicator}); + } + + [[nodiscard]] DhtCallContext createContextFromSender() const { + DhtCallContext context; + context.incomingSourceDescriptor = this->mockSender; + return context; + } +}; + +TEST_F(ContentDeliveryRpcLocalTest, ServerSendStreamMessage) { + const auto message = createStreamMessage( + R"({"hello":"WORLD"})", + StreamPartIDUtils::parse("random-graph#0"), + toEthereumAddress("0x1234567890123456789012345678901234567890")); + this->rpcLocal->sendStreamMessage(message, createContextFromSender()); + EXPECT_EQ(this->duplicateCheckCalls, 1); + EXPECT_EQ(this->broadcastCalls, 1); + EXPECT_EQ(this->markForInspectionCalls, 1); +} + +TEST_F(ContentDeliveryRpcLocalTest, ServerLeaveStreamPartNotice) { + LeaveStreamPartNotice leaveNotice; + leaveNotice.set_streampartid(StreamPartIDUtils::parse("stream#0")); + leaveNotice.set_isentrypoint(false); + this->rpcLocal->leaveStreamPartNotice( + leaveNotice, createContextFromSender()); + EXPECT_EQ(this->onLeaveNoticeCalls, 1); } diff --git a/packages/streamr-trackerless-network/test/unit/ContentDeliveryRpcRemoteTest.cpp b/packages/streamr-trackerless-network/test/unit/ContentDeliveryRpcRemoteTest.cpp index fc697172..491d4d1d 100644 --- a/packages/streamr-trackerless-network/test/unit/ContentDeliveryRpcRemoteTest.cpp +++ b/packages/streamr-trackerless-network/test/unit/ContentDeliveryRpcRemoteTest.cpp @@ -1,9 +1,103 @@ +// Ported from packages/trackerless-network/test/integration/ +// ContentDeliveryRpcRemote.test.ts (v103.8.0-rc.3): the client side of +// content delivery reaches a server's registered notification handlers. +// Adaptation: the TS test wires the two ListeningRpcCommunicators over +// SimulatorTransports; the C++ house pattern for RPC-remote tests wires +// two plain RpcCommunicators back to back (see streamr-dht's +// DhtNodeRpcRemoteTest), which exercises the same client/server RPC +// path without the simulator moving parts. +// +// NB: NetworkRpc types are consumed ONLY through the +// streamr.trackerlessnetwork.protos module (no textual NetworkRpc.pb.h +// include) — see TestUtilsTest.cpp for the clangd rationale. +#include +#include #include +#include "packages/dht/protos/DhtRpc.pb.h" +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.protorpc.RpcCommunicator; +import streamr.protorpc.protos; import streamr.trackerlessnetwork.ContentDeliveryRpcRemote; +import streamr.trackerlessnetwork.NetworkRpcClient; +import streamr.trackerlessnetwork.TestUtils; +import streamr.trackerlessnetwork.protos; +import streamr.dht.DhtCallContext; +import streamr.dht.protos; +import streamr.utils.EthereumAddress; +import streamr.utils.StreamPartID; + +using ::dht::PeerDescriptor; +using ::protorpc::RpcMessage; +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::protorpc::RpcCommunicator; +using streamr::trackerlessnetwork::ContentDeliveryRpcRemote; +using streamr::trackerlessnetwork::testutils::createMockPeerDescriptor; +using streamr::trackerlessnetwork::testutils::createStreamMessage; +using streamr::utils::blockingWait; +using streamr::utils::StreamPartIDUtils; +using streamr::utils::toEthereumAddress; + +using ContentDeliveryRpcClient = + streamr::protorpc::ContentDeliveryRpcClient; +using RpcCommunicatorType = RpcCommunicator; -using streamr::trackerlessnetwork::ContentDeliveryRpcRemote; // NOLINT +class ContentDeliveryRpcRemoteTest : public ::testing::Test { +protected: + RpcCommunicatorType clientCommunicator; + RpcCommunicatorType serverCommunicator; + PeerDescriptor clientNode = createMockPeerDescriptor(); + PeerDescriptor serverNode = createMockPeerDescriptor(); + size_t recvCounter = 0; + std::optional rpcRemote; + + void SetUp() override { + this->serverCommunicator.registerRpcNotification( + "sendStreamMessage", + [this]( + const StreamMessage& /*message*/, + const DhtCallContext& /*context*/) { this->recvCounter++; }); + this->serverCommunicator.registerRpcNotification( + "leaveStreamPartNotice", + [this]( + const LeaveStreamPartNotice& /*message*/, + const DhtCallContext& /*context*/) { this->recvCounter++; }); + this->clientCommunicator.setOutgoingMessageCallback( + [this]( + const RpcMessage& message, + const std::string& /*requestId*/, + const DhtCallContext& /*context*/) { + this->serverCommunicator.handleIncomingMessage( + message, DhtCallContext()); + }); + this->serverCommunicator.setOutgoingMessageCallback( + [this]( + const RpcMessage& message, + const std::string& /*requestId*/, + const DhtCallContext& /*context*/) { + this->clientCommunicator.handleIncomingMessage( + message, DhtCallContext()); + }); + this->rpcRemote.emplace( + this->clientNode, + this->serverNode, + ContentDeliveryRpcClient(this->clientCommunicator)); + } +}; + +TEST_F(ContentDeliveryRpcRemoteTest, SendStreamMessage) { + const auto msg = createStreamMessage( + R"({"hello":"WORLD"})", + StreamPartIDUtils::parse("test-stream#0"), + toEthereumAddress("0x1234567890123456789012345678901234567890")); + blockingWait(this->rpcRemote->sendStreamMessage(msg)); + EXPECT_EQ(this->recvCounter, 1); +} -TEST(ContentDeliveryRpcRemoteTest, ItCanBeInstantiated) { - // ContentDeliveryRpcRemote contentDeliveryRpcRemote; +TEST_F(ContentDeliveryRpcRemoteTest, LeaveNotice) { + blockingWait(this->rpcRemote->leaveStreamPartNotice( + StreamPartIDUtils::parse("test#0"), false)); + EXPECT_EQ(this->recvCounter, 1); } diff --git a/packages/streamr-trackerless-network/test/unit/NumberPairTest.cpp b/packages/streamr-trackerless-network/test/unit/NumberPairTest.cpp new file mode 100644 index 00000000..60c62d08 --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/NumberPairTest.cpp @@ -0,0 +1,29 @@ +// Ported from packages/trackerless-network/test/unit/NumberPair.test.ts +// (v103.8.0-rc.3): ordering semantics of the (a, b) pairs the duplicate +// message detector compares message references with. +#include + +import streamr.trackerlessnetwork.DuplicateMessageDetector; + +using streamr::trackerlessnetwork::NumberPair; + +TEST(NumberPairTest, EqualTo) { + EXPECT_EQ(NumberPair(5, 2).equalTo(NumberPair(5, 3)), false); + EXPECT_EQ(NumberPair(5, 2).equalTo(NumberPair(5, 2)), true); +} + +TEST(NumberPairTest, GreaterThan) { + EXPECT_EQ(NumberPair(5, 2).greaterThan(NumberPair(6, 2)), false); + EXPECT_EQ(NumberPair(5, 2).greaterThan(NumberPair(5, 3)), false); + EXPECT_EQ(NumberPair(5, 2).greaterThan(NumberPair(5, 2)), false); + EXPECT_EQ(NumberPair(5, 2).greaterThan(NumberPair(5, 1)), true); + EXPECT_EQ(NumberPair(5, 2).greaterThan(NumberPair(3, 2)), true); +} + +TEST(NumberPairTest, GreaterThanOrEqual) { + EXPECT_EQ(NumberPair(5, 2).greaterThanOrEqual(NumberPair(6, 2)), false); + EXPECT_EQ(NumberPair(5, 2).greaterThanOrEqual(NumberPair(5, 3)), false); + EXPECT_EQ(NumberPair(5, 2).greaterThanOrEqual(NumberPair(5, 2)), true); + EXPECT_EQ(NumberPair(5, 2).greaterThanOrEqual(NumberPair(5, 1)), true); + EXPECT_EQ(NumberPair(5, 2).greaterThanOrEqual(NumberPair(3, 2)), true); +} diff --git a/packages/streamr-trackerless-network/test/unit/StreamPartIdToDataKeyTest.cpp b/packages/streamr-trackerless-network/test/unit/StreamPartIdToDataKeyTest.cpp new file mode 100644 index 00000000..2a35b9fb --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/StreamPartIdToDataKeyTest.cpp @@ -0,0 +1,22 @@ +// Ported from packages/trackerless-network/test/unit/ +// StreamPartIDDataKey.test.ts (v103.8.0-rc.3): the DHT data key derived +// from a stream part id is a 160-bit (40 hex character) address. +// The exact-value assertion is a C++ addition: the key must match the +// TS derivation byte for byte (SHA1 of the stream part id), or C++ and +// TS nodes would store the same stream part's entry points under +// different DHT keys. +#include + +import streamr.trackerlessnetwork.streamPartIdToDataKey; +import streamr.utils.StreamPartID; + +using streamr::trackerlessnetwork::streamPartIdToDataKey; +using streamr::utils::StreamPartIDUtils; + +TEST(StreamPartIdToDataKeyTest, GeneratedKeyLengthIsCorrect) { + const auto streamPartId = StreamPartIDUtils::parse("stream#0"); + const auto dataKey = streamPartIdToDataKey(streamPartId); + EXPECT_EQ(dataKey.length(), 40); + // sha1("stream#0") — pins wire compatibility with the TS derivation. + EXPECT_EQ(dataKey, "23a32b02463a236285eb026d76657155b2ccb1f1"); +} diff --git a/packages/streamr-trackerless-network/test/unit/TemporaryConnectionRpcLocalTest.cpp b/packages/streamr-trackerless-network/test/unit/TemporaryConnectionRpcLocalTest.cpp new file mode 100644 index 00000000..dd63d3cc --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/TemporaryConnectionRpcLocalTest.cpp @@ -0,0 +1,64 @@ +// Ported from packages/trackerless-network/test/unit/ +// TemporaryConnectionRpcLocal.test.ts (v103.8.0-rc.3): openConnection +// registers the caller in the temporary-node list, closeConnection +// removes it again. +// +// NB: NetworkRpc types are consumed ONLY through the +// streamr.trackerlessnetwork.protos module (no textual NetworkRpc.pb.h +// include) — see TestUtilsTest.cpp for the clangd rationale. +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" + +import streamr.trackerlessnetwork.TemporaryConnectionRpcLocal; +import streamr.trackerlessnetwork.TestUtils; +import streamr.trackerlessnetwork.protos; +import streamr.dht.DhtCallContext; +import streamr.dht.FakeTransport; +import streamr.dht.Identifiers; +import streamr.dht.ListeningRpcCommunicator; +import streamr.dht.protos; +import streamr.utils.StreamPartID; + +using ::dht::PeerDescriptor; +using streamr::dht::Identifiers; +using streamr::dht::ServiceID; +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::dht::transport::FakeTransport; +using streamr::dht::transport::ListeningRpcCommunicator; +using streamr::trackerlessnetwork::TemporaryConnectionRpcLocal; +using streamr::trackerlessnetwork::TemporaryConnectionRpcLocalOptions; +using streamr::trackerlessnetwork::testutils::createMockPeerDescriptor; +using streamr::trackerlessnetwork::testutils::MockConnectionLocker; +using streamr::utils::StreamPartIDUtils; + +class TemporaryConnectionRpcLocalTest : public ::testing::Test { +protected: + PeerDescriptor peerDescriptor = createMockPeerDescriptor(); + FakeTransport transport{peerDescriptor, [](const auto& /*message*/) {}}; + ListeningRpcCommunicator rpcCommunicator{ServiceID{"mock"}, transport}; + MockConnectionLocker connectionLocker; + std::optional rpcLocal; + + void SetUp() override { + this->rpcLocal.emplace( + TemporaryConnectionRpcLocalOptions{ + .localPeerDescriptor = this->peerDescriptor, + .streamPartId = StreamPartIDUtils::parse("mock#0"), + .rpcCommunicator = this->rpcCommunicator, + .connectionLocker = this->connectionLocker}); + } +}; + +TEST_F(TemporaryConnectionRpcLocalTest, OpenAndCloseConnection) { + const auto caller = createMockPeerDescriptor(); + const auto callerId = Identifiers::getNodeIdFromPeerDescriptor(caller); + DhtCallContext context; + context.incomingSourceDescriptor = caller; + + this->rpcLocal->openConnection(TemporaryConnectionRequest{}, context); + EXPECT_TRUE(this->rpcLocal->getNodes().get(callerId).has_value()); + + this->rpcLocal->closeConnection(CloseTemporaryConnection{}, context); + EXPECT_FALSE(this->rpcLocal->getNodes().get(callerId).has_value()); +} diff --git a/packages/streamr-trackerless-network/test/utils/TestUtils.cppm b/packages/streamr-trackerless-network/test/utils/TestUtils.cppm index f1907d88..312d783c 100644 --- a/packages/streamr-trackerless-network/test/utils/TestUtils.cppm +++ b/packages/streamr-trackerless-network/test/utils/TestUtils.cppm @@ -64,6 +64,18 @@ inline StreamMessage createStreamMessage( return msg; } +// Ported from createMockPeerDescriptor() (test/utils/utils.ts): a +// descriptor with a random node id, sufficient for tests that only need +// identity (mirrors the streamr-dht test util of the same name). +inline PeerDescriptor createMockPeerDescriptor() { + PeerDescriptor descriptor; + descriptor.set_nodeid( + streamr::dht::Identifiers::getRawFromDhtAddress( + streamr::dht::Identifiers::createRandomDhtAddress())); + descriptor.set_type(::dht::NodeType::NODEJS); + return descriptor; +} + // Ported from mockConnectionLocker (test/utils/utils.ts): a no-op // ConnectionLocker for components that require one but whose locking // behavior is irrelevant to the test. diff --git a/packages/streamr-trackerless-network/vcpkg.json b/packages/streamr-trackerless-network/vcpkg.json index 42227bb1..5bdf61e1 100644 --- a/packages/streamr-trackerless-network/vcpkg.json +++ b/packages/streamr-trackerless-network/vcpkg.json @@ -6,7 +6,8 @@ "protobuf", "folly", "magic-enum", - "boost-algorithm" + "boost-algorithm", + "openssl" ], "features": { "monorepo": {