vsg-voltha-onboarding:
Adding multiple test scenarios and fixing some issues in merge.
Change-Id: If7c0ec2d013817acbf3f7c2c6b0c50ccf54bcab6
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: