last fixes + style fixing

Author: slampy97

tests/coordination/test_coordination_client.py

@@ -217,7 +217,6 @@ async def second_lock_task():
 
             await asyncio.sleep(0)
 
-
         await asyncio.wait_for(lock2_acquired.wait(), timeout=timeout)
 
         lock2_release.set()
@@ -275,14 +274,11 @@ async def test_coordination_reconnect_async(self, async_coordination_node):
 
         lock = client.lock("test_lock", node_path)
 
-        # create semaphore
         res = await lock.create(init_limit=1, init_data=b"")
         assert res.status == StatusCode.SUCCESS
 
-        # break connection (simulate network drop)
         await lock._reconnector._stream.close()
 
-        # next call must succeed after reconnect
         desc = await lock.describe()
         assert desc.status == StatusCode.SUCCESS
         assert desc.name == "test_lock"

ydb/_grpc/grpcwrapper/common_utils.py

@@ -220,7 +220,7 @@ async def _start_sync_driver(self, driver: Driver, stub, method):
         self._stream_call = stream_call
         self.from_server_grpc = SyncToAsyncIterator(stream_call.__iter__(), self._wait_executor)
 
-    async def receive(self, timeout: Optional[int] = None) -> Any:
+    async def receive(self, timeout: Optional[int] = None, is_coordination_calls=False) -> Any:
         # todo handle grpc exceptions and convert it to internal exceptions
         try:
             if timeout is None:
@@ -235,8 +235,8 @@ async def get_response():
         except (grpc.RpcError, grpc.aio.AioRpcError) as e:
             raise connection._rpc_error_handler(self._connection_state, e)
 
-        # coordination grpc calls dont have status field
-        # issues._process_response(grpc_message)
+        if not is_coordination_calls:
+            issues._process_response(grpc_message)
 
         if self._connection_state != "has_received_messages":
             self._connection_state = "has_received_messages"

ydb/aio/coordination/stream.py

@@ -34,7 +34,7 @@ async def start_session(self, path: str, timeout_millis: int):
         try:
             while True:
                 try:
-                    resp = await self._stream.receive(timeout=3)
+                    resp = await self._stream.receive(timeout=3, is_coordination_calls=True)
                 except asyncio.TimeoutError:
                     raise issues.Error("Timeout waiting for SessionStart response")
                 except StopAsyncIteration:
@@ -73,7 +73,7 @@ async def _reader_loop(self):
         try:
             while True:
                 try:
-                    resp = await self._stream.receive(timeout=3)
+                    resp = await self._stream.receive(timeout=3, is_coordination_calls=True)
                 except asyncio.TimeoutError:
                     continue
                 except asyncio.CancelledError: