How to use the moler.device.unixlocal.UnixLocal function in moler

To help you get started, we’ve selected a few moler examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github nokia / moler / test / device / test_device.py View on Github external
def test_device_add_neighbour_device_without_bidirectional(buffer_connection):
    from moler.device.unixlocal import UnixLocal
    dev1 = UnixLocal(io_connection=buffer_connection)
    dev2 = UnixLocal(io_connection=buffer_connection)

    dev1.add_neighbour_device(neighbour_device=dev2, bidirectional=False)

    neighbour_devices = dev1.get_neighbour_devices(device_type=UnixLocal)
    assert 1 == len(neighbour_devices)

    neighbour_devices = dev2.get_neighbour_devices(device_type=UnixLocal)
    assert 0 == len(neighbour_devices)
github nokia / moler / moler / device / proxy_pc.py View on Github external
def _get_packages_for_state(self, state, observer):
        """
        Get available packages contain cmds and events for each state.
        :param state: device state.
        :param observer: observer type, available: cmd, events
        :return: available cmds or events for specific device state.
        """
        available = super(ProxyPc, self)._get_packages_for_state(state, observer)

        if not available:
            if state == ProxyPc.proxy_pc:
                available = {UnixLocal.cmds: ['moler.cmd.unix'],
                             UnixLocal.events: ['moler.events.shared', 'moler.events.unix']}
            if available:
                return available[observer]

        return available
github nokia / moler / moler / device / proxy_pc.py View on Github external
def _prepare_state_hops_with_proxy_pc(self):
        """
        Prepare non direct transitions for each state for State Machine with proxy_pc state.
        :return: non direct transitions for each state with proxy_pc state.
        """
        state_hops = {
            UnixLocal.not_connected: {
                ProxyPc.proxy_pc: ProxyPc.unix_local,
            },
            UnixLocal.unix_local_root: {
                ProxyPc.proxy_pc: ProxyPc.unix_local,
                ProxyPc.not_connected: ProxyPc.unix_local,
            },
            ProxyPc.proxy_pc: {
                ProxyPc.not_connected: ProxyPc.unix_local,
                ProxyPc.unix_local_root: ProxyPc.unix_local,
            },
        }
        return state_hops
github nokia / moler / moler / device / unixlocal.py View on Github external
def _prepare_state_prompts(self):
        """
        Prepare textual prompt for each state.
        :return: None.
        """
        super(UnixLocal, self)._prepare_state_prompts()

        state_prompts = {
            UnixLocal.unix_local:
                self._configurations[UnixLocal.connection_hops][UnixLocal.unix_local_root][UnixLocal.unix_local][
                    "command_params"]["expected_prompt"],
            UnixLocal.unix_local_root:
                self._configurations[UnixLocal.connection_hops][UnixLocal.unix_local][UnixLocal.unix_local_root][
                    "command_params"]["expected_prompt"],
        }
        self._update_dict(self._state_prompts, state_prompts)
github nokia / moler / moler / device / unixlocal.py View on Github external
def _get_packages_for_state(self, state, observer):
        """
        Get available packages contain cmds and events for each state.
        :param state: device state.
        :param observer: observer type, available: cmd, events
        :return: available cmds or events for specific device state.
        """
        available = list()
        if state == UnixLocal.unix_local or state == UnixLocal.unix_local_root:
            available = {UnixLocal.cmds: ['moler.cmd.unix'],
                         UnixLocal.events: ['moler.events.shared', 'moler.events.unix']}
            return available[observer]
        return available
github nokia / moler / moler / device / atremote.py View on Github external
def _prepare_state_hops_without_proxy_pc(self):
        """
        Prepare non direct transitions for each state for State Machine without proxy_pc state.
        :return: non direct transitions for each state without proxy_pc state.
        """
        state_hops = {
            TextualDevice.not_connected: {
                UnixLocal.unix_local_root: UnixLocal.unix_local,
                UnixRemote.unix_remote: UnixLocal.unix_local,
                UnixRemote.unix_remote_root: UnixLocal.unix_local,
                AtRemote.at_remote: UnixLocal.unix_local,
            },
            UnixLocal.unix_local: {
                UnixRemote.unix_remote_root: UnixRemote.unix_remote,
                AtRemote.at_remote: UnixRemote.unix_remote,
            },
            UnixLocal.unix_local_root: {
                TextualDevice.not_connected: UnixLocal.unix_local,
                UnixRemote.unix_remote: UnixLocal.unix_local,
                UnixRemote.unix_remote_root: UnixLocal.unix_local,
                AtRemote.at_remote: UnixLocal.unix_local,
            },
            UnixRemote.unix_remote: {
                TextualDevice.not_connected: UnixLocal.unix_local,
github nokia / moler / moler / device / proxy_pc2.py View on Github external
def want_local_unix_state(io_type=None, io_connection=None):
    """
    Check if device is intended to work with local machine or remote ones only.
    :return: True for local.
    """
    if io_type == "terminal":
        return True
    if isinstance(io_connection, ThreadedTerminal):
        return True
    else:  # all remote-access connections (tcp, udp, telnet, ssh); even connecting to localhost
        return False


@six.add_metaclass(abc.ABCMeta)
class ProxyPc2(UnixLocal):

    def __init__(self, sm_params, name=None, io_connection=None, io_type=None, variant=None, io_constructor_kwargs=None,
                 initial_state=None):
        """
        Create Unix device communicating over io_connection
        :param sm_params: dict with parameters of state machine for device
        :param name: name of device
        :param io_connection: External-IO connection having embedded moler-connection
        :param io_type: type of connection - tcp, udp, ssh, telnet, ...
        :param variant: connection implementation variant, ex. 'threaded', 'twisted', 'asyncio', ...
                        (if not given then default one is taken)
        :param io_constructor_kwargs: additional parameter into constructor of selected connection type
                        (if not given then default one is taken)
        :param initial_state: name of initial state. State machine tries to enter this state just after creation.
        """
        self._use_local_unix_state = want_local_unix_state(io_type, io_connection)
github nokia / moler / moler / device / proxy_pc.py View on Github external
def _prepare_state_hops_with_proxy_pc(self):
        """
        Prepare non direct transitions for each state for State Machine with proxy_pc state.
        :return: non direct transitions for each state with proxy_pc state.
        """
        state_hops = {
            UnixLocal.not_connected: {
                ProxyPc.proxy_pc: ProxyPc.unix_local,
            },
            UnixLocal.unix_local_root: {
                ProxyPc.proxy_pc: ProxyPc.unix_local,
                ProxyPc.not_connected: ProxyPc.unix_local,
            },
            ProxyPc.proxy_pc: {
                ProxyPc.not_connected: ProxyPc.unix_local,
                ProxyPc.unix_local_root: ProxyPc.unix_local,
            },
        }
        return state_hops
github nokia / moler / moler / device / unixlocal.py View on Github external
def _prepare_state_prompts(self):
        """
        Prepare textual prompt for each state.
        :return: None.
        """
        super(UnixLocal, self)._prepare_state_prompts()

        state_prompts = {
            UnixLocal.unix_local:
                self._configurations[UnixLocal.connection_hops][UnixLocal.unix_local_root][UnixLocal.unix_local][
                    "command_params"]["expected_prompt"],
            UnixLocal.unix_local_root:
                self._configurations[UnixLocal.connection_hops][UnixLocal.unix_local][UnixLocal.unix_local_root][
                    "command_params"]["expected_prompt"],
        }
        self._update_dict(self._state_prompts, state_prompts)
github nokia / moler / moler / device / proxy_pc.py View on Github external
Moler's device has 2 main responsibilities:
- be the factory that returns commands of that device
- be the state machine that controls which commands may run in given state
"""

__author__ = 'Michal Ernst'
__copyright__ = 'Copyright (C) 2018-2019, Nokia'
__email__ = 'michal.ernst@nokia.com'
import six
import abc

from moler.device.unixlocal import UnixLocal


@six.add_metaclass(abc.ABCMeta)
class ProxyPc(UnixLocal):
    proxy_pc = "PROXY_PC"

    def __init__(self, sm_params, name=None, io_connection=None, io_type=None, variant=None, io_constructor_kwargs=None,
                 initial_state=None):
        """
        Create Unix device communicating over io_connection
        :param sm_params: dict with parameters of state machine for device
        :param name: name of device
        :param io_connection: External-IO connection having embedded moler-connection
        :param io_type: type of connection - tcp, udp, ssh, telnet, ...
        :param variant: connection implementation variant, ex. 'threaded', 'twisted', 'asyncio', ...
                        (if not given then default one is taken)
        :param io_constructor_kwargs: additional parameter into constructor of selected connection type
                        (if not given then default one is taken)
        :param initial_state: name of initial state. State machine tries to enter this state just after creation.
        """