59 lines
1.5 KiB
Python
59 lines
1.5 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, 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)
|
|
|
|
|
|
def test_core():
|
|
sc = core.Scope()
|
|
assert sc.value == 0
|
|
|