blob: f3eb65c2e42e67098a480cabbde75e8e5cfd0532 [file] [log] [blame]
# 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 ('
__author__ = 'Lehan Meng ('
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
# 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
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
the command sent to Bruno system
to query the device information
the device attribute
which can be get from this command
the logging message to be shown in log
return the command results
# Send command to query Bruno device information
# 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)
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.
version: the expected version string to check with device software
True: if the current version matches the given version
False: if otherwise
if version == self.GetCurrentVersion():
return True
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 ='nameserver (.*)', line)
if m:
dns =
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 ='server (.*)', line)
if m:
ts =
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 ='http[s]?://[\w\d\./-]*', cmd_line)
if m:
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.
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 =
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,
return result
def VerifyCurrentACSURL(self, url):
"""Check if the current ACS URL matches the given URL.
url: the expected url string to check with device ACS URL
True: if the current url matches the given url
False: otherwise
if url == self.GetACSUrl():
return True
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
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)
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.
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
# get time stamp from a Device log line
m = re.match('\\[[\s]*(\d*\\.\d{3})\\]', line)
if m:
# Match, return the time stamp
# 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
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
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):
if 'detected' in line:
print 'here'
matching =
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