forked from goodboy/tractor
				
			Port remaining tests to pass func refs
							parent
							
								
									493f2efb50
								
							
						
					
					
						commit
						0eba5f4708
					
				| 
						 | 
				
			
			@ -137,7 +137,7 @@ def test_cancel_single_subactor(arb_addr, mechanism):
 | 
			
		|||
            portal = await nursery.start_actor(
 | 
			
		||||
                'nothin', rpc_module_paths=[__name__],
 | 
			
		||||
            )
 | 
			
		||||
            assert (await portal.run(__name__, 'do_nothing')) is None
 | 
			
		||||
            assert (await portal.run(do_nothing)) is None
 | 
			
		||||
 | 
			
		||||
            if mechanism == 'nursery_cancel':
 | 
			
		||||
                # would hang otherwise
 | 
			
		||||
| 
						 | 
				
			
			@ -173,7 +173,7 @@ async def test_cancel_infinite_streamer(start_method):
 | 
			
		|||
 | 
			
		||||
            # this async for loop streams values from the above
 | 
			
		||||
            # async generator running in a separate process
 | 
			
		||||
            async for letter in await portal.run(__name__, 'stream_forever'):
 | 
			
		||||
            async for letter in await portal.run(stream_forever):
 | 
			
		||||
                print(letter)
 | 
			
		||||
 | 
			
		||||
    # we support trio's cancellation system
 | 
			
		||||
| 
						 | 
				
			
			@ -247,7 +247,8 @@ async def test_some_cancels_all(num_actors_and_errs, start_method, loglevel):
 | 
			
		|||
                    # if this function fails then we should error here
 | 
			
		||||
                    # and the nursery should teardown all other actors
 | 
			
		||||
                    try:
 | 
			
		||||
                        await portal.run(__name__, func.__name__, **kwargs)
 | 
			
		||||
                        await portal.run(func, **kwargs)
 | 
			
		||||
 | 
			
		||||
                    except tractor.RemoteActorError as err:
 | 
			
		||||
                        assert err.type == err_type
 | 
			
		||||
                        # we only expect this first error to propogate
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -58,7 +58,7 @@ async def subs(
 | 
			
		|||
 | 
			
		||||
    async with tractor.find_actor(pub_actor_name) as portal:
 | 
			
		||||
        stream = await portal.run(
 | 
			
		||||
            __name__, 'pubber',
 | 
			
		||||
            pubber,
 | 
			
		||||
            topics=which,
 | 
			
		||||
            seed=seed,
 | 
			
		||||
        )
 | 
			
		||||
| 
						 | 
				
			
			@ -76,7 +76,7 @@ async def subs(
 | 
			
		|||
        await stream.aclose()
 | 
			
		||||
 | 
			
		||||
        stream = await portal.run(
 | 
			
		||||
            __name__, 'pubber',
 | 
			
		||||
            pubber,
 | 
			
		||||
            topics=['odd'],
 | 
			
		||||
            seed=seed,
 | 
			
		||||
        )
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -70,9 +70,9 @@ async def test_movie_theatre_convo(start_method):
 | 
			
		|||
            rpc_module_paths=[__name__],
 | 
			
		||||
        )
 | 
			
		||||
 | 
			
		||||
        print(await portal.run(__name__, 'movie_theatre_question'))
 | 
			
		||||
        print(await portal.run(movie_theatre_question))
 | 
			
		||||
        # call the subactor a 2nd time
 | 
			
		||||
        print(await portal.run(__name__, 'movie_theatre_question'))
 | 
			
		||||
        print(await portal.run(movie_theatre_question))
 | 
			
		||||
 | 
			
		||||
        # the async with will block here indefinitely waiting
 | 
			
		||||
        # for our actor "frank" to complete, we cancel 'frank'
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -50,7 +50,7 @@ async def context_stream(ctx, sequence):
 | 
			
		|||
    assert cs.cancelled_caught
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
async def stream_from_single_subactor(stream_func_name):
 | 
			
		||||
async def stream_from_single_subactor(stream_func):
 | 
			
		||||
    """Verify we can spawn a daemon actor and retrieve streamed data.
 | 
			
		||||
    """
 | 
			
		||||
    async with tractor.find_actor('streamerd') as portals:
 | 
			
		||||
| 
						 | 
				
			
			@ -67,8 +67,7 @@ async def stream_from_single_subactor(stream_func_name):
 | 
			
		|||
                seq = range(10)
 | 
			
		||||
 | 
			
		||||
                stream = await portal.run(
 | 
			
		||||
                    __name__,
 | 
			
		||||
                    stream_func_name,  # one of the funcs above
 | 
			
		||||
                    stream_func,  # one of the funcs above
 | 
			
		||||
                    sequence=list(seq),  # has to be msgpack serializable
 | 
			
		||||
                )
 | 
			
		||||
                # it'd sure be nice to have an asyncitertools here...
 | 
			
		||||
| 
						 | 
				
			
			@ -96,7 +95,7 @@ async def stream_from_single_subactor(stream_func_name):
 | 
			
		|||
 | 
			
		||||
 | 
			
		||||
@pytest.mark.parametrize(
 | 
			
		||||
    'stream_func', ['async_gen_stream', 'context_stream']
 | 
			
		||||
    'stream_func', [async_gen_stream, context_stream]
 | 
			
		||||
)
 | 
			
		||||
def test_stream_from_single_subactor(arb_addr, start_method, stream_func):
 | 
			
		||||
    """Verify streaming from a spawned async generator.
 | 
			
		||||
| 
						 | 
				
			
			@ -104,7 +103,7 @@ def test_stream_from_single_subactor(arb_addr, start_method, stream_func):
 | 
			
		|||
    tractor.run(
 | 
			
		||||
        partial(
 | 
			
		||||
            stream_from_single_subactor,
 | 
			
		||||
            stream_func_name=stream_func,
 | 
			
		||||
            stream_func=stream_func,
 | 
			
		||||
        ),
 | 
			
		||||
        arbiter_addr=arb_addr,
 | 
			
		||||
        start_method=start_method,
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
		Loading…
	
		Reference in New Issue