M tests/tasks.py +5 -0
@@ 6,6 6,11 @@ from rework import api, io
@api.task
+def basic(task):
+ print('This task is a 10. So basic...')
+
+
+@api.task
def print_sleep_and_go_away(task):
print('Hello, world')
time.sleep(.2)
M tests/test_cli.py +1 -0
@@ 19,6 19,7 @@ def test_list_operations(engine, cli, cl
r = cli('list-operations', engine.url)
assert """
+<X> `basic` default path(.../tests/tasks.py) host(<X>.<X>.<X>.<X>)
<X> `print_sleep_and_go_away` default path(.../tests/tasks.py) host(<X>.<X>.<X>.<X>)
<X> `run_in_non_default_domain` nondefault path(.../tests/tasks.py) host(<X>.<X>.<X>.<X>)
<X> `raw_input` default path(.../tests/tasks.py) host(<X>.<X>.<X>.<X>)
M tests/test_monitor.py +1 -0
@@ 41,6 41,7 @@ def test_basic_task_operations(engine, c
]
assert [
('allocate_and_leak_mbytes', 'tasks.py'),
+ ('basic', 'tasks.py'),
('capture_logs', 'tasks.py'),
('fancy_inputs_outputs', 'tasks.py'),
('flush_captured_stdout', 'tasks.py'),
A => tests/test_schedule.py +22 -0
@@ 0,0 1,22 @@
+from rework import api
+from rework.testutils import workers
+
+
+def test_web_scheduling(engine, cleanup, capsys):
+ api.prepare(
+ engine,
+ 'basic',
+ rule='* * * * * *',
+ metadata={'user': 'WEBUI'},
+ _anyrule = True
+ )
+ with workers(engine) as mon:
+ mon.step()
+ printed = capsys.readouterr()
+ error = (
+ " can\'t adapt type \'dict\'\n"
+ "[SQL: select id from rework.operation where name = %(name)s and modname = %(modname)s ]"
+ "\n[parameters: {\'name\': \'basic\', \'modname\': {\'user\': \'WEBUI\'}}]"
+ )
+
+ assert error in printed[-1]