Source code for cis_interface.tests.test_runner

import os
import unittest
import signal
import uuid
from cis_interface import runner, tools, platform
from cis_interface.tests import CisTestBase, assert_raises
# from cis_interface.tests import yamls as sc_yamls
from cis_interface.examples import yamls as ex_yamls


[docs]def test_get_runner(): r"""Use get_runner to start a run.""" namespace = "test_get_runner_%s" % str(uuid.uuid4) cr = runner.get_runner([ex_yamls['hello']['python']], namespace=namespace) cr.run() cr.sleep()
# def test_runner_error(): # r"""Start a runner for a model with an error.""" # cr = runner.get_runner([sc_yamls['error']]) # cr.run() # Spawning fake Ctrl-C works locally for windows, but causes hang on appveyor
[docs]@unittest.skipIf(platform._is_win, "Signal processing not sorted on windows") def test_Arunner_interrupt(): r"""Start a runner then stop it with a keyboard interrupt.""" cr = runner.get_runner([ex_yamls['hello']['python']]) if platform._is_win: # pragma: debug cr.debug_log() cr.loadDrivers() cr.startDrivers() cr.set_signal_handler() tools.kill(os.getpid(), signal.SIGINT) tools.kill(os.getpid(), signal.SIGINT) cr.reset_signal_handler() cr.waitModels() cr.closeChannels() cr.cleanup() if platform._is_win: # pragma: debug cr.reset_log()
[docs]def test_runner_terminate(): r"""Start a runner, then stop it early.""" cr = runner.get_runner([ex_yamls['hello']['python']]) cr.loadDrivers() cr.startDrivers() cr.printStatus() cr.terminate()
[docs]def test_runner_error(): r"""Test error on missing yaml.""" assert_raises(IOError, runner.CisRunner, ['fake_yaml.yml'], 'test_cis_run')
[docs]class TestCisRunner(CisTestBase): r"""Tests of the CisRunner class."""
[docs] def setup(self, *args, **kwargs): super(TestCisRunner, self).setup(*args, **kwargs) self.runner = runner.CisRunner([ex_yamls['hello']['python']], 'test_cis_run')
[docs] def test_createIODriver(self): r"""Test createInputDriver and createOutputDriver.""" yml = {'name': 'fake_IODriver', 'args': 'fake_channel', 'driver': 'InputDriver', 'working_dir': os.getcwd(), 'icomm_kws': {'comm': [{'name': 'fake_IODriver'}]}, 'ocomm_kws': {'comm': [{'name': 'fake_IODriver'}]}, 'kwargs': {}} assert_raises(ValueError, self.runner.createInputDriver, yml) yml['driver'] = 'OutputDriver' assert_raises(ValueError, self.runner.createOutputDriver, yml)