update pytest version to prevent warning

Author: Valeriya Popova

test-requirements.txt

@@ -26,8 +26,8 @@ pycparser==2.20
 PyNaCl==1.4.0
 pyparsing==2.4.7
 pyrsistent==0.18.0
-pytest==6.2.4
-pytest-asyncio==0.15.1
+pytest==7.2.2
+pytest-asyncio==0.21.0
 pytest-docker-compose==3.2.1
 python-dotenv==0.18.0
 PyYAML==5.4.1

ydb/_topic_reader/datatypes_test.py

@@ -175,7 +175,7 @@ def add_notify(future, notified_offset):
             ),
         ],
     )
-    def test_add_waiter(
+    async def test_add_waiter(
         self,
         session,
         original: List[PartitionSession.CommitAckWaiter],
@@ -188,13 +188,13 @@ def test_add_waiter(
         assert result == session._ack_waiters
         assert res.future.done() == is_done
 
-    def test_close_notify_waiters(self, session):
+    async def test_close_notify_waiters(self, session):
         waiter = session.add_waiter(session.committed_offset + 1)
         session.close()
 
         with pytest.raises(topic_reader_asyncio.TopicReaderCommitToExpiredPartition):
             waiter.future.result()
 
-    def test_close_twice(self, session):
+    async def test_close_twice(self, session):
         session.close()
         session.close()

ydb/_topic_reader/topic_reader_asyncio.py

@@ -606,7 +606,8 @@ async def flush(self):
         for session in self._partition_sessions.values():
             futures.extend(w.future for w in session._ack_waiters)
 
-        await asyncio.gather(*futures)
+        if futures:
+            await asyncio.wait(futures)
 
     async def close(self):
         if self._closed:
@@ -622,4 +623,6 @@ async def close(self):
 
         for task in self._background_tasks:
             task.cancel()
-        await asyncio.wait(self._background_tasks)
+
+        if self._background_tasks:
+            await asyncio.wait(self._background_tasks)

ydb/_topic_reader/topic_reader_asyncio_test.py

@@ -411,6 +411,8 @@ async def test_commit_ranges_for_received_messages(
         received = stream_reader_started.receive_batch_nowait().messages
         assert received == [m2]
 
+        await stream_reader_started.close()
+
     # noinspection PyTypeChecker
     @pytest.mark.parametrize(
         "batch,data_out",

ydb/_topic_writer/topic_writer_asyncio.py

@@ -384,17 +384,21 @@ async def _connection_loop(self):
                     await stream_writer.close()
                 for task in tasks:
                     task.cancel()
-                await asyncio.wait(tasks)
+                if tasks:
+                    await asyncio.wait(tasks)
 
     async def _encode_loop(self):
-        while True:
-            messages = await self._messages_for_encode.get()
-            while not self._messages_for_encode.empty():
-                messages.extend(self._messages_for_encode.get_nowait())
+        try:
+            while True:
+                messages = await self._messages_for_encode.get()
+                while not self._messages_for_encode.empty():
+                    messages.extend(self._messages_for_encode.get_nowait())
 
-            batch_codec = await self._codec_selector(messages)
-            await self._encode_data_inplace(batch_codec, messages)
-            self._add_messages_to_send_queue(messages)
+                batch_codec = await self._codec_selector(messages)
+                await self._encode_data_inplace(batch_codec, messages)
+                self._add_messages_to_send_queue(messages)
+        except BaseException as err:
+            self._stop(err)
 
     async def _encode_data_inplace(self, codec: PublicCodec, messages: List[InternalMessage]):
         if codec == PublicCodec.RAW:
@@ -531,10 +535,9 @@ async def _send_loop(self, writer: "WriterAsyncIOStream"):
                     writer.write([m])
         except Exception as e:
             self._stop(e)
-        finally:
-            pass
+            raise
 
-    def _stop(self, reason: Exception):
+    def _stop(self, reason: BaseException):
         if reason is None:
             raise Exception("writer stop reason can not be None")
 
@@ -554,7 +557,7 @@ async def flush(self):
             return
 
         # wait last message
-        await asyncio.wait((self._messages_future[-1],))
+        await asyncio.wait(self._messages_future)
 
 
 class WriterAsyncIOStream:
@@ -647,7 +650,7 @@ async def _start(self, stream: IGrpcWrapperAsyncIO, init_message: StreamWriteMes
     @staticmethod
     def _ensure_ok(message: WriterMessagesFromServerToClient):
         if not message.status.is_success():
-            raise TopicWriterError("status error from server in writer: %s", message.status)
+            raise TopicWriterError(f"status error from server in writer: {message.status}")
 
     def write(self, messages: List[InternalMessage]):
         if self._closed:

ydb/_topic_writer/topic_writer_asyncio_test.py

@@ -183,6 +183,8 @@ async def test_update_token(self, stream: StreamMock):
         receive_task.cancel()
         await asyncio.wait([receive_task])
 
+        await writer.close()
+
 
 @pytest.mark.asyncio
 class TestWriterAsyncIOReconnector:
@@ -439,8 +441,11 @@ async def test_auto_seq_no(self, default_driver, default_settings, get_stream_wr
 
         await reconnector.close(flush=False)
 
-    async def test_deny_double_seqno(self, reconnector: WriterAsyncIOReconnector):
+    async def test_deny_double_seqno(self, reconnector: WriterAsyncIOReconnector, get_stream_writer):
+        writer = get_stream_writer()
+
         await reconnector.write_with_ack_future([PublicMessage(seqno=10, data="123")])
+        writer.from_server.put_nowait(self.make_default_ack_message(seq_no=10))
 
         with pytest.raises(TopicWriterError):
             await reconnector.write_with_ack_future([PublicMessage(seqno=9, data="123")])
@@ -449,8 +454,9 @@ async def test_deny_double_seqno(self, reconnector: WriterAsyncIOReconnector):
             await reconnector.write_with_ack_future([PublicMessage(seqno=10, data="123")])
 
         await reconnector.write_with_ack_future([PublicMessage(seqno=11, data="123")])
+        writer.from_server.put_nowait(self.make_default_ack_message(seq_no=11))
 
-        await reconnector.close(flush=False)
+        await reconnector.close(flush=True)
 
     @freezegun.freeze_time("2022-01-13 20:50:00", tz_offset=0)
     async def test_auto_created_at(self, default_driver, default_settings, get_stream_writer):
@@ -571,6 +577,8 @@ async def test_encode_data_inplace(
             assert mess.codec == codec
             assert mess.get_bytes() == expected_datas[index]
 
+        await reconnector.close(flush=True)
+
     async def test_custom_encoder(self, default_driver, default_settings, get_stream_writer):
         codec = 10001