openstack
/
congress-tempest-plugin
Code Issues Proposed changes

Merge "Replace the usage of 'admin_manager' with 'os_admin'"

This commit is contained in:
Jenkins 2017-07-11 20:17:36 +00:00 committed by Gerrit Code Review
parent 4d21599164 eea4104993
commit dd3cf4017e
13 changed files with 108 additions and 103 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
tests/scenario/congress_datasources/_test_keystonev2.py
View File

@ -38,18 +38,19 @@ class TestKeystoneV2Driver(manager_congress.ScenarioPolicyBase):
def setUp(cls):
super(TestKeystoneV2Driver, cls).setUp()
cls.os = clients.Manager(cls.admin_manager.auth_provider.credentials)
cls.keystone = cls.os.identity_client
cls.tenants_client = cls.os.tenants_client
cls.roles_client = cls.os.roles_client
cls.users_client = cls.os.users_client
cls.os_primary = clients.Manager(
cls.os_admin.auth_provider.credentials)
cls.keystone = cls.os_primary.identity_client
cls.tenants_client = cls.os_primary.tenants_client
cls.roles_client = cls.os_primary.roles_client
cls.users_client = cls.os_primary.users_client
cls.datasource_id = manager_congress.get_datasource_id(
cls.admin_manager.congress_client, 'keystone')
cls.os_admin.congress_client, 'keystone')
@decorators.attr(type='smoke')
def test_keystone_users_table(self):
user_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'users')['columns'])
user_id_col = next(i for i, c in enumerate(user_schema)
if c['name'] == 'id')
@ -63,7 +64,7 @@ class TestKeystoneV2Driver(manager_congress.ScenarioPolicyBase):
user_map[user['id']] = user
results = (
self.admin_manager.congress_client.list_datasource_rows(
self.os_admin.congress_client.list_datasource_rows(
self.datasource_id, 'users'))
for row in results['results']:
try:
@ -92,7 +93,7 @@ class TestKeystoneV2Driver(manager_congress.ScenarioPolicyBase):
@decorators.attr(type='smoke')
def test_keystone_roles_table(self):
role_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'roles')['columns'])
role_id_col = next(i for i, c in enumerate(role_schema)
if c['name'] == 'id')
@ -106,7 +107,7 @@ class TestKeystoneV2Driver(manager_congress.ScenarioPolicyBase):
roles_map[role['id']] = role
results = (
self.admin_manager.congress_client.list_datasource_rows(
self.os_admin.congress_client.list_datasource_rows(
self.datasource_id, 'roles'))
for row in results['results']:
try:
@ -128,7 +129,7 @@ class TestKeystoneV2Driver(manager_congress.ScenarioPolicyBase):
@decorators.attr(type='smoke')
def test_keystone_tenants_table(self):
tenant_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'tenants')['columns'])
tenant_id_col = next(i for i, c in enumerate(tenant_schema)
if c['name'] == 'id')
@ -142,7 +143,7 @@ class TestKeystoneV2Driver(manager_congress.ScenarioPolicyBase):
tenants_map[tenant['id']] = tenant
results = (
self.admin_manager.congress_client.list_datasource_rows(
self.os_admin.congress_client.list_datasource_rows(
self.datasource_id, 'tenants'))
for row in results['results']:
try:

10
tests/scenario/congress_datasources/test_aodh.py
View File

@ -36,9 +36,9 @@ class TestAodhDriver(manager_congress.ScenarioPolicyBase):
def setUp(cls):
super(TestAodhDriver, cls).setUp()
cls.alarms_client = cls.admin_manager.alarms_client
cls.alarms_client = cls.os_admin.alarms_client
cls.datasource_id = manager_congress.get_datasource_id(
cls.admin_manager.congress_client, 'aodh')
cls.os_admin.congress_client, 'aodh')
@decorators.attr(type='smoke')
def test_aodh_alarms_table(self):
@ -52,7 +52,7 @@ class TestAodhDriver(manager_congress.ScenarioPolicyBase):
enabled=False,
threshold_rule=rule)
alarms_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'alarms')['columns'])
alarms_id_col = next(i for i, c in enumerate(alarms_schema)
if c['name'] == 'alarm_id')
@ -66,10 +66,10 @@ class TestAodhDriver(manager_congress.ScenarioPolicyBase):
alarm_map[alarm['alarm_id']] = alarm
results = (
self.admin_manager.congress_client.list_datasource_rows(
self.os_admin.congress_client.list_datasource_rows(
self.datasource_id, 'alarms'))
rule_data = (
self.admin_manager.congress_client.list_datasource_rows(
self.os_admin.congress_client.list_datasource_rows(
self.datasource_id, 'alarms.threshold_rule'))['results']
for row in results['results']:

8
tests/scenario/congress_datasources/test_ceilometer.py
View File

@ -36,14 +36,14 @@ class TestCeilometerDriver(manager_congress.ScenarioPolicyBase):
def setUp(cls):
super(TestCeilometerDriver, cls).setUp()
cls.telemetry_client = cls.admin_manager.telemetry_client
cls.telemetry_client = cls.os_admin.telemetry_client
cls.datasource_id = manager_congress.get_datasource_id(
cls.admin_manager.congress_client, 'ceilometer')
cls.os_admin.congress_client, 'ceilometer')
@decorators.attr(type='smoke')
def test_ceilometer_meters_table(self):
meter_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'meters')['columns'])
meter_id_col = next(i for i, c in enumerate(meter_schema)
if c['name'] == 'meter_id')
@ -57,7 +57,7 @@ class TestCeilometerDriver(manager_congress.ScenarioPolicyBase):
meter_map[meter['meter_id']] = meter
results = (
self.admin_manager.congress_client.list_datasource_rows(
self.os_admin.congress_client.list_datasource_rows(
self.datasource_id, 'meters'))
for row in results['results']:
try:

11
tests/scenario/congress_datasources/test_cinder.py
View File

@ -45,10 +45,11 @@ class TestCinderDriver(manager_congress.ScenarioPolicyBase):
def setUp(cls):
super(TestCinderDriver, cls).setUp()
cls.os = clients.Manager(cls.admin_manager.auth_provider.credentials)
cls.cinder = cls.os.volumes_v2_client
cls.os_primary = clients.Manager(
cls.os_admin.auth_provider.credentials)
cls.cinder = cls.os_primary.volumes_v2_client
cls.datasource_id = manager_congress.get_datasource_id(
cls.admin_manager.congress_client, 'cinder')
cls.os_admin.congress_client, 'cinder')
res = cls.cinder.create_volume(size=1, description=None, name='v0',
consistencygroup_id=None, metadata={})
LOG.debug('result of creating new volume: %s', res)
@ -56,7 +57,7 @@ class TestCinderDriver(manager_congress.ScenarioPolicyBase):
@decorators.attr(type='smoke')
def test_cinder_volumes_table(self):
volume_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'volumes')['columns'])
volume_id_col = next(i for i, c in enumerate(volume_schema)
if c['name'] == 'id')
@ -71,7 +72,7 @@ class TestCinderDriver(manager_congress.ScenarioPolicyBase):
volumes_map[volume['id']] = volume
results = (
self.admin_manager.congress_client.list_datasource_rows(
self.os_admin.congress_client.list_datasource_rows(
self.datasource_id, 'volumes'))
LOG.debug('congress cinder volumes table: %s', results)
# check that congress and cinder return the same volume IDs

2
tests/scenario/congress_datasources/test_doctor.py
View File

@ -28,7 +28,7 @@ class TestDoctorDriver(manager_congress.ScenarioPolicyBase):
'driver': 'doctor',
'config': None,
}
self.client = self.admin_manager.congress_client
self.client = self.os_admin.congress_client
response = self.client.create_datasource(doctor_setting)
self.datasource_id = response['id']

13
tests/scenario/congress_datasources/test_glancev2.py
View File

@ -43,16 +43,17 @@ class TestGlanceV2Driver(manager_congress.ScenarioPolicyBase):
def setUp(cls):
super(TestGlanceV2Driver, cls).setUp()
cls.os = clients.Manager(cls.admin_manager.auth_provider.credentials)
cls.glancev2 = cls.os.image_client_v2
cls.os_primary = clients.Manager(
cls.os_admin.auth_provider.credentials)
cls.glancev2 = cls.os_primary.image_client_v2
cls.datasource_id = manager_congress.get_datasource_id(
cls.admin_manager.congress_client, 'glancev2')
cls.os_admin.congress_client, 'glancev2')
@decorators.attr(type='smoke')
@test.services('image')
def test_glancev2_images_table(self):
image_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'images')['columns'])
image_id_col = next(i for i, c in enumerate(image_schema)
if c['name'] == 'id')
@ -66,7 +67,7 @@ class TestGlanceV2Driver(manager_congress.ScenarioPolicyBase):
image_map[image['id']] = image
results = (
self.admin_manager.congress_client.list_datasource_rows(
self.os_admin.congress_client.list_datasource_rows(
self.datasource_id, 'images'))
for row in results['results']:
try:
@ -107,7 +108,7 @@ class TestGlanceV2Driver(manager_congress.ScenarioPolicyBase):
image_tag_map[image['id']] = image['tags']
results = (
self.admin_manager.congress_client.list_datasource_rows(
self.os_admin.congress_client.list_datasource_rows(
self.datasource_id, 'tags'))
for row in results['results']:
image_id, tag = row['data'][0], row['data'][1]

31
tests/scenario/congress_datasources/test_keystonev3.py
View File

@ -38,19 +38,20 @@ class TestKeystoneV3Driver(manager_congress.ScenarioPolicyBase):
def setUp(cls):
super(TestKeystoneV3Driver, cls).setUp()
cls.os = clients.Manager(cls.admin_manager.auth_provider.credentials)
cls.keystone = cls.os.identity_v3_client
cls.projects_client = cls.os.projects_client
cls.domains_client = cls.os.domains_client
cls.roles_client = cls.os.roles_v3_client
cls.users_client = cls.os.users_v3_client
cls.os_primary = clients.Manager(
cls.os_admin.auth_provider.credentials)
cls.keystone = cls.os_primary.identity_v3_client
cls.projects_client = cls.os_primary.projects_client
cls.domains_client = cls.os_primary.domains_client
cls.roles_client = cls.os_primary.roles_v3_client
cls.users_client = cls.os_primary.users_v3_client
cls.datasource_id = manager_congress.get_datasource_id(
cls.admin_manager.congress_client, 'keystonev3')
cls.os_admin.congress_client, 'keystonev3')
@test.attr(type='smoke')
def test_keystone_users_table(self):
user_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'users')['columns'])
user_id_col = next(i for i, c in enumerate(user_schema)
if c['name'] == 'id')
@ -64,7 +65,7 @@ class TestKeystoneV3Driver(manager_congress.ScenarioPolicyBase):
user_map[user['id']] = user
results = (
self.admin_manager.congress_client.list_datasource_rows(
self.os_admin.congress_client.list_datasource_rows(
self.datasource_id, 'users'))
for row in results['results']:
try:
@ -91,7 +92,7 @@ class TestKeystoneV3Driver(manager_congress.ScenarioPolicyBase):
@test.attr(type='smoke')
def test_keystone_roles_table(self):
role_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'roles')['columns'])
role_id_col = next(i for i, c in enumerate(role_schema)
if c['name'] == 'id')
@ -105,7 +106,7 @@ class TestKeystoneV3Driver(manager_congress.ScenarioPolicyBase):
roles_map[role['id']] = role
results = (
self.admin_manager.congress_client.list_datasource_rows(
self.os_admin.congress_client.list_datasource_rows(
self.datasource_id, 'roles'))
for row in results['results']:
try:
@ -127,7 +128,7 @@ class TestKeystoneV3Driver(manager_congress.ScenarioPolicyBase):
@test.attr(type='smoke')
def test_keystone_domains_table(self):
domains_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'domains')['columns'])
domain_id_col = next(i for i, c in enumerate(domains_schema)
if c['name'] == 'id')
@ -141,7 +142,7 @@ class TestKeystoneV3Driver(manager_congress.ScenarioPolicyBase):
domains_map[domain['id']] = domain
results = (
self.admin_manager.congress_client.list_datasource_rows(
self.os_admin.congress_client.list_datasource_rows(
self.datasource_id, 'domains'))
for row in results['results']:
try:
@ -163,7 +164,7 @@ class TestKeystoneV3Driver(manager_congress.ScenarioPolicyBase):
@test.attr(type='smoke')
def test_keystone_projects_table(self):
projects_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'projects')['columns'])
project_id_col = next(i for i, c in enumerate(projects_schema)
if c['name'] == 'id')
@ -177,7 +178,7 @@ class TestKeystoneV3Driver(manager_congress.ScenarioPolicyBase):
projects_map[project['id']] = project
results = (
self.admin_manager.congress_client.list_datasource_rows(
self.os_admin.congress_client.list_datasource_rows(
self.datasource_id, 'projects'))
for row in results['results']:
try:

2
tests/scenario/congress_datasources/test_murano.py
View File

@ -46,7 +46,7 @@ class TestMuranoDriver(manager_congress.ScenarioPolicyBase):
def setUp(self):
super(TestMuranoDriver, self).setUp()
self.congress_client = (
self.admin_manager.congress_client)
self.os_admin.congress_client)
@decorators.attr(type='smoke')
@test.services('compute')

49
tests/scenario/congress_datasources/test_neutronv2.py
View File

@ -46,14 +46,15 @@ class TestNeutronV2Driver(manager_congress.ScenarioPolicyBase):
def setUp(cls):
super(TestNeutronV2Driver, cls).setUp()
cls.os = clients.Manager(cls.admin_manager.auth_provider.credentials)
cls.networks_client = cls.os.networks_client
cls.subnets_client = cls.os.subnets_client
cls.ports_client = cls.os.ports_client
cls.security_groups_client = cls.os.security_groups_client
cls.routers_client = cls.os.routers_client
cls.os_primary = clients.Manager(
cls.os_admin.auth_provider.credentials)
cls.networks_client = cls.os_primary.networks_client
cls.subnets_client = cls.os_primary.subnets_client
cls.ports_client = cls.os_primary.ports_client
cls.security_groups_client = cls.os_primary.security_groups_client
cls.routers_client = cls.os_primary.routers_client
cls.datasource_id = manager_congress.get_datasource_id(
cls.admin_manager.congress_client, 'neutronv2')
cls.os_admin.congress_client, 'neutronv2')
@decorators.attr(type='smoke')
@test.services('network')
@ -66,7 +67,7 @@ class TestNeutronV2Driver(manager_congress.ScenarioPolicyBase):
for network in networks['networks']:
network_map[network['id']] = network
client = self.admin_manager.congress_client
client = self.os_admin.congress_client
client.request_refresh(self.datasource_id)
time.sleep(1)
@ -92,15 +93,15 @@ class TestNeutronV2Driver(manager_congress.ScenarioPolicyBase):
@test.services('network')
def test_neutronv2_ports_tables(self):
port_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'ports')['columns'])
port_sec_binding_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'security_group_port_bindings')['columns'])
fixed_ips_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'fixed_ips')['columns'])
@helper.retry_on_exception
@ -110,7 +111,7 @@ class TestNeutronV2Driver(manager_congress.ScenarioPolicyBase):
for port in ports_from_neutron['ports']:
port_map[port['id']] = port
client = self.admin_manager.congress_client
client = self.os_admin.congress_client
client.request_refresh(self.datasource_id)
time.sleep(1)
@ -169,19 +170,19 @@ class TestNeutronV2Driver(manager_congress.ScenarioPolicyBase):
@test.services('network')
def test_neutronv2_subnets_tables(self):
subnet_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'subnets')['columns'])
host_routes_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'host_routes')['columns'])
dns_nameservers_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'dns_nameservers')['columns'])
allocation_pools_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'allocation_pools')['columns'])
@helper.retry_on_exception
@ -191,7 +192,7 @@ class TestNeutronV2Driver(manager_congress.ScenarioPolicyBase):
for subnet in subnets_from_neutron['subnets']:
subnet_map[subnet['id']] = subnet
client = self.admin_manager.congress_client
client = self.os_admin.congress_client
client.request_refresh(self.datasource_id)
time.sleep(1)
@ -264,11 +265,11 @@ class TestNeutronV2Driver(manager_congress.ScenarioPolicyBase):
@test.services('network')
def test_neutronv2_routers_tables(self):
router_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'routers')['columns'])
ext_gw_info_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'external_gateway_infos')['columns'])
@helper.retry_on_exception
@ -278,7 +279,7 @@ class TestNeutronV2Driver(manager_congress.ScenarioPolicyBase):
for router in routers_from_neutron['routers']:
router_map[router['id']] = router
client = self.admin_manager.congress_client
client = self.os_admin.congress_client
client.request_refresh(self.datasource_id)
time.sleep(1)
@ -318,7 +319,7 @@ class TestNeutronV2Driver(manager_congress.ScenarioPolicyBase):
@test.services('network')
def test_neutronv2_security_groups_table(self):
sg_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'security_groups')['columns'])
@helper.retry_on_exception
@ -329,7 +330,7 @@ class TestNeutronV2Driver(manager_congress.ScenarioPolicyBase):
for security_group in security_groups_neutron['security_groups']:
security_groups_map[security_group['id']] = security_group
client = self.admin_manager.congress_client
client = self.os_admin.congress_client
client.request_refresh(self.datasource_id)
time.sleep(1)
@ -355,7 +356,7 @@ class TestNeutronV2Driver(manager_congress.ScenarioPolicyBase):
@test.services('network')
def test_neutronv2_security_group_rules_table(self):
sgrs_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'security_group_rules')['columns'])
@helper.retry_on_exception
@ -367,7 +368,7 @@ class TestNeutronV2Driver(manager_congress.ScenarioPolicyBase):
for sgr in sg['security_group_rules']:
sgrs_map[sgr['id']] = sgr
client = self.admin_manager.congress_client
client = self.os_admin.congress_client
client.request_refresh(self.datasource_id)
time.sleep(1)

8
tests/scenario/congress_datasources/test_nova.py
View File

@ -46,7 +46,7 @@ class TestNovaDriver(manager_congress.ScenarioPolicyBase):
self.keypairs = {}
self.servers = []
self.datasource_id = manager_congress.get_datasource_id(
self.admin_manager.congress_client, 'nova')
self.os_admin.congress_client, 'nova')
@decorators.attr(type='smoke')
@test.services('compute', 'network')
@ -54,7 +54,7 @@ class TestNovaDriver(manager_congress.ScenarioPolicyBase):
self._setup_network_and_servers()
server_schema = (
self.admin_manager.congress_client.show_datasource_table_schema(
self.os_admin.congress_client.show_datasource_table_schema(
self.datasource_id, 'servers')['columns'])
# Convert some of the column names.
@ -77,7 +77,7 @@ class TestNovaDriver(manager_congress.ScenarioPolicyBase):
@helper.retry_on_exception
def _check_data_table_nova_servers():
results = (
self.admin_manager.congress_client.list_datasource_rows(
self.os_admin.congress_client.list_datasource_rows(
self.datasource_id, 'servers'))
for row in results['results']:
match = True
@ -118,7 +118,7 @@ class TestNovaDriver(manager_congress.ScenarioPolicyBase):
flavor_id_map[flavor['id']] = flavor
results = (
self.admin_manager.congress_client.list_datasource_rows(
self.os_admin.congress_client.list_datasource_rows(
self.datasource_id, 'flavors'))
# TODO(alexsyip): Not sure what the following OS-FLV-EXT-DATA:
# prefix is for.

6
tests/scenario/manager.py
View File

@ -522,7 +522,7 @@ class NetworkScenarioTest(ScenarioTest):
:returns: True if subnet with cidr already exist in tenant
False else
"""
cidr_in_use = self.admin_manager.subnets_client.list_subnets(
cidr_in_use = self.os_admin.subnets_client.list_subnets(
tenant_id=tenant_id, cidr=cidr)['subnets']
return len(cidr_in_use) != 0
@ -571,7 +571,7 @@ class NetworkScenarioTest(ScenarioTest):
return subnet
def _get_server_port_id_and_ip4(self, server, ip_addr=None):
ports = self.admin_manager.ports_client.list_ports(
ports = self.os_admin.ports_client.list_ports(
device_id=server['id'], fixed_ip=ip_addr)['ports']
# A port can have more than one IP address in some cases.
# If the network is dual-stack (IPv4 + IPv6), this port is associated
@ -601,7 +601,7 @@ class NetworkScenarioTest(ScenarioTest):
return port_map[0]
def _get_network_by_name(self, network_name):
net = self.admin_manager.networks_client.list_networks(
net = self.os_admin.networks_client.list_networks(
name=network_name)['networks']
self.assertNotEqual(len(net), 0,
"Unable to get network by name: %s" % network_name)

18
tests/scenario/manager_congress.py
View File

@ -54,13 +54,13 @@ class ScenarioPolicyBase(manager.NetworkScenarioTest):
@classmethod
def setup_required_clients(cls, auth_prov):
# Get congress client
cls.admin_manager.congress_client = policy_client.PolicyClient(
cls.os_admin.congress_client = policy_client.PolicyClient(
auth_prov, "policy", CONF.identity.region)
# Get telemtery_client
if getattr(CONF.service_available, 'ceilometer', False):
import ceilometer.tests.tempest.service.client as telemetry_client
cls.admin_manager.telemetry_client = (
cls.os_admin.telemetry_client = (
telemetry_client.TelemetryClient(
auth_prov,
CONF.telemetry.catalog_type, CONF.identity.region,
@ -69,7 +69,7 @@ class ScenarioPolicyBase(manager.NetworkScenarioTest):
# Get alarms client
if getattr(CONF.service_available, 'aodh_plugin', False):
import aodh.tests.tempest.service.client as alarms_client
cls.admin_manager.alarms_client = (
cls.os_admin.alarms_client = (
alarms_client.AlarmingClient(
auth_prov,
CONF.alarming_plugin.catalog_type, CONF.identity.region,
@ -94,21 +94,21 @@ class ScenarioPolicyBase(manager.NetworkScenarioTest):
checking the result of list_[networks,routers,subnets].
"""
seen_nets = self.admin_manager.networks_client.list_networks()
seen_nets = self.os_admin.networks_client.list_networks()
seen_names = [n['name'] for n in seen_nets['networks']]
seen_ids = [n['id'] for n in seen_nets['networks']]
self.assertIn(self.network['name'], seen_names)
self.assertIn(self.network['id'], seen_ids)
if self.subnet:
seen_subnets = self.admin_manager.subnets_client.list_subnets()
seen_subnets = self.os_admin.subnets_client.list_subnets()
seen_net_ids = [n['network_id'] for n in seen_subnets['subnets']]
seen_subnet_ids = [n['id'] for n in seen_subnets['subnets']]
self.assertIn(self.network['id'], seen_net_ids)
self.assertIn(self.subnet['id'], seen_subnet_ids)
if self.router:
seen_routers = self.admin_manager.routers_client.list_routers()
seen_routers = self.os_admin.routers_client.list_routers()
seen_router_ids = [n['id'] for n in seen_routers['routers']]
seen_router_names = [n['name'] for n in seen_routers['routers']]
self.assertIn(self.router['name'],
@ -118,7 +118,7 @@ class ScenarioPolicyBase(manager.NetworkScenarioTest):
def check_datasource_no_error(self, datasource_name):
"""Check that datasource has no error on latest update"""
ds_status = self.admin_manager.congress_client.list_datasource_status(
ds_status = self.os_admin.congress_client.list_datasource_status(
datasource_name)
if (ds_status['initialized'] == 'True' and
ds_status['number_of_updates'] != '0' and
@ -209,7 +209,7 @@ class ScenarioPolicyBase(manager.NetworkScenarioTest):
floating_ip, server = self.floating_ip_tuple
# get internal ports' ips:
# get all network ports in the new network
ports = self.admin_manager.ports_client.list_ports(
ports = self.os_admin.ports_client.list_ports(
tenant_id=server['tenant_id'], network_id=network.id)['ports']
internal_ips = (p['fixed_ips'][0]['ip_address'] for p in ports
@ -224,7 +224,7 @@ class ScenarioPolicyBase(manager.NetworkScenarioTest):
LOG.info(msg)
return
subnet = self.admin_manager.subnets_client.list_subnets(
subnet = self.os_admin.subnets_client.list_subnets(
network_id=CONF.network.public_network_id)['subnets']
self.assertEqual(1, len(subnet), "Found %d subnets" % len(subnet))

28
tests/scenario/test_congress_basic_ops.py
View File

@ -52,8 +52,8 @@ class TestPolicyBasicOps(manager_congress.ScenarioPolicyBase):
policy_name = "nova_%s" % ''.join(random.choice(string.lowercase)
for x in range(10))
body = {"name": policy_name}
resp = self.admin_manager.congress_client.create_policy(body)
self.addCleanup(self.admin_manager.congress_client.delete_policy,
resp = self.os_admin.congress_client.create_policy(body)
self.addCleanup(self.os_admin.congress_client.delete_policy,
resp['id'])
return resp['name']
@ -64,7 +64,7 @@ class TestPolicyBasicOps(manager_congress.ScenarioPolicyBase):
body['name'] = rule_name
if comment:
body['comment'] = comment
client = self.admin_manager.congress_client
client = self.os_admin.congress_client
response = client.create_policy_rule(policy_name, body)
if response:
self.addCleanup(client.delete_policy_rule, policy_name,
@ -95,8 +95,8 @@ class TestPolicyBasicOps(manager_congress.ScenarioPolicyBase):
metadata = {'testkey1': 'value3'}
res = {'meta': {'testkey1': 'value3'}}
server = self._create_test_server()
congress_client = self.admin_manager.congress_client
servers_client = self.admin_manager.servers_client
congress_client = self.os_admin.congress_client
servers_client = self.os_admin.servers_client
policy = self._create_random_policy()
service = 'nova'
action = 'servers.set_meta'
@ -130,19 +130,19 @@ class TestPolicyBasicOps(manager_congress.ScenarioPolicyBase):
"security_group_id), neutronv2:security_groups("
"security_group_id, tenant_id1, security_group_name,"
"description)"}
results = self.admin_manager.congress_client.create_policy_rule(
results = self.os_admin.congress_client.create_policy_rule(
'classification', body)
rule_id = results['id']
self.addCleanup(
self.admin_manager.congress_client.delete_policy_rule,
self.os_admin.congress_client.delete_policy_rule,
'classification', rule_id)
# Find the ports of on this server
ports = self.admin_manager.ports_client.list_ports(
ports = self.os_admin.ports_client.list_ports(
device_id=self.servers[0]['id'])['ports']
def check_data():
results = self.admin_manager.congress_client.list_policy_rows(
results = self.os_admin.congress_client.list_policy_rows(
'classification', 'port_security_group')
for row in results['results']:
if (row['data'][0] == ports[0]['id'] and
@ -163,7 +163,7 @@ class TestPolicyBasicOps(manager_congress.ScenarioPolicyBase):
@decorators.attr(type='smoke')
@test.services('compute', 'network')
def test_reactive_enforcement(self):
servers_client = self.admin_manager.servers_client
servers_client = self.os_admin.servers_client
server_name = 'server_under_test'
server = self._create_test_server(name=server_name)
policy_name = self._create_random_policy()
@ -254,10 +254,10 @@ class TestCongressDataSources(manager_congress.ScenarioPolicyBase):
@helper.retry_on_exception
def _check_all_datasources_are_initialized():
datasources = self.admin_manager.congress_client.list_datasources()
datasources = self.os_admin.congress_client.list_datasources()
for datasource in datasources['results']:
results = (
self.admin_manager.congress_client.list_datasource_status(
self.os_admin.congress_client.list_datasource_status(
datasource['id']))
if results['initialized'] != 'True':
return False
@ -273,10 +273,10 @@ class TestCongressDataSources(manager_congress.ScenarioPolicyBase):
@helper.retry_on_exception
def check_data():
datasources = self.admin_manager.congress_client.list_datasources()
datasources = self.os_admin.congress_client.list_datasources()
for datasource in datasources['results']:
results = (
self.admin_manager.congress_client.list_datasource_tables(
self.os_admin.congress_client.list_datasource_tables(
datasource['id']))
# NOTE(arosen): if there are no results here we return false as
# there is something wrong with a driver as it doesn't expose