forked from goodboy/tractor
				
			Add timeout guard around caller side context open
							parent
							
								
									b21f2e16ad
								
							
						
					
					
						commit
						808d7ae2c6
					
				| 
						 | 
					@ -265,42 +265,44 @@ async def test_callee_closes_ctx_after_stream_open():
 | 
				
			||||||
            enable_modules=[__name__],
 | 
					            enable_modules=[__name__],
 | 
				
			||||||
        )
 | 
					        )
 | 
				
			||||||
 | 
					
 | 
				
			||||||
        async with portal.open_context(
 | 
					        with trio.fail_after(2):
 | 
				
			||||||
            close_ctx_immediately,
 | 
					            async with portal.open_context(
 | 
				
			||||||
 | 
					                close_ctx_immediately,
 | 
				
			||||||
 | 
					
 | 
				
			||||||
            # flag to avoid waiting the final result
 | 
					                # flag to avoid waiting the final result
 | 
				
			||||||
            # cancel_on_exit=True,
 | 
					                # cancel_on_exit=True,
 | 
				
			||||||
 | 
					
 | 
				
			||||||
        ) as (ctx, sent):
 | 
					            ) as (ctx, sent):
 | 
				
			||||||
 | 
					
 | 
				
			||||||
            assert sent is None
 | 
					                assert sent is None
 | 
				
			||||||
 | 
					
 | 
				
			||||||
            with trio.fail_after(0.5):
 | 
					                with trio.fail_after(0.5):
 | 
				
			||||||
                async with ctx.open_stream() as stream:
 | 
					                    async with ctx.open_stream() as stream:
 | 
				
			||||||
 | 
					
 | 
				
			||||||
                    # should fall through since ``StopAsyncIteration``
 | 
					                        # should fall through since ``StopAsyncIteration``
 | 
				
			||||||
                    # should be raised through translation of
 | 
					                        # should be raised through translation of
 | 
				
			||||||
                    # a ``trio.EndOfChannel`` by
 | 
					                        # a ``trio.EndOfChannel`` by
 | 
				
			||||||
                    # ``trio.abc.ReceiveChannel.__anext__()``
 | 
					                        # ``trio.abc.ReceiveChannel.__anext__()``
 | 
				
			||||||
                    async for _ in stream:
 | 
					                        async for _ in stream:
 | 
				
			||||||
                        assert 0
 | 
					                            assert 0
 | 
				
			||||||
                    else:
 | 
					                        else:
 | 
				
			||||||
 | 
					
 | 
				
			||||||
                        # verify stream is now closed
 | 
					                            # verify stream is now closed
 | 
				
			||||||
                        try:
 | 
					                            try:
 | 
				
			||||||
                            await stream.receive()
 | 
					                                await stream.receive()
 | 
				
			||||||
                        except trio.EndOfChannel:
 | 
					                            except trio.EndOfChannel:
 | 
				
			||||||
 | 
					                                pass
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					                # TODO: should be just raise the closed resource err
 | 
				
			||||||
 | 
					                # directly here to enforce not allowing a re-open
 | 
				
			||||||
 | 
					                # of a stream to the context (at least until a time of
 | 
				
			||||||
 | 
					                # if/when we decide that's a good idea?)
 | 
				
			||||||
 | 
					                try:
 | 
				
			||||||
 | 
					                    with trio.fail_after(0.5):
 | 
				
			||||||
 | 
					                        async with ctx.open_stream() as stream:
 | 
				
			||||||
                            pass
 | 
					                            pass
 | 
				
			||||||
 | 
					                except trio.ClosedResourceError:
 | 
				
			||||||
            # TODO: should be just raise the closed resource err
 | 
					 | 
				
			||||||
            # directly here to enforce not allowing a re-open
 | 
					 | 
				
			||||||
            # of a stream to the context (at least until a time of
 | 
					 | 
				
			||||||
            # if/when we decide that's a good idea?)
 | 
					 | 
				
			||||||
            try:
 | 
					 | 
				
			||||||
                async with ctx.open_stream() as stream:
 | 
					 | 
				
			||||||
                    pass
 | 
					                    pass
 | 
				
			||||||
            except trio.ClosedResourceError:
 | 
					 | 
				
			||||||
                pass
 | 
					 | 
				
			||||||
 | 
					
 | 
				
			||||||
        await portal.cancel_actor()
 | 
					        await portal.cancel_actor()
 | 
				
			||||||
 | 
					
 | 
				
			||||||
| 
						 | 
					
 | 
				
			||||||
		Loading…
	
		Reference in New Issue