forked from goodboy/tractor
Initial examples testing attempt
A per #98 we need tests for examples from the docs as they would be run by a user copy and pasting the code. This adds a small system for loading examples from an "examples/" directory and executing them in a subprocess while checking the output. We can use this to also verify end-to-end expected logging output on std streams (ex. logging on stderr). To expand this further we can parameterize the test list using the contents of the examples directory instead of hardcoding the script names as I've done here initially. Also, fix up the current readme examples to have the required/proper `if __name__ == '__main__'` script guard.example_tests
parent
4bd3a14a68
commit
c6f3ab5ae2
|
@ -0,0 +1,44 @@
|
|||
import tractor
|
||||
from functools import partial
|
||||
|
||||
_this_module = __name__
|
||||
the_line = 'Hi my name is {}'
|
||||
|
||||
|
||||
tractor.log.get_console_log("INFO")
|
||||
|
||||
|
||||
async def hi():
|
||||
return the_line.format(tractor.current_actor().name)
|
||||
|
||||
|
||||
async def say_hello(other_actor):
|
||||
async with tractor.wait_for_actor(other_actor) as portal:
|
||||
return await portal.run(_this_module, 'hi')
|
||||
|
||||
|
||||
async def main():
|
||||
"""Main tractor entry point, the "master" process (for now
|
||||
acts as the "director").
|
||||
"""
|
||||
async with tractor.open_nursery() as n:
|
||||
print("Alright... Action!")
|
||||
|
||||
donny = await n.run_in_actor(
|
||||
'donny',
|
||||
say_hello,
|
||||
# arguments are always named
|
||||
other_actor='gretchen',
|
||||
)
|
||||
gretchen = await n.run_in_actor(
|
||||
'gretchen',
|
||||
say_hello,
|
||||
other_actor='donny',
|
||||
)
|
||||
print(await gretchen.result())
|
||||
print(await donny.result())
|
||||
print("CUTTTT CUUTT CUT!!! Donny!! You're supposed to say...")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
tractor.run(main)
|
|
@ -0,0 +1,65 @@
|
|||
"""
|
||||
Let's make sure them docs work yah?
|
||||
"""
|
||||
from contextlib import contextmanager
|
||||
import os
|
||||
import sys
|
||||
import subprocess
|
||||
import platform
|
||||
import pprint
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture(scope='session')
|
||||
def confdir():
|
||||
dirname = os.path.dirname
|
||||
dirpath = os.path.abspath(
|
||||
dirname(dirname(os.path.realpath(__file__)))
|
||||
)
|
||||
return dirpath
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def run_example_in_subproc(loglevel, testdir, arb_addr):
|
||||
|
||||
@contextmanager
|
||||
def run(script_code):
|
||||
script_file = testdir.makefile('.py', script_code)
|
||||
cmdargs = [
|
||||
sys.executable,
|
||||
str(script_file),
|
||||
]
|
||||
kwargs = dict()
|
||||
if platform.system() == 'Windows':
|
||||
# without this, tests hang on windows forever
|
||||
kwargs['creationflags'] = subprocess.CREATE_NEW_PROCESS_GROUP
|
||||
|
||||
proc = testdir.popen(
|
||||
cmdargs,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
**kwargs,
|
||||
)
|
||||
assert not proc.returncode
|
||||
yield proc
|
||||
proc.wait()
|
||||
assert proc.returncode == 0
|
||||
|
||||
yield run
|
||||
|
||||
|
||||
def test_a_trynamic_first_scene(confdir, run_example_in_subproc):
|
||||
ex_file = os.path.join(confdir, 'examples', 'a_trynamic_first_scene.py')
|
||||
with open(ex_file, 'r') as ex:
|
||||
code = ex.read()
|
||||
|
||||
with run_example_in_subproc(code) as proc:
|
||||
proc.wait()
|
||||
err, _ = proc.stderr.read(), proc.stdout.read()
|
||||
|
||||
# if we get some gnarly output let's aggregate and raise
|
||||
if err and b'Error' in err:
|
||||
raise Exception(err.decode())
|
||||
|
||||
assert proc.returncode == 0
|
Loading…
Reference in New Issue