80 lines
2.2 KiB
Python
80 lines
2.2 KiB
Python
# py-scopes: tests.test_core
|
|
|
|
""" Tests for scopes.core modules, including e.g. actor, message"""
|
|
|
|
import asyncio
|
|
import pytest
|
|
|
|
from scopes import core
|
|
from scopes.core import actor, environ
|
|
from scopes.core.message import Message
|
|
from scopes.util import keywords as k
|
|
|
|
@pytest.fixture
|
|
def collector():
|
|
return []
|
|
|
|
|
|
def show(msg, coll):
|
|
coll.append(msg)
|
|
|
|
def handle_basic(coll):
|
|
async def handle(msg):
|
|
match msg:
|
|
case k.show: show(msg, coll)
|
|
case k.switch: return handle_alt(coll)
|
|
case _: show(msg, coll)
|
|
return handle
|
|
|
|
def handle_alt(coll):
|
|
async def handle(msg):
|
|
match msg:
|
|
case k.show: show('switched', coll)
|
|
case _: await asyncio.sleep(0.01); show(msg, coll)
|
|
return handle
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_actor(collector):
|
|
async with asyncio.TaskGroup() as group:
|
|
q1 = actor.create(handle_basic(collector), group)
|
|
q2 = actor.create(handle_basic(collector), group)
|
|
msgs = [Message(k.test, k.hello), k.show, k.switch, k.show, "actor 1: Goodbye"]
|
|
await asyncio.gather(*[q1.put(m) for m in msgs])
|
|
await q2.put("actor 2")
|
|
await q2.put(k.show)
|
|
await q1.put(k.quit)
|
|
await q2.put(k.quit)
|
|
#print(collector)
|
|
assert len(collector) == 6
|
|
assert collector[-1] == 'actor 1: Goodbye'
|
|
|
|
|
|
def test_message():
|
|
msg = Message(k.test, k.data, 'person', '4711',
|
|
data=dict(firstName='Bob', lastName='Miller'))
|
|
assert msg.head['class'] is k.person
|
|
assert msg.addr() == (k.test, k.person, '4711')
|
|
assert msg.categ() == (k.test, k.person)
|
|
|
|
|
|
@pytest.fixture
|
|
async def meta_env(collector):
|
|
async def collect(msg):
|
|
collector.append(msg)
|
|
return actor.create(collect)
|
|
|
|
@pytest.mark.anyio
|
|
async def test_core(meta_env, collector):
|
|
env = environ.create(meta_env)
|
|
await env.put(Message(k.env, k.hello))
|
|
await asyncio.sleep(0.01)
|
|
assert len(collector) == 1
|
|
assert collector[0].domain is k.env
|
|
zero = core.create_zero(env)
|
|
await zero.put(Message(k.csys, k.hello))
|
|
await asyncio.sleep(0.01)
|
|
assert len(collector) == 2
|
|
assert collector[1].domain is k.csys
|
|
#print(collector)
|
|
|