Test-cordvtn:
Adding new test scenarios with nova instances and validation
using cord vtn app.
Validating connectivity across instances for public and mgmt network
Also checking for net connectivity in some scenarios.
Change-Id: I7505bb9b6782c9b9cc69bc901d3e5be7d398a2f2
diff --git a/src/test/cordvtn/cordvtnTest.py b/src/test/cordvtn/cordvtnTest.py
index 8c2f49b..bb5087f 100644
--- a/src/test/cordvtn/cordvtnTest.py
+++ b/src/test/cordvtn/cordvtnTest.py
@@ -19,7 +19,7 @@
import keystoneclient.apiclient.exceptions
import neutronclient.v2_0.client as nclient
import neutronclient.common.exceptions
-#import novaclient.v1_1.client as novaclient
+import novaclient.v1_1.client as novaclient
from novaclient import client as nova_client
from multiprocessing import Pool
from neutronclient.v2_0 import client as neutron_client
@@ -30,6 +30,7 @@
from CordLogger import CordLogger
from TestManifest import TestManifest
from OnosFlowCtrl import OnosFlowCtrl
+from scapy.all import *
from credentials import *
from VSGAccess import VSGAccess
from SSHTestAgent import SSHTestAgent
@@ -152,7 +153,7 @@
def get_compute_nodes(cls):
credentials = get_nova_credentials_v2()
novaclient = nova_client.Client('2', **credentials)
- print(novaclient.hypervisors.list())
+ print novaclient.hypervisors.list()
return novaclient.hypervisors.list()
def create_network(i):
@@ -163,10 +164,10 @@
while True:
try:
net = neutron.create_network(body=json)
- print('\nnetwork-' + str(i) + ' created')
+ print '\nnetwork-' + str(i) + ' created'
return net
except Exception as e:
- print(e)
+ print e
continue
def create_tenant(tenant_name):
@@ -182,7 +183,7 @@
user_name = tenant_name + '-user-' + str(j)
user_data.append(create_user(user_name, tenant_id))
- print(" Tenant and User Created")
+ print " Tenant and User Created"
tenant_data = {'tenant_name': tenant_name,
'tenant_id': tenant_id,
@@ -442,7 +443,7 @@
match = True
return match
if match is True:
- print("Network search is successful")
+ print"Network search is successful"
return match
def get_key_value(self, d, key = None, value = None,):
@@ -458,7 +459,7 @@
if type(d[i]) is dict:
match,ret_k,ret_v = self.get_key_value(d[i], key, value)
if match is True:
- print("Network creation is successful")
+ print "Network creation is successful"
break
else:
for k, v in d.items():
@@ -559,23 +560,38 @@
return resp
def nova_instance_creation_and_validation(self,net_name,nova_obj,instance_name,image_name, flavor_id):
+ print nova_obj.images.list()
image = nova_obj.images.find(name=image_name)
flavor = nova_obj.flavors.find(name=flavor_id)
- network = nova_obj.networks.find(label=net_name)
- print(network.id)
+
+ nics_list = ""
+ if len(net_name) == 2:
+ network_1 = nova_obj.networks.find(label=net_name[0])
+ network_2 = nova_obj.networks.find(label=net_name[1])
+ print network_1.id
+ print network_2.id
+ nics_list = [{'net-id':network_1.id},{'net-id':network_2.id}]
+ else:
+ network_1 = nova_obj.networks.find(label=net_name)
+ print network_1.id
+ nics_list = [{'net-id':network_1.id}]
server = nova_obj.servers.create(name = instance_name,
image = image.id,
flavor = flavor.id,
- nics = [{'net-id':network.id}])
+ nics = nics_list,
+ userdata = "#cloud-config \n password: ubuntu \n chpasswd: { expire: False }\n ssh_pwauth: True")
+ # key_name = keypair.name)
server_details = nova_obj.servers.find(id=server.id)
print('Server is launched and status is %s' %server_details.status)
if server_details.status == 'BUILD':
- time.sleep(20)
+ time.sleep(120)
server_details = nova_obj.servers.find(id=server.id)
print('After delay server status is %s state'%server_details.status)
if server_details.status == 'ERROR':
print('Server status is still in %s state'%server_details.status)
+ server_boot_up_log = nova_obj.servers.get_console_output(server.id)
+ print 'Server boot Up console log \n%s'%server_boot_up_log
return server_details
def create_net_subnet_nova_instance(self,net_name, subnet_name, instance_vm_details,management_type):
@@ -591,7 +607,7 @@
creds = get_nova_credentials()
nova = nova_client.Client('2', **creds)
new_instance_details = self.nova_instance_creation_and_validation(net_name,nova,instance_vm_details[0],instance_vm_details[1],instance_vm_details[2])
- #assert_equal(new_instance_details.status, 'ACTIVE')
+ assert_equal(new_instance_details.status, 'ACTIVE')
return [nova, new_instance_details]
def nova_instance_tenants_access_check(self, target_tenants_details, compute_details = None, source_tenants_details = None , check_type = 'Ping_from_compute'):
@@ -608,37 +624,40 @@
compute_ip = compute_details.ip
## TO DO New instance is not getting subnet ip, hence checking only mysite-vsg1 vm from compute node
- if target_tenants_details.addresses == {}:
- target_tenants_details.addresses = '10.1.0.1'
+ if target_tenants_details == {}:
+ target_tenants_details = '10.1.0.1'
## TO DO New instance is not getting subnet ip, hence checking only mysite-vsg1 vm from compute node
if source_tenants_details is not None:
- if source_tenants_details.addresses == {}:
- source_tenants_details.addresses = '10.1.0.1'
+ if source_tenants_details == {}:
+ source_tenants_details = '10.1.0.1'
if check_type == "Ping_from_compute":
- cmd = "ping -c 3 {0}".format(target_tenants_details.addresses)
+ cmd2 = "ping -c 3 {0}".format(target_tenants_details)
ssh_agent = SSHTestAgent(host = compute_ip)
- status, output = ssh_agent.run_cmd(cmd, timeout = 5)
+ status, output = ssh_agent.run_cmd(cmd2, timeout = 5)
+ print output
if source_tenants_details is not None:
if check_type == "Ping_from_source_tenant":
- cmd = "ping -c 3 {0}".format(target_tenants_details.addresses)
- ssh_cmd = 'ssh {} {}'.format(source_tenants_details.addresses, cmd)
- print(ssh_cmd)
+ cmd = "ping -c 3 {0}".format(target_tenants_details)
+ ssh_cmd = 'ssh {} {}'.format(source_tenants_details, cmd)
ssh_agent = SSHTestAgent(host = compute_ip)
- status, output = ssh_agent.run_cmd(cmd, timeout = 5)
+ status, output = ssh_agent.run_cmd(ssh_cmd, timeout = 5)
+ print output
if check_type == "Ping_to_external":
cmd = "ping -c 3 google.com"
+ ssh_cmd = 'ssh {} {}'.format(target_tenants_details, cmd)
ssh_agent = SSHTestAgent(host = compute_ip)
- status, output = ssh_agent.run_cmd(cmd, timeout = 5)
+ status, output = ssh_agent.run_cmd(ssh_cmd, timeout = 5)
+ print output
if status == True and output:
- print("Ping is successful")
+ print "Ping is successful"
output = output.strip()
- else:
- print("Ping is not successful")
+ elif status == False:
+ print "Ping is not successful"
output = None
return [status, output]
@@ -646,9 +665,9 @@
def nova_instance_deletion(self, nova_obj, server_details):
results_nova_instance_deletion=nova_obj.servers.delete(server_details.id)
if results_nova_instance_deletion == None:
- print("Nova instance is deleted")
+ print"Nova instance is deleted"
else:
- print("Nova instance is not deleted")
+ print"Nova instance is not deleted"
return results_nova_instance_deletion
def test_cordvtn_neutron_network_creation_and_validation_on_head_node_with_neutron_service(self):
@@ -788,7 +807,7 @@
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork whose ip is under management network
+ 1. Create subnetwork whose ip is under management network
2. Run sync command for cordvtn
3. Do GET Rest API and validate creation of network
4. Pushed the network type as management local to onos
@@ -825,11 +844,11 @@
self.neutron_network_deletion(test_net_name)
assert_equal(net_type_json['message'], 'null already exists')
- def test_cordvtn_management_network_creation_with_launching_nova_instance_and_validation_on_head_node_with_nova_service(self):
+ def test_cordvtn_with_management_network_creation_launching_nova_instance_and_validation_on_head_node_with_nova_service(self):
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork whose ip is under management network
+ 1. Create subnetwork whose ip is under management network
3. Do GET Rest API and validate creation of network
4. Create new nova instance under management network
5. Validate new nova instance is created on nova service
@@ -855,11 +874,11 @@
self.nova_instance_deletion(nova, new_instance_details)
assert_equal(new_instance_details.status, 'ACTIVE')
- def test_cordvtn_neutron_public_network_creation_and_validation_on_head_node_with_neutron_service(self):
+ def test_cordvtn_with_public_network_creation_and_validation_on_head_node_with_neutron_service(self):
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork who ip is under management network
+ 1. Create subnetwork who ip is under management network
2. Run sync command for cordvtn
3. Do GET Rest API and validate creation of network
4. Validate network synch with created network in cord-onos
@@ -873,11 +892,11 @@
self.neutron_network_deletion(test_net_name)
assert_equal(sub_result[0], True)
- def test_cordvtn_neutron_public_network_creation_and_validation_on_onos(self):
+ def test_cordvtn_with_public_network_creation_and_validation_on_onos(self):
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork whoes ip is under management network
+ 1. Create subnetwork whoes ip is under management network
2. Run sync command for cordvtn
3. Do GET Rest API and validate creation of network
"""
@@ -907,11 +926,11 @@
assert_equal(sub_net_id[0], True)
break
- def test_cordvtn_neutron_public_network_creation_and_post_network_type_as_public_to_onos(self):
+ def test_cordvtn_with_public_network_creation_and_post_network_type_as_public_to_onos(self):
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork whose ip is under management network
+ 1. Create subnetwork whose ip is under management network
2. Run sync command for cordvtn
3. Do GET Rest API and validate creation of network
4. Pushed the network type as management local to onos
@@ -953,7 +972,7 @@
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork whose ip is under public network
+ 1. Create subnetwork whose ip is under public network
3. Do GET Rest API and validate creation of network
4. Create new nova instance under public network
5. Validate new nova instance is created on nova service
@@ -978,11 +997,11 @@
self.nova_instance_deletion(nova, new_instance_details)
assert_equal(new_instance_details.status, 'ACTIVE')
- def test_cordvtn_neutron_private_network_creation_and_validation_on_head_node_with_neutron_service(self):
+ def test_cordvtn_with_private_network_creation_and_validation_on_head_node_with_neutron_service(self):
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork who ip is under private network
+ 1. Create subnetwork who ip is under private network
2. Run sync command for cordvtn
3. Do GET Rest API and validate creation of network
4. Validate network synch with created network in cord-onos
@@ -996,11 +1015,11 @@
self.neutron_network_deletion(test_net_name)
assert_equal(sub_result[0], True)
- def test_cordvtn_neutron_private_network_creation_and_validation_on_onos(self):
+ def test_cordvtn_with_private_network_creation_and_validation_on_onos(self):
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork whoes ip is under management network
+ 1. Create subnetwork whoes ip is under management network
2. Run sync command for cordvtn
3. Do GET Rest API and validate creation of network
"""
@@ -1031,11 +1050,11 @@
assert_equal(sub_net_id[0], True)
break
- def test_cordvtn_neutron_private_network_creation_and_post_network_type_as_private_to_onos(self):
+ def test_cordvtn_with_private_network_creation_and_post_network_type_as_private_to_onos(self):
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork whose ip is under management network
+ 1. Create subnetwork whose ip is under management network
2. Run sync command for cordvtn
3. Do GET Rest API and validate creation of network
4. Pushed the network type as management local to onos
@@ -1073,11 +1092,11 @@
self.neutron_network_deletion(test_net_name)
assert_equal(net_type_json['message'], 'null already exists')
- def test_cordvtn_private_network_creation_with_launching_nova_instance_and_validation_on_head_node_with_nova_service(self):
+ def test_cordvtn_with_private_network_creation_launching_nova_instance_and_validating_on_head_node_with_nova_service(self):
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork whose ip is under private network
+ 1. Create subnetwork whose ip is under private network
3. Do GET Rest API and validate creation of network
4. Create new nova instance under private network
5. Validate new nova instance is created on nova service
@@ -1102,11 +1121,11 @@
self.nova_instance_deletion(nova, new_instance_details)
assert_equal(new_instance_details.status, 'ACTIVE')
- def test_cordvtn_management_network_instance_ping_form_host_machine_or_compute_node(self):
+ def test_cordvtn_management_network_instance_and_validate_connectivity_from_host_machine_or_compute_node(self):
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork whose ip is under management network
+ 1. Create subnetwork whose ip is under management network
3. Do GET Rest API and validate creation of network
4. Create new nova instance under management network
5. Validate new nova instance is created on nova service
@@ -1116,7 +1135,8 @@
test_sub_net_cidr = ["management","172.27.0.0/24", "172.27.0.20", "172.27.0.21"]
test_management_type = "management_local"
instance_vm_name = 'vtn_test_17_nova_instance_management_net'
- image_name = "vsg-1.1"
+ #image_name = "vsg-1.1"
+ image_name = "trusty-server-multi-nic"
flavor_id = 'm1.small'
result = self.neutron_network_creation_and_validation(test_net_name)
assert_equal(result, True)
@@ -1130,19 +1150,22 @@
creds = get_nova_credentials()
nova = nova_client.Client('2', **creds)
new_instance_details = self.nova_instance_creation_and_validation(test_net_name,nova,instance_vm_name,image_name,flavor_id)
- #assert_equal(new_instance_details.status, 'ACTIVE')
+ assert_equal(new_instance_details.status, 'ACTIVE')
compute_details = self.get_compute_nodes()
- print(new_instance_details.addresses)
+ print new_instance_details.addresses
+ time.sleep(60)
status, output = self.nova_instance_tenants_access_check(new_instance_details)
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(5)
self.neutron_network_deletion(test_net_name)
self.nova_instance_deletion(nova, new_instance_details)
assert_equal(status, True)
- def test_cordvtn_management_network_instance_ping_to_external_network(self):
+ def test_cordvtn_for_management_network_instance_and_validate_connectivity_to_external_network(self):
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork whose ip is under management network
+ 1. Create subnetwork whose ip is under management network
3. Do GET Rest API and validate creation of network
4. Create new nova instance under management network
5. Validate new nova instance is created on nova service
@@ -1166,19 +1189,19 @@
creds = get_nova_credentials()
nova = nova_client.Client('2', **creds)
new_instance_details = self.nova_instance_creation_and_validation(test_net_name,nova,instance_vm_name,image_name,flavor_id)
- #assert_equal(new_instance_details.status, 'ACTIVE')
+ assert_equal(new_instance_details.status, 'ACTIVE')
compute_details = self.get_compute_nodes()
- print(new_instance_details.addresses)
+ print new_instance_details.addresses
status, output = self.nova_instance_tenants_access_check(new_instance_details, check_type = "Ping_to_external")
self.neutron_network_deletion(test_net_name)
self.nova_instance_deletion(nova, new_instance_details)
assert_equal(status, False)
- def test_cordvtn_management_network_two_instance_ping_between_two_instances(self):
+ def test_cordvtn_with_management_network_creating_two_instances_and_validate_connectivity_between_two(self):
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork whose ip is under management network
+ 1. Create subnetwork whose ip is under management network
3. Do GET Rest API and validate creation of network
4. Create first nova instance under management network
5. Validate first nova instance is created on nova service
@@ -1206,23 +1229,23 @@
nova = nova_client.Client('2', **creds)
first_nova_instance_details = self.nova_instance_creation_and_validation(test_net_name,nova,first_instance_vm_name,image_name,flavor_id)
second_nova_instance_details = self.nova_instance_creation_and_validation(test_net_name,nova,second_instance_vm_name,image_name,flavor_id)
- #assert_equal(first_instance_details.status, 'ACTIVE')
- #assert_equal(second_instance_details.status, 'ACTIVE')
+ assert_equal(first_instance_details.status, 'ACTIVE')
+ assert_equal(second_instance_details.status, 'ACTIVE')
compute_details = self.get_compute_nodes()
- print('New nova instance ip addresses are ')
- print(first_nova_instance_details.addresses)
- print(second_nova_instance_details.addresses)
+ print 'New nova instance ip addresses are '
+ print first_nova_instance_details.addresses
+ print second_nova_instance_details.addresses
status, output = self.nova_instance_tenants_access_check(first_nova_instance_details,source_tenants_details = second_nova_instance_details, check_type = "Ping_from_source_tenant")
self.neutron_network_deletion(test_net_name)
self.nova_instance_deletion(nova, first_nova_instance_details)
self.nova_instance_deletion(nova, second_nova_instance_details)
assert_equal(status, False)
- def test_cordvtn_two_management_network_instances_ping_between_two_networks_via_management_network(self):
+ def test_cordvtn_creating_two_management_network_instances_and_validate_connectivity_between_two_networks_via_management_network(self):
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork whose ip is under management network
+ 1. Create subnetwork whose ip is under management network
3. Do GET Rest API and validate creation of network
4. Create new nova instance under management network
5. Validate new nova instance is created on nova service
@@ -1242,12 +1265,12 @@
nova_netA, nova_instance_details_netA = self.create_net_subnet_nova_instance(test_netA_name, test_sub_netA_cidr, netA_instance_vm_details, test_management_type)
nova_netB, nova_instance_details_netB = self.create_net_subnet_nova_instance(test_netB_name, test_sub_netB_cidr, netB_instance_vm_details, test_management_type)
- #assert_equal(nova_instance_details_netA.status, 'ACTIVE')
- #assert_equal(nova_instance_details_netB.status, 'ACTIVE')
+ assert_equal(nova_instance_details_netA.status, 'ACTIVE')
+ assert_equal(nova_instance_details_netB.status, 'ACTIVE')
compute_details = self.get_compute_nodes()
- print('New nova instance ip addresses are ')
- print(nova_instance_details_netA.addresses)
- print(nova_instance_details_netB.addresses)
+ print 'New nova instance ip addresses are '
+ print nova_instance_details_netA.addresses
+ print nova_instance_details_netB.addresses
status, output = self.nova_instance_tenants_access_check(nova_instance_details_netA, source_tenants_details = nova_instance_details_netB,check_type = "Ping_from_source_tenant")
self.neutron_network_deletion(test_netA_name)
self.nova_instance_deletion(nova_netA, nova_instance_details_netA)
@@ -1255,12 +1278,11 @@
self.nova_instance_deletion(nova_netB, nova_instance_details_netB)
assert_equal(status, False)
- def test_cordvtn_public_network_instance_ping_form_host_machine_or_compute_node(self):
-
+ def test_cordvtn_creating_public_network_instance_and_validate_connectivity_from_host_machine_or_compute_node(self):
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork whose ip is under public network
+ 1. Create subnetwork whose ip is under public network
3. Do GET Rest API and validate creation of network
4. Create new nova instance under public network
5. Validate new nova instance is created on nova service
@@ -1284,19 +1306,19 @@
creds = get_nova_credentials()
nova = nova_client.Client('2', **creds)
new_instance_details = self.nova_instance_creation_and_validation(test_net_name,nova,instance_vm_name,image_name,flavor_id)
- #assert_equal(new_instance_details.status, 'ACTIVE')
+ assert_equal(new_instance_details.status, 'ACTIVE')
compute_details = self.get_compute_nodes()
- print(new_instance_details.addresses)
+ print new_instance_details.addresses
status, output = self.nova_instance_tenants_access_check(new_instance_details)
self.neutron_network_deletion(test_net_name)
self.nova_instance_deletion(nova, new_instance_details)
assert_equal(status, False)
- def test_cordvtn_public_network_instance_ping_to_external_network(self):
+ def test_cordvtn_creating_public_network_instance_and_validate_connectivity_to_external_network(self):
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork whose ip is under public network
+ 1. Create subnetwork whose ip is under public network
3. Do GET Rest API and validate creation of network
4. Create new nova instance under public network
5. Validate new nova instance is created on nova service
@@ -1320,20 +1342,19 @@
creds = get_nova_credentials()
nova = nova_client.Client('2', **creds)
new_instance_details = self.nova_instance_creation_and_validation(test_net_name,nova,instance_vm_name,image_name,flavor_id)
- #assert_equal(new_instance_details.status, 'ACTIVE')
+ assert_equal(new_instance_details.status, 'ACTIVE')
compute_details = self.get_compute_nodes()
- print(new_instance_details.addresses)
+ print new_instance_details.addresses
status, output = self.nova_instance_tenants_access_check(new_instance_details, check_type = "Ping_to_external")
self.neutron_network_deletion(test_net_name)
self.nova_instance_deletion(nova, new_instance_details)
assert_equal(status, True)
-
- def test_cordvtn_public_network_two_instance_ping_between_two_instances(self):
+ def test_cordvtn_creating_public_network_with_two_instances_and_validate_connectivity_between_two(self):
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork whose ip is under public network
+ 1. Create subnetwork whose ip is under public network
3. Do GET Rest API and validate creation of network
4. Create first nova instance under public network
5. Validate first nova instance is created on nova service
@@ -1361,23 +1382,23 @@
nova = nova_client.Client('2', **creds)
first_nova_instance_details = self.nova_instance_creation_and_validation(test_net_name,nova,first_instance_vm_name,image_name,flavor_id)
second_nova_instance_details = self.nova_instance_creation_and_validation(test_net_name,nova,second_instance_vm_name,image_name,flavor_id)
- #assert_equal(first_instance_details.status, 'ACTIVE')
- #assert_equal(second_instance_details.status, 'ACTIVE')
+ assert_equal(first_instance_details.status, 'ACTIVE')
+ assert_equal(second_instance_details.status, 'ACTIVE')
compute_details = self.get_compute_nodes()
- print('New nova instance ip addresses are ')
- print(first_nova_instance_details.addresses)
- print(second_nova_instance_details.addresses)
+ print 'New nova instance ip addresses are '
+ print first_nova_instance_details.addresses
+ print second_nova_instance_details.addresses
status, output = self.nova_instance_tenants_access_check(first_nova_instance_details,source_tenants_details = second_nova_instance_details, check_type = "Ping_from_source_tenant")
self.neutron_network_deletion(test_net_name)
self.nova_instance_deletion(nova, first_nova_instance_details)
self.nova_instance_deletion(nova, second_nova_instance_details)
assert_equal(status, False)
- def test_cordvtn_two_public_network_instances_ping_between_two_networks_via_public_network(self):
+ def test_cordvtn_creating_two_public_network_instances_and_check_connectvity_between_two_networks_via_public_network(self):
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork whose ip is under public network
+ 1. Create subnetwork whose ip is under public network
3. Do GET Rest API and validate creation of network
4. Create new nova instance under public network
5. Validate new nova instance is created on nova service
@@ -1397,12 +1418,12 @@
nova_netA, nova_instance_details_netA = self.create_net_subnet_nova_instance(test_netA_name, test_sub_netA_cidr, netA_instance_vm_details, test_management_type)
nova_netB, nova_instance_details_netB = self.create_net_subnet_nova_instance(test_netB_name, test_sub_netB_cidr, netB_instance_vm_details, test_management_type)
- #assert_equal(nova_instance_details_netA.status, 'ACTIVE')
- #assert_equal(nova_instance_details_netB.status, 'ACTIVE')
+ assert_equal(nova_instance_details_netA.status, 'ACTIVE')
+ assert_equal(nova_instance_details_netB.status, 'ACTIVE')
compute_details = self.get_compute_nodes()
- print('New nova instance ip addresses are ')
- print(nova_instance_details_netA.addresses)
- print(nova_instance_details_netB.addresses)
+ print 'New nova instance ip addresses are '
+ print nova_instance_details_netA.addresses
+ print nova_instance_details_netB.addresses
status, output = self.nova_instance_tenants_access_check(nova_instance_details_netA, source_tenants_details = nova_instance_details_netB,check_type = "Ping_from_source_tenant")
self.neutron_network_deletion(test_netA_name)
self.nova_instance_deletion(nova_netA, nova_instance_details_netA)
@@ -1410,11 +1431,11 @@
self.nova_instance_deletion(nova_netB, nova_instance_details_netB)
assert_equal(status, False)
- def test_cordvtn_private_network_instance_ping_form_host_machine_or_compute_node(self):
+ def test_cordvtn_creating_private_network_instance_and_validate_connectivity_from_host_machine_or_compute_node(self):
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork whose ip is under private network
+ 1. Create subnetwork whose ip is under private network
3. Do GET Rest API and validate creation of network
4. Create new nova instance under private network
5. Validate new nova instance is created on nova service
@@ -1438,19 +1459,19 @@
creds = get_nova_credentials()
nova = nova_client.Client('2', **creds)
new_instance_details = self.nova_instance_creation_and_validation(test_net_name,nova,instance_vm_name,image_name,flavor_id)
- #assert_equal(new_instance_details.status, 'ACTIVE')
+ assert_equal(new_instance_details.status, 'ACTIVE')
compute_details = self.get_compute_nodes()
- print(new_instance_details.addresses)
+ print new_instance_details.addresses
status, output = self.nova_instance_tenants_access_check(new_instance_details)
self.neutron_network_deletion(test_net_name)
self.nova_instance_deletion(nova, new_instance_details)
assert_equal(status, False)
- def test_cordvtn_private_network_instance_ping_to_external_network(self):
+ def test_cordvtn_creating_private_network_instance_and_validate_connectivity_to_external_network(self):
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork whose ip is under private network
+ 1. Create subnetwork whose ip is under private network
3. Do GET Rest API and validate creation of network
4. Create new nova instance under private network
5. Validate new nova instance is created on nova service
@@ -1474,19 +1495,19 @@
creds = get_nova_credentials()
nova = nova_client.Client('2', **creds)
new_instance_details = self.nova_instance_creation_and_validation(test_net_name,nova,instance_vm_name,image_name,flavor_id)
- #assert_equal(new_instance_details.status, 'ACTIVE')
+ assert_equal(new_instance_details.status, 'ACTIVE')
compute_details = self.get_compute_nodes()
- print(new_instance_details.addresses)
+ print new_instance_details.addresses
status, output = self.nova_instance_tenants_access_check(new_instance_details, check_type = "Ping_to_external")
self.neutron_network_deletion(test_net_name)
self.nova_instance_deletion(nova, new_instance_details)
assert_equal(status, False)
- def test_cordvtn_private_network_two_instance_ping_between_two_instances(self):
+ def test_cordvtn_creating_private_network_with_two_instances_and_check_connectvity_between_two_instances(self):
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork whose ip is under private network
+ 1. Create subnetwork whose ip is under private network
3. Do GET Rest API and validate creation of network
4. Create first nova instance under private network
5. Validate first nova instance is created on nova service
@@ -1514,23 +1535,23 @@
nova = nova_client.Client('2', **creds)
first_nova_instance_details = self.nova_instance_creation_and_validation(test_net_name,nova,first_instance_vm_name,image_name,flavor_id)
second_nova_instance_details = self.nova_instance_creation_and_validation(test_net_name,nova,second_instance_vm_name,image_name,flavor_id)
- #assert_equal(first_instance_details.status, 'ACTIVE')
- #assert_equal(second_instance_details.status, 'ACTIVE')
+ assert_equal(first_instance_details.status, 'ACTIVE')
+ assert_equal(second_instance_details.status, 'ACTIVE')
compute_details = self.get_compute_nodes()
- print('New nova instance ip addresses are ')
- print(first_nova_instance_details.addresses)
- print(second_nova_instance_details.addresses)
+ print 'New nova instance ip addresses are '
+ print first_nova_instance_details.addresses
+ print second_nova_instance_details.addresses
status, output = self.nova_instance_tenants_access_check(first_nova_instance_details,source_tenants_details = second_nova_instance_details, check_type = "Ping_from_source_tenant")
self.neutron_network_deletion(test_net_name)
self.nova_instance_deletion(nova, first_nova_instance_details)
self.nova_instance_deletion(nova, second_nova_instance_details)
assert_equal(status, True)
- def test_cordvtn_two_private_network_instances_ping_between_two_networks_via_private_network(self):
+ def test_cordvtn_creating_two_private_network_instances_and_validate_connectivity_between_two_networks_via_private_network(self):
"""
Algo:
0. Create Test-Net,
- 1. Cretae subnetwork whose ip is under private network
+ 1. Create subnetwork whose ip is under private network
3. Do GET Rest API and validate creation of network
4. Create new nova instance under private network
5. Validate new nova instance is created on nova service
@@ -1550,12 +1571,12 @@
nova_netA, nova_instance_details_netA = self.create_net_subnet_nova_instance(test_netA_name, test_sub_netA_cidr, netA_instance_vm_details, test_management_type)
nova_netB, nova_instance_details_netB = self.create_net_subnet_nova_instance(test_netB_name, test_sub_netB_cidr, netB_instance_vm_details, test_management_type)
- #assert_equal(nova_instance_details_netA.status, 'ACTIVE')
- #assert_equal(nova_instance_details_netB.status, 'ACTIVE')
+ assert_equal(nova_instance_details_netA.status, 'ACTIVE')
+ assert_equal(nova_instance_details_netB.status, 'ACTIVE')
compute_details = self.get_compute_nodes()
- print('New nova instance ip addresses are ')
- print(nova_instance_details_netA.addresses)
- print(nova_instance_details_netB.addresses)
+ print 'New nova instance ip addresses are '
+ print nova_instance_details_netA.addresses
+ print nova_instance_details_netB.addresses
status, output = self.nova_instance_tenants_access_check(nova_instance_details_netA, source_tenants_details = nova_instance_details_netB,check_type = "Ping_from_source_tenant")
self.neutron_network_deletion(test_netA_name)
self.nova_instance_deletion(nova_netA, nova_instance_details_netA)
@@ -1563,6 +1584,728 @@
self.nova_instance_deletion(nova_netB, nova_instance_details_netB)
assert_equal(status, False)
+ def test_cordvtn_creating_management_and_public_network_instances_and_validate_connectivity_from_host_machine_or_compute_node_and_validate_connectivity_to_internet(self):
+ """
+ Algo:
+ 0. Create Test-Net,
+ 1. Create subnetwork whose ip is under management network
+ 3. Do GET Rest API and validate creation of network
+ 4. Create new nova instance under management network
+ 5. Validate new nova instance is created on nova service
+ 6. Verify ping is getting successful from compute node to nova instance which is created in step 4.
+ """
+ test_second_network_name = ['vtn_test_29_net_management','vtn_test_29_net_public']
+ test_sub_second_network_cidr = [["management","172.27.0.0/24", "172.27.0.20", "172.27.0.21"], ["public","10.6.1.192/26",'10.6.1.193']]
+ test_management_type = ["management_local", 'public']
+ instance_vm_name = 'vtn_test_29_nova_instance_management_net'
+# image_name = "vsg-1.1"
+ image_name = "trusty-server-multi-nic"
+ flavor_id = 'm1.small'
+ for test_net_name in test_second_network_name:
+ result = self.neutron_network_creation_and_validation(test_net_name)
+ assert_equal(result, True)
+ neutron_creds = self.get_neutron_credentials()
+ neutron = neutronclient.Client(**neutron_creds)
+ #for test_net_name,test_sub_net_cidr in test_second_network_name test_sub_2networks_cidr:
+ for i in range(0,2):
+ networks = neutron.list_networks(name=test_second_network_name[i])
+ network_id = self.get_key_value(d=networks, key = 'id')
+ sub_result = self.neutron_subnet_creation_and_validation(test_second_network_name[i],test_sub_2networks_cidr[i])
+ assert_equal(sub_result[0], True)
+ net_type_post = self.sub_network_type_post_to_onos(test_second_network_name[i], test_management_type[i])
+ creds = get_nova_credentials()
+ nova = nova_client.Client('2', **creds)
+ print nova.security_groups.list()
+ new_instance_details = self.nova_instance_creation_and_validation(test_second_network_name,nova,instance_vm_name,image_name,flavor_id)
+ time.sleep(60)
+ assert_equal(new_instance_details.status, 'ACTIVE')
+ compute_details = self.get_compute_nodes()
+ print new_instance_details.addresses
+ address = new_instance_details.addresses
+ print 'Nova instance management ip = %s and public ip %s'%(address[test_second_network_name[0]][0]['addr'],address[test_2networks_name[1]][0]['addr'])
+ print address[test_second_network_name[0]][0]['addr']
+ print nova.security_groups.list()
+ print address[test_second_network_name[1]][0]['addr']
+ print nova.security_groups.list()
+ secgroup = nova.security_groups.find(name="default")
+# nova.security_group_rules.create(secgroup.id,ip_protocol="tcp",
+ #from_port="22",
+ #to_port="22",
+ # cidr="0.0.0.0/0",)
+
+ # nova.security_group_rules.create(secgroup.id,
+ # ip_protocol="icmp",
+ # from_port=-1,
+ # cidr="0.0.0.0/0",
+ # to_port=-1)
+ print nova.security_groups.list()
+
+ status_1, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'],check_type = "Ping_to_external")
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, True)
+ assert_equal(status_2, True)
+
+ def test_cordvtn_creating_management_and_public_network_instance_with_and_without_pause_and_validate_connectivity_from_host_machine_or_compute_node_and_validate_connectivity_to_internet(self):
+ """
+ Algo:
+ 0. Create Test-Net,
+ 1. Create subnetwork whose ip is under management network
+ 3. Do GET Rest API and validate creation of network
+ 4. Create new nova instance under management network
+ 5. Validate new nova instance is created on nova service
+ 6. Verify ping is getting successful from compute node to nova instance which is created in step 4.
+ 7. Now pause the nova instance and check connectivity
+ 8. Now unpause the nova instance and check connectivity
+ """
+ test_second_network_name = ['vtn_test_30_net_management','vtn_test_30_net_public']
+ test_sub_second_network_cidr = [["management","172.27.0.0/24", "172.27.0.20", "172.27.0.21"], ["public","10.6.1.192/26",'10.6.1.193']]
+ test_management_type = ["management_local", 'public']
+ instance_vm_name = 'vtn_test_30_nova_instance_management_net'
+# image_name = "vsg-1.1"
+ image_name = "trusty-server-multi-nic"
+ flavor_id = 'm1.small'
+ for test_net_name in test_second_network_name:
+ result = self.neutron_network_creation_and_validation(test_net_name)
+ assert_equal(result, True)
+ neutron_creds = self.get_neutron_credentials()
+ neutron = neutronclient.Client(**neutron_creds)
+ #for test_net_name,test_sub_net_cidr in test_second_network_name test_sub_2networks_cidr:
+ for i in range(0,2):
+ networks = neutron.list_networks(name=test_second_network_name[i])
+ network_id = self.get_key_value(d=networks, key = 'id')
+ sub_result = self.neutron_subnet_creation_and_validation(test_second_network_name[i],test_sub_2networks_cidr[i])
+ assert_equal(sub_result[0], True)
+ net_type_post = self.sub_network_type_post_to_onos(test_second_network_name[i], test_management_type[i])
+ creds = get_nova_credentials()
+ nova = nova_client.Client('2', **creds)
+ print nova.security_groups.list()
+ new_instance_details = self.nova_instance_creation_and_validation(test_second_network_name,nova,instance_vm_name,image_name,flavor_id)
+ time.sleep(60)
+ assert_equal(new_instance_details.status, 'ACTIVE')
+ compute_details = self.get_compute_nodes()
+ print new_instance_details.addresses
+ address = new_instance_details.addresses
+ print 'Nova instance management ip = %s and public ip %s'%(address[test_second_network_name[0]][0]['addr'],address[test_2networks_name[1]][0]['addr'])
+ print address[test_second_network_name[0]][0]['addr']
+ print nova.security_groups.list()
+ print address[test_second_network_name[1]][0]['addr']
+ print nova.security_groups.list()
+ secgroup = nova.security_groups.find(name="default")
+# nova.security_group_rules.create(secgroup.id,ip_protocol="tcp",
+ #from_port="22",
+ #to_port="22",
+ # cidr="0.0.0.0/0",)
+
+ # nova.security_group_rules.create(secgroup.id,
+ # ip_protocol="icmp",
+ # from_port=-1,
+ # cidr="0.0.0.0/0",
+ # to_port=-1)
+ print nova.security_groups.list()
+
+ status_1, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'],check_type = "Ping_to_external")
+ if status_1 is False or status_2 is False:
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, True)
+ assert_equal(status_2, True)
+ new_instance_details.pause()
+ time.sleep(60)
+ status_1, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'],check_type = "Ping_to_external")
+ if status_1 is True or status_2 is True:
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, False)
+ assert_equal(status_2, False)
+ new_instance_details.unpause()
+ print 'Nova instance is paused and unpasued now checking connectivity'
+ time.sleep(60)
+ status_1, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'],check_type = "Ping_to_external")
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, True)
+ assert_equal(status_2, True)
+
+ def test_cordvtn_creating_management_and_public_network_instance_doing_suspend_and_resume_validating_connectivity_from_host_machine_or_compute_node_and_validate_connectivity_to_internet(self):
+ """
+ Algo:
+ 0. Create Test-Net,
+ 1. Create subnetwork whose ip is under management network
+ 3. Do GET Rest API and validate creation of network
+ 4. Create new nova instance under management network
+ 5. Validate new nova instance is created on nova service
+ 6. Verify ping is getting successful from compute node to nova instance which is created in step 4.
+ 7. Now suspend the nova instance and check connectivity
+ 8. Now resume the nova instance and check connectivity
+ """
+ test_second_network_name = ['vtn_test_31_net_management','vtn_test_31_net_public']
+ test_sub_second_network_cidr = [["management","172.27.0.0/24", "172.27.0.20", "172.27.0.21"], ["public","10.6.1.192/26",'10.6.1.193']]
+ test_management_type = ["management_local", 'public']
+ instance_vm_name = 'vtn_test_31_nova_instance_management_net'
+# image_name = "vsg-1.1"
+ image_name = "trusty-server-multi-nic"
+ flavor_id = 'm1.small'
+ for test_net_name in test_second_network_name:
+ result = self.neutron_network_creation_and_validation(test_net_name)
+ assert_equal(result, True)
+ neutron_creds = self.get_neutron_credentials()
+ neutron = neutronclient.Client(**neutron_creds)
+ #for test_net_name,test_sub_net_cidr in test_second_network_name test_sub_2networks_cidr:
+ for i in range(0,2):
+ networks = neutron.list_networks(name=test_second_network_name[i])
+ network_id = self.get_key_value(d=networks, key = 'id')
+ sub_result = self.neutron_subnet_creation_and_validation(test_second_network_name[i],test_sub_2networks_cidr[i])
+ assert_equal(sub_result[0], True)
+ net_type_post = self.sub_network_type_post_to_onos(test_second_network_name[i], test_management_type[i])
+ creds = get_nova_credentials()
+ nova = nova_client.Client('2', **creds)
+ print nova.security_groups.list()
+ new_instance_details = self.nova_instance_creation_and_validation(test_second_network_name,nova,instance_vm_name,image_name,flavor_id)
+ time.sleep(60)
+ assert_equal(new_instance_details.status, 'ACTIVE')
+ compute_details = self.get_compute_nodes()
+ print new_instance_details.addresses
+ address = new_instance_details.addresses
+ print 'Nova instance management ip = %s and public ip %s'%(address[test_second_network_name[0]][0]['addr'],address[test_2networks_name[1]][0]['addr'])
+ print address[test_second_network_name[0]][0]['addr']
+ print nova.security_groups.list()
+ print address[test_second_network_name[1]][0]['addr']
+ print nova.security_groups.list()
+ secgroup = nova.security_groups.find(name="default")
+# nova.security_group_rules.create(secgroup.id,ip_protocol="tcp",
+ #from_port="22",
+ #to_port="22",
+ # cidr="0.0.0.0/0",)
+
+ # nova.security_group_rules.create(secgroup.id,
+ # ip_protocol="icmp",
+ # from_port=-1,
+ # cidr="0.0.0.0/0",
+ # to_port=-1)
+ print nova.security_groups.list()
+
+ status_1, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'],check_type = "Ping_to_external")
+ if status_1 is False or status_2 is False:
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, True)
+ assert_equal(status_2, True)
+ new_instance_details.suspend()
+ time.sleep(60)
+ status_1, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'],check_type = "Ping_to_external")
+ if status_1 is True or status_2 is True:
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, False)
+ assert_equal(status_2, False)
+ new_instance_details.resume()
+ print 'Nova instance is suspend and rsumed now checking connectivity'
+ time.sleep(60)
+ status_1, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'],check_type = "Ping_to_external")
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, True)
+ assert_equal(status_2, True)
+
+ def test_cordvtn_creating_mgmt_and_public_network_instance_with_stopping_and_starting_instances_and_checking_connectvity_from_host_machine_or_compute_node_and_validate_connectivity_to_internet(self):
+ """
+ Algo:
+ 0. Create Test-Net,
+ 1. Create subnetwork whose ip is under management network
+ 3. Do GET Rest API and validate creation of network
+ 4. Create new nova instance under management network
+ 5. Validate new nova instance is created on nova service
+ 6. Verify ping is getting successful from compute node to nova instance which is created in step 4.
+ 7. Now stop the nova instance and check connectivity
+ 8. Now start the nova instance and check connectivity
+ """
+ test_second_network_name = ['vtn_test_32_net_management','vtn_test_32_net_public']
+ test_sub_second_network_cidr = [["management","172.27.0.0/24", "172.27.0.20", "172.27.0.21"], ["public","10.6.1.192/26",'10.6.1.193']]
+ test_management_type = ["management_local", 'public']
+ instance_vm_name = 'vtn_test_32_nova_instance_management_net'
+# image_name = "vsg-1.1"
+ image_name = "trusty-server-multi-nic"
+ flavor_id = 'm1.small'
+ for test_net_name in test_second_network_name:
+ result = self.neutron_network_creation_and_validation(test_net_name)
+ assert_equal(result, True)
+ neutron_creds = self.get_neutron_credentials()
+ neutron = neutronclient.Client(**neutron_creds)
+ #for test_net_name,test_sub_net_cidr in test_second_network_name test_sub_2networks_cidr:
+ for i in range(0,2):
+ networks = neutron.list_networks(name=test_second_network_name[i])
+ network_id = self.get_key_value(d=networks, key = 'id')
+ sub_result = self.neutron_subnet_creation_and_validation(test_second_network_name[i],test_sub_2networks_cidr[i])
+ assert_equal(sub_result[0], True)
+ net_type_post = self.sub_network_type_post_to_onos(test_second_network_name[i], test_management_type[i])
+ creds = get_nova_credentials()
+ nova = nova_client.Client('2', **creds)
+ print nova.security_groups.list()
+ new_instance_details = self.nova_instance_creation_and_validation(test_second_network_name,nova,instance_vm_name,image_name,flavor_id)
+ time.sleep(60)
+ assert_equal(new_instance_details.status, 'ACTIVE')
+ compute_details = self.get_compute_nodes()
+ print new_instance_details.addresses
+ address = new_instance_details.addresses
+ print 'Nova instance management ip = %s and public ip %s'%(address[test_second_network_name[0]][0]['addr'],address[test_2networks_name[1]][0]['addr'])
+ print address[test_second_network_name[0]][0]['addr']
+ print nova.security_groups.list()
+ print address[test_second_network_name[1]][0]['addr']
+ print nova.security_groups.list()
+ secgroup = nova.security_groups.find(name="default")
+# nova.security_group_rules.create(secgroup.id,ip_protocol="tcp",
+ #from_port="22",
+ #to_port="22",
+ # cidr="0.0.0.0/0",)
+
+ # nova.security_group_rules.create(secgroup.id,
+ # ip_protocol="icmp",
+ # from_port=-1,
+ # cidr="0.0.0.0/0",
+ # to_port=-1)
+ print nova.security_groups.list()
+
+ status_1, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'],check_type = "Ping_to_external")
+ if status_1 is False or status_2 is False:
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, True)
+ assert_equal(status_2, True)
+ new_instance_details.stop()
+ time.sleep(60)
+ status_1, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'],check_type = "Ping_to_external")
+ if status_1 is True or status_2 is True:
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, False)
+ assert_equal(status_2, False)
+ new_instance_details.start()
+ print 'Nova instance is stopped and started now checking connectivity'
+ time.sleep(60)
+ status_1, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'],check_type = "Ping_to_external")
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, True)
+ assert_equal(status_2, True)
+
+
+ def test_cordvtn_creating_mgmt_and_private_network_instance_and_validate_connectivity_from_host_machine_or_compute_node_and_validate_connectivity_to_internet(self):
+ """
+ Algo:
+ 0. Create Test-Net,
+ 1. Create subnetwork whose ip is under management network
+ 3. Do GET Rest API and validate creation of network
+ 4. Create new nova instance under management network
+ 5. Validate new nova instance is created on nova service
+ 6. Verify ping is getting successful from compute node to nova instance which is created in step 4.
+ """
+ test_second_network_name = ['vtn_test_33_net_management','vtn_test_33_net_private']
+ test_sub_second_network_cidr = [["management","172.27.0.0/24", "172.27.0.20", "172.27.0.21"], ["private","10.160.160.192/26",'10.160.160.193']]
+ test_management_type = ["management_local", 'private']
+ instance_vm_name = 'vtn_test_33_nova_instance_management_net'
+# image_name = "vsg-1.1"
+ image_name = "trusty-server-multi-nic"
+ flavor_id = 'm1.small'
+ for test_net_name in test_second_network_name:
+ result = self.neutron_network_creation_and_validation(test_net_name)
+ assert_equal(result, True)
+ neutron_creds = self.get_neutron_credentials()
+ neutron = neutronclient.Client(**neutron_creds)
+ #for test_net_name,test_sub_net_cidr in test_second_network_name test_sub_2networks_cidr:
+ for i in range(0,2):
+ networks = neutron.list_networks(name=test_second_network_name[i])
+ network_id = self.get_key_value(d=networks, key = 'id')
+ sub_result = self.neutron_subnet_creation_and_validation(test_second_network_name[i],test_sub_2networks_cidr[i])
+ assert_equal(sub_result[0], True)
+ net_type_post = self.sub_network_type_post_to_onos(test_second_network_name[i], test_management_type[i])
+ creds = get_nova_credentials()
+ nova = nova_client.Client('2', **creds)
+ print nova.security_groups.list()
+ new_instance_details = self.nova_instance_creation_and_validation(test_second_network_name,nova,instance_vm_name,image_name,flavor_id)
+ time.sleep(60)
+ assert_equal(new_instance_details.status, 'ACTIVE')
+ compute_details = self.get_compute_nodes()
+ print new_instance_details.addresses
+ address = new_instance_details.addresses
+ print 'Nova instance management ip = %s and private ip %s'%(address[test_second_network_name[0]][0]['addr'],address[test_2networks_name[1]][0]['addr'])
+ print address[test_second_network_name[0]][0]['addr']
+ print nova.security_groups.list()
+ print address[test_second_network_name[1]][0]['addr']
+ print nova.security_groups.list()
+ secgroup = nova.security_groups.find(name="default")
+# nova.security_group_rules.create(secgroup.id,ip_protocol="tcp",
+ #from_port="22",
+ #to_port="22",
+ # cidr="0.0.0.0/0",)
+
+ # nova.security_group_rules.create(secgroup.id,
+ # ip_protocol="icmp",
+ # from_port=-1,
+ # cidr="0.0.0.0/0",
+ # to_port=-1)
+ print nova.security_groups.list()
+ status_1, output = self.nova_instance_tenants_access_check(address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(address[test_second_network_name[1]][0]['addr'],check_type = "Ping_to_external")
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, True)
+ assert_equal(status_2, False)
+
+ def test_cordvtn_creating_mgmt_and_private_network_with_two_instances_and_validate_connectivity_from_host_machine_or_compute_node_and_check_connectivity_to_other_instance(self):
+ """
+ Algo:
+ 0. Create Test-Net,
+ 1. Create subnetwork whose ip is under management network
+ 3. Do GET Rest API and validate creation of network
+ 4. Create new nova instance under management network
+ 5. Validate new nova instance is created on nova service
+ 6. Verify ping is getting successful from compute node to nova instance which is created in step 4.
+ """
+ test_second_network_name = ['vtn_test_34_net_management','vtn_test_34_net_private']
+ test_sub_second_network_cidr = [["management","172.27.0.0/24", "172.27.0.20", "172.27.0.21"], ["private","10.160.160.192/26",'10.160.160.193']]
+ test_management_type = ["management_local", 'private']
+ first_instance_vm_name = 'vtn_test_34_nova_first_instance_management_net'
+ second_instance_vm_name = 'vtn_test_34_nova_second_instance_management_net'
+# image_name = "vsg-1.1"
+ image_name = "trusty-server-multi-nic"
+ flavor_id = 'm1.small'
+ for test_net_name in test_second_network_name:
+ result = self.neutron_network_creation_and_validation(test_net_name)
+ assert_equal(result, True)
+ neutron_creds = self.get_neutron_credentials()
+ neutron = neutronclient.Client(**neutron_creds)
+ #for test_net_name,test_sub_net_cidr in test_second_network_name test_sub_2networks_cidr:
+ for i in range(0,2):
+ networks = neutron.list_networks(name=test_second_network_name[i])
+ network_id = self.get_key_value(d=networks, key = 'id')
+ sub_result = self.neutron_subnet_creation_and_validation(test_second_network_name[i],test_sub_2networks_cidr[i])
+ assert_equal(sub_result[0], True)
+ net_type_post = self.sub_network_type_post_to_onos(test_second_network_name[i], test_management_type[i])
+ creds = get_nova_credentials()
+ nova = nova_client.Client('2', **creds)
+ print nova.security_groups.list()
+ new_first_instance_details = self.nova_instance_creation_and_validation(test_second_network_name,nova,first_instance_vm_name,image_name,flavor_id)
+ new_second_instance_details = self.nova_instance_creation_and_validation(test_second_network_name,nova,second_instance_vm_name,image_name,flavor_id)
+ time.sleep(60)
+ assert_equal(new_first_instance_details.status, 'ACTIVE')
+ assert_equal(new_second_instance_details.status, 'ACTIVE')
+ compute_details = self.get_compute_nodes()
+ first_instance_address = new_first_instance_details.addresses
+ second_instance_address = new_second_instance_details.addresses
+ print 'Nova first instance management ip = %s and private ip %s'%(first_instance_address[test_second_network_name[0]][0]['addr'],first_instance_address[test_2networks_name[1]][0]['addr'])
+ print 'Nova second instance management ip = %s and private ip %s'%(second_instance_address[test_second_network_name[0]][0]['addr'],second_instance_address[test_2networks_name[1]][0]['addr'])
+ secgroup = nova.security_groups.find(name="default")
+# nova.security_group_rules.create(secgroup.id,ip_protocol="tcp",
+ #from_port="22",
+ #to_port="22",
+ # cidr="0.0.0.0/0",)
+
+ # nova.security_group_rules.create(secgroup.id,
+ # ip_protocol="icmp",
+ # from_port=-1,
+ # cidr="0.0.0.0/0",
+ # to_port=-1)
+ print nova.security_groups.list()
+
+ status_1, output = self.nova_instance_tenants_access_check(first_instance_address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(first_instance_address[test_second_network_name[0]][0]['addr'],source_tenants_details = second_instance_address[test_2networks_name[1]][0]['addr'],check_type = "Ping_from_source_tenant")
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, True)
+ assert_equal(status_2, True)
+
+ def test_cordvtn_creating_mgmt_and_private_network_with_two_instances_with_and_without_pause_validating_connectivity_from_host_machine_or_compute_node_and_validating_connectivity_to_other_instance(self):
+ """
+ Algo:
+ 0. Create Test-Net,
+ 1. Create subnetwork whose ip is under management network
+ 3. Do GET Rest API and validate creation of network
+ 4. Create new nova instance under management network
+ 5. Validate new nova instance is created on nova service
+ 6. Verify ping is getting successful from compute node to nova instance which is created in step 4.
+ 7. Now pause one of the nova instance and check connectivity
+ 8. Now start the same nova instance and check connectivity
+ """
+ test_second_network_name = ['vtn_test_35_net_management','vtn_test_35_net_private']
+ test_sub_second_network_cidr = [["management","172.27.0.0/24", "172.27.0.20", "172.27.0.21"], ["private","10.160.160.192/26",'10.160.160.193']]
+ test_management_type = ["management_local", 'private']
+ first_instance_vm_name = 'vtn_test_35_nova_first_instance_management_net'
+ second_instance_vm_name = 'vtn_test_35_nova_second_instance_management_net'
+# image_name = "vsg-1.1"
+ image_name = "trusty-server-multi-nic"
+ flavor_id = 'm1.small'
+ for test_net_name in test_second_network_name:
+ result = self.neutron_network_creation_and_validation(test_net_name)
+ assert_equal(result, True)
+ neutron_creds = self.get_neutron_credentials()
+ neutron = neutronclient.Client(**neutron_creds)
+ #for test_net_name,test_sub_net_cidr in test_second_network_name test_sub_2networks_cidr:
+ for i in range(0,2):
+ networks = neutron.list_networks(name=test_second_network_name[i])
+ network_id = self.get_key_value(d=networks, key = 'id')
+ sub_result = self.neutron_subnet_creation_and_validation(test_second_network_name[i],test_sub_2networks_cidr[i])
+ assert_equal(sub_result[0], True)
+ net_type_post = self.sub_network_type_post_to_onos(test_second_network_name[i], test_management_type[i])
+ creds = get_nova_credentials()
+ nova = nova_client.Client('2', **creds)
+ print nova.security_groups.list()
+ new_first_instance_details = self.nova_instance_creation_and_validation(test_second_network_name,nova,first_instance_vm_name,image_name,flavor_id)
+ new_second_instance_details = self.nova_instance_creation_and_validation(test_second_network_name,nova,second_instance_vm_name,image_name,flavor_id)
+ time.sleep(60)
+ assert_equal(new_first_instance_details.status, 'ACTIVE')
+ assert_equal(new_second_instance_details.status, 'ACTIVE')
+ compute_details = self.get_compute_nodes()
+ first_instance_address = new_first_instance_details.addresses
+ second_instance_address = new_second_instance_details.addresses
+ print 'Nova first instance management ip = %s and private ip %s'%(first_instance_address[test_second_network_name[0]][0]['addr'],first_instance_address[test_2networks_name[1]][0]['addr'])
+ print 'Nova second instance management ip = %s and private ip %s'%(second_instance_address[test_second_network_name[0]][0]['addr'],second_instance_address[test_2networks_name[1]][0]['addr'])
+ secgroup = nova.security_groups.find(name="default")
+# nova.security_group_rules.create(secgroup.id,ip_protocol="tcp",
+ #from_port="22",
+ #to_port="22",
+ # cidr="0.0.0.0/0",)
+
+ # nova.security_group_rules.create(secgroup.id,
+ # ip_protocol="icmp",
+ # from_port=-1,
+ # cidr="0.0.0.0/0",
+ # to_port=-1)
+ print nova.security_groups.list()
+ status_1, output = self.nova_instance_tenants_access_check(first_instance_address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(first_instance_address[test_second_network_name[0]][0]['addr'],source_tenants_details = second_instance_address[test_2networks_name[1]][0]['addr'],check_type = "Ping_from_source_tenant")
+ if status_1 is False or status_2 is False:
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, True)
+ assert_equal(status_2, True)
+ new_first_instance_details.pause()
+ time.sleep(60)
+ status_1, output = self.nova_instance_tenants_access_check(first_instance_address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(first_instance_address[test_second_network_name[0]][0]['addr'],source_tenants_details = second_instance_address[test_2networks_name[1]][0]['addr'],check_type = "Ping_from_source_tenant")
+ if status_1 is True or status_2 is True:
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, False)
+ assert_equal(status_2, False)
+ new_first_instance_details.unpause()
+ print 'Nova instance is paused and unpased now checking connectivity'
+ time.sleep(60)
+ status_1, output = self.nova_instance_tenants_access_check(first_instance_address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(first_instance_address[test_second_network_name[0]][0]['addr'],source_tenants_details = second_instance_address[test_2networks_name[1]][0]['addr'],check_type = "Ping_from_source_tenant")
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, True)
+ assert_equal(status_2, True)
+
+ def test_cordvtn_creating_mgmt_and_private_network_with_two_instances_and_doing_suspend_and_resume_validating_connectivity_from_host_machine_or_compute_node_and_validating_connectivity_to_other_instance(self):
+ """
+ Algo:
+ 0. Create Test-Net,
+ 1. Create subnetwork whose ip is under management network
+ 3. Do GET Rest API and validate creation of network
+ 4. Create new nova instance under management network
+ 5. Validate new nova instance is created on nova service
+ 6. Verify ping is getting successful from compute node to nova instance which is created in step 4.
+ 7. Now suspend one of the nova instance and check connectivity
+ 8. Now resume the same nova instance and check connectivity
+ """
+ test_second_network_name = ['vtn_test_36_net_management','vtn_test_36_net_private']
+ test_sub_second_network_cidr = [["management","172.27.0.0/24", "172.27.0.20", "172.27.0.21"], ["private","10.160.160.192/26",'10.160.160.193']]
+ test_management_type = ["management_local", 'private']
+ first_instance_vm_name = 'vtn_test_36_nova_first_instance_management_net'
+ second_instance_vm_name = 'vtn_test_36_nova_second_instance_management_net'
+# image_name = "vsg-1.1"
+ image_name = "trusty-server-multi-nic"
+ flavor_id = 'm1.small'
+ for test_net_name in test_second_network_name:
+ result = self.neutron_network_creation_and_validation(test_net_name)
+ assert_equal(result, True)
+ neutron_creds = self.get_neutron_credentials()
+ neutron = neutronclient.Client(**neutron_creds)
+ #for test_net_name,test_sub_net_cidr in test_second_network_name test_sub_2networks_cidr:
+ for i in range(0,2):
+ networks = neutron.list_networks(name=test_second_network_name[i])
+ network_id = self.get_key_value(d=networks, key = 'id')
+ sub_result = self.neutron_subnet_creation_and_validation(test_second_network_name[i],test_sub_2networks_cidr[i])
+ assert_equal(sub_result[0], True)
+ net_type_post = self.sub_network_type_post_to_onos(test_second_network_name[i], test_management_type[i])
+ creds = get_nova_credentials()
+ nova = nova_client.Client('2', **creds)
+ print nova.security_groups.list()
+ new_first_instance_details = self.nova_instance_creation_and_validation(test_second_network_name,nova,first_instance_vm_name,image_name,flavor_id)
+ new_second_instance_details = self.nova_instance_creation_and_validation(test_second_network_name,nova,second_instance_vm_name,image_name,flavor_id)
+ time.sleep(60)
+ assert_equal(new_first_instance_details.status, 'ACTIVE')
+ assert_equal(new_second_instance_details.status, 'ACTIVE')
+ compute_details = self.get_compute_nodes()
+ first_instance_address = new_first_instance_details.addresses
+ second_instance_address = new_second_instance_details.addresses
+ print 'Nova first instance management ip = %s and private ip %s'%(first_instance_address[test_second_network_name[0]][0]['addr'],first_instance_address[test_2networks_name[1]][0]['addr'])
+ print 'Nova second instance management ip = %s and private ip %s'%(second_instance_address[test_second_network_name[0]][0]['addr'],second_instance_address[test_2networks_name[1]][0]['addr'])
+ secgroup = nova.security_groups.find(name="default")
+# nova.security_group_rules.create(secgroup.id,ip_protocol="tcp",
+ #from_port="22",
+ #to_port="22",
+ # cidr="0.0.0.0/0",)
+
+ # nova.security_group_rules.create(secgroup.id,
+ # ip_protocol="icmp",
+ # from_port=-1,
+ # cidr="0.0.0.0/0",
+ # to_port=-1)
+ print nova.security_groups.list()
+ status_1, output = self.nova_instance_tenants_access_check(first_instance_address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(first_instance_address[test_second_network_name[0]][0]['addr'],source_tenants_details = second_instance_address[test_2networks_name[1]][0]['addr'],check_type = "Ping_from_source_tenant")
+ if status_1 is False or status_2 is False:
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, True)
+ assert_equal(status_2, True)
+ new_first_instance_details.suspend()
+ time.sleep(60)
+ status_1, output = self.nova_instance_tenants_access_check(first_instance_address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(first_instance_address[test_second_network_name[0]][0]['addr'],source_tenants_details = second_instance_address[test_2networks_name[1]][0]['addr'],check_type = "Ping_from_source_tenant")
+ if status_1 is True or status_2 is True:
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, False)
+ assert_equal(status_2, False)
+ new_first_instance_details.resume()
+ print 'Nova instance is suspend and resume now checking connectivity'
+ time.sleep(60)
+ status_1, output = self.nova_instance_tenants_access_check(first_instance_address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(first_instance_address[test_second_network_name[0]][0]['addr'],source_tenants_details = second_instance_address[test_2networks_name[1]][0]['addr'],check_type = "Ping_from_source_tenant")
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, True)
+ assert_equal(status_2, True)
+
+ def test_cordvtn_creating_mgmt_and_private_network_with_two_instances_applying_stop_and_start_validating_connectivity_from_host_machine_or_compute_node_and_validating_connectivity_to_other_instance(self):
+ """
+ Algo:
+ 0. Create Test-Net,
+ 1. Create subnetwork whose ip is under management network
+ 3. Do GET Rest API and validate creation of network
+ 4. Create new nova instance under management network
+ 5. Validate new nova instance is created on nova service
+ 6. Verify ping is getting successful from compute node to nova instance which is created in step 4.
+ 7. Now stop one of the nova instance and check connectivity
+ 8. Now start the same nova instance and check connectivity
+ """
+ test_second_network_name = ['vtn_test_37_net_management','vtn_test_37_net_private']
+ test_sub_second_network_cidr = [["management","172.27.0.0/24", "172.27.0.20", "172.27.0.21"], ["private","10.160.160.192/26",'10.160.160.193']]
+ test_management_type = ["management_local", 'private']
+ first_instance_vm_name = 'vtn_test_37_nova_first_instance_management_net'
+ second_instance_vm_name = 'vtn_test_37_nova_second_instance_management_net'
+# image_name = "vsg-1.1"
+ image_name = "trusty-server-multi-nic"
+ flavor_id = 'm1.small'
+ for test_net_name in test_second_network_name:
+ result = self.neutron_network_creation_and_validation(test_net_name)
+ assert_equal(result, True)
+ neutron_creds = self.get_neutron_credentials()
+ neutron = neutronclient.Client(**neutron_creds)
+ #for test_net_name,test_sub_net_cidr in test_second_network_name test_sub_2networks_cidr:
+ for i in range(0,2):
+ networks = neutron.list_networks(name=test_second_network_name[i])
+ network_id = self.get_key_value(d=networks, key = 'id')
+ sub_result = self.neutron_subnet_creation_and_validation(test_second_network_name[i],test_sub_2networks_cidr[i])
+ assert_equal(sub_result[0], True)
+ net_type_post = self.sub_network_type_post_to_onos(test_second_network_name[i], test_management_type[i])
+ creds = get_nova_credentials()
+ nova = nova_client.Client('2', **creds)
+ print nova.security_groups.list()
+ new_first_instance_details = self.nova_instance_creation_and_validation(test_second_network_name,nova,first_instance_vm_name,image_name,flavor_id)
+ new_second_instance_details = self.nova_instance_creation_and_validation(test_second_network_name,nova,second_instance_vm_name,image_name,flavor_id)
+ time.sleep(60)
+ assert_equal(new_first_instance_details.status, 'ACTIVE')
+ assert_equal(new_second_instance_details.status, 'ACTIVE')
+ compute_details = self.get_compute_nodes()
+ first_instance_address = new_first_instance_details.addresses
+ second_instance_address = new_second_instance_details.addresses
+ print 'Nova first instance management ip = %s and private ip %s'%(first_instance_address[test_second_network_name[0]][0]['addr'],first_instance_address[test_2networks_name[1]][0]['addr'])
+ print 'Nova second instance management ip = %s and private ip %s'%(second_instance_address[test_second_network_name[0]][0]['addr'],second_instance_address[test_2networks_name[1]][0]['addr'])
+ secgroup = nova.security_groups.find(name="default")
+ print nova.security_groups.list()
+ status_1, output = self.nova_instance_tenants_access_check(first_instance_address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(first_instance_address[test_second_network_name[0]][0]['addr'],source_tenants_details = second_instance_address[test_2networks_name[1]][0]['addr'],check_type = "Ping_from_source_tenant")
+ if status_1 is False or status_2 is False:
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, True)
+ assert_equal(status_2, True)
+ new_first_instance_details.stop()
+ time.sleep(60)
+ status_1, output = self.nova_instance_tenants_access_check(first_instance_address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(first_instance_address[test_second_network_name[0]][0]['addr'],source_tenants_details = second_instance_address[test_2networks_name[1]][0]['addr'],check_type = "Ping_from_source_tenant")
+ if status_1 is True or status_2 is True:
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, False)
+ assert_equal(status_2, False)
+ new_first_instance_details.start()
+ print 'Nova instance is stopped and started now checking connectivity'
+ time.sleep(60)
+ status_1, output = self.nova_instance_tenants_access_check(first_instance_address[test_second_network_name[0]][0]['addr'])
+ status_2, output = self.nova_instance_tenants_access_check(first_instance_address[test_second_network_name[0]][0]['addr'],source_tenants_details = second_instance_address[test_2networks_name[1]][0]['addr'],check_type = "Ping_from_source_tenant")
+ self.nova_instance_deletion(nova, new_instance_details)
+ time.sleep(3)
+ self.neutron_network_deletion(test_second_network_name[0])
+ self.neutron_network_deletion(test_second_network_name[1])
+ assert_equal(status_1, True)
+ assert_equal(status_2, True)
+
def test_cordvtn_with_neutron_network_creation_and_validation_on_head_node_with_neutron_service(self):
"""
Algo:
@@ -1674,193 +2417,6 @@
result = self.search_value(data, device_id)
assert_equal(result, True)
- def test_cordvtn_basic_tenant(self):
-
- tenant_1= create_tenant("CORD_Subscriber_Test_Tenant_1")
- if tenant1 != 0:
- print("Creation of CORD Subscriber Test Tenant 1")
-
- tenant_2 = create_tenant("CORD_Subscriber_Test_Tenant_2")
- if tenant2 != 0:
- print("Creation of CORD Subscriber Test Tenant 2")
-
- create_net(tenant_1,"a1")
- create_subnet(tenant_1,"a1","as1","10.0.1.0/24")
-
- create_net(tenant_2,"a2")
- create_subnet(tenant_2,"a2","as1","10.0.2.0/24")
-
- netid_1 = get_id(tenant_1,"net","a1")
- netid_2 = get_id(tenant_2,"net","a2")
-
- nova_boot(tenant_1,"vm1",netid=netid)
- nova_boot(tenant_2,"vm1",netid=netid)
-
- nova_wait_boot(tenant_1,"vm1", "ACTIVE")
- nova_wait_boot(tenant_2,"vm1", "ACTIVE")
-
- router_create(tenant_1,"r1")
- router_interface_add(tenant_1,"r1","as1")
- router_create(tenant_2,"r1")
- router_interface_add(tenant_2,"r1","as1")
-
- create_net(tenant_1,"x1","","--router:external=True")
- create_net(tenant_2,"x1","","--router:external=True")
-
- router_gateway_set(tenant_1,"r1","x1")
- router_gateway_set(tenant_2,"r1","x1")
-
- subnetid_1 = get_id(tenant_1,"subnet","as1")
- subnetid_2 = get_id(tenant_2,"subnet","as1")
- port_create(tenant_1,"p1","a1","10.0.1.100",subnetid_1)
- port_create(tenant_2,"p1","a1","10.0.1.100",subnetid_2)
-
- port_id_1 = get_id(tenant_1,"port","p1")
- port_id_2 = get_id(tenant_2,"port","p1")
- status = validate_vtn_flows()
- assert_equal(status, True)
-
- def test_cordvtn_for_creation_of_network(self):
-
- ret1 = create_tenant(netA)
- if ret1 != 0:
- print("Creation of Tenant netA Failed")
-
- ret2 = create_tenant(netB)
- if ret2 != 0:
- print("Creation of Tenant netB Failed")
- network = {'name': self.network_name, 'admin_state_up': True}
- self.neutron.create_network({'network':network})
- log_test.info("Created network:{0}".format(self.network_name))
- status = validate_vtn_flows()
- assert_equal(status, True)
-
- def test_cordvtn_to_create_net_work_with_subnet(self):
- ret1 = create_tenant(netA)
- if ret1 != 0:
- print("Creation of Tenant netA Failed")
-
- ret2 = create_tenant(netB)
- if ret2 != 0:
- print("Creation of Tenant netB Failed")
- network_name = self.network_name
- network = {'name': network_name, 'admin_state_up': True}
- network_info = self.neutron.create_network({'network':network})
- network_id = network_info['network']['id']
-
- log_test.info("Created network:{0}".format(network_id))
- self.network_ids.append(network_id)
- subnet_count = 1
- for cidr in self.subnet_cidrs:
- gateway_ip = str(list(cidr)[1])
- subnet = {"network_id": network_id, "ip_version":4,
- "cidr":str(cidr), "enable_dhcp":True,
- "host_routes":[{"destination":"0.0.0.0/0", "nexthop":gateway_ip}]
- }
- subnet = {"name":"subnet-"+str(subnet_count), "network_id": network_id, "ip_version":4, "cidr":str(cidr), "enable_dhcp":True}
- print(subnet)
- self.neutron.create_subnet({'subnet':subnet})
- log_test.info("Created subnet:{0}".format(str(cidr)))
- if not self.number_of_subnet - 1:
- break
- self.number_of_subnet -= 1
- subnet_count += 1
- status = validate_vtn_flows()
- assert_equal(status, True)
-
- def test_cordvtn_subnet_limit(self):
- ret1 = create_tenant(netA)
- if ret1 != 0:
- print("Creation of Tenant netA Failed")
-
- ret2 = create_tenant(netB)
- if ret2 != 0:
- print("Creation of Tenant netB Failed")
- network_name = uuid.uuid4().get_hex()
- network = {'name': network_name, 'admin_state_up': True}
- network_info = self.neutron.create_network({'network':network})
- log_test.info("Created network:{0}".format(network_name))
- network_id = network_info['network']['id']
- self.network_ids.append(network_id)
- subnet_cidrs = ['11.2.2.0/29', '11.2.2.8/29']
- for cidr in subnet_cidrs:
- subnet = {"network_id": network_id, "ip_version":4, "cidr": cidr}
- subnet_info = self.neutron.create_subnet({'subnet':subnet})
- subnet_id = subnet_info['subnet']['id']
- log_test.info("Created subnet:{0}".format(cidr))
- while True:
- port = {"network_id": network_id, "admin_state_up": True}
- port_info = self.neutron.create_port({'port':port})
- port_id = port_info['port']['id']
- self.port_ids.append(port_id)
- log_test.info("Created Port:{0}".format(port_info['port']['id']))
- if not self.quota_limit:
- break
- self.quota_limit -= 1
- status = validate_vtn_flows()
- assert_equal(status, True)
-
- def test_cordvtn_floatingip_limit(self):
- ret1 = create_tenant(netA)
- if ret1 != 0:
- print("Creation of Tenant netA Failed")
-
- ret2 = create_tenant(netB)
- if ret2 != 0:
- print("Creation of Tenant netB Failed")
- while True:
- floatingip = {"floating_network_id": self.floating_nw_id}
- fip_info = self.neutron.create_floatingip({'floatingip':floatingip})
- fip_id = fip_info['floatingip']['id']
- log_test.info("Created Floating IP:{0}".format(fip_id))
- self.fip_ids.append(fip_id)
- if not self.quota_limit:
- break
- self.quota_limit -= 1
- status = validate_vtn_flows()
- assert_equal(status, True)
-
- def test_cordvtn_for_10_neutron_networks(self):
- ret1 = create_tenant(netA)
- if ret1 != 0:
- print("Creation of Tenant netA Failed")
-
- ret2 = create_tenant(netB)
- if ret2 != 0:
- print("Creation of Tenant netB Failed")
- pool = Pool(processes=10)
- ret = os.system("neutron quote-update --network 15")
- if ret1 != 0:
- print("Neutron network install failed")
- for i in range(1, 11):
- pool.apply_async(create_network, (i, ))
-
- pool.close()
- pool.join()
- status = validate_vtn_flows()
- assert_equal(status, True)
-
- def test_cordvtn_for_100_neutron_networks(self):
- ret1 = create_tenant(netA)
- if ret1 != 0:
- print("Creation of Tenant netA Failed")
-
- ret2 = create_tenant(netB)
- if ret2 != 0:
- print("Creation of Tenant netB Failed")
- pool = Pool(processes=10)
-
- ret = os.system("neutron quote-update --network 105")
- if ret1 != 0:
- print("Neutron network install failed")
- for i in range(1, 101):
- pool.apply_async(create_network, (i, ))
-
- pool.close()
- pool.join()
- status = validate_vtn_flows()
- assert_equal(status, True)
-
def test_cordvtn_creating_virtual_private_network(self):
"""
Algo: