Source code for labgrid.driver.infodriver

import logging
import re

import attr

from ..factory import target_factory
from ..step import step
from ..protocol import InfoProtocol, CommandProtocol
from .common import Driver

[docs]class InfoDriver(Driver, InfoProtocol): """ InfoDriver implementing the InfoProtocol on top of CommandProtocol drivers """ # TODO: rework CommandProtocol binding to select correct underlying driver # (No UBoot/BareboxDriver, SSH > Serial,…) bindings = { 'command': CommandProtocol } def __attrs_post_init__(self): super().__attrs_post_init__() self.logger = logging.getLogger("{}".format(self)) @Driver.check_active @step(args=['interface'])
[docs] def get_ip(self, interface="eth0"): """Returns the IP of the supplied interface""" try: ip_string = self.command.run_check("ip -o -4 addr show") except ExecutionError: self.logger.debug('No ip address found') return None regex = re.compile( r"""\d+: # Match the leading number \s+(?P<if>\w+) # Match whitespace and interfacename \s+inet\s+(?P<ip>[\d.]+) # Match IP Adress /(?P<prefix>\d+) # Match prefix .*global # Match global scope, not host scope""", re.X ) self.logger.debug('IP String: %s', ip_string) result = {} for line in ip_string: match = regex.match(line) if match: match = match.groupdict() self.logger.debug("Match dict: %s", match) result[match['if']] = match['ip'] self.logger.debug("Complete result: %s", result) if result: return result[interface] else: return None
@Driver.check_active @step(args=['service'])
[docs] def get_service_status(self, service): """Returns True if service is active, False in all other cases""" _, _, exitcode = "systemctl --quiet is-active {}".format(service) ) return exitcode == 0
@Driver.check_active @step(result=True)
[docs] def get_hostname(self): try: hostname_string = self.command.run_check("hostname") except ExecutionError: self.logger.debug('Hostname unavailable') return None self.logger.debug('Hostname String: %s', hostname_string) return hostname_string[0]