Changes to delete olt subscriber and remove voltha device after subscriber tests are done.
The voltha device enable tests now also cleanup/disable the device after the test is complete.
Change-Id: I2596bcc4664ddee818a01693600d528e3f834264
diff --git a/src/test/cordSubscriber/cordSubscriberTest.py b/src/test/cordSubscriber/cordSubscriberTest.py
index f4f5df8..a713199 100644
--- a/src/test/cordSubscriber/cordSubscriberTest.py
+++ b/src/test/cordSubscriber/cordSubscriberTest.py
@@ -35,7 +35,7 @@
from threadPool import ThreadPool
from portmaps import g_subscriber_port_map
from OltConfig import *
-from CordTestServer import cord_test_onos_restart, cord_test_shell
+from CordTestServer import cord_test_onos_restart, cord_test_shell, cord_test_radius_restart
from CordTestUtils import log_test, get_controller
from CordLogger import CordLogger
from CordTestConfig import setup_module
@@ -209,6 +209,7 @@
table_app_file = os.path.join(test_path, '..', 'apps/ciena-cordigmp-multitable-2.0-SNAPSHOT.oar')
app_file = os.path.join(test_path, '..', 'apps/ciena-cordigmp-2.0-SNAPSHOT.oar')
olt_app_file = os.path.join(test_path, '..', 'apps/olt-app-1.2-SNAPSHOT.oar')
+ olt_app_name = 'org.onosproject.olt'
onos_config_path = os.path.join(test_path, '..', 'setup/onos-config')
olt_conf_file = os.getenv('OLT_CONFIG_FILE', os.path.join(test_path, '..', 'setup/olt_config.json'))
cpqd_path = os.path.join(test_path, '..', 'setup')
@@ -498,11 +499,17 @@
return self.test_status
def tls_verify(self, subscriber):
+ def tls_fail_cb():
+ log_test.info('TLS verification failed')
if subscriber.has_service('TLS'):
+ OnosCtrl('org.opencord.aaa').deactivate()
time.sleep(2)
- tls = TLSAuthTest(intf = subscriber.rx_intf)
+ OnosCtrl('org.opencord.aaa').activate()
+ time.sleep(5)
+ tls = TLSAuthTest(fail_cb = tls_fail_cb, intf = subscriber.rx_intf)
log_test.info('Running subscriber %s tls auth test' %subscriber.name)
tls.runTest()
+ assert_equal(tls.failTest, False)
self.test_status = True
return self.test_status
else:
@@ -2602,13 +2609,25 @@
- def remove_olt(self):
- OnosCtrl.uninstall_app(self.olt_app_file)
+ def remove_olt(self, switch_map):
+ controller = get_controller()
+ auth = ('karaf', 'karaf')
+ #remove subscriber for every port on all the voltha devices
+ for device, device_map in switch_map.iteritems():
+ uni_ports = device_map['ports']
+ uplink_vlan = device_map['uplink_vlan']
+ for port in uni_ports:
+ rest_url = 'http://{}:8181/onos/olt/oltapp/{}/{}'.format(controller,
+ device,
+ port)
+ resp = requests.delete(rest_url, auth = auth)
+ if resp.status_code not in [204, 202, 200]:
+ log_test.error('Error deleting subscriber for device %s on port %s' %(device, port))
+ else:
+ log_test.info('Deleted subscriber for device %s on port %s' %(device, port))
def config_olt(self, switch_map):
controller = get_controller()
- OnosCtrl.install_app(self.olt_app_file)
- time.sleep(5)
auth = ('karaf', 'karaf')
#configure subscriber for every port on all the voltha devices
for device, device_map in switch_map.iteritems():
@@ -2634,28 +2653,46 @@
if self.VOLTHA_OLT_TYPE.startswith('ponsim'):
ponsim_address = '{}:50060'.format(self.VOLTHA_HOST)
log_test.info('Enabling ponsim olt')
- status = voltha.enable_device(self.VOLTHA_OLT_TYPE, address = ponsim_address)
+ device_id, status = voltha.enable_device(self.VOLTHA_OLT_TYPE, address = ponsim_address)
else:
log_test.info('Enabling OLT instance for %s with mac %s' %(self.VOLTHA_OLT_TYPE, self.VOLTHA_OLT_MAC))
- status = voltha.enable_device(self.VOLTHA_OLT_TYPE, self.VOLTHA_OLT_MAC)
+ device_id, status = voltha.enable_device(self.VOLTHA_OLT_TYPE, self.VOLTHA_OLT_MAC)
+ assert_not_equal(device_id, None)
+ if status == False:
+ voltha.disable_device(device_id, delete = True)
assert_equal(status, True)
time.sleep(10)
- switch_map = voltha.config(fake = self.VOLTHA_CONFIG_FAKE)
- if not switch_map:
- log_test.info('No voltha devices found')
- return
- log_test.info('Adding subscribers through OLT app')
- self.config_olt(switch_map)
- time.sleep(5)
- self.num_subscribers = 1
- self.num_channels = 1
- services = ('TLS', 'IGMP')
- test_status = self.subscriber_join_verify(num_subscribers = self.num_subscribers,
- num_channels = self.num_channels,
- cbs = (self.tls_verify, self.dhcp_next_verify,
- self.voltha_igmp_next_verify, self.traffic_verify),
- port_list = self.generate_port_list(self.num_subscribers,
- self.num_channels),
- services = services)
- assert_equal(test_status, True)
+ switch_map = None
+ olt_configured = False
+ try:
+ switch_map = voltha.config(fake = self.VOLTHA_CONFIG_FAKE)
+ if not switch_map:
+ log_test.info('No voltha devices found')
+ return
+ log_test.info('Installing OLT app')
+ OnosCtrl.install_app(self.olt_app_file)
+ time.sleep(5)
+ log_test.info('Adding subscribers through OLT app')
+ self.config_olt(switch_map)
+ olt_configured = True
+ time.sleep(5)
+ self.num_subscribers = 1
+ self.num_channels = 1
+ services = ('TLS', 'IGMP')
+ test_status = self.subscriber_join_verify(num_subscribers = self.num_subscribers,
+ num_channels = self.num_channels,
+ cbs = (self.tls_verify, self.dhcp_next_verify,
+ self.voltha_igmp_next_verify, self.traffic_verify),
+ port_list = self.generate_port_list(self.num_subscribers,
+ self.num_channels),
+ services = services)
+ assert_equal(test_status, True)
+ finally:
+ if switch_map is not None:
+ if olt_configured is True:
+ self.remove_olt(switch_map)
+ voltha.disable_device(device_id, delete = True)
+ time.sleep(10)
+ log_test.info('Uninstalling OLT app')
+ OnosCtrl.uninstall_app(self.olt_app_name)
diff --git a/src/test/utils/VolthaCtrl.py b/src/test/utils/VolthaCtrl.py
index 86d17e9..d018edb 100644
--- a/src/test/utils/VolthaCtrl.py
+++ b/src/test/utils/VolthaCtrl.py
@@ -192,7 +192,7 @@
url = '{}/local/devices'.format(self.rest_url)
if olt_mac is None and address is None:
log.error('Either olt mac or address needs to be specified')
- return False
+ return None, False
if olt_mac is not None:
device_config = { 'type' : olt_type, 'mac_address' : olt_mac }
else:
@@ -204,25 +204,39 @@
log.info('Pre-provisioning %s with address %s' %(olt_type, address))
resp = requests.post(url, data = json.dumps(device_config))
if resp.ok is not True or resp.status_code != 200:
- return False
+ return None, False
device_id = resp.json()['id']
log.info('Enabling device %s' %(device_id))
enable_url = '{}/{}/enable'.format(url, device_id)
resp = requests.post(enable_url)
if resp.ok is not True or resp.status_code != 200:
- return False
+ return None, False
#get operational status
time.sleep(5)
log.info('Checking operational status for device %s' %(device_id))
resp = requests.get('{}/{}'.format(url, device_id))
if resp.ok is not True or resp.status_code != 200:
- return False
+ return device_id, False
device_info = resp.json()
if device_info['oper_status'] != 'ACTIVE' or \
device_info['admin_state'] != 'ENABLED' or \
device_info['connect_status'] != 'REACHABLE':
- return False
+ return device_id, False
+ return device_id, True
+
+ def disable_device(self, device_id, delete = True):
+ log.info('Disabling device %s' %(device_id))
+ disable_url = '{}/local/devices/{}/disable'.format(self.rest_url, device_id)
+ resp = requests.post(disable_url)
+ if resp.ok is not True or resp.status_code != 200:
+ return False
+ if delete is True:
+ log.info('Deleting device %s' %(device_id))
+ delete_url = '{}/local/devices/{}/delete'.format(self.rest_url, device_id)
+ resp = requests.delete(delete_url)
+ if resp.status_code not in [204, 202, 200]:
+ return False
return True
def get_operational_status(self, device_id):
diff --git a/src/test/voltha/volthaTest.py b/src/test/voltha/volthaTest.py
index db65d69..b74753e 100644
--- a/src/test/voltha/volthaTest.py
+++ b/src/test/voltha/volthaTest.py
@@ -1,6 +1,7 @@
import os
import sys
import unittest
+import time
from nose.tools import *
from CordTestConfig import setup_module
from CordTestUtils import log_test
@@ -18,16 +19,26 @@
def setUpClass(cls):
cls.voltha = VolthaCtrl(cls.VOLTHA_HOST, rest_port = cls.VOLTHA_REST_PORT)
- def test_olt_enable(self):
+ def test_olt_enable_disable(self):
log_test.info('Enabling OLT type %s, MAC %s' %(self.OLT_TYPE, self.OLT_MAC))
- status = self.voltha.enable_device(self.OLT_TYPE, self.OLT_MAC)
- assert_equal(status, True)
+ device_id, status = self.voltha.enable_device(self.OLT_TYPE, self.OLT_MAC)
+ assert_not_equal(device_id, None)
+ try:
+ assert_equal(status, True)
+ time.sleep(10)
+ finally:
+ self.voltha.disable_device(device_id, delete = True)
- def test_ponsim_enable(self):
+ def test_ponsim_enable_disable(self):
log_test.info('Enabling ponsim_olt')
ponsim_address = '{}:50060'.format(self.VOLTHA_HOST)
- status = self.voltha.enable_device('ponsim_olt', address = ponsim_address)
- assert_equal(status, True)
+ device_id, status = self.voltha.enable_device('ponsim_olt', address = ponsim_address)
+ assert_not_equal(device_id, None)
+ try:
+ assert_equal(status, True)
+ time.sleep(10)
+ finally:
+ self.voltha.disable_device(device_id, delete = True)
def test_subscriber_with_voltha_for_eap_tls_authentication(self):
"""