73 lines
1.8 KiB
Python
73 lines
1.8 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, 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)
|
|
#print(collector)
|
|
assert len(collector) == 6
|
|
assert collector[-1] == 'actor 1: Goodbye'
|
|
|
|
|
|
def test_message():
|
|
msg = message.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
|
|
def coll():
|
|
return []
|
|
|
|
@pytest.fixture
|
|
async def meta_env(coll):
|
|
async def collect(msg):
|
|
coll.append(msg)
|
|
return actor.create(collect)
|
|
|
|
@pytest.mark.anyio
|
|
async def test_core(meta_env, coll):
|
|
env_scope = environ.create(meta_env)
|
|
assert env_scope.value == 0
|
|
#print(meta_env)
|
|
await asyncio.sleep(0.01)
|
|
|