| #!/usr/bin/python |
| # Copyright 2012 Google Inc. All Rights Reserved. |
| |
| """This class defines devices. |
| |
| This file defines the base class of all devices, |
| and specific device classes, such as Bruno, Spirent, etc. |
| """ |
| |
| __author__ = 'Lin Xue (linxue@google.com)' |
| __author__ = 'Lehan Meng (lmeng@google.com)' |
| |
| |
| import re |
| |
| import ip |
| |
| |
| class Device(object): |
| """Base class of all devices. |
| |
| information of device class: |
| addr: address of device |
| user: user name of device |
| pwd: password of device |
| dev_name: device name |
| cmd_prompt: command prompt of device |
| (eg. For late iguana of Bruno, it is gfibertv#; |
| before iguana, it is none) |
| other: other information of device |
| """ |
| |
| def GetDevInfo(self): |
| """This function returns device information.""" |
| return (self.dev_info['dev_name'], self.dev_info['other']) |
| |
| def SetLogging(self, logging): |
| """This function set logging for device.""" |
| self.log = logging |
| return |
| # End of Device class |
| |
| |
| class Bruno(Device): |
| """Bruno device class.""" |
| |
| def __init__(self, **kwargs): |
| """Constructor for Bruno device. |
| |
| device name: Bruno, and other device informations |
| """ |
| self.dev_info = {'addr': 'None', |
| 'user': 'None', |
| 'pwd': 'None', |
| 'dev_name': 'None', |
| 'cmd_prompt': 'None', |
| 'other': 'None'} |
| |
| self.dev_info['dev_name'] = 'Bruno' |
| for s in kwargs: |
| self.dev_info[s] = kwargs[s] |
| self.dev_ssh = None |
| |
| # Bruno Device attributes |
| # serial number |
| self.dev_sn = None |
| # image version |
| self.dev_ver = None |
| # ip address |
| self.dev_ip = None |
| # system time |
| self.dev_time = None |
| # ACS url |
| self.dev_acs = None |
| # downloading image |
| self.dev_image = None |
| # product class |
| self.dev_class = None |
| # EPG primary |
| self.dev_epgpri = None |
| # EPG secondary |
| self.dev_epgsec = None |
| return |
| |
| def GetDevCmd(self, cmd='', dev_attr='', dev_log=''): |
| """This function sends command to Bruno. |
| |
| Then it gets back the results |
| If success, save the results and write log |
| If fail, show Warning message |
| Args: |
| cmd: |
| the command sent to Bruno system |
| to query the device information |
| dev_attr: |
| the device attribute |
| which can be get from this command |
| dev_log: |
| the logging message to be shown in log |
| Returns: |
| return the command results |
| """ |
| # Send command to query Bruno device information |
| self.dev_ssh.SendCmd(cmd) |
| |
| # Get back results |
| line = self.dev_ssh.GetCmdOutput(2)[1] |
| |
| if line is None: |
| info = self.log.CreateErrorInfo('Warning', |
| 'Can not get ' + dev_log) |
| self.log.SendLine(None, info) |
| else: |
| info = self.log.CreateProgressInfo('---', |
| 'Get ' + dev_log + ': ' + line) |
| self.log.SendLine(None, info) |
| dev_attr = line |
| return line.strip() |
| |
| def GetSerialNumber(self): |
| """This function returns Bruno device serial number.""" |
| |
| result = self.GetDevCmd(r'hnvram -r 1st_serial_number', self.dev_sn, |
| 'serial number') |
| |
| return result |
| |
| def GetCurrentVersion(self): |
| """This function returns Bruno device current image version.""" |
| |
| result = self.GetDevCmd(r'more /etc/version', self.dev_ver, |
| 'current image version') |
| |
| return result |
| |
| def VerifyCurrentVersion(self, version): |
| """Check if the current software version matches the given version. |
| |
| Args: |
| version: the expected version string to check with device software |
| Returns: |
| True: if the current version matches the given version |
| False: if otherwise |
| """ |
| if version == self.GetCurrentVersion(): |
| return True |
| else: |
| return False |
| |
| def GetIPaddress(self): |
| """This function returns Bruno device IP address.""" |
| # Obsoleted |
| result = self.GetDevCmd(r'ifconfig br0', self.dev_ip, |
| 'IP address') |
| |
| return result |
| |
| def GetDNSServer(self): |
| """This function returns device DNS server name.""" |
| self.dev_ssh.SendCmd('cat /etc/resolv.conf') |
| cmd_list = self.dev_ssh.GetCmdOutput() |
| dns = None |
| for line in cmd_list: |
| m = re.search('nameserver (.*)', line) |
| if m: |
| dns = m.group(1) |
| break |
| return dns |
| |
| def GetTimeServer(self): |
| """This function returns device DNS server name.""" |
| self.dev_ssh.SendCmd('cat /etc/ntpd.conf') |
| cmd_list = self.dev_ssh.GetCmdOutput() |
| ts = None |
| for line in cmd_list: |
| m = re.search('server (.*)', line) |
| if m: |
| ts = m.group(1) |
| break |
| return ts |
| |
| def GetGVSBHost(self): |
| """This function returns device GVSB host server name.""" |
| self.dev_ssh.SendCmd('cat /tmp/gvsbhost') |
| cmd_line = self.dev_ssh.GetCmdOutput(2)[0] |
| m = re.search('http[s]?://[\w\d\./-]*', cmd_line) |
| if m: |
| return m.group(0) |
| return None |
| |
| def GetIPv4Address(self): |
| """This function returns Bruno device IPv4 address.""" |
| self.dev_ssh.SendCmd('ip addr show dev br0') |
| cmd_list = self.dev_ssh.GetCmdOutput() |
| address = None |
| for line in cmd_list: |
| if 'inet ' in line and 'scope global' in line: |
| address = ip.IPADDR(line).IsLikeIpv4Address() |
| |
| if not address: |
| info = self.log.CreateErrorInfo('critical', |
| 'Can not detect IPv4 address on device!') |
| self.log.SendLine(None, info) |
| return address |
| |
| def GetIPv6Address(self): |
| """This function returns Bruno device IPv6 address. |
| |
| This function will return a list of addresses. Sometimes device may |
| have multiple IPv6 addresses. |
| |
| Returns: |
| return the obtained IPv6 address |
| """ |
| self.dev_ssh.SendCmd('ip addr show dev br0') |
| cmd_list = self.dev_ssh.GetCmdOutput() |
| address = [] |
| for line in cmd_list: |
| if 'inet6 ' in line and 'scope global' in line: |
| print line |
| m = re.match('inet6 ([\da-f:]+)/\d+ scope global \w*', line.strip()) |
| if not m: |
| info = self.log.CreateErrorInfo( |
| 'critical', 'Can not detect valid IPv6 address ' |
| 'on device! ' + line) |
| self.log.SendLine(None, info) |
| return address |
| addr = m.group(1) |
| address.append(addr) |
| return address |
| |
| def GetDevTime(self): |
| """This function returns Bruno device current date.""" |
| |
| result = self.GetDevCmd(r'date', self.dev_time, |
| 'current date') |
| return result |
| |
| def GetACSUrl(self): |
| """This function returns Bruno device ACS URL.""" |
| |
| result = self.GetDevCmd(r'cat /tmp/cwmp/acs_url', self.dev_acs, |
| 'ACS URL') |
| return result |
| |
| def VerifyCurrentACSURL(self, url): |
| """Check if the current ACS URL matches the given URL. |
| |
| Args: |
| url: the expected url string to check with device ACS URL |
| Returns: |
| True: if the current url matches the given url |
| False: otherwise |
| """ |
| if url == self.GetACSUrl(): |
| return True |
| else: |
| return False |
| |
| def GetDnldImage(self): |
| """This function returns Bruno device current downloading image.""" |
| |
| result = self.GetDevCmd(r'ls -l /rw/tr69/dnld', self.dev_image, |
| 'current downloading image') |
| |
| return result |
| |
| def GetProductClass(self): |
| """This function returns Bruno device product class.""" |
| |
| result = self.GetDevCmd(r'cat /etc/platform', self.dev_class, |
| 'product class') |
| |
| return result |
| |
| def GetEpgPrimary(self): |
| """This function returns Bruno device primary EPG.""" |
| |
| result = self.GetDevCmd(r'cat /tmp/epgprimary', self.dev_epgpri, |
| 'EPG primary') |
| |
| return result |
| |
| def GetEpgSecond(self): |
| """This function returns Bruno device secondary EPG.""" |
| |
| result = self.GetDevCmd(r'cat /tmp/epgsecondary', self.dev_epgsec, |
| 'EPG secondary') |
| |
| return result |
| |
| def GetDevInfo(self): |
| """This function returns all Bruno device information.""" |
| # Device name |
| info = self.log.CreateProgressInfo('---', |
| 'The device is ' |
| + self.dev_info['dev_name']) |
| self.log.SendLine(None, info) |
| |
| # Other device informations |
| self.GetSerialNumber() |
| self.GetCurrentVersion() |
| self.GetIPv4Address() |
| self.GetDevTime() |
| self.GetACSUrl() |
| self.GetDnldImage() |
| self.GetProductClass() |
| self.GetEpgPrimary() |
| self.GetEpgSecond() |
| |
| return |
| |
| def GetTimeStamp(self, line=None): |
| """Extract the time information of most current event from Device log. |
| |
| i.e., the time that the last event happens since reboot (in seconds) |
| Args: |
| line: a line of device log which contains event time information. |
| If it has value of 'None', then retrieve the last a few lines from |
| device log to extract time information. |
| Returns: |
| return the time stamp string if succeed, return -1 otherwise. |
| """ |
| if not line: |
| # get time stamp from Device log file |
| self.dev_ssh.SendCmd('dmesg | grep cwmpd:') |
| log_list = self.dev_ssh.GetCmdOutput() |
| |
| for line in log_list: |
| m = re.match('\\[[\s]*(\d*\\.\d{3})\\]', line) |
| if m: |
| # Match, return the time stamp |
| return m.group(1) |
| else: |
| # get time stamp from a Device log line |
| m = re.match('\\[[\s]*(\d*\\.\d{3})\\]', line) |
| if m: |
| # Match, return the time stamp |
| return m.group(1) |
| # No time stamp found |
| return -1 |
| |
| def FindLineInLog(self, log_list, time_stamp, pattern_line): |
| """Match a particular string from device logs. |
| |
| Find a particular string or line from device log |
| Args: |
| log_list: the device log to search from |
| time_stamp: to search events that are after this time |
| pattern_line: the string or line that need to be matched in device log |
| Returns: |
| True: if 'line' matched |
| False: 'line' not found in device log |
| """ |
| m = re.compile(pattern_line) |
| for line in log_list: |
| if float(self.GetTimeStamp(line)) > float(time_stamp): |
| #Debug |
| if 'detected' in line: |
| print 'here' |
| |
| matching = m.search(line) |
| if matching: |
| return True |
| return False |
| # End of Bruno device class |
| |
| |
| class Spirent(Device): |
| """Spirent device class.""" |
| |
| def __init__(self, **kwargs): |
| """Constructor for Spirent device. |
| |
| device name: Spirent, and other device informations |
| """ |
| |
| self.dev_info['dev_name'] = 'Spirent' |
| for s in ('addr', 'user', 'pwd', 'cmd_prompt', 'other'): |
| if s in kwargs: |
| self.dev_info[s] = kwargs[s] |
| |
| # End of Spirent device class |
