| # |
| # Copyright 2016-present Ciena Corporation |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| # |
| import unittest |
| import os,sys |
| import os |
| import keystoneclient.v2_0.client as ksclient |
| import keystoneclient.apiclient.exceptions |
| import neutronclient.v2_0.client as nclient |
| import neutronclient.common.exceptions |
| import novaclient.v1_1.client as novaclient |
| from multiprocessing import Pool |
| from nose.tools import assert_equal |
| from CordLogger import CordLogger |
| |
| class cordvtn_exchange(CordLogger): |
| |
| app = 'org.opencord.cordvtn' |
| |
| @classmethod |
| def setUpClass(cls): |
| cls.olt = OltConfig() |
| cls.port_map, _ = cls.olt.olt_port_map() |
| if not cls.port_map: |
| cls.port_map = g_subscriber_port_map |
| cls.iface = cls.port_map[1] |
| |
| def setUp(self): |
| ''' Activate the cord vtn app''' |
| super(dhcp_exchange, self).setUp() |
| self.maxDiff = None ##for assert_equal compare outputs on failure |
| self.onos_ctrl = OnosCtrl(self.app) |
| status, _ = self.onos_ctrl.activate() |
| assert_equal(status, True) |
| time.sleep(3) |
| |
| def tearDown(self): |
| '''Deactivate the cord vtn app''' |
| self.onos_ctrl.deactivate() |
| super(dhcp_exchange, self).tearDown() |
| |
| def onos_load_config(self, config): |
| status, code = OnosCtrl.config(config) |
| if status is False: |
| log.info('JSON request returned status %d' %code) |
| assert_equal(status, True) |
| time.sleep(3) |
| |
| def create_tenant(tenant_name): |
| new_tenant = keystone.tenants.create(tenant_name=tenant_name, |
| description="CORD Tenant \ |
| created", |
| enabled=True) |
| tenant_id = new_tenant.id |
| tenant_status = True |
| user_data = [] |
| for j in range(2): |
| j += 1 |
| user_name = tenant_name + '-user-' + str(j) |
| user_data.append(create_user(user_name, tenant_id)) |
| |
| print " Tenant and User Created" |
| |
| tenant_data = {'tenant_name': tenant_name, |
| 'tenant_id': tenant_id, |
| 'status': tenant_status} |
| return tenant_data |
| |
| def create_user(user_name, tenant_id): |
| new_user = keystone.users.create(name=user_name, |
| password="ubuntu", |
| tenant_id=tenant_id) |
| print(' - Created User %s' % user_name) |
| keystone.roles.add_user_role(new_user, member_role, tenant_id) |
| if assign_admin: |
| admin_user = keystone.users.find(name='admin') |
| admin_role = keystone.roles.find(name='admin') |
| keystone.roles.add_user_role(admin_user, admin_role, tenant_id) |
| user_data = {'name': new_user.name, |
| 'id': new_user.id} |
| return user_data |
| |
| def delete_tenant(tenant_name): |
| tenant = keystone.tenants.find(name=tenant_name) |
| for j in range(2): |
| j += 1 |
| user_name = tenant_name + '-user-' + str(j) |
| delete_user(user_name, tenant.id) |
| tenant.delete() |
| print(' - Deleted Tenant %s ' % tenant_name) |
| return True |
| |
| def delete_user(user_name, tenant_id): |
| user = keystone.users.find(name=user_name) |
| user.delete() |
| |
| print(' - Deleted User %s' % user_name) |
| return True |
| |
| def get_neutron_credentials(): |
| d = {} |
| d['username'] = os.environ['OS_USERNAME'] |
| d['password'] = os.environ['OS_PASSWORD'] |
| d['auth_url'] = os.environ['OS_AUTH_URL'] |
| d['tenant_name'] = os.environ['OS_TENANT_NAME'] |
| return d |
| |
| |
| def create_network(i): |
| neutron_credentials = get_neutron_credentials() |
| neutron = neutron_client.Client(**neutron_credentials) |
| json = {'network': {'name': 'network-' + str(i), |
| 'admin_state_up': True}} |
| while True: |
| neutron.create_network(body=json) |
| print '\nnetwork-' + str(i) + ' created' |
| break |
| |
| pool = Pool(processes=5) |
| os.system("neutron quota-update --network 105") |
| for i in range(1,5): |
| pool.apply_async(create_network, (i, )) |
| pool.close() |
| pool.join() |
| |
| def test_cordvtn_basic_tenant(self): |
| pass |
| |
| def test_cordvtn_mgmt_network(self): |
| pass |
| |
| def test_cordvtn_data_network(self): |
| pass |
| |
| def test_cordvtn_public_network(self): |
| pass |
| |
| def test_cordvtn_in_same_network(self): |
| pass |
| |
| def test_cordvtn_local_mgmt_network(self): |
| pass |
| |
| def test_cordvtn_service_dependency(self): |
| pass |
| |
| def test_cordvtn_service_dependency_with_xos(self): |
| pass |
| |
| def test_cordvtn_vsg_xos_service_profile(self): |
| pass |
| |
| def test_cordvtn_access_agent(self): |
| pass |
| |
| def test_cordvtn_network_creation(self): |
| pass |
| |
| def test_cordvtn_removing_service_network(self): |
| pass |
| |
| def test_cordvtn_web_application(self): |
| pass |
| |
| def test_cordvtn_service_port(self): |
| pass |