45 lines
1.1 KiB
Python
45 lines
1.1 KiB
Python
# py-scopes: tests.test_actor
|
|
|
|
""" Tests for scopes.core.actor"""
|
|
|
|
import asyncio
|
|
import pytest
|
|
|
|
from scopes.core import actor, message
|
|
from scopes.util import keywords as k
|
|
|
|
collector = []
|
|
|
|
def show(msg):
|
|
collector.append(msg)
|
|
|
|
async def handle_basic(msg):
|
|
match msg:
|
|
case k.show: show(msg)
|
|
case k.switch: return handle_alt
|
|
case _: show(msg)
|
|
|
|
async def handle_alt(msg):
|
|
match msg:
|
|
case k.show: show('switched')
|
|
case _: await asyncio.sleep(0.01); show(msg)
|
|
|
|
@pytest.mark.anyio
|
|
async def test_actor():
|
|
async with asyncio.TaskGroup() as group:
|
|
q1 = actor.create(handle_basic, group)
|
|
q2 = actor.create(handle_basic, group)
|
|
await q1.put(message.Message(k.test, k.hello))
|
|
await q1.put(k.show)
|
|
await q1.put(k.switch)
|
|
await q1.put(k.show)
|
|
await q1.put("actor 1: Goodbye")
|
|
await q2.put("actor 2")
|
|
await q2.put(k.show)
|
|
await q1.put(k.quit)
|
|
await q2.put(k.quit)
|
|
#await asyncio.wait([t1, t2])
|
|
print(collector)
|
|
assert len(collector) == 6
|
|
assert collector[-1] == 'actor 1: Goodbye'
|
|
|