labgrid.util package

Submodules

labgrid.util.agent module

labgrid.util.agent.b2s(b)[source]
labgrid.util.agent.s2b(s)[source]
class labgrid.util.agent.Agent[source]

Bases: object

__init__()[source]
register(name, func)[source]
run()[source]
__dict__ = mappingproxy({'_send': <staticmethod object>, '__init__': <function Agent.__init__>, '__weakref__': <attribute '__weakref__' of 'Agent' objects>, '__dict__': <attribute '__dict__' of 'Agent' objects>, 'run': <function Agent.run>, '__module__': 'labgrid.util.agent', 'register': <function Agent.register>, '__doc__': None})
__module__ = 'labgrid.util.agent'
__weakref__

list of weak references to the object (if defined)

labgrid.util.agent.handle_test(*args, **kwargs)[source]
labgrid.util.agent.handle_error(message)[source]
labgrid.util.agent.handle_usbtmc(index, cmd, read=False)[source]
labgrid.util.agent.main()[source]

labgrid.util.agentwrapper module

labgrid.util.agentwrapper.b2s(b)[source]
labgrid.util.agentwrapper.s2b(s)[source]
exception labgrid.util.agentwrapper.AgentError[source]

Bases: Exception

__module__ = 'labgrid.util.agentwrapper'
__weakref__

list of weak references to the object (if defined)

exception labgrid.util.agentwrapper.AgentException[source]

Bases: Exception

__module__ = 'labgrid.util.agentwrapper'
__weakref__

list of weak references to the object (if defined)

class labgrid.util.agentwrapper.AgentWrapper(host)[source]

Bases: object

class Proxy(wrapper, name)[source]

Bases: object

__init__(wrapper, name)[source]
__call__(*args, **kwargs)[source]
__dict__ = mappingproxy({'__dict__': <attribute '__dict__' of 'Proxy' objects>, '__call__': <function AgentWrapper.Proxy.__call__>, '__init__': <function AgentWrapper.Proxy.__init__>, '__doc__': None, '__weakref__': <attribute '__weakref__' of 'Proxy' objects>, '__module__': 'labgrid.util.agentwrapper'})
__module__ = 'labgrid.util.agentwrapper'
__weakref__

list of weak references to the object (if defined)

__init__(host)[source]
__del__()[source]
__getattr__(name)[source]
call(method, *args, **kwargs)[source]
__dict__ = mappingproxy({'__doc__': None, '__dict__': <attribute '__dict__' of 'AgentWrapper' objects>, 'close': <function AgentWrapper.close>, 'Proxy': <class 'labgrid.util.agentwrapper.AgentWrapper.Proxy'>, '__module__': 'labgrid.util.agentwrapper', 'call': <function AgentWrapper.call>, '__init__': <function AgentWrapper.__init__>, '__weakref__': <attribute '__weakref__' of 'AgentWrapper' objects>, '__del__': <function AgentWrapper.__del__>, '__getattr__': <function AgentWrapper.__getattr__>})
__module__ = 'labgrid.util.agentwrapper'
__weakref__

list of weak references to the object (if defined)

close()[source]

labgrid.util.dict module

labgrid.util.dict.diff_dict(old, new)[source]

Compares old and new dictionaries, yielding for each difference (key, old_value, new_value). None is used for missing values.

labgrid.util.dict.flat_dict(d)[source]
labgrid.util.dict.filter_dict(d, cls, warn=False)[source]

Returns a copy a dictionary which only contains the attributes defined on an attrs class.

labgrid.util.exceptions module

exception labgrid.util.exceptions.NoValidDriverError(msg) → None[source]

Bases: Exception

__attrs_attrs__ = (Attribute(name='msg', default=NOTHING, validator=<instance_of validator for type <class 'str'>>, repr=True, cmp=True, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False),)
__init__(msg) → None
__module__ = 'labgrid.util.exceptions'
__repr__()

Automatically created by attrs.

__weakref__

list of weak references to the object (if defined)

labgrid.util.expect module

class labgrid.util.expect.PtxExpect(driver, logfile=None, timeout=30, cwd=None)[source]

Bases: pexpect.pty_spawn.spawn

labgrid Wrapper of the pexpect module.

This class provides pexpect functionality for the ConsoleProtocol classes. driver: ConsoleProtocol object to be passed in

__init__(driver, logfile=None, timeout=30, cwd=None)[source]
send(s)[source]

Write to underlying transport, return number of bytes written

read_nonblocking(size=1, timeout=-1)[source]

Pexpects needs a nonblocking read function, simply use pyserial with a timeout of 0

__module__ = 'labgrid.util.expect'

labgrid.util.helper module

labgrid.util.helper.get_free_port()[source]

Helper function to always return an unused port.

labgrid.util.helper.get_user()[source]

labgrid.util.managedfile module

class labgrid.util.managedfile.ManagedFile(local_path, resource) → None[source]

Bases: object

The ManagedFile allows the synchronisation of a file to a remote host. It has to be created with the to be synced file and the target resource as argument:

::

from labgrid.util.managedfile import ManagedFile

ManagedFile(“/tmp/examplefile”, <your-resource>)

Synchronisation is done with the sync_to_resource method.

__attrs_post_init__()[source]
sync_to_resource()[source]

sync the file to the host specified in a resource

Raises:ExecutionError – if the SSH connection/copy fails
get_remote_path()[source]

Retrieve the remote file path

Returns:path to the file on the remote host
Return type:str
get_hash()[source]

Retrieve the hash of the file

Returns:SHA256 hexdigest of the file
Return type:str
__attrs_attrs__ = (Attribute(name='local_path', default=NOTHING, validator=<instance_of validator for type <class 'str'>>, repr=True, cmp=True, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=<function ManagedFile.<lambda>>, kw_only=False), Attribute(name='resource', default=NOTHING, validator=<instance_of validator for type <class 'labgrid.resource.common.Resource'>>, repr=True, cmp=True, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False))
__dict__ = mappingproxy({'__ne__': <function EthernetPortExport.__ne__>, '__repr__': <function ManagedFile.__repr__>, '__doc__': ' The ManagedFile allows the synchronisation of a file to a remote host.\n It has to be created with the to be synced file and the target resource as\n argument:\n\n ::\n from labgrid.util.managedfile import ManagedFile\n\n ManagedFile("/tmp/examplefile", <your-resource>)\n\n\n Synchronisation is done with the sync_to_resource method.\n ', '__gt__': <function ManagedFile.__gt__>, '__le__': <function ManagedFile.__le__>, '__ge__': <function ManagedFile.__ge__>, 'get_hash': <function ManagedFile.get_hash>, '__hash__': None, '__dict__': <attribute '__dict__' of 'ManagedFile' objects>, 'sync_to_resource': <function ManagedFile.sync_to_resource>, '__init__': <function ManagedFile.__init__>, '__attrs_post_init__': <function ManagedFile.__attrs_post_init__>, '__weakref__': <attribute '__weakref__' of 'ManagedFile' objects>, '__module__': 'labgrid.util.managedfile', '__attrs_attrs__': (Attribute(name='local_path', default=NOTHING, validator=<instance_of validator for type <class 'str'>>, repr=True, cmp=True, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=<function ManagedFile.<lambda>>, kw_only=False), Attribute(name='resource', default=NOTHING, validator=<instance_of validator for type <class 'labgrid.resource.common.Resource'>>, repr=True, cmp=True, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False)), '__lt__': <function ManagedFile.__lt__>, 'get_remote_path': <function ManagedFile.get_remote_path>, '__eq__': <function ManagedFile.__eq__>})
__eq__(other)
__ge__(other)

Automatically created by attrs.

__gt__(other)

Automatically created by attrs.

__hash__ = None
__init__(local_path, resource) → None
__le__(other)

Automatically created by attrs.

__lt__(other)

Automatically created by attrs.

__module__ = 'labgrid.util.managedfile'
__ne__(other)

Check equality and either forward a NotImplemented or return the result negated.

__repr__()

Automatically created by attrs.

__weakref__

list of weak references to the object (if defined)

labgrid.util.marker module

labgrid.util.marker.gen_marker()[source]

labgrid.util.proxy module

labgrid.util.qmp module

class labgrid.util.qmp.QMPMonitor(monitor_out, monitor_in) → None[source]

Bases: object

__attrs_post_init__()[source]
execute(command)[source]
__attrs_attrs__ = (Attribute(name='monitor_out', default=NOTHING, validator=None, repr=True, cmp=True, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False), Attribute(name='monitor_in', default=NOTHING, validator=None, repr=True, cmp=True, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False))
__dict__ = mappingproxy({'_negotiate_capabilities': <function QMPMonitor._negotiate_capabilities>, '__repr__': <function QMPMonitor.__repr__>, '__doc__': None, 'execute': <function QMPMonitor.execute>, '__dict__': <attribute '__dict__' of 'QMPMonitor' objects>, '_read_parse_json': <function QMPMonitor._read_parse_json>, '__init__': <function QMPMonitor.__init__>, '__attrs_post_init__': <function QMPMonitor.__attrs_post_init__>, '__weakref__': <attribute '__weakref__' of 'QMPMonitor' objects>, '__module__': 'labgrid.util.qmp', '__attrs_attrs__': (Attribute(name='monitor_out', default=NOTHING, validator=None, repr=True, cmp=True, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False), Attribute(name='monitor_in', default=NOTHING, validator=None, repr=True, cmp=True, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False))})
__init__(monitor_out, monitor_in) → None
__module__ = 'labgrid.util.qmp'
__repr__()

Automatically created by attrs.

__weakref__

list of weak references to the object (if defined)

exception labgrid.util.qmp.QMPError(msg) → None[source]

Bases: Exception

__attrs_attrs__ = (Attribute(name='msg', default=NOTHING, validator=<instance_of validator for type <class 'str'>>, repr=True, cmp=True, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False),)
__init__(msg) → None
__module__ = 'labgrid.util.qmp'
__repr__()

Automatically created by attrs.

__weakref__

list of weak references to the object (if defined)

labgrid.util.ssh module

class labgrid.util.ssh.SSHConnection(host) → None[source]

Bases: object

SSHConnections are individual connections to hosts managed by a control socket. In addition to command execution this class also provides an interface to manage port forwardings. These are used in the remote infrastructure to tunnel multiple connections over one SSH link.

A public identity infrastructure is assumed, no extra username or passwords are supported.

__attrs_post_init__()[source]
run_command(command)[source]

Run a command over the SSHConnection

Parameters:command (string) – The command to run
Returns:exitcode of the command
Return type:int
get_file(remote_file, local_file)[source]

Get a file from the remote host

put_file(local_file, remote_path)[source]

Put a file onto the remote host

add_port_forward(remote_host, remote_port)[source]

forward command

remove_port_forward(remote_host, remote_port)[source]

cancel command

connect()[source]
disconnect()[source]
isconnected()[source]
__attrs_attrs__ = (Attribute(name='host', default=NOTHING, validator=<instance_of validator for type <class 'str'>>, repr=True, cmp=True, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False), Attribute(name='_connected', default=False, validator=<instance_of validator for type <class 'bool'>>, repr=True, cmp=True, hash=None, init=False, metadata=mappingproxy({}), type=None, converter=None, kw_only=False), Attribute(name='_tmpdir', default=Factory(factory=<function SSHConnection.<lambda>>, takes_self=False), validator=<instance_of validator for type <class 'str'>>, repr=True, cmp=True, hash=None, init=False, metadata=mappingproxy({}), type=None, converter=None, kw_only=False), Attribute(name='_forwards', default=Factory(factory=<class 'dict'>, takes_self=False), validator=None, repr=True, cmp=True, hash=None, init=False, metadata=mappingproxy({}), type=None, converter=None, kw_only=False))
__dict__ = mappingproxy({'__ne__': <function EthernetPortExport.__ne__>, '_start_own_master': <function SSHConnection._start_own_master>, 'remove_port_forward': <function SSHConnection.remove_port_forward>, '_run_socket_command': <function SSHConnection._run_socket_command>, '__gt__': <function SSHConnection.__gt__>, '__dict__': <attribute '__dict__' of 'SSHConnection' objects>, '__ge__': <function SSHConnection.__ge__>, 'get_file': <function SSHConnection.get_file>, '_start_keepalive': <function SSHConnection._start_keepalive>, 'run_command': <function SSHConnection.run_command>, '__hash__': None, '_check_connected': <function SSHConnection._check_connected>, '__attrs_post_init__': <function SSHConnection.__attrs_post_init__>, '__weakref__': <attribute '__weakref__' of 'SSHConnection' objects>, '_get_ssh_args': <function SSHConnection._get_ssh_args>, '_check_keepalive': <function SSHConnection._check_keepalive>, '__lt__': <function SSHConnection.__lt__>, '_check_external_master': <function SSHConnection._check_external_master>, '_get_ssh_control_args': <function SSHConnection._get_ssh_control_args>, '_disconnect': <function SSHConnection._disconnect>, '_stop_own_master': <function SSHConnection._stop_own_master>, '_get_ssh_base_args': <function SSHConnection._get_ssh_base_args>, '__doc__': 'SSHConnections are individual connections to hosts managed by a control\n socket. In addition to command execution this class also provides an\n interface to manage port forwardings. These are used in the remote\n infrastructure to tunnel multiple connections over one SSH link.\n\n A public identity infrastructure is assumed, no extra username or passwords\n are supported.', '__le__': <function SSHConnection.__le__>, 'disconnect': <function SSHConnection.disconnect>, 'put_file': <function SSHConnection.put_file>, '__eq__': <function SSHConnection.__eq__>, '_open_connection': <function SSHConnection._open_connection>, '_stop_keepalive': <function SSHConnection._stop_keepalive>, '__init__': <function SSHConnection.__init__>, 'add_port_forward': <function SSHConnection.add_port_forward>, '__module__': 'labgrid.util.ssh', '__attrs_attrs__': (Attribute(name='host', default=NOTHING, validator=<instance_of validator for type <class 'str'>>, repr=True, cmp=True, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False), Attribute(name='_connected', default=False, validator=<instance_of validator for type <class 'bool'>>, repr=True, cmp=True, hash=None, init=False, metadata=mappingproxy({}), type=None, converter=None, kw_only=False), Attribute(name='_tmpdir', default=Factory(factory=<function SSHConnection.<lambda>>, takes_self=False), validator=<instance_of validator for type <class 'str'>>, repr=True, cmp=True, hash=None, init=False, metadata=mappingproxy({}), type=None, converter=None, kw_only=False), Attribute(name='_forwards', default=Factory(factory=<class 'dict'>, takes_self=False), validator=None, repr=True, cmp=True, hash=None, init=False, metadata=mappingproxy({}), type=None, converter=None, kw_only=False)), 'connect': <function SSHConnection.connect>, '__repr__': <function SSHConnection.__repr__>, 'isconnected': <function SSHConnection.isconnected>})
__eq__(other)
__ge__(other)

Automatically created by attrs.

__gt__(other)

Automatically created by attrs.

__hash__ = None
__init__(host) → None
__le__(other)

Automatically created by attrs.

__lt__(other)

Automatically created by attrs.

__module__ = 'labgrid.util.ssh'
__ne__(other)

Check equality and either forward a NotImplemented or return the result negated.

__repr__()

Automatically created by attrs.

__weakref__

list of weak references to the object (if defined)

exception labgrid.util.ssh.ForwardError(msg) → None[source]

Bases: Exception

__attrs_attrs__ = (Attribute(name='msg', default=NOTHING, validator=<instance_of validator for type <class 'str'>>, repr=True, cmp=True, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False),)
__eq__(other)
__ge__(other)

Automatically created by attrs.

__gt__(other)

Automatically created by attrs.

__hash__ = None
__init__(msg) → None
__le__(other)

Automatically created by attrs.

__lt__(other)

Automatically created by attrs.

__module__ = 'labgrid.util.ssh'
__ne__(other)

Check equality and either forward a NotImplemented or return the result negated.

__repr__()

Automatically created by attrs.

__weakref__

list of weak references to the object (if defined)

labgrid.util.timeout module

class labgrid.util.timeout.Timeout(timeout=120.0) → None[source]

Bases: object

Reperents a timeout (as a deadline)

__attrs_post_init__()[source]
remaining
expired
__attrs_attrs__ = (Attribute(name='timeout', default=120.0, validator=<instance_of validator for type <class 'float'>>, repr=True, cmp=True, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False),)
__dict__ = mappingproxy({'__dict__': <attribute '__dict__' of 'Timeout' objects>, '__repr__': <function Timeout.__repr__>, '__init__': <function Timeout.__init__>, 'remaining': <property object>, '__weakref__': <attribute '__weakref__' of 'Timeout' objects>, '__module__': 'labgrid.util.timeout', '__attrs_attrs__': (Attribute(name='timeout', default=120.0, validator=<instance_of validator for type <class 'float'>>, repr=True, cmp=True, hash=None, init=True, metadata=mappingproxy({}), type=None, converter=None, kw_only=False),), '__attrs_post_init__': <function Timeout.__attrs_post_init__>, 'expired': <property object>, '__doc__': 'Reperents a timeout (as a deadline)'})
__init__(timeout=120.0) → None
__module__ = 'labgrid.util.timeout'
__repr__()

Automatically created by attrs.

__weakref__

list of weak references to the object (if defined)

labgrid.util.yaml module

labgrid.util.yaml.load(stream)[source]
labgrid.util.yaml.dump(data, stream=None)[source]
labgrid.util.yaml.resolve_templates(data, mapping)[source]

Iterate recursively over data and call substitute(mapping) on all Templates.