SEBA-752 fixing atests in voltha-1.7
Change-Id: Id99a24a72e35baa31096750bf7fb12424017124f
diff --git a/tests/atests/common/authentication.py b/tests/atests/common/authentication.py
index aedbbf2..a03f17e 100644
--- a/tests/atests/common/authentication.py
+++ b/tests/atests/common/authentication.py
@@ -19,7 +19,6 @@
"""
import time
-import os
import subprocess
import testCaseUtils
import logging
@@ -37,37 +36,16 @@
self.dirs['log'] = None
self.dirs['root'] = None
self.dirs['voltha'] = None
-
+
+ self.__onuCount = None
self.__rgName = testCaseUtils.discover_rg_pod_name()
- self.__radiusName = None
- self.__radiusIp = None
-
+
def a_set_log_dirs(self, root_dir, voltha_dir, log_dir):
testCaseUtils.config_dirs(self, log_dir, root_dir, voltha_dir)
- def discover_freeradius_pod_name(self):
- self.__radiusName = testCaseUtils.extract_pod_name('freeradius').strip()
- logging.info('freeradius Name = %s' % self.__radiusName)
-
- def discover_freeradius_ip_addr(self):
- ipAddr = testCaseUtils.extract_radius_ip_addr(self.__radiusName)
- assert ipAddr, 'No IP address listed for freeradius'
- self.__radiusIp = ipAddr.strip()
- logging.info('freeradius IP = %s' % self.__radiusIp)
-
- def set_current_freeradius_ip_in_aaa_json(self):
- status = testCaseUtils.modify_radius_ip_in_json_using_sed(self, self.__radiusIp)
- assert (status == 0), 'Setting Radius Ip in Json File did not return Success'
-
- def alter_aaa_application_configuration_in_onos_using_aaa_json(self):
- logging.info('Altering the Onos NetCfg AAA apps with Freeradius IP address')
- logging.debug('curl --user karaf:karaf -X POST -H "Content-Type: application/json" '
- 'http://localhost:30120/onos/v1/network/configuration/apps/ -d @%s/tests/atests/build/aaa_json'
- % testCaseUtils.get_dir(self, 'voltha'))
- os.system('curl --user karaf:karaf -X POST -H "Content-Type: application/json" '
- 'http://localhost:30120/onos/v1/network/configuration/apps/ -d @%s/tests/atests/build/aaa_json'
- % testCaseUtils.get_dir(self, 'voltha'))
-
+ def a_configure(self, onu_count):
+ self.__onuCount = onu_count
+
def execute_authentication_on_rg(self):
logging.info('Running Radius Authentication from RG')
process_output = open('%s/%s' % (testCaseUtils.get_dir(self, 'log'), self.AUTHENTICATE_FILENAME), 'w')
@@ -126,32 +104,40 @@
testCaseUtils.print_log_file(self, self.AUTHENTICATE_FILENAME)
- def verify_authentication_should_have_started(self):
+ def authentication_should_have_started(self):
statusLines = testCaseUtils.get_fields_from_grep_command(self, 'CTRL-EVENT-EAP-STARTED', self.AUTHENTICATE_FILENAME)
assert statusLines, 'Authentication was not started'
- def verify_authentication_should_have_completed(self):
+ def authentication_should_have_completed(self):
statusLines = testCaseUtils.get_fields_from_grep_command(self, 'CTRL-EVENT-EAP-SUCCESS', self.AUTHENTICATE_FILENAME)
assert statusLines, 'Authentication was not completed successfully'
- def verify_authentication_should_have_disconnected(self):
+ def authentication_should_have_disconnected(self):
statusLines = testCaseUtils.get_fields_from_grep_command(self, 'CTRL-EVENT-DISCONNECTED', self.AUTHENTICATE_FILENAME)
assert statusLines, 'Authentication was not disconnected'
- def verify_authentication_should_have_terminated(self):
+ def authentication_should_have_terminated(self):
statusLines = testCaseUtils.get_fields_from_grep_command(self, 'CTRL-EVENT-TERMINATING', self.AUTHENTICATE_FILENAME)
assert statusLines, 'Authentication was not terminated'
-
-def run_test(root_dir, voltha_dir, log_dir):
+ def should_have_all_onus_authenticated(self):
+ testCaseUtils.send_command_to_onos_cli(testCaseUtils.get_dir(self, 'log'),
+ 'voltha_onu_auth.log', 'aaa-users')
+ statusLines = testCaseUtils.get_fields_from_grep_command(self, 'AUTHORIZED', 'voltha_onu_auth.log')
+ lines = statusLines.splitlines()
+ auth_count = len(lines)
+ assert self.__onuCount == auth_count, 'There are only %s ONUS Authenticated' % auth_count
+
+
+def run_test(onu_count, root_dir, voltha_dir, log_dir, simtype):
auth = Authentication()
auth.a_set_log_dirs(root_dir, voltha_dir, log_dir)
- auth.discover_freeradius_pod_name()
- auth.discover_freeradius_ip_addr()
- auth.set_current_freeradius_ip_in_aaa_json()
- auth.alter_aaa_application_configuration_in_onos_using_aaa_json()
- auth.execute_authentication_on_rg()
- auth.verify_authentication_should_have_started()
- auth.verify_authentication_should_have_completed()
- auth.verify_authentication_should_have_disconnected()
- auth.verify_authentication_should_have_terminated()
+ auth.a_configure(onu_count)
+ if simtype == 'ponsim':
+ auth.execute_authentication_on_rg()
+ auth.authentication_should_have_started()
+ auth.authentication_should_have_completed()
+ auth.authentication_should_have_disconnected()
+ auth.authentication_should_have_terminated()
+ elif simtype == 'bbsim':
+ auth.should_have_all_onus_authenticated()
diff --git a/tests/atests/common/auto_test.py b/tests/atests/common/auto_test.py
index 6e21842..be52134 100755
--- a/tests/atests/common/auto_test.py
+++ b/tests/atests/common/auto_test.py
@@ -31,7 +31,7 @@
import logging
DEFAULT_LOG_DIR = '/tmp/voltha_test_results'
-DEFAULT_ADAPTER = 'ponsim'
+DEFAULT_SIMTYPE = 'ponsim'
logging.basicConfig(level=logging.INFO)
@@ -66,18 +66,18 @@
return root_dir, voltha_dir, log_dir
-def adapter_init(adapter=DEFAULT_ADAPTER):
+def simtype_init(simtype=DEFAULT_SIMTYPE):
"""
- :param adapter: ponsim or bbsim
+ :param simtype: ponsim or bbsim
:return: olt_type, onu_type, olt_host_ip, onu_count
"""
- if adapter == 'ponsim':
+ if simtype == 'ponsim':
olt_type = 'ponsim_olt'
onu_type = 'ponsim_onu'
- olt_host_ip = 'olt.voltha.svc'
+ olt_host_ip = 'olt0.voltha.svc'
onu_count = 1
- elif adapter == 'bbsim':
+ elif simtype == 'bbsim':
olt_type = 'openolt'
onu_type = 'brcm_openomci_onu'
olt_host_ip = 'bbsim.voltha.svc'
@@ -102,21 +102,22 @@
parser = argparse.ArgumentParser(description='VOLTHA Automated Testing')
parser.add_argument('-l', dest='logDir', default=DEFAULT_LOG_DIR,
help='log directory (default: %s).' % DEFAULT_LOG_DIR)
- parser.add_argument('-a', dest='adapter', choices=['ponsim', 'bbsim'], default=DEFAULT_ADAPTER,
- help='adapter (default: %s).' % DEFAULT_ADAPTER)
+ parser.add_argument('-a', dest='simtype', choices=['ponsim', 'bbsim'], default=DEFAULT_SIMTYPE,
+ help='simtype (default: %s).' % DEFAULT_SIMTYPE)
args = parser.parse_args()
ROOT_DIR, VOLTHA_DIR, LOG_DIR = dir_init(args.logDir)
- OLT_TYPE, ONU_TYPE, OLT_HOST_IP, ONU_COUNT = adapter_init(args.adapter)
+ OLT_TYPE, ONU_TYPE, OLT_HOST_IP, ONU_COUNT = simtype_init(args.simtype)
- volthaMngr.voltha_initialize(ROOT_DIR, VOLTHA_DIR, LOG_DIR, args.adapter)
+ volthaMngr.voltha_initialize(ROOT_DIR, VOLTHA_DIR, LOG_DIR, args.simtype)
preprovisioning.run_test(OLT_HOST_IP, 50060, OLT_TYPE, ONU_TYPE, ONU_COUNT, LOG_DIR)
- time.sleep(60)
- discovery.run_test(OLT_HOST_IP, OLT_TYPE, ONU_TYPE, ONU_COUNT, LOG_DIR)
- if args.adapter == 'ponsim':
- authentication.run_test(ROOT_DIR, VOLTHA_DIR, LOG_DIR)
+ discovery.run_test(OLT_HOST_IP, OLT_TYPE, ONU_TYPE, ONU_COUNT, LOG_DIR)
+
+ authentication.run_test(ONU_COUNT, ROOT_DIR, VOLTHA_DIR, LOG_DIR, args.simtype)
+
+ if args.simtype == 'ponsim':
dhcp.run_test(ROOT_DIR, VOLTHA_DIR, LOG_DIR)
unicast.run_test(ONU_TYPE, ONU_COUNT, ROOT_DIR, VOLTHA_DIR, LOG_DIR)
diff --git a/tests/atests/common/dhcp.py b/tests/atests/common/dhcp.py
index a9acd8e..391c2dc 100644
--- a/tests/atests/common/dhcp.py
+++ b/tests/atests/common/dhcp.py
@@ -74,7 +74,7 @@
statusLines = testCaseUtils.get_fields_from_grep_command(self, 'IP_PROTO:17', 'voltha_onos_flows.log')
assert statusLines, 'No DHCP Detection flows'
lines = statusLines.splitlines()
- assert len(lines) == 2, 'Expected 2 DHCP Detection Flows but result was %s' % len(lines)
+ assert len(lines) >= 2, 'Expected at least 2 DHCP Detection Flows but result was %s' % len(lines)
for line in lines:
self.__fields = testCaseUtils.parse_fields(line, ',')
inPortStr = self.__fields[5].strip()
diff --git a/tests/atests/common/preprovisioning.py b/tests/atests/common/preprovisioning.py
index fb68634..2b077c9 100755
--- a/tests/atests/common/preprovisioning.py
+++ b/tests/atests/common/preprovisioning.py
@@ -28,7 +28,7 @@
"""
This class implements voltha pre-provisioning test
"""
-
+
def __init__(self):
self.dirs = dict()
self.dirs['log'] = None
@@ -99,6 +99,7 @@
assert statusLines, 'No Olt listed under devices'
self.__fields = testCaseUtils.parse_fields(statusLines, '|')
assert self.check_states(self.__oltType), 'States of %s does match expected' % self.__oltType
+ hostPortCount = False
for field in self.__fields:
if field.strip() == self.__oltIpAddress + ':' + str(self.__oltPort):
hostPortCount = True
@@ -128,6 +129,23 @@
'voltha_devices_after_enable.log', 'devices')
testCaseUtils.print_log_file(self, 'voltha_devices_after_enable.log')
+ def proceed(self):
+ logging.info('Pre-provisioning hold')
+ onuOnline = 0
+ portType = None
+ if self.__oltType == 'ponsim_olt':
+ portType = 'PON_OLT'
+ elif self.__oltType == 'openolt':
+ portType = 'ETHERNET_UNI'
+ while onuOnline < self.__onuCount:
+ testCaseUtils.send_command_to_voltha_cli(testCaseUtils.get_dir(self, 'log'), 'voltha_olt_device.log', 'device ' +
+ self.__oltDeviceId, 'voltha_olt_ports.log', 'ports')
+ statusLines = testCaseUtils.get_fields_from_grep_command(self, portType, 'voltha_olt_ports.log')
+ lines = statusLines.splitlines()
+ onuOnline = len(lines)
+ time.sleep(5)
+ logging.info('All ONUs now Online!')
+
def run_test(olt_ip_address, olt_port, olt_type, onu_type, onu_count, log_dir):
preprovisioning = Preprovisioning()
@@ -142,3 +160,4 @@
preprovisioning.query_devices_after_enabling()
preprovisioning.check_olt_fields_after_enabling()
preprovisioning.check_onu_fields_after_enabling()
+ preprovisioning.proceed()
diff --git a/tests/atests/common/run_robot.sh b/tests/atests/common/run_robot.sh
index 84e6bcd..17847a1 100755
--- a/tests/atests/common/run_robot.sh
+++ b/tests/atests/common/run_robot.sh
@@ -16,13 +16,13 @@
SRC_DIR="$( cd -P "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
VOLTHA_DIR="$SRC_DIR/../../.."
-echo "Run Robot Framework TEST. Log: $1, Adapter: ${2:-ponsim}"
+echo "Run Voltha Test Automation Suite. Log: $1, Simtype: ${2:-ponsim}"
cd ${VOLTHA_DIR}
source env.sh
if [[ "${2:-ponsim}" == "ponsim" ]]
then
- robot -d $1 -v LOG_DIR:$1/voltha_test_results -v ADAPTER:ponsim ./tests/atests/robot/voltha_automated_test_suite.robot
+ robot -d $1 -v LOG_DIR:$1/voltha_test_results -v SIMTYPE:ponsim ./tests/atests/robot/voltha_automated_test_suite.robot
elif [[ "${2}" == "bbsim" ]]
then
- robot -d $1 -v LOG_DIR:$1/voltha_test_results -v ADAPTER:bbsim -e ponsim ./tests/atests/robot/voltha_automated_test_suite.robot
+ robot -d $1 -v LOG_DIR:$1/voltha_test_results -v SIMTYPE:bbsim -e ponsim ./tests/atests/robot/voltha_automated_test_suite.robot
fi
\ No newline at end of file
diff --git a/tests/atests/common/testCaseUtils.py b/tests/atests/common/testCaseUtils.py
index 1fe4f82..9b9b67d 100755
--- a/tests/atests/common/testCaseUtils.py
+++ b/tests/atests/common/testCaseUtils.py
@@ -171,4 +171,4 @@
def discover_rg_pod_name():
- return extract_pod_name('rg-').strip()
+ return extract_pod_name('rg0').strip()
diff --git a/tests/atests/common/unicast.py b/tests/atests/common/unicast.py
index 69d3d1a..1622b99 100644
--- a/tests/atests/common/unicast.py
+++ b/tests/atests/common/unicast.py
@@ -78,7 +78,7 @@
def execute_tcpdump(self):
logging.info('Execute tcpdump')
process_output = open('%s/%s' % (testCaseUtils.get_dir(self, 'log'), self.TCPDUMP_FILENAME), 'w')
- tcpdump = subprocess.Popen(['sudo', '/usr/sbin/tcpdump', '-nei', 'pon1'],
+ tcpdump = subprocess.Popen(['sudo', '/usr/sbin/tcpdump', '-nei', 'nni0'],
stdout=process_output,
stderr=process_output)
self.__tcpdumpPid = tcpdump.pid
diff --git a/tests/atests/common/volthaMngr.py b/tests/atests/common/volthaMngr.py
index 1751fa4..84e937f 100755
--- a/tests/atests/common/volthaMngr.py
+++ b/tests/atests/common/volthaMngr.py
@@ -23,7 +23,7 @@
import subprocess
import testCaseUtils
import logging
-
+import time
class VolthaMngr(object):
@@ -32,35 +32,38 @@
This class implements voltha startup/shutdown callable helper functions
"""
- DEFAULT_ADAPTER = 'ponsim'
+ DEFAULT_SIMTYPE = 'ponsim'
def __init__(self):
self.dirs = dict()
self.dirs['root'] = None
self.dirs['voltha'] = None
self.dirs['log'] = None
-
+
+ self.__radiusName = None
+ self.__radiusIp = None
+
def v_set_log_dirs(self, root_dir, voltha_dir, log_dir):
testCaseUtils.config_dirs(self, log_dir, root_dir, voltha_dir)
- def start_all_pods(self, adapter=DEFAULT_ADAPTER):
- proc1 = subprocess.Popen([testCaseUtils.get_dir(self, 'root') + '/build.sh', 'start', adapter],
+ def start_all_pods(self, simtype=DEFAULT_SIMTYPE):
+ proc1 = subprocess.Popen([testCaseUtils.get_dir(self, 'root') + '/build.sh', 'start', simtype],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
output = proc1.communicate()[0]
print(output)
proc1.stdout.close()
- def stop_all_pods(self, adapter=DEFAULT_ADAPTER):
- proc1 = subprocess.Popen([testCaseUtils.get_dir(self, 'root') + '/build.sh', 'stop', adapter],
+ def stop_all_pods(self, simtype=DEFAULT_SIMTYPE):
+ proc1 = subprocess.Popen([testCaseUtils.get_dir(self, 'root') + '/build.sh', 'stop', simtype],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
output = proc1.communicate()[0]
print(output)
proc1.stdout.close()
- def reset_kube_adm(self, adapter=DEFAULT_ADAPTER):
- proc1 = subprocess.Popen([testCaseUtils.get_dir(self, 'root') + '/build.sh', 'clear', adapter],
+ def reset_kube_adm(self, simtype=DEFAULT_SIMTYPE):
+ proc1 = subprocess.Popen([testCaseUtils.get_dir(self, 'root') + '/build.sh', 'clear', simtype],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
output = proc1.communicate()[0]
@@ -98,13 +101,50 @@
os.system('/usr/bin/kubectl logs -n %s -f %s onos > %s/%s.log 2>&1 &' %
(Namespace, podName, testCaseUtils.get_dir(self, 'log'), podName))
elif 'calico-node' in podName:
- os.system('/usr/bin/kubectl logs -n %s -f %s calico-node > %s/%s.log 2>&1 &' %
- (Namespace, podName, testCaseUtils.get_dir(self, 'log'), podName))
+ os.system('/usr/bin/kubectl logs -n %s -f %s calico-node > %s/%s.log 2>&1 &' %
+ (Namespace, podName, testCaseUtils.get_dir(self, 'log'), podName))
else:
os.system('/usr/bin/kubectl logs -n %s -f %s > %s/%s.log 2>&1 &' %
(Namespace, podName, testCaseUtils.get_dir(self, 'log'), podName))
-
+ def discover_freeradius_pod_name(self):
+ self.__radiusName = testCaseUtils.extract_pod_name('freeradius').strip()
+ logging.info('freeradius Name = %s' % self.__radiusName)
+
+ def discover_freeradius_ip_addr(self):
+ ipAddr = testCaseUtils.extract_radius_ip_addr(self.__radiusName)
+ assert ipAddr, 'No IP address listed for freeradius'
+ self.__radiusIp = ipAddr.strip()
+ logging.info('freeradius IP = %s' % self.__radiusIp)
+
+ def prepare_current_freeradius_ip(self):
+ status = testCaseUtils.modify_radius_ip_in_json_using_sed(self, self.__radiusIp)
+ assert (status == 0), 'Setting Radius Ip in Json File did not return Success'
+
+ def alter_freeradius_ip_in_onos_aaa_application_configuration(self):
+ logging.info('Altering the Onos NetCfg AAA apps with Freeradius IP address')
+ logging.debug('curl --user karaf:karaf -X POST -H "Content-Type: application/json" '
+ 'http://localhost:30120/onos/v1/network/configuration/apps/ -d @%s/tests/atests/build/aaa_json'
+ % testCaseUtils.get_dir(self, 'voltha'))
+ os.system('curl --user karaf:karaf -X POST -H "Content-Type: application/json" '
+ 'http://localhost:30120/onos/v1/network/configuration/apps/ -d @%s/tests/atests/build/aaa_json'
+ % testCaseUtils.get_dir(self, 'voltha'))
+
+ def activate_aaa_app_in_onos(self):
+ logging.info('Activating AAA Application on Onos')
+ testCaseUtils.send_command_to_onos_cli(testCaseUtils.get_dir(self, 'log'),
+ 'voltha_aaa_application_activate.log', 'app activate aaa')
+ statusLines = testCaseUtils.get_fields_from_grep_command(self, 'Activated', 'voltha_aaa_application_activate.log')
+ assert statusLines, 'AAA Application failed to be Activated'
+
+ def deactivate_aaa_app_in_onos(self):
+ logging.info('Deactivating AAA Application on Onos')
+ testCaseUtils.send_command_to_onos_cli(testCaseUtils.get_dir(self, 'log'),
+ 'voltha_aaa_application_deactivate.log', 'app deactivate aaa')
+ statusLines = testCaseUtils.get_fields_from_grep_command(self, 'Deactivated', 'voltha_aaa_application_deactivate.log')
+ assert statusLines, 'AAA Application failed to be Deactivated'
+
+
def get_all_running_pods():
allRunningPods = []
proc1 = subprocess.Popen(['/usr/bin/kubectl', 'get', 'pods', '--all-namespaces'],
@@ -127,11 +167,18 @@
return allRunningPods
-def voltha_initialize(root_dir, voltha_dir, log_dir, adapter):
+def voltha_initialize(root_dir, voltha_dir, log_dir, simtype):
voltha = VolthaMngr()
voltha.v_set_log_dirs(root_dir, voltha_dir, log_dir)
- voltha.stop_all_pods(adapter)
- voltha.reset_kube_adm(adapter)
- voltha.start_all_pods(adapter)
+ voltha.stop_all_pods(simtype)
+ voltha.reset_kube_adm(simtype)
+ voltha.start_all_pods(simtype)
voltha.alter_onos_net_cfg()
voltha.collect_pod_logs()
+ voltha.discover_freeradius_pod_name()
+ voltha.discover_freeradius_ip_addr()
+ voltha.prepare_current_freeradius_ip()
+ voltha.alter_freeradius_ip_in_onos_aaa_application_configuration()
+ voltha.deactivate_aaa_app_in_onos()
+ time.sleep(5)
+ voltha.activate_aaa_app_in_onos()