labgrid.util package¶
Submodules¶
labgrid.util.agent module¶
-
class
labgrid.util.agent.
Agent
[source]¶ Bases:
object
-
__dict__
= mappingproxy({'__init__': <function Agent.__init__>, '__dict__': <attribute '__dict__' of 'Agent' objects>, 'register': <function Agent.register>, '__doc__': None, '__weakref__': <attribute '__weakref__' of 'Agent' objects>, 'run': <function Agent.run>, '__module__': 'labgrid.util.agent', '_send': <staticmethod object>})¶
-
__module__
= 'labgrid.util.agent'¶
-
__weakref__
¶ list of weak references to the object (if defined)
-
labgrid.util.agentwrapper module¶
-
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
-
__dict__
= mappingproxy({'__call__': <function AgentWrapper.Proxy.__call__>, '__init__': <function AgentWrapper.Proxy.__init__>, '__dict__': <attribute '__dict__' of 'Proxy' objects>, '__doc__': None, '__module__': 'labgrid.util.agentwrapper', '__weakref__': <attribute '__weakref__' of 'Proxy' objects>})¶
-
__module__
= 'labgrid.util.agentwrapper'¶
-
__weakref__
¶ list of weak references to the object (if defined)
-
-
__dict__
= mappingproxy({'__init__': <function AgentWrapper.__init__>, 'call': <function AgentWrapper.call>, '__doc__': None, '__module__': 'labgrid.util.agentwrapper', 'Proxy': <class 'labgrid.util.agentwrapper.AgentWrapper.Proxy'>, '__getattr__': <function AgentWrapper.__getattr__>, 'close': <function AgentWrapper.close>, '__weakref__': <attribute '__weakref__' of 'AgentWrapper' objects>, '__dict__': <attribute '__dict__' of 'AgentWrapper' objects>, '__del__': <function AgentWrapper.__del__>})¶
-
__module__
= 'labgrid.util.agentwrapper'¶
-
__weakref__
¶ list of weak references to the object (if defined)
-
class
labgrid.util.dict module¶
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
-
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.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.
-
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({'__lt__': <function ManagedFile.__lt__>, '__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)), '__init__': <function ManagedFile.__init__>, '__dict__': <attribute '__dict__' of 'ManagedFile' objects>, '__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 ', '__repr__': <function ManagedFile.__repr__>, '__module__': 'labgrid.util.managedfile', 'sync_to_resource': <function ManagedFile.sync_to_resource>, '__gt__': <function ManagedFile.__gt__>, '__ne__': <function EthernetPortExport.__ne__>, '__eq__': <function ManagedFile.__eq__>, '__le__': <function ManagedFile.__le__>, 'get_remote_path': <function ManagedFile.get_remote_path>, '__attrs_post_init__': <function ManagedFile.__attrs_post_init__>, '__weakref__': <attribute '__weakref__' of 'ManagedFile' objects>, '__ge__': <function ManagedFile.__ge__>, 'get_hash': <function ManagedFile.get_hash>, '__hash__': None})¶
-
__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.proxy module¶
labgrid.util.qmp module¶
-
class
labgrid.util.qmp.
QMPMonitor
(monitor_out, monitor_in) → None[source]¶ Bases:
object
-
__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({'__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__': <function QMPMonitor.__init__>, '__dict__': <attribute '__dict__' of 'QMPMonitor' objects>, '__doc__': None, '__repr__': <function QMPMonitor.__repr__>, '__module__': 'labgrid.util.qmp', '_negotiate_capabilities': <function QMPMonitor._negotiate_capabilities>, '_read_parse_json': <function QMPMonitor._read_parse_json>, 'execute': <function QMPMonitor.execute>, '__attrs_post_init__': <function QMPMonitor.__attrs_post_init__>, '__weakref__': <attribute '__weakref__' of 'QMPMonitor' objects>})¶
-
__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.
-
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
-
__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({'__repr__': <function SSHConnection.__repr__>, '__lt__': <function SSHConnection.__lt__>, '__init__': <function SSHConnection.__init__>, '__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.', '__gt__': <function SSHConnection.__gt__>, '_check_external_master': <function SSHConnection._check_external_master>, '_start_keepalive': <function SSHConnection._start_keepalive>, '__eq__': <function SSHConnection.__eq__>, '_start_own_master': <function SSHConnection._start_own_master>, 'connect': <function SSHConnection.connect>, 'disconnect': <function SSHConnection.disconnect>, 'remove_port_forward': <function SSHConnection.remove_port_forward>, '__weakref__': <attribute '__weakref__' of 'SSHConnection' objects>, 'run_command': <function SSHConnection.run_command>, 'put_file': <function SSHConnection.put_file>, '__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__': <attribute '__dict__' of 'SSHConnection' objects>, '_check_connected': <function SSHConnection._check_connected>, 'add_port_forward': <function SSHConnection.add_port_forward>, '__ne__': <function EthernetPortExport.__ne__>, '_get_ssh_control_args': <function SSHConnection._get_ssh_control_args>, '__module__': 'labgrid.util.ssh', '_check_keepalive': <function SSHConnection._check_keepalive>, '_stop_own_master': <function SSHConnection._stop_own_master>, '_open_connection': <function SSHConnection._open_connection>, '_run_socket_command': <function SSHConnection._run_socket_command>, 'get_file': <function SSHConnection.get_file>, '__hash__': None, 'isconnected': <function SSHConnection.isconnected>, '_get_ssh_args': <function SSHConnection._get_ssh_args>, '_disconnect': <function SSHConnection._disconnect>, '_get_ssh_base_args': <function SSHConnection._get_ssh_base_args>, '__attrs_post_init__': <function SSHConnection.__attrs_post_init__>, '__ge__': <function SSHConnection.__ge__>, '_stop_keepalive': <function SSHConnection._stop_keepalive>, '__le__': <function SSHConnection.__le__>})¶
-
__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)
-
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({'__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),), '__init__': <function Timeout.__init__>, '__dict__': <attribute '__dict__' of 'Timeout' objects>, '__doc__': 'Reperents a timeout (as a deadline)', '__repr__': <function Timeout.__repr__>, 'remaining': <property object>, 'expired': <property object>, '__attrs_post_init__': <function Timeout.__attrs_post_init__>, '__weakref__': <attribute '__weakref__' of 'Timeout' objects>, '__module__': 'labgrid.util.timeout'})¶
-
__init__
(timeout=120.0) → None¶
-
__module__
= 'labgrid.util.timeout'¶
-
__repr__
()¶ Automatically created by attrs.
-
__weakref__
¶ list of weak references to the object (if defined)
-