Add 6sec timeout around `test_simple_rpc` suite for macos

ns_aware
Gud Boi 2026-02-24 20:28:21 -05:00
parent c5af2fa778
commit 706a4b761b
1 changed files with 53 additions and 41 deletions

View File

@ -1,15 +1,25 @@
"""
Bidirectional streaming.
'''
Audit the simplest inter-actor bidirectional (streaming)
msg patterns.
"""
'''
from __future__ import annotations
from typing import (
Callable,
TYPE_CHECKING,
)
import pytest
import trio
import tractor
if TYPE_CHECKING:
from tractor import (
Portal,
)
@tractor.context
async def simple_rpc(
ctx: tractor.Context,
data: int,
@ -39,15 +49,14 @@ async def simple_rpc(
@tractor.context
async def simple_rpc_with_forloop(
ctx: tractor.Context,
data: int,
) -> None:
"""Same as previous test but using ``async for`` syntax/api.
"""
'''
Same as previous test but using `async for` syntax/api.
'''
# signal to parent that we're up
await ctx.started(data + 1)
@ -74,15 +83,18 @@ async def simple_rpc_with_forloop(
'server_func',
[simple_rpc, simple_rpc_with_forloop],
)
def test_simple_rpc(server_func, use_async_for):
def test_simple_rpc(
server_func: Callabe,
use_async_for: bool,
):
'''
The simplest request response pattern.
'''
async def main():
async with tractor.open_nursery() as n:
portal = await n.start_actor(
with trio.fail_after(6):
async with tractor.open_nursery() as an:
portal: Portal = await an.start_actor(
'rpc_server',
enable_modules=[__name__],
)