@@ -702,10 +702,11 @@ async def test_read_session_termination(self, stream: S2Stream):
702702 )
703703
704704 batches = []
705- async for batch in stream .read_session (
705+ async with stream .read_session (
706706 start = SeqNum (0 ), limit = ReadLimit (count = 2 )
707- ):
708- batches .append (batch )
707+ ) as session :
708+ async for batch in session :
709+ batches .append (batch )
709710
710711 assert len (batches ) >= 1
711712 assert len (batches [0 ].records ) == 2
@@ -717,8 +718,9 @@ async def test_read_session_read_existing_then_tails(self, stream: S2Stream):
717718
718719 with pytest .raises (asyncio .TimeoutError ):
719720 async with asyncio .timeout (2 ):
720- async for batch in stream .read_session (start = SeqNum (0 )):
721- received .extend (batch .records )
721+ async with stream .read_session (start = SeqNum (0 )) as session :
722+ async for batch in session :
723+ received .extend (batch .records )
722724
723725 assert len (received ) == 2
724726 assert received [0 ].body == b"a"
@@ -734,8 +736,9 @@ async def append_later():
734736 try :
735737 with pytest .raises (asyncio .TimeoutError ):
736738 async with asyncio .timeout (5 ):
737- async for batch in stream .read_session (start = SeqNum (0 )):
738- received .extend (batch .records )
739+ async with stream .read_session (start = SeqNum (0 )) as session :
740+ async for batch in session :
741+ received .extend (batch .records )
739742 finally :
740743 append_later_task .cancel ()
741744 with suppress (asyncio .CancelledError ):
@@ -746,24 +749,27 @@ async def append_later():
746749 async def test_read_session_tails (self , stream : S2Stream ):
747750 with pytest .raises (asyncio .TimeoutError ):
748751 async with asyncio .timeout (1 ):
749- async for _ in stream .read_session (start = SeqNum (0 )):
750- pass
752+ async with stream .read_session (start = SeqNum (0 )) as session :
753+ async for _ in session :
754+ pass
751755
752756 async def test_read_session_clamp_to_tail_tails (self , stream : S2Stream ):
753757 await stream .append (AppendInput (records = [Record (body = b"data" )]))
754758 received = []
755759 with pytest .raises (asyncio .TimeoutError ):
756760 async with asyncio .timeout (1 ):
757- async for batch in stream .read_session (
761+ async with stream .read_session (
758762 start = SeqNum (100 ), clamp_to_tail = True
759- ):
760- received .extend (batch .records )
763+ ) as session :
764+ async for batch in session :
765+ received .extend (batch .records )
761766 assert received == []
762767
763768 async def test_read_session_beyond_tail_errors (self , stream : S2Stream ):
764769 with pytest .raises (ReadUnwrittenError ) as exc_info :
765- async for _ in stream .read_session (start = SeqNum (100 )):
766- pass
770+ async with stream .read_session (start = SeqNum (100 )) as session :
771+ async for _ in session :
772+ pass
767773 assert exc_info .value .tail .seq_num == 0
768774
769775
0 commit comments