Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_log_error_when_not_abs_path_for_configuation_path_was_used(moler_config):
from moler.exceptions import MolerException
conn_config = os.path.join(os.pardir, "resources", "device_config.yml")
with pytest.raises(MolerException) as err:
moler_config.load_config(config=conn_config, config_type='yaml')
assert "Loading configuration requires absolute path and not '../resources/device_config.yml'" in str(err.value)
def test_log_error_when_not_abs_path_for_configuation_path_was_used(moler_config):
from moler.exceptions import MolerException
conn_config = os.path.join(os.pardir, "resources", "device_config.yml")
with pytest.raises(MolerException) as err:
moler_config.load_config(config=conn_config, config_type='yaml')
assert "Loading configuration requires absolute path and not '../resources/device_config.yml'" in str(err.value)
'UNIX_LOCAL_THE_SAME_PROMPTS': {
'DEVICE_CLASS': 'moler.device.unixlocal.UnixLocal',
'INITIAL_STATE': 'UNIX_LOCAL',
'CONNECTION_HOPS': {
"UNIX_LOCAL": {
"UNIX_LOCAL_ROOT": {
"command_params": {
"expected_prompt": "^moler_bash#"
}
}
}
}
}
}
}
with pytest.raises(MolerException) as err:
moler_config.load_config(config=conn_config, config_type='dict')
device_factory.get_device(name='UNIX_LOCAL_THE_SAME_PROMPTS')
assert "Incorrect device configuration. The same prompts for state" in str(err.value)
feed_started,
subscribed_data_receiver=None))
self.logger.debug("scheduled feed() - future: {}".format(conn_observer_future))
await feed_started.wait()
self.logger.debug("feed() started - future: {}:{}".format(instance_id(conn_observer_future),
conn_observer_future))
return conn_observer_future
thread4async = get_asyncio_loop_thread()
start_timeout = 0.5
try:
connection_observer_future = thread4async.run_async_coroutine(start_feeder(), timeout=start_timeout)
except MolerTimeout:
err_msg = "Failed to start observer feeder within {} sec".format(start_timeout)
self.logger.error(err_msg)
exc = MolerException(err_msg)
connection_observer.set_exception(exception=exc)
return None
self.logger.debug("runner submit() returning - future: {}:{}".format(instance_id(connection_observer_future),
connection_observer_future))
return connection_observer_future
def __init__(self, command, message):
err_msg = "Command '{}.{}' ('{}') failed with >>{}<<.".format(command.__class__.__module__, command.__class__.__name__,
command.command_string, message)
self.command = command
super(CommandFailure, self).__init__(err_msg)
class CommandWrongState(MolerException):
def __init__(self, command, expected_state, current_state):
err_msg = "Command '{}' tried to run in state '{}' but created in '{}'.".format(command.command_string,
current_state, expected_state)
self.command = command
super(CommandWrongState, self).__init__(err_msg)
class EventWrongState(MolerException):
def __init__(self, command, expected_state, current_state):
err_msg = "Event '{}' tried to run in state '{}' but created in '{}'.".format(command.command_string,
current_state, expected_state)
self.command = command
super(EventWrongState, self).__init__(err_msg)
class MolerStatusException(MolerException):
def __init__(self, msg):
super(MolerStatusException, self).__init__(msg)
class DeviceFailure(MolerException):
def __init__(self, device, message):
self.device = device
err_msg = "Device '{}' failed with '{}'.".format(device, message)
# TODO: do we need it? Just mapping to asyncio/concurrent.futures naming?
class CancelledError(MolerException):
pass
class NoResultSinceCancelCalled(CancelledError):
def __init__(self, connection_observer):
"""Create instance of NoResultSinceCancelCalled exception"""
err_msg = 'for {}'.format(connection_observer)
super(NoResultSinceCancelCalled, self).__init__(err_msg)
self.connection_observer = connection_observer
# TODO: do we need it? Just mapping to asyncio naming?
class InvalidStateError(MolerException):
pass
class ResultNotAvailableYet(InvalidStateError):
def __init__(self, connection_observer):
"""Create instance of ResultNotAvailableYet exception"""
err_msg = 'for {}'.format(connection_observer)
super(ResultNotAvailableYet, self).__init__(err_msg)
self.connection_observer = connection_observer
class ConnectionObserverNotStarted(InvalidStateError):
def __init__(self, connection_observer):
"""Create instance of ConnectionObserverNotStarted exception"""
err_msg = 'for {}'.format(connection_observer)
super(ConnectionObserverNotStarted, self).__init__(err_msg)
def read_yaml_configfile(path):
"""
Read and convert YAML into dictionary
:param path: location of yaml file
:return: configuration as a python dictionary
"""
if os.path.isabs(path):
with read_configfile(path) as content:
return yaml.load(content, Loader=yaml.FullLoader)
else:
error = "Loading configuration requires absolute path and not '{}'".format(path)
raise MolerException(error)
class CommandTimeout(ConnectionObserverTimeout):
pass
class NoCommandStringProvided(MolerException):
def __init__(self, command):
"""Create instance of NoCommandStringProvided exception"""
fix_info = 'fill .command_string member before starting command'
err_msg = 'for {}\nYou should {}'.format(command, fix_info)
super(NoCommandStringProvided, self).__init__(err_msg)
self.command = command
class NoDetectPatternProvided(MolerException):
def __init__(self, command):
"""Create instance of NoDetectPatternProvided exception"""
fix_info = 'fill .detect_patterns member before starting event'
err_msg = 'for {}\nYou should {}'.format(command, fix_info)
super(NoDetectPatternProvided, self).__init__(err_msg)
self.command = command
class NoConnectionProvided(MolerException):
def __init__(self, connection_observer):
"""Create instance of NoConnectionProvided exception"""
err_msg = 'for {}'.format(connection_observer)
super(NoConnectionProvided, self).__init__(err_msg)
self.connection_observer = connection_observer
random.shuffle(source_states)
random.shuffle(target_states)
print(states)
print(len(states) * len(states))
for source_state in source_states:
for target_state in target_states:
try:
print('{} -> {}'.format(device.state, source_state))
device.goto_state(source_state)
print('{} -> {}'.format(source_state, target_state))
device.goto_state(target_state)
except Exception as exc:
raise MolerException(
"Cannot trigger change state: '{}' -> '{}'\n{}".format(source_state, target_state, exc))
"""Create instance of NoDetectPatternProvided exception"""
fix_info = 'fill .detect_patterns member before starting event'
err_msg = 'for {}\nYou should {}'.format(command, fix_info)
super(NoDetectPatternProvided, self).__init__(err_msg)
self.command = command
class NoConnectionProvided(MolerException):
def __init__(self, connection_observer):
"""Create instance of NoConnectionProvided exception"""
err_msg = 'for {}'.format(connection_observer)
super(NoConnectionProvided, self).__init__(err_msg)
self.connection_observer = connection_observer
class CommandFailure(MolerException):
def __init__(self, command, message):
err_msg = "Command '{}.{}' ('{}') failed with >>{}<<.".format(command.__class__.__module__, command.__class__.__name__,
command.command_string, message)
self.command = command
super(CommandFailure, self).__init__(err_msg)
class CommandWrongState(MolerException):
def __init__(self, command, expected_state, current_state):
err_msg = "Command '{}' tried to run in state '{}' but created in '{}'.".format(command.command_string,
current_state, expected_state)
self.command = command
super(CommandWrongState, self).__init__(err_msg)
class EventWrongState(MolerException):