Merge "Some tools for auto-probing attached bruno devices."
diff --git a/device.py b/device.py
index 61d9ad3..c02fa33 100644
--- a/device.py
+++ b/device.py
@@ -7,16 +7,11 @@
"""This file defines the base class of all devices,
and specific device classes, such as Bruno, Spirent, etc.
-
"""
-import logging
-import os
-
-import ssh
class Device(object):
- """Base class of all devices
+ """Base class of all devices.
information of device class:
addr: address of device
user: user name of device
@@ -27,16 +22,24 @@
before iguana, it is none)
other: other information of device
"""
- devInfo = {'addr': 'None',
- 'user': 'None',
- 'pwd': 'None',
- 'dev_name': 'None',
- 'cmd_prompt': 'None',
- 'other': 'None'}
- def getDevInfo(self):
+ dev_info = {'addr': 'None',
+ 'user': 'None',
+ 'pwd': 'None',
+ 'dev_name': 'None',
+ 'cmd_prompt': 'None',
+ 'other': 'None'}
+
+ def GetDevInfo(self):
"""This function returns device information."""
- return (self.devInfo['dev_name'], self.devInfo['other'])
+
+ 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
@@ -44,133 +47,165 @@
class Bruno(Device):
"""Bruno device class."""
- dev_ssh = None
-
def __init__(self, **kwargs):
""" Constructor for Bruno device
device name: Bruno, and other device informations
"""
- self.devInfo['dev_name'] = 'Bruno'
- for s in ('addr', 'user', 'pwd', 'cmd_prompt', 'other'):
- if s in kwargs:
- self.devInfo[s] = kwargs[s]
+ self.dev_info['dev_name'] = 'Bruno'
+ for s in kwargs:
+ self.dev_info[s] = kwargs[s]
- self.log = logging.Logging()
+ self.dev_ssh = None
- def createSSH(self):
- """This function creates SSH tunnel to Bruno device."""
- self.bruno = Bruno(usr=self.devInfo['user'],
- addr=self.devInfo['addr'],
- pwd=self.devInfo['pwd'],
- cmd_prompt=self.devInfo['cmd_prompt'])
+ # 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
- self.dev_ssh = ssh.SSH(self.bruno)
+ return
- if self.dev_ssh.ssh() < 0:
- info = self.log.createErrorInfo('critical',
- 'Cannot establish ssh tunnel to Device! '
- 'Exit!')
- self.log.sendLine(None, info)
- os.sys.exit(-1)
+ 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
+
+ """
+
+ # 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('---',
- 'ssh session to Device '
- 'successfully established!')
- self.log.sendLine(None, info)
- return self.dev_ssh
+ info = self.log.CreateProgressInfo('---',
+ 'Get ' + dev_log + ': ' + line)
+ self.log.SendLine(None, info)
+ dev_attr = line
- def closeSSH(self):
- """This function closes SSH tunnel to Bruno device."""
- self.dev_ssh.close()
+ return line
- def getSerialNumber(self):
+ def GetSerialNumber(self):
"""This function returns Bruno device serial number."""
- self.dev_ssh.sendCmd(r'hnvram -r 1st_serial_number')
- line = self.dev_ssh.getCmdOutput(2)[1]
- if line is None:
- info = self.log.createErrorInfo('Warning',
- 'Can not get serial number')
- self.log.sendLine(None, info)
- else:
- info = self.log.createProgressInfo('---',
- 'Get serial number: ' + line)
- self.log.sendLine(None, info)
- def getCurrentVersion(self):
- """This function returns Bruno device current software image version."""
- self.dev_ssh.sendCmd(r'more /etc/version')
- line = self.dev_ssh.getCmdOutput(2)[1]
- if line is None:
- info = self.log.createErrorInfo('Warning',
- 'Can not get current image version')
- self.log.sendLine(None, info)
- else:
- info = self.log.createProgressInfo('---',
- 'Get current image version: ' + line)
- self.log.sendLine(None, info)
+ result = self.GetDevCmd(r'hnvram -r 1st_serial_number', self.dev_sn,
+ 'serial number')
- def getIPAddress(self):
+ 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 GetIPAddress(self):
"""This function returns Bruno device IP address."""
- self.dev_ssh.sendCmd(r'ifconfig br0')
- line = self.dev_ssh.getCmdOutput(2)[1]
- if line is None:
- info = self.log.createErrorInfo('Warning',
- 'Can not get ip address')
- self.log.sendLine(None, info)
- else:
- info = self.log.createProgressInfo('---',
- 'Get IP address: ' + line)
- self.log.sendLine(None, info)
- def getDevTime(self):
+ result = self.GetDevCmd(r'ifconfig br0', self.dev_ip,
+ 'IP address')
+
+ return result
+
+ def GetDevTime(self):
"""This function returns Bruno device current date."""
- self.dev_ssh.sendCmd(r'date')
- line = self.dev_ssh.getCmdOutput(2)[1]
- if line is None:
- info = self.log.createErrorInfo('Warning',
- 'Can not get current date')
- self.log.sendLine(None, info)
- else:
- info = self.log.createProgressInfo('---',
- 'Get current date: ' + line)
- self.log.sendLine(None, info)
- def getACSUrl(self):
+ result = self.GetDevCmd(r'date', self.dev_time,
+ 'current date')
+
+ return result
+
+ def GetACSUrl(self):
"""This function returns Bruno device ACS URL."""
- self.dev_ssh.sendCmd(r'cat /tmp/cwmp/acs_url')
- line = self.dev_ssh.getCmdOutput(2)[1]
- if line is None:
- info = self.log.createErrorInfo('Warning',
- 'Can not get ACS url')
- self.log.sendLine(None, info)
- else:
- info = self.log.createProgressInfo('---',
- 'Get ACS URL: ' + line)
- self.log.sendLine(None, info)
- def getDnldImage(self):
+ result = self.GetDevCmd(r'cat /tmp/cwmp/acs_url', self.dev_acs,
+ 'ACS URL')
+
+ return result
+
+ def GetDnldImage(self):
"""This function returns Bruno device current downloading image."""
- self.dev_ssh.sendCmd(r'ls -l /rw/tr69/dnld')
- line = self.dev_ssh.getCmdOutput(2)[1]
- if line is None:
- info = self.log.createErrorInfo('Warning',
- 'Can not get current downloading image')
- self.log.sendLine(None, info)
- else:
- info = self.log.createProgressInfo('---',
- 'Get current downloading image: '
- + line)
- self.log.sendLine(None, info)
- def getDevInfo(self):
+ 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."""
- print "This device is: " + self.devInfo['dev_name']
- self.getSerialNumber()
- self.getCurrentVersion()
- self.getIPAddress()
- self.getDevTime()
- self.getACSUrl()
- self.getDnldImage()
+ # 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.GetIPAddress()
+ self.GetDevTime()
+ self.GetACSUrl()
+ self.GetDnldImage()
+ self.GetProductClass()
+ self.GetEpgPrimary()
+ self.GetEpgSecond()
+
+ return
# End of Bruno device class
@@ -183,9 +218,9 @@
device name: Spirent, and other device informations
"""
- self.devInfo['dev_name'] = 'Spirent'
+ self.dev_info['dev_name'] = 'Spirent'
for s in ('addr', 'user', 'pwd', 'cmd_prompt', 'other'):
if s in kwargs:
- self.devInfo[s] = kwargs[s]
+ self.dev_info[s] = kwargs[s]
# End of Spirent device class
diff --git a/ip.py b/ip.py
index 858f240..8afd7cd 100644
--- a/ip.py
+++ b/ip.py
@@ -10,6 +10,7 @@
"""
+
import re
import socket
@@ -28,7 +29,7 @@
def __init__(self, ip_address):
self.ipaddr = ip_address
- def is_like_ipv4_address(self):
+ def IsLikeIpv4Address(self):
"""To check if the given string has something like
ip address, eg. 4 numbers with 3 .
return the ip address in the string if yes,
@@ -41,7 +42,7 @@
else:
return None
- def is_valid_ipv4_address(self):
+ def IsValidIpv4Address(self):
"""To check if the ipv4 address is valid or not."""
try:
socket.inet_aton(self.ipaddr)
@@ -57,11 +58,11 @@
print 'Error: ' + self.ipaddr + ' is an invalid IPv4 address'
return False
- def is_ipv4_address(self):
+ def IsIpv4Address(self):
"""To check if the given string is valid ipv4 address."""
return self.is_like_ipv4_address() and self.is_valid_ipv4_address()
- def is_valid_ipv6_address(self):
+ def IsValidIpv6Address(self):
"""To check if the ipv6 address is valid or not."""
try:
socket.inet_pton(socket.AF_INET6, self.ipaddr)
diff --git a/networking.py b/networking.py
old mode 100755
new mode 100644
index ce40ae1..2289738
--- a/networking.py
+++ b/networking.py
@@ -30,8 +30,12 @@
testID 11722376
Verify Bruno IPv6 address
+testID 11723377
+Verify IPv6 SLAAC function over Ethernet/MoCA
+
testID 11724362
Verify IGMP/Multicast traffic Pass Through
+
"""
import os
@@ -39,14 +43,15 @@
import time
import pexpect
-import testCase
import device
import ip
+import ssh
+import testCase
class NetworkingTest(testCase.TestCase):
- """ Class for test cases of Bruno networking features
+ """Class for test cases of Bruno networking features.
Configuration Parameters:
testID: 11874104/11792168/11724358/11792169
@@ -54,126 +59,174 @@
addr = Bruno's IP address
user = root
pwd = google
- cmd_prompt = gfibertv#
+ bruno_prompt = gfibertv#
file = cmdOutput.txt
title = Networking
"""
- verify_progress = 0
-
- def get_Dev(self):
- """Get all device information."""
+ def Init(self):
+ """Initialize the networking feature tests."""
+ # Get all device information.
self.bruno = device.Bruno(user=self.params['user'],
addr=self.params['addr'],
pwd=self.params['pwd'],
- cmd_prompt=self.params['cmd_prompt'])
+ bruno_prompt=self.params['bruno_prompt'],
+ addr_ipv6=self.params['addr_ipv6'])
- def ssh_Tunnel(self):
+ # Pass logging instance to device
+ self.bruno.SetLogging(self.log)
+
+ return
+
+ def SshTunnel(self):
"""Create ssh tunnel for the test."""
- self.p_ssh = self.bruno.createSSH()
+ self.p_ssh = ssh.SSH(self.bruno)
- def verifyIP(self):
- """Test Bruno IP address
+ if self.p_ssh.Ssh() < 0:
+ info = self.log.CreateErrorInfo('critical',
+ 'Cannot establish ssh tunnel to Device! '
+ 'Exit!')
+ self.log.SendLine(None, info)
+ os.sys.exit(-1)
+
+ else:
+ # Set bruno device ssh instance
+ self.bruno.dev_ssh = self.p_ssh
+
+ info = self.log.CreateProgressInfo('---',
+ 'ssh session to Device '
+ 'successfully established!')
+ self.log.SendLine(None, info)
+
+ return
+
+ def VerifyIP(self):
+ """Test Bruno IP address.
testID: 11874104
"""
# Use ifconfig to check interface IP address
- self.p_ssh.sendCmd(r'ifconfig br0')
- line = self.p_ssh.getCmdOutput(3)[1]
+ self.p_ssh.SendCmd(r'ifconfig br0')
+ line = self.p_ssh.GetCmdOutput(3)[1]
# Get Bruno IP address
- address = ip.IPADDR(line).is_like_ipv4_address()
+ address = ip.IPADDR(line).IsLikeIpv4Address()
if address is not None:
- ip.IPADDR(address).is_valid_ipv4_address()
- info = self.log.createProgressInfo('50% ',
+ ip.IPADDR(address).IsValidIpv4Address()
+ info = self.log.CreateProgressInfo('50% ',
'Verify IP: Bruno has IP address '
+ address)
- self.log.sendLine(self.TESTINFO, info)
+ self.log.SendLine(self.test_info, info)
else:
- info = self.log.createErrorInfo('critical',
+ info = self.log.CreateErrorInfo('critical',
'Error! Bruno does not have IP address!'
' Exit!')
- self.log.sendLine(self.TESTINFO, info)
+ self.log.SendLine(self.test_info, info)
os.sys.exit(-1)
# Get Bruno netmask
netmask = re.search(r'mask [0-9]+(?:\.[0-9]+){3}', line).group()
- address = ip.IPADDR(netmask).is_like_ipv4_address()
+ address = ip.IPADDR(netmask).IsLikeIpv4Address()
if address is not None:
- ip.IPADDR(address).is_valid_ipv4_address()
- info = self.log.createProgressInfo('100% ',
+ ip.IPADDR(address).IsValidIpv4Address()
+ info = self.log.CreateProgressInfo('100% ',
'Verify IP: Bruno has netmask: '
+ address)
- self.log.sendLine(self.TESTINFO, info)
+ self.log.SendLine(self.test_info, info)
else:
- info = self.log.createErrorInfo('critical',
+ info = self.log.CreateErrorInfo('critical',
'Error! Bruno does not have netmask!'
' Exit!')
- self.log.sendLine(self.TESTINFO, info)
+ self.log.SendLine(self.test_info, info)
os.sys.exit(-1)
return
- def verifyDHCP(self):
- """Test Bruno communication with RG through DHCP service
+ def VerifyIPv6(self):
+ """Test Bruno IPv6 address.
+ testID: 11722376
+ """
+ # Use ip addr to check interface IPv6 address
+ self.p_ssh.SendCmd(r'ip addr show dev br0')
+
+ for line in reversed(self.p_ssh.p_ssh.before.splitlines()[-10:]):
+ ipv6_addr = re.search('inet6.*dynamic', line)
+ # Get Bruno IPv6 address
+ if ipv6_addr is not None:
+ info = self.log.CreateProgressInfo('100% ',
+ 'Verify IPv6: '
+ 'Bruno has IPv6 address: '
+ + ipv6_addr.group())
+ self.log.SendLine(self.test_info, info)
+ break
+ else:
+ info = self.log.CreateErrorInfo('intermediate',
+ 'Wanring! '
+ 'Bruno did not have IPv6 address')
+ self.log.SendLine(self.test_info, info)
+
+ return
+
+ def VerifyDHCP(self):
+ """Test Bruno communication with RG through DHCP service.
testID 11724358
"""
# check DHCP communications (DHCP REQUEST, DHCP ACK) between Bruno and RG
- self.p_ssh.sendCmd(r'dmesg | grep DHCPREQUEST')
- dhcp_request = self.p_ssh.getCmdOutput(3)[1]
- self.p_ssh.sendCmd(r'dmesg | grep DHCPACK')
- dhcp_ack = self.p_ssh.getCmdOutput(3)[1]
+ self.p_ssh.SendCmd(r'dmesg | grep DHCPREQUEST')
+ dhcp_request = self.p_ssh.GetCmdOutput(3)[1]
+ self.p_ssh.SendCmd(r'dmesg | grep DHCPACK')
+ dhcp_ack = self.p_ssh.GetCmdOutput(3)[1]
# check DHCP REQUEST
dhcp_request_time = re.search(r'[0-9]+.[0-9]+', dhcp_request)
- address = ip.IPADDR(dhcp_request).is_like_ipv4_address()
+ address = ip.IPADDR(dhcp_request).IsLikeIpv4Address()
if address is not None:
- ip.IPADDR(address).is_valid_ipv4_address()
- info = self.log.createProgressInfo('20% ',
+ ip.IPADDR(address).IsValidIpv4Address()
+ info = self.log.CreateProgressInfo('20% ',
'Verify DHCP: at time '
- + dhcp_request_time.group()
- +', Bruno sent DHCP REQUEST to '
+ + dhcp_request_time.group() +
+ ', Bruno sent DHCP REQUEST to '
+ address)
- self.log.sendLine(self.TESTINFO, info)
+ self.log.SendLine(self.test_info, info)
else:
- info = self.log.createErrorInfo('critical',
+ info = self.log.CreateErrorInfo('critical',
'Error! Bruno did not send DHCP requst!'
' Exit!')
- self.log.sendLine(self.TESTINFO, info)
+ self.log.SendLine(self.test_info, info)
os.sys.exit(-1)
# check DHCP ACK
dhcp_ack_time = re.search(r'[0-9]+.[0-9]+', dhcp_ack)
- address = ip.IPADDR(dhcp_ack).is_like_ipv4_address()
+ address = ip.IPADDR(dhcp_ack).IsLikeIpv4Address()
if address is not None:
- ip.IPADDR(address).is_valid_ipv4_address()
- info = self.log.createProgressInfo('40% ',
+ ip.IPADDR(address).IsValidIpv4Address()
+ info = self.log.CreateProgressInfo('40% ',
'Verify DHCP: at time '
- + dhcp_ack_time.group()
- +', Bruno got DHCP ACK from '
+ + dhcp_ack_time.group() +
+ ', Bruno got DHCP ACK from '
+ address)
- self.log.sendLine(self.TESTINFO, info)
+ self.log.SendLine(self.test_info, info)
else:
- info = self.log.createErrorInfo('critical',
+ info = self.log.CreateErrorInfo('critical',
'Error! Bruno did not get DHCP ACK!'
' Exit!')
- self.log.sendLine(self.TESTINFO, info)
+ self.log.SendLine(self.test_info, info)
os.sys.exit(-1)
# check DHCP renewal time
- self.p_ssh.sendCmd(r'dmesg | grep renewal')
- line = self.p_ssh.getCmdOutput(3)[1]
+ self.p_ssh.SendCmd(r'dmesg | grep renewal')
+ line = self.p_ssh.GetCmdOutput(3)[1]
renewtime = re.search(r'[0-9]+ seconds', line)
if renewtime is not None:
- info = self.log.createProgressInfo('60% ',
+ info = self.log.CreateProgressInfo('60% ',
'Verify DHCP: Bruno DHCP service '
'renewal time is '
+ renewtime.group())
- self.log.sendLine(self.TESTINFO, info)
+ self.log.SendLine(self.test_info, info)
else:
- info = self.log.createErrorInfo('critical',
+ info = self.log.CreateErrorInfo('critical',
'Error! Bruno does not have DHCP '
'renewal time! ')
- self.log.sendLine(self.TESTINFO, info)
+ self.log.SendLine(self.test_info, info)
os.sys.exit(-1)
# DHCP ACK time before renewal
@@ -181,244 +234,263 @@
# check if Bruno can renew DHCP after renew time
renew_time = re.search(r'[0-9]+', renewtime.group()).group()
- info = self.log.createProgressInfo('80% ',
- 'Verify DHCP: Wait ' + renew_time +
+ info = self.log.CreateProgressInfo('80% ',
+ 'Verify DHCP: Wait '
+ + renew_time +
' seconds and see if Bruno can get'
' DHCP renew... ')
- self.log.sendLine(self.TESTINFO, info)
+ self.log.SendLine(self.test_info, info)
# sleep for renew_time seconds
time.sleep(int(renew_time))
- self.p_ssh.sendCmd(r'dmesg | grep DHCPACK')
- line = self.p_ssh.getCmdOutput(3)[1]
+ self.p_ssh.SendCmd(r'dmesg | grep DHCPACK')
+ line = self.p_ssh.GetCmdOutput(3)[1]
# DHCP ACK time after renewal
- dhcp_after_renew = re.search(r'[0-9]+.[0-9]+', line).group()
+ dhcp_after_renew = re.search(r'[0-9]+.[0-9]+', line)
+ if dhcp_after_renew is not None:
+ dhcp_after_renew = dhcp_after_renew.group()
- address = ip.IPADDR(line).is_like_ipv4_address()
+ address = ip.IPADDR(line).IsLikeIpv4Address()
# if DHCP ACK before renewal and DHCP ACK after renewal are different,
# then Bruno got DHCP renewed
if address is not None and dhcp_after_renew != dhcp_before_renew:
- ip.IPADDR(address).is_valid_ipv4_address()
- info = self.log.createProgressInfo('100% ',
- 'Verify DHCP: Bruno got DHCP renewal,'
- ' at ' + dhcp_after_renew +
- ' seconds from ' + address)
- self.log.sendLine(self.TESTINFO, info)
+ ip.IPADDR(address).IsValidIpv4Address()
+ info = self.log.CreateProgressInfo('100% ',
+ 'Verify DHCP: Bruno got '
+ 'DHCP renewal, at '
+ + dhcp_after_renew +
+ ' seconds from'
+ + address)
+ self.log.SendLine(self.test_info, info)
else:
- info = self.log.createErrorInfo('critical',
+ info = self.log.CreateErrorInfo('critical',
'Error! Bruno did not get DHCP renewal! '
'Exit!')
- self.log.sendLine(self.TESTINFO, info)
+ self.log.SendLine(self.test_info, info)
os.sys.exit(-1)
return
- def verifyDNS(self):
- """Check Bruno DNS server information
+ def VerifyDNS(self):
+ """Check Bruno DNS server information.
+ This function observes the DNS messages exchanged between RG and Bruno
testID:11792169
"""
# Check DNS server information on Bruno
- self.p_ssh.sendCmd(r'cat /etc/resolv.conf')
- lines = self.p_ssh.getCmdOutput(3)
+ self.p_ssh.SendCmd(r'dmesg | grep dns')
+ lines = self.p_ssh.GetCmdOutput(4)
for line in lines:
- address = ip.IPADDR(line).is_like_ipv4_address()
- if address is not None:
- ip.IPADDR(address).is_valid_ipv4_address()
- info = self.log.createProgressInfo('100% ',
- 'Bruno has DNS server ' + address)
- self.log.sendLine(self.TESTINFO, info)
+ dns4 = re.search(r'dns4.*', line)
+ if dns4 is not None:
+ info = self.log.CreateProgressInfo('100% ',
+ 'Bruno has DNS server '
+ + dns4.group())
+ self.log.SendLine(self.test_info, info)
break
else:
- info = self.log.createErrorInfo('critical',
+ info = self.log.CreateErrorInfo('critical',
'Error! Bruno does not have DNS server! '
'Exit!')
- self.log.sendLine(self.TESTINFO, info)
+ self.log.SendLine(self.test_info, info)
os.sys.exit(-1)
return
- def verifyNTP(self):
- """Check Bruno NTP information
+ def VerifyNTP(self):
+ """Check Bruno NTP information.
+ This function observes the NTP messages exchanged between RG and Bruno
testID:11874105
"""
- # Check DNS server information on Bruno
- self.p_ssh.sendCmd(r'dmesg | grep synced')
- line = self.p_ssh.getCmdOutput(3)[1]
- sync = re.search(r'synced', line)
- if sync.group() is not None:
- info = self.log.createProgressInfo('100% ',
- 'Bruno\'s NTP service is synchronised')
- self.log.sendLine(self.TESTINFO, info)
+ # Check NTP server information on Bruno
+ self.p_ssh.SendCmd(r'dmesg | grep ntpd')
+
+ for line in reversed(self.p_ssh.p_ssh.before.splitlines()[-10:]):
+ sync = re.search(r'reply from .*', line)
+ if sync is not None:
+ info = self.log.CreateProgressInfo('100% ',
+ 'Bruno NTP is synchronised. '
+ 'NTP info: '
+ + sync.group())
+ self.log.SendLine(self.test_info, info)
+ break
else:
- info = self.log.createErrorInfo('intermediate',
+ info = self.log.CreateErrorInfo('intermediate',
'Warning! Bruno\'s NTP service has not '
'been synchronised yet.')
- self.log.sendLine(self.TESTINFO, info)
+ self.log.SendLine(self.test_info, info)
return
- def checkTraffic(self, protocol='', isMOCA=0):
- """This function checks TCP/UDP/ICMP traffic on Bruno interface
- will be called from function verifyTCPUDPICMP()
+ def TcpdumpCmd(self, cmd='', pattern='', testcase='', traffic='', timeout=0):
+ """This function send TCPDUMP command to Bruno prompt
+ Then it gets back the TCPDUMP results
+ If expected traffic is found, send success log
+ If no expected results, or TIMEOUT, or EOF, send Warning or Error
- Args:
- protocol:
- to check traffic from specific protocol
- tcp: to check tcp traffic
- udp: to check udp traffic
- icmp: to check icmp traffic
- isMOCA:
- to check specific interface on Bruno
- 0: Ethernet interface eth0
- 1: MOCA interface eth1
- Bruno interfaces:
- eth0: ethernet port
- eth1: MOCA port
+ Args:
+ cmd:
+ The TCPDUMP command which will be sent to detect traffic
+ pattern:
+ The search pattern in the results
+ For example, for IPv6 traffic results, it is IPv6
+ testcase:
+ The test case we are running now
+ traffic:
+ The traffic we are meausuring in one test case
+ timeout:
+ Maximun wait time for the traffic
"""
- # Set the protocol to be checked this time
- # protocol_cmd is the specific string in tcpdump for each protocol
- # expect_pattern is the specific regex pattern to be searched
- # in the expect return string
- if protocol == 'tcp':
- protocol_cmd = 'tcp'
- expect_pattern = 'IP.*ack'
- elif protocol == 'udp':
- protocol_cmd = 'udp'
- expect_pattern = 'UDP'
- elif protocol == 'icmp':
- protocol_cmd = 'icmp'
- expect_pattern = 'ICMP'
- else:
- # Now support TCP/UDP/ICMP
- info = self.log.createErrorInfo('intermediate',
- 'Warning! Protocol not supported ')
- self.log.sendLine(self.TESTINFO, info)
+ # Send tcpdump command
+ self.p_ssh.p_ssh.sendline(cmd)
- # Set interface to be checked this time
- # interface_cmd is the interface string for tcpdump
- # interface_output is the interface string for output
- if isMOCA == 0:
- interface_cmd = 'eth0'
- interface_output = 'Ethernet port eth0'
- else:
- interface_cmd = 'eth1'
- interface_output = 'MOCA port eth1'
+ # Wait timeout seconds to see if there is any result
+ info = self.log.CreateProgressInfo('50%',
+ 'Verify ' + testcase +': Wait for '
+ + str(timeout) + ' seconds'
+ ' for ' + traffic + ' traffic...')
+ self.log.SendLine(self.test_info, info)
- # Set the progress output for generate traffic
- # after each output, verify progress need to be increased
- if isMOCA == 0:
- info = self.log.createProgressInfo(str(self.verify_progress) + '%',
- 'Verify ' + protocol +
- ': Generate ' + protocol +
- ' traffic ... now manual, '
- 'need Spirient traffic generator ')
- self.log.sendLine(self.TESTINFO, info)
- self.verify_progress += 10
-
- # Send tcpdump command to prompt
- self.p_ssh.p_ssh.sendline('tcpdump -i ' + interface_cmd + ' -c 10 ' + protocol_cmd)
- i = self.p_ssh.p_ssh.expect([self.bruno.devInfo['cmd_prompt'], pexpect.EOF, pexpect.TIMEOUT], 5)
+ i = self.p_ssh.p_ssh.expect([self.bruno.dev_info['bruno_prompt'],
+ pexpect.EOF, pexpect.TIMEOUT], timeout)
if i == 0:
# Got the tcpdump return results
- for result in reversed(self.p_ssh.p_ssh.before.splitlines()[-10:]):
- protocol_flag = re.search(expect_pattern, result)
- if protocol_flag is not None:
- # Found the specific protocol traffic in the tcpdump results
- info = self.log.createProgressInfo(str(self.verify_progress) + '%',
- 'Verify ' + protocol +
- ': Bruno got ' + protocol +
- ' traffic went through ' +
- interface_output)
- self.log.sendLine(self.TESTINFO, info)
- self.verify_progress += 10
+ for result in self.p_ssh.p_ssh.before.splitlines()[-20:]:
+ ra = re.search(pattern, result)
+ if ra is not None:
+ info = self.log.CreateProgressInfo('100%',
+ 'Verify ' + testcase +
+ ': Bruno got '
+ + traffic +
+ ' traffic come through!')
+ self.log.SendLine(self.test_info, info)
break
else:
# Did not find the specific protocol traffic in the tcpdump results
- info = self.log.createErrorInfo('intermediate',
- 'Warning! packet not found '
- 'Bruno does not have ' + protocol +
- ' traffic went through ' + interface_output)
- self.log.sendLine(self.TESTINFO, info)
- self.verify_progress += 10
+ info = self.log.CreateErrorInfo('intermediate',
+ 'Warning! Bruno does not have '
+ + traffic + ' traffic come through!')
+ self.log.SendLine(self.test_info, info)
else:
- # Got TIMEOUT or EOF, no results got from tcpdump
- info = self.log.createErrorInfo('intermediate',
- 'Warning! TIMEOUT or EOF '
- 'Bruno does not have ' + protocol +
- ' traffic went through ' + interface_output)
- self.log.sendLine(self.TESTINFO, info)
- self.verify_progress += 10
+ # TIMEOUT or EOF
+ info = self.log.CreateErrorInfo('intermediate',
+ 'Warning! TIMEOUT, Bruno does not have '
+ + traffic + ' traffic come through!')
+ self.log.SendLine(self.test_info, info)
# If it is timeout or EOF, ctl + c to end the command
self.p_ssh.p_ssh.sendline('\x03')
# Get rid of the command prompt for next protocol command
- self.p_ssh.p_ssh.expect(self.bruno.devInfo['cmd_prompt'])
- self.p_ssh.p_ssh.expect(self.bruno.devInfo['cmd_prompt'])
+ self.p_ssh.p_ssh.expect(self.bruno.dev_info['bruno_prompt'])
+ self.p_ssh.p_ssh.expect(self.bruno.dev_info['bruno_prompt'])
- def verifyTCPUDPICMP(self):
+ return
+
+ def VerifyTCPUDPICMP(self):
"""Verify TCP/UDP/ICMP traffic can go through Bruno
testID: 11721380
+ thress steps:
+ 1. check if TCP can come through
+ 2. check if UDP can come through
+ 3. check if ICMP can come through
+
+ Note: need to make sure traffic is generated first
+ no need to manually generate traffic
+ need to add Spirient automation traffic generation
"""
# Check TCP traffic
- # TCP: first check ethernet port (eth0) on Bruno
- self.checkTraffic('tcp', 0)
-
- # TCP: then check MOCA port (eth1) on Bruno
- self.checkTraffic('tcp', 1)
+ # Use tcpdump to check on br0 for TCP packets
+ self.TcpdumpCmd('tcpdump -i br0 -c 1 tcp', 'IP.*ack',
+ 'TCPUDPICMP', 'TCP', 10)
# Check UDP traffic
- # UDP: first check ethernet port (eth0) on Bruno
- self.checkTraffic('udp', 0)
-
- # UDP: then check MOCA port (eth1) on Bruno
- self.checkTraffic('udp', 1)
+ # Use tcpdump to check on br0 for UDP packets
+ self.TcpdumpCmd('tcpdump -i br0 -c 1 udp', 'UDP',
+ 'TCPUDPICMP', 'UDP', 10)
# Check ICMP traffic
- # ICMP: first check ethernet port (eth0) on Bruno
- self.checkTraffic('icmp', 0)
-
- # ICMP: then check MOCA port (eth1) on Bruno
- self.checkTraffic('icmp', 1)
+ # Use tcpdump to check on br0 for ICMP packets
+ self.TcpdumpCmd('tcpdump -i br0 -c 1 icmp', 'ICMP',
+ 'TCPUDPICMP', 'ICMP', 10)
return
- def run(self):
+ def VerifySLAAC(self):
+ """Check Bruno IPv6 SLAAC function over Ethernet/MoCA.
+ testID:11723377
+ Two steps:
+ 1. check if Router Advertisement can come through
+ 2. check if IPv6 packages can come through
+ """
+
+ # Check Router Advertisement message
+ # Router Advertisement message is in Neighbor Discovery ICMPv6 message
+ # Type Field Value 134
+ # Send tcpdump command to check Router Advertisement message
+ self.TcpdumpCmd('tcpdump -s0 -n -c 1 ip6[40+0]==134', 'IPv6',
+ 'SLACC', 'Router Advertisement message', 30)
+
+ # Send tcpdump command to check IPv6 traffic
+ self.TcpdumpCmd('tcpdump -s0 -n -c 1 ip6', 'IPv6',
+ 'SLACC', 'IPv6', 30)
+
+ def VerifyIgmpMulticast(self):
+ """Check Bruno IGMP/Multicast traffic pass through.
+ testID:11724362
+ Two steps:
+ 1. check if IGMP can come through
+ 2. check if Multicast can come through
+ """
+
+ # Send tcpdump command to check IGMP traffic
+ self.TcpdumpCmd('tcpdump -c 1 igmp', 'igmp',
+ 'IGMPMULTICAST', 'IGMP', 30)
+
+ # Send tcpdump command to check Multicast traffic
+ # Bruno multicast address is 255.0.0.0/8
+ self.TcpdumpCmd('tcpdump -c 1 ip net 225.0.0.0/8', 'UDP',
+ 'IGMPMULTICAST', 'Multicast 225.0.0.0/8', 30)
+
+ return
+
+ def Run(self):
"""Run the test case."""
####### Add your code here -- BEGIN #######
- print "Test Started..."
+ print 'Test Started...'
- self.get_Dev()
+ self.Init()
time.sleep(2) # time delay between commands issued to Device
+ self.SshTunnel()
- self.ssh_Tunnel()
+ print 'Get dev info ...'
+ self.bruno.GetDevInfo()
- print "Get dev info ..."
- self.bruno.getDevInfo()
+ self.VerifyIP()
+ self.VerifyIPv6()
+ self.VerifyDHCP()
+ self.VerifyNTP()
+ self.VerifyDNS()
+ self.VerifyTCPUDPICMP()
+ self.VerifySLAAC()
+ self.VerifyIgmpMulticast()
- self.verifyIP()
- self.verifyNTP()
- self.verifyDHCP()
- self.verifyDNS()
- self.verifyTCPUDPICMP()
-
- info = self.log.createResultInfo('Pass',
+ info = self.log.CreateResultInfo('Pass',
'Finish all networking '
'feature test cases')
- self.log.sendLine(self.TESTINFO, info)
- print "Test Completed..."
+ self.log.SendLine(self.test_info, info)
+ print 'Test Completed...'
####### Add your code here -- END #######
def destructor(self):
"""This is the destructor function to call for the user extened class."""
self.p_ssh.close()
-if __name__ == "__main__":
+if __name__ == '__main__':
"""now do all the networking tests together."""
test = NetworkingTest(testID='11364', key_word='Networking')
