vsg-voltha-onboarding:
Adding multiple test scenarios and fixing some issues in merge.
Change-Id: If7c0ec2d013817acbf3f7c2c6b0c50ccf54bcab6
diff --git a/src/test/onboarding/onboardingTest.py b/src/test/onboarding/onboardingTest.py
index 8243b9f..c324320 100644
--- a/src/test/onboarding/onboardingTest.py
+++ b/src/test/onboarding/onboardingTest.py
@@ -1221,23 +1221,8 @@
volt_subscriber_info = volt_subscriber_info)
log.info('Created reserved subscriber %s' %(subId))
- def test_vsg_xos_subscriber_create_all(self):
- for index in xrange(len(self.subscriber_info)):
- #check if the index exists
- subId = self.vsg_xos_subscriber_id(index)
- if subId and subId != '0':
- self.vsg_xos_subscriber_delete(index, subId = subId)
- subId = self.vsg_xos_subscriber_create(index)
- log.info('Created Subscriber %s' %(subId))
-
- def test_vsg_xos_subscriber_delete_all(self):
- for index in xrange(len(self.subscriber_info)):
- subId = self.vsg_xos_subscriber_id(index)
- if subId and subId != '0':
- self.vsg_xos_subscriber_delete(index, subId = subId)
-
@deferred(500)
- def test_exampleservice_xos_subcriber_access_exampleservice(self,index=0):
+ def test_exampleservice_xos_subscriber_access_exampleservice(self,index=0):
"""
Algo:
1. Create two vcpe instances using XOS
@@ -1270,7 +1255,7 @@
return df
@deferred(500)
- def test_exampleservice_multiple_subcribers_access_same_service(self,index1=0,index2=1):
+ def test_exampleservice_multiple_subscribers_access_same_service(self,index1=0,index2=1):
"""
Algo:
1. Create two vcpe instances using XOS
diff --git a/src/test/voltha/volthaTest.py b/src/test/voltha/volthaTest.py
index e5b7c6d..fae699c 100644
--- a/src/test/voltha/volthaTest.py
+++ b/src/test/voltha/volthaTest.py
@@ -912,6 +912,7 @@
reactor.callLater(0, tls_flow_check_on_two_subscriber_same_olt_device, df)
return df
+ @deferred(TESTCASE_TIMEOUT)
def test_two_subscribers_with_voltha_for_eap_tls_authentication_initiating_invalid_tls_packets_for_one_subscriber(self):
"""
Test Method:
@@ -925,6 +926,52 @@
7. Verify that invalid subscriber are not authenticated successfully.
"""
+ df = defer.Deferred()
+ def tls_flow_check_on_two_subscriber_same_olt_device(df):
+ aaa_app = ["org.opencord.aaa"]
+ log_test.info('Enabling ponsim_olt')
+ ponsim_address = '{}:50060'.format(self.VOLTHA_HOST)
+ device_id, status = self.voltha.enable_device('ponsim_olt', address = ponsim_address)
+ devices_list = self.voltha.get_devices()
+ log_test.info('All available devices on voltha = %s'%devices_list['items'])
+
+ onu_device_id = devices_list['items'][1]['id']
+ assert_not_equal(device_id, None)
+ voltha = VolthaCtrl(self.VOLTHA_HOST,
+ rest_port = self.VOLTHA_REST_PORT,
+ uplink_vlan_map = self.VOLTHA_UPLINK_VLAN_MAP)
+ time.sleep(10)
+ switch_map = None
+ olt_configured = False
+ switch_map = voltha.config(fake = self.VOLTHA_CONFIG_FAKE)
+ 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)
+ devices_list = self.voltha.get_devices()
+ thread1 = threading.Thread(target = self.tls_flow_check, args = (self.INTF_RX_DEFAULT,))
+ thread2 = threading.Thread(target = self.tls_flow_check, args = (self.INTF_2_RX_DEFAULT, "no_cert",))
+ thread1.start()
+ time.sleep(randint(1,2))
+ log_test.info('Initiating tls auth packets from one more subscriber on same olt device which is deteced on voltha')
+ thread2.start()
+ time.sleep(10)
+ thread1.join()
+ thread2.join()
+ try:
+ # assert_equal(status, True)
+ assert_equal(self.success, True)
+ time.sleep(10)
+ finally:
+ self.voltha.disable_device(device_id, delete = True)
+ df.callback(0)
+ reactor.callLater(0, tls_flow_check_on_two_subscriber_same_olt_device, df)
+ return df
+
+ @deferred(TESTCASE_TIMEOUT)
def test_two_subscribers_with_voltha_for_eap_tls_authentication_initiating_invalid_cert_for_one_subscriber(self):
"""
Test Method:
@@ -938,6 +985,52 @@
7. Verify that invalid subscriber are not authenticated successfully.
"""
+ df = defer.Deferred()
+ def tls_flow_check_on_two_subscriber_same_olt_device(df):
+ aaa_app = ["org.opencord.aaa"]
+ log_test.info('Enabling ponsim_olt')
+ ponsim_address = '{}:50060'.format(self.VOLTHA_HOST)
+ device_id, status = self.voltha.enable_device('ponsim_olt', address = ponsim_address)
+ devices_list = self.voltha.get_devices()
+ log_test.info('All available devices on voltha = %s'%devices_list['items'])
+
+ onu_device_id = devices_list['items'][1]['id']
+ assert_not_equal(device_id, None)
+ voltha = VolthaCtrl(self.VOLTHA_HOST,
+ rest_port = self.VOLTHA_REST_PORT,
+ uplink_vlan_map = self.VOLTHA_UPLINK_VLAN_MAP)
+ time.sleep(10)
+ switch_map = None
+ olt_configured = False
+ switch_map = voltha.config(fake = self.VOLTHA_CONFIG_FAKE)
+ 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)
+ devices_list = self.voltha.get_devices()
+ thread1 = threading.Thread(target = self.tls_flow_check, args = (self.INTF_RX_DEFAULT,))
+ thread2 = threading.Thread(target = self.tls_flow_check, args = (self.INTF_2_RX_DEFAULT, "invalid_cert",))
+ thread1.start()
+ time.sleep(randint(1,2))
+ log_test.info('Initiating tls auth packets from one more subscriber on same olt device which is deteced on voltha')
+ thread2.start()
+ time.sleep(10)
+ thread1.join()
+ thread2.join()
+ try:
+ # assert_equal(status, True)
+ assert_equal(self.success, True)
+ time.sleep(10)
+ finally:
+ self.voltha.disable_device(device_id, delete = True)
+ df.callback(0)
+ reactor.callLater(0, tls_flow_check_on_two_subscriber_same_olt_device, df)
+ return df
+
+ @deferred(TESTCASE_TIMEOUT)
def test_two_subscribers_with_voltha_for_eap_tls_authentication_with_one_uni_port_disabled(self):
"""
Test Method:
@@ -953,6 +1046,52 @@
10. Verify that other subscriber authenticated successfully.
"""
+ df = defer.Deferred()
+ def tls_flow_check_on_two_subscriber_same_olt_device(df):
+ aaa_app = ["org.opencord.aaa"]
+ log_test.info('Enabling ponsim_olt')
+ ponsim_address = '{}:50060'.format(self.VOLTHA_HOST)
+ device_id, status = self.voltha.enable_device('ponsim_olt', address = ponsim_address)
+ devices_list = self.voltha.get_devices()
+ log_test.info('All available devices on voltha = %s'%devices_list['items'])
+
+ onu_device_id = devices_list['items'][1]['id']
+ assert_not_equal(device_id, None)
+ voltha = VolthaCtrl(self.VOLTHA_HOST,
+ rest_port = self.VOLTHA_REST_PORT,
+ uplink_vlan_map = self.VOLTHA_UPLINK_VLAN_MAP)
+ time.sleep(10)
+ switch_map = None
+ olt_configured = False
+ switch_map = voltha.config(fake = self.VOLTHA_CONFIG_FAKE)
+ 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)
+ devices_list = self.voltha.get_devices()
+ thread1 = threading.Thread(target = self.tls_flow_check, args = (self.INTF_RX_DEFAULT,))
+ thread2 = threading.Thread(target = self.tls_flow_check, args = (self.INTF_2_RX_DEFAULT, "uni_port_admin_down",))
+ thread1.start()
+ time.sleep(randint(1,2))
+ log_test.info('Initiating tls auth packets from one more subscriber on same olt device which is deteced on voltha')
+ thread2.start()
+ time.sleep(10)
+ thread1.join()
+ thread2.join()
+ try:
+ # assert_equal(status, True)
+ assert_equal(self.success, True)
+ time.sleep(10)
+ finally:
+ self.voltha.disable_device(device_id, delete = True)
+ df.callback(0)
+ reactor.callLater(0, tls_flow_check_on_two_subscriber_same_olt_device, df)
+ return df
+
+ @deferred(TESTCASE_TIMEOUT)
def test_subscriber_with_voltha_for_dhcp_request(self):
"""
Test Method:
diff --git a/src/test/vsg/vsgTest.py b/src/test/vsg/vsgTest.py
index e53c4e3..439b94f 100644
--- a/src/test/vsg/vsgTest.py
+++ b/src/test/vsg/vsgTest.py
@@ -803,7 +803,7 @@
return df
@deferred(120)
- def test_vsg_multiple_subscribers_for_same_vcpe_instace(self):
+ def test_vsg_multiple_subscribers_for_same_vcpe_instance(self):
"""
Test Method:
1. Create a vcpe instance
@@ -2964,7 +2964,7 @@
def test_vsg_xos_subscriber_external_connectivity_after_vsg_vm_is_restarted(self, index=0):
"""
Test Method:
- 1.Create two vcpe instances in two different vsg vms using XOS
+ 1.Create subscriber
2.Verify external connectivity through vcpe instances from cord-tester
3.Bring down first vSG vm
4.Verify external network cant be reachable form first vcpe interface
@@ -3009,7 +3009,7 @@
return df
@deferred(780)
- def test_vsg_xos_multiple_subscribers_external_connectivity_if_two_vsgs_stop_and_start(self,index1=4,index2=6):
+ def test_vsg_xos_multiple_subscribers_external_connectivity_if_two_vsgs_stop_and_start(self, index1=4, index2=6):
"""
Test Method:
1.Create two vcpe instances in two different vsg vms using XOS
@@ -3062,7 +3062,7 @@
3.Verify external network can be reachable from cord-tester
4.Add an iptable rule to drop packets destined to external network in vcpe
5.Verify now external network cant be reachable
- 6.Delele the iptale in vcpe instance
+ 6.Delele the iptable in vcpe instance
7.Verify external network can be reachable from cord-tester
"""
df = defer.Deferred()