[C++ SDK] Refactor topic tests (part 3) (#19925)

Author: Gazizonoki

.github/last_commit.txt

@@ -1 +1 @@
-63058f09d3560f26085ffea238bc097700aeb8f2
+73c2345cf95acef59e91b8bf03a41539b4c9194d

src/client/federated_topic/ut/basic_usage_ut.cpp

@@ -1,7 +1,7 @@
 #include <ydb-cpp-sdk/client/federated_topic/federated_topic.h>
 #include <src/client/federated_topic/impl/federated_write_session.h>
 
-#include <src/client/topic/ut/ut_utils/managed_executor.h>
+#include <tests/integration/topic/utils/managed_executor.h>
 
 #include <src/client/persqueue_public/persqueue.h>
 

src/client/topic/ut/basic_usage_ut.cpp

@@ -1,5 +1,7 @@
-#include "ut_utils/managed_executor.h"
 #include "ut_utils/topic_sdk_test_setup.h"
+
+#include <tests/integration/topic/utils/managed_executor.h>
+
 #include <src/client/persqueue_public/ut/ut_utils/ut_utils.h>
 
 #include <ydb-cpp-sdk/client/topic/client.h>
@@ -18,19 +20,19 @@
 
 #include <util/stream/zlib.h>
 
-#include <future>
 
+using namespace std::chrono_literals;
 
 static const bool EnableDirectRead = !std::string{std::getenv("PQ_EXPERIMENTAL_DIRECT_READ") ? std::getenv("PQ_EXPERIMENTAL_DIRECT_READ") : ""}.empty();
 
 
-namespace NYdb::NTopic::NTests {
+namespace NYdb::inline V3::NTopic::NTests {
 
-void WriteAndReadToEndWithRestarts(TReadSessionSettings readSettings, TWriteSessionSettings writeSettings, const std::string& message, ui32 count, TTopicSdkTestSetup& setup, TIntrusivePtr<TManagedExecutor> decompressor) {
+void WriteAndReadToEndWithRestarts(TReadSessionSettings readSettings, TWriteSessionSettings writeSettings, const std::string& message, std::uint32_t count, TTopicSdkTestSetup& setup, TIntrusivePtr<TManagedExecutor> decompressor) {
     auto client = setup.MakeClient();
     auto session = client.CreateSimpleBlockingWriteSession(writeSettings);
 
-    for (ui32 i = 1; i <= count; ++i) {
+    for (std::uint32_t i = 1; i <= count; ++i) {
         bool res = session->Write(message);
         UNIT_ASSERT(res);
     }
@@ -44,7 +46,7 @@ void WriteAndReadToEndWithRestarts(TReadSessionSettings readSettings, TWriteSess
 
     auto WaitTasks = [&](auto f, size_t c) {
         while (f() < c) {
-            Sleep(TDuration::MilliSeconds(100));
+            std::this_thread::sleep_for(100ms);
         };
     };
     auto WaitPlannedTasks = [&](auto e, size_t count) {
@@ -69,7 +71,7 @@ void WriteAndReadToEndWithRestarts(TReadSessionSettings readSettings, TWriteSess
         size_t completed = e->GetExecutedCount();
 
         setup.GetServer().KillTopicPqrbTablet(setup.GetTopicPath());
-        Sleep(TDuration::MilliSeconds(100));
+        std::this_thread::sleep_for(100ms);
 
         e->StartFuncs(tasks);
         WaitExecutedTasks(e, completed + n);
@@ -90,7 +92,7 @@ void WriteAndReadToEndWithRestarts(TReadSessionSettings readSettings, TWriteSess
 
     ReadSession = topicClient.CreateReadSession(readSettings);
 
-    ui32 i = 0;
+    std::uint32_t i = 0;
     while (AtomicGet(lastOffset) + 1 < count) {
         RunTasks(decompressor, {i++});
     }
@@ -109,7 +111,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
         auto decompressor = CreateThreadPoolManagedExecutor(1);
 
         TReadSessionSettings readSettings;
-        TTopicReadSettings topic = TEST_TOPIC;
+        TTopicReadSettings topic = setup.GetTopicPath();
         topic.AppendPartitionIds(0);
         readSettings
             .WithoutConsumer()
@@ -121,13 +123,13 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
 
         TWriteSessionSettings writeSettings;
         writeSettings
-            .Path(TEST_TOPIC)
+            .Path(setup.GetTopicPath())
             .MessageGroupId(TEST_MESSAGE_GROUP_ID)
             .Codec(NTopic::ECodec::RAW)
             .CompressionExecutor(compressor);
 
 
-        ui32 count = 700;
+        std::uint32_t count = 700;
         std::string message(2'000, 'x');
 
         WriteAndReadToEndWithRestarts(readSettings, writeSettings, message, count, setup, decompressor);
@@ -140,21 +142,21 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
 
         TReadSessionSettings readSettings;
         readSettings
-            .ConsumerName(TEST_CONSUMER)
+            .ConsumerName(setup.GetConsumerName())
             .MaxMemoryUsageBytes(1_MB)
             .DecompressionExecutor(decompressor)
-            .AppendTopics(TEST_TOPIC)
+            .AppendTopics(setup.GetTopicPath())
             // .DirectRead(EnableDirectRead)
             ;
 
         TWriteSessionSettings writeSettings;
         writeSettings
-            .Path(TEST_TOPIC).MessageGroupId(TEST_MESSAGE_GROUP_ID)
-            .Codec(NTopic::ECodec::RAW)
+            .Path(setup.GetTopicPath()).MessageGroupId(TEST_MESSAGE_GROUP_ID)
+            .Codec(ECodec::RAW)
             .CompressionExecutor(compressor);
 
 
-        ui32 count = 700;
+        std::uint32_t count = 700;
         std::string message(2'000, 'x');
 
         WriteAndReadToEndWithRestarts(readSettings, writeSettings, message, count, setup, decompressor);
@@ -164,25 +166,25 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
 
         TTopicSdkTestSetup setup(TEST_CASE_NAME);
 
-        NTopic::TWriteSessionSettings writeSettings;
+        TWriteSessionSettings writeSettings;
         writeSettings.Path(setup.GetTopicPath()).MessageGroupId(TEST_MESSAGE_GROUP_ID);
         writeSettings.Path(setup.GetTopicPath()).ProducerId(TEST_MESSAGE_GROUP_ID);
-        writeSettings.Codec(NTopic::ECodec::RAW);
-        NTopic::IExecutor::TPtr executor = new NTopic::TSyncExecutor();
+        writeSettings.Codec(ECodec::RAW);
+        IExecutor::TPtr executor = new TSyncExecutor();
         writeSettings.CompressionExecutor(executor);
 
-        ui64 count = 100u;
+        std::uint64_t count = 100u;
 
         auto client = setup.MakeClient();
         auto session = client.CreateSimpleBlockingWriteSession(writeSettings);
 
-        TString messageBase = "message----";
+        std::string messageBase = "message----";
 
         for (auto i = 0u; i < count; i++) {
             auto res = session->Write(messageBase);
             UNIT_ASSERT(res);
             if (i % 10 == 0) {
-                setup.GetServer().KillTopicPqTablets(setup.GetTopicPath());
+                setup.GetServer().KillTopicPqTablets(setup.GetFullTopicPath());
             }
         }
         session->Close();