add functional test

Author: nshestakov

ydb/tests/functional/transfer/main.cpp

@@ -13,6 +13,7 @@
 using namespace NYdb;
 using namespace NYdb::NQuery;
 using namespace NYdb::NTopic;
+using namespace NYdb::NReplication;
 
 namespace {
 
@@ -188,14 +189,22 @@ struct MainTestCase {
     }
 
     auto DescribeTransfer() {
-        NYdb::NReplication::TReplicationClient client(Driver);
+        TReplicationClient client(Driver);
 
-        NYdb::NReplication::TDescribeReplicationSettings settings;
+        TDescribeReplicationSettings settings;
         settings.IncludeStats(true);
 
         return client.DescribeReplication(TString("/") + GetEnv("YDB_DATABASE") + "/" + TransferName, settings);
     }
 
+    auto DescribeTopic() {
+        TDescribeTopicSettings settings;
+        settings.IncludeLocation(true);
+        settings.IncludeStats(true);
+
+        return TopicClient.DescribeTopic(TopicName, settings);
+    }
+
     void Write(const TMessage& message) {
         TWriteSessionSettings writeSettings;
         writeSettings.Path(TopicName);
@@ -255,7 +264,7 @@ struct MainTestCase {
                 break;
             }
 
-            UNIT_ASSERT_C(attempt, "Unable to wait replication result");
+            UNIT_ASSERT_C(attempt, "Unable to wait transfer result");
             Sleep(TDuration::Seconds(1));
         }
     }
@@ -828,6 +837,63 @@ Y_UNIT_TEST_SUITE(Transfer)
         {
             auto result = testCase.DescribeTransfer().ExtractValueSync();
             UNIT_ASSERT_C(!result.IsSuccess(), result.GetIssues().ToOneLineString());
+            UNIT_ASSERT_VALUES_EQUAL(EStatus::SCHEME_ERROR, result.GetStatus());
+        }
+    }
+
+    Y_UNIT_TEST(CreateAndDropConsumer)
+    {
+        MainTestCase testCase;
+        testCase.CreateTable(R"(
+                CREATE TABLE `%s` (
+                    Key Uint64 NOT NULL,
+                    Message Utf8 NOT NULL,
+                    PRIMARY KEY (Key)
+                )  WITH (
+                    STORE = COLUMN
+                );
+            )");
+        
+        testCase.CreateTopic();
+        testCase.CreateTransfer(R"(
+                $l = ($x) -> {
+                    return [
+                        <|
+                            Key:CAST($x._offset AS Uint64),
+                            Message:CAST($x._data AS Utf8)
+                        |>
+                    ];
+                };
+            )");
+
+        for (size_t i = 20; i--; ) {
+            auto result = testCase.DescribeTopic().ExtractValueSync();
+            UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToOneLineString());
+            auto& consumers = result.GetTopicDescription().GetConsumers();
+            if (1 == consumers.size()) {
+                UNIT_ASSERT_VALUES_EQUAL(1, consumers.size());
+                Cerr << "Consumer name is '" << consumers[0].GetConsumerName() << "'" << Endl << Flush;
+                UNIT_ASSERT_VALUES_EQUAL_C(35, consumers[0].GetConsumerName().size(), "Consumer name is random uuid");
+                break;
+            }
+
+            UNIT_ASSERT_C(i, "Unable to wait consumer has been created");
+            Sleep(TDuration::Seconds(1));
+        }
+
+        testCase.DropTransfer();
+
+        for (size_t i = 20; i--; ) {
+            auto result = testCase.DescribeTopic().ExtractValueSync();
+            UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToOneLineString());
+            auto& consumers = result.GetTopicDescription().GetConsumers();
+            if (0 == consumers.size()) {
+                UNIT_ASSERT_VALUES_EQUAL(0, consumers.size());
+                break;
+            }
+
+            UNIT_ASSERT_C(i, "Unable to wait consumer has been removed");
+            Sleep(TDuration::Seconds(1));
         }
     }