tests

Author: vgvoleg

tests/topics/test_control_plane.py

@@ -39,14 +39,23 @@ async def test_describe_topic(self, driver, topic_path: str, topic_consumer):
 
         assert has_consumer
 
-    async def test_alter_topic(self, driver, topic_path):
+    async def test_alter_not_existed_topic(self, driver, topic_path):
         client = driver.topic_client
 
-        await client.alter_topic(topic_path)
-
         with pytest.raises(issues.SchemeError):
             await client.alter_topic(topic_path + "-not-exist")
 
+    async def test_alter_existed_topic(self, driver, topic_path):
+        client = driver.topic_client
+
+        topic_before = await client.describe_topic(topic_path)
+
+        target_min_active_partitions = topic_before.min_active_partitions + 1
+        await client.alter_topic(topic_path, set_min_active_partitions=target_min_active_partitions)
+
+        topic_after = await client.describe_topic(topic_path)
+        assert topic_after.min_active_partitions == target_min_active_partitions
+
 
 class TestTopicClientControlPlane:
     def test_create_topic(self, driver_sync, database):
@@ -81,10 +90,19 @@ def test_describe_topic(self, driver_sync, topic_path: str, topic_consumer):
 
         assert has_consumer
 
-    def test_alter_topic(self, driver_sync, topic_path):
+    def test_alter_not_existed_topic(self, driver_sync, topic_path):
         client = driver_sync.topic_client
 
-        client.alter_topic(topic_path)
-
         with pytest.raises(issues.SchemeError):
             client.alter_topic(topic_path + "-not-exist")
+
+    def test_alter_existed_topic(self, driver_sync, topic_path):
+        client = driver_sync.topic_client
+
+        topic_before = client.describe_topic(topic_path)
+
+        target_min_active_partitions = topic_before.min_active_partitions + 1
+        client.alter_topic(topic_path, set_min_active_partitions=target_min_active_partitions)
+
+        topic_after = client.describe_topic(topic_path)
+        assert topic_after.min_active_partitions == target_min_active_partitions

ydb/_grpc/grpcwrapper/ydb_topic.py

@@ -1063,13 +1063,15 @@ class AlterTopicRequest(IToProto, IFromPublic):
     set_metering_mode: "MeteringMode"
 
     def to_proto(self) -> ydb_topic_pb2.AlterTopicRequest:
+        supported_codecs = self.set_supported_codecs.to_proto() if self.set_supported_codecs.codecs else None
+
         return ydb_topic_pb2.AlterTopicRequest(
             path=self.path,
             add_consumers=[consumer.to_proto() for consumer in self.add_consumers],
             alter_partitioning_settings=self.alter_partitioning_settings.to_proto(),
             set_retention_period=proto_duration_from_timedelta(self.set_retention_period),
             set_retention_storage_mb=self.set_retention_storage_mb,
-            set_supported_codecs=self.set_supported_codecs.to_proto(),
+            set_supported_codecs=supported_codecs,
             set_partition_write_burst_bytes=self.set_partition_write_burst_bytes,
             set_partition_write_speed_bytes_per_second=self.set_partition_write_speed_bytes_per_second,
             alter_attributes=self.alter_attributes,

ydb/_grpc/grpcwrapper/ydb_topic_public_types.py

@@ -37,7 +37,7 @@ class AlterTopicRequestParams:
     set_partition_count_limit: Optional[int]
     add_consumers: Optional[List[Union["PublicConsumer", str]]]
     alter_consumers: Optional[List[Union["PublicAlterConsumer", str]]]
-    drop_consumers: Optional[List[str]]  # TODO: clarify
+    drop_consumers: Optional[List[str]]
     alter_attributes: Optional[Dict[str, str]]
     set_metering_mode: Optional["PublicMeteringMode"]
     set_partition_write_speed_bytes_per_second: Optional[int]

ydb/_grpc/grpcwrapper/ydb_topic_test.py

@@ -1,4 +1,25 @@
+import datetime
+import typing
+
+from google.protobuf.internal import containers
+from google.protobuf import duration_pb2
+from google.protobuf.json_format import MessageToDict
+
 from ydb._grpc.grpcwrapper.ydb_topic import OffsetsRange
+from .ydb_topic import AlterTopicRequest
+from .ydb_topic_public_types import (
+    AlterTopicRequestParams,
+    PublicAlterConsumer,
+    PublicConsumer,
+    PublicMeteringMode,
+    PublicCodec,
+)
+
+# Workaround for good IDE and universal for runtime
+if typing.TYPE_CHECKING:
+    from ..v4.protos import ydb_topic_pb2
+else:
+    from ..common.protos import ydb_topic_pb2
 
 
 def test_offsets_range_intersected():
@@ -17,3 +38,59 @@ def test_offsets_range_intersected():
     ]:
         assert OffsetsRange(test[0], test[1]).is_intersected_with(OffsetsRange(test[2], test[3]))
         assert OffsetsRange(test[2], test[3]).is_intersected_with(OffsetsRange(test[0], test[1]))
+
+
+def test_alter_topic_request_from_public_to_proto():
+    # Specify all fields with all possible input ways
+    params = {
+        "path": "topic_name",
+        "add_consumers": [
+            "new_consumer_1",
+            PublicConsumer("new_consumer_2"),
+        ],
+        "alter_consumers": [
+            "old_consumer_1",
+            PublicAlterConsumer("old_consumer_2"),
+        ],
+        "drop_consumers": ["redundant_consumer"],
+        "set_retention_period": datetime.timedelta(weeks=4),
+        "set_retention_storage_mb": 4,
+        "set_supported_codecs": [1, PublicCodec(2)],
+        "set_partition_write_burst_bytes": 8,
+        "set_partition_write_speed_bytes_per_second": 15,
+        "alter_attributes": {"key": "value"},
+        "set_metering_mode": 1,
+        "set_min_active_partitions": 2,
+        "set_partition_count_limit": 4,
+    }
+
+    params_public = AlterTopicRequestParams(**params)
+    request = AlterTopicRequest.from_public(params_public)
+    request_proto = request.to_proto()
+
+    msg_dict = MessageToDict(request_proto, preserving_proto_field_name=True)
+
+    assert msg_dict["path"] == params["path"]
+    assert len(msg_dict["add_consumers"]) == len(params["add_consumers"])
+    assert len(msg_dict["alter_consumers"]) == len(params["alter_consumers"])
+    assert len(msg_dict["drop_consumers"]) == len(params["drop_consumers"])
+    assert msg_dict["alter_attributes"] == params["alter_attributes"]
+
+    assert (
+        int(msg_dict["alter_partitioning_settings"]["set_min_active_partitions"]) == params["set_min_active_partitions"]
+    )
+    assert (
+        int(msg_dict["alter_partitioning_settings"]["set_partition_count_limit"]) == params["set_partition_count_limit"]
+    )
+
+    assert int(msg_dict["set_partition_write_burst_bytes"]) == params["set_partition_write_burst_bytes"]
+    assert (
+        int(msg_dict["set_partition_write_speed_bytes_per_second"])
+        == params["set_partition_write_speed_bytes_per_second"]
+    )
+    assert msg_dict["set_retention_period"] == str(int(params["set_retention_period"].total_seconds())) + "s"
+    assert int(msg_dict["set_retention_storage_mb"]) == params["set_retention_storage_mb"]
+
+    assert msg_dict["set_metering_mode"] == "METERING_MODE_RESERVED_CAPACITY"
+
+    assert msg_dict["set_supported_codecs"]["codecs"] == params["set_supported_codecs"]