fix linter

Author: rekby

tests/topics/test_topic_reader.py

@@ -168,7 +168,7 @@ def decode(b: bytes):
 @pytest.mark.asyncio
 class TestBugFixesAsync:
     async def test_issue_297_bad_handle_stop_partition(
-            self, driver, topic_consumer, topic_with_two_partitions_path: str
+        self, driver, topic_consumer, topic_with_two_partitions_path: str
     ):
         async def wait(fut):
             return await asyncio.wait_for(fut, timeout=10)
@@ -189,29 +189,19 @@ async def wait(fut):
         # Start second reader for same topic, same consumer, partition 1
         reader1 = driver.topic_client.reader(topic, consumer=topic_consumer)
 
-        await asyncio.sleep(1)
-
-        async with driver.topic_client.writer(topic, partition_id=0) as writer:
-            await writer.write_with_ack("--")
-        async with driver.topic_client.writer(topic, partition_id=1) as writer:
-            await writer.write_with_ack("--")
-
-        await reader0.receive_message()
-        await reader0.receive_message()
-
         # receive uncommited message
         await reader1.receive_message()
 
         # write one message for every partition
         async with driver.topic_client.writer(topic, partition_id=0) as writer:
             await writer.write_with_ack("10")
-        async with driver.topic_client.writer(topic, partition_id=0) as writer:
+        async with driver.topic_client.writer(topic, partition_id=1) as writer:
             await writer.write_with_ack("11")
 
         msg0 = await wait(reader0.receive_message())
         msg1 = await wait(reader1.receive_message())
 
-        datas = [msg0.data.decode(), msg1.data.decode]
+        datas = [msg0.data.decode(), msg1.data.decode()]
         datas.sort()
 
         assert datas == ["10", "11"]

ydb/_grpc/grpcwrapper/ydb_topic.py

@@ -591,9 +591,14 @@ def from_proto(
                 )
 
     @dataclass
-    class PartitionSessionStatusRequest:
+    class PartitionSessionStatusRequest(IToProto):
         partition_session_id: int
 
+        def to_proto(self) -> ydb_topic_pb2.StreamReadMessage.PartitionSessionStatusRequest:
+            return ydb_topic_pb2.StreamReadMessage.PartitionSessionStatusRequest(
+                partition_session_id=self.partition_session_id
+            )
+
     @dataclass
     class PartitionSessionStatusResponse(IFromProto):
         partition_session_id: int
@@ -662,9 +667,14 @@ def from_proto(
             )
 
     @dataclass
-    class StopPartitionSessionResponse:
+    class StopPartitionSessionResponse(IToProto):
         partition_session_id: int
 
+        def to_proto(self) -> ydb_topic_pb2.StreamReadMessage.StopPartitionSessionResponse:
+            return ydb_topic_pb2.StreamReadMessage.StopPartitionSessionResponse(
+                partition_session_id=self.partition_session_id,
+            )
+
     @dataclass
     class FromClient(IToProto):
         client_message: "ReaderMessagesFromClientToServer"
@@ -684,6 +694,10 @@ def to_proto(self) -> ydb_topic_pb2.StreamReadMessage.FromClient:
                 res.update_token_request.CopyFrom(self.client_message.to_proto())
             elif isinstance(self.client_message, StreamReadMessage.StartPartitionSessionResponse):
                 res.start_partition_session_response.CopyFrom(self.client_message.to_proto())
+            elif isinstance(self.client_message, StreamReadMessage.StopPartitionSessionResponse):
+                res.stop_partition_session_response.CopyFrom(self.client_message.to_proto())
+            elif isinstance(self.client_message, StreamReadMessage.PartitionSessionStatusRequest):
+                res.start_partition_session_response.CopyFrom(self.client_message.to_proto())
             else:
                 raise NotImplementedError("Unknown message type: %s" % type(self.client_message))
             return res

ydb/_topic_reader/topic_reader_asyncio.py

@@ -2,7 +2,6 @@
 
 import asyncio
 import concurrent.futures
-import copy
 import gzip
 import typing
 from asyncio import Task

ydb/topic.py

@@ -168,7 +168,7 @@ def reader(
         if not decoder_executor:
             decoder_executor = self._executor
 
-        args = locals()
+        args = locals().copy()
         del args["self"]
 
         settings = TopicReaderSettings(**args)
@@ -188,7 +188,7 @@ def writer(
         encoders: Optional[Mapping[_ydb_topic_public_types.PublicCodec, Callable[[bytes], bytes]]] = None,
         encoder_executor: Optional[concurrent.futures.Executor] = None,  # default shared client executor pool
     ) -> TopicWriterAsyncIO:
-        args = locals()
+        args = locals().copy()
         del args["self"]
 
         settings = TopicWriterSettings(**args)