raise cls.skipException(msg)
cls.identity_admin_client = cls.os_adm.identity_client
- def _check_quotas(self, new_quotas):
+ @test.attr(type='gate')
+ @test.idempotent_id('2390f766-836d-40ef-9aeb-e810d78207fb')
+ def test_quotas(self):
# Add a tenant to conduct the test
test_tenant = data_utils.rand_name('test_tenant_')
test_description = data_utils.rand_name('desc_')
tenant_id = tenant['id']
self.addCleanup(self.identity_admin_client.delete_tenant, tenant_id)
+ new_quotas = {'network': 0, 'security_group': 0}
+
# Change quotas for tenant
quota_set = self.admin_client.update_quotas(tenant_id,
**new_quotas)
non_default_quotas = self.admin_client.list_quotas()
for q in non_default_quotas['quotas']:
self.assertNotEqual(tenant_id, q['tenant_id'])
-
- @test.attr(type='gate')
- @test.idempotent_id('2390f766-836d-40ef-9aeb-e810d78207fb')
- def test_quotas(self):
- new_quotas = {'network': 0, 'security_group': 0}
- self._check_quotas(new_quotas)
-
- @test.idempotent_id('a7add2b1-691e-44d6-875f-697d9685f091')
- @test.requires_ext(extension='lbaas', service='network')
- @test.attr(type='gate')
- def test_lbaas_quotas(self):
- new_quotas = {'vip': 1, 'pool': 2,
- 'member': 3, 'health_monitor': 4}
- self._check_quotas(new_quotas)
cls.subnets = []
cls.ports = []
cls.routers = []
- cls.pools = []
- cls.vips = []
- cls.members = []
- cls.health_monitors = []
cls.vpnservices = []
cls.ikepolicies = []
cls.floating_ips = []
for router in cls.routers:
cls._try_delete_resource(cls.delete_router,
router)
-
- # Clean up health monitors
- for health_monitor in cls.health_monitors:
- cls._try_delete_resource(cls.client.delete_health_monitor,
- health_monitor['id'])
- # Clean up members
- for member in cls.members:
- cls._try_delete_resource(cls.client.delete_member,
- member['id'])
- # Clean up vips
- for vip in cls.vips:
- cls._try_delete_resource(cls.client.delete_vip,
- vip['id'])
- # Clean up pools
- for pool in cls.pools:
- cls._try_delete_resource(cls.client.delete_pool,
- pool['id'])
# Clean up metering label rules
for metering_label_rule in cls.metering_label_rules:
cls._try_delete_resource(
cls.floating_ips.append(fip)
return fip
- @classmethod
- def create_pool(cls, name, lb_method, protocol, subnet):
- """Wrapper utility that returns a test pool."""
- body = cls.client.create_pool(
- name=name,
- lb_method=lb_method,
- protocol=protocol,
- subnet_id=subnet['id'])
- pool = body['pool']
- cls.pools.append(pool)
- return pool
-
- @classmethod
- def update_pool(cls, name):
- """Wrapper utility that returns a test pool."""
- body = cls.client.update_pool(name=name)
- pool = body['pool']
- return pool
-
- @classmethod
- def create_vip(cls, name, protocol, protocol_port, subnet, pool):
- """Wrapper utility that returns a test vip."""
- body = cls.client.create_vip(name=name,
- protocol=protocol,
- protocol_port=protocol_port,
- subnet_id=subnet['id'],
- pool_id=pool['id'])
- vip = body['vip']
- cls.vips.append(vip)
- return vip
-
- @classmethod
- def update_vip(cls, name):
- body = cls.client.update_vip(name=name)
- vip = body['vip']
- return vip
-
- @classmethod
- def create_member(cls, protocol_port, pool, ip_version=None):
- """Wrapper utility that returns a test member."""
- ip_version = ip_version if ip_version is not None else cls._ip_version
- member_address = "fd00::abcd" if ip_version == 6 else "10.0.9.46"
- body = cls.client.create_member(address=member_address,
- protocol_port=protocol_port,
- pool_id=pool['id'])
- member = body['member']
- cls.members.append(member)
- return member
-
- @classmethod
- def update_member(cls, admin_state_up):
- body = cls.client.update_member(admin_state_up=admin_state_up)
- member = body['member']
- return member
-
- @classmethod
- def create_health_monitor(cls, delay, max_retries, Type, timeout):
- """Wrapper utility that returns a test health monitor."""
- body = cls.client.create_health_monitor(delay=delay,
- max_retries=max_retries,
- type=Type,
- timeout=timeout)
- health_monitor = body['health_monitor']
- cls.health_monitors.append(health_monitor)
- return health_monitor
-
- @classmethod
- def update_health_monitor(cls, admin_state_up):
- body = cls.client.update_vip(admin_state_up=admin_state_up)
- health_monitor = body['health_monitor']
- return health_monitor
-
@classmethod
def create_router_interface(cls, router_id, subnet_id):
"""Wrapper utility that returns a router interface."""
# The following list represents resource names that do not require
# changing underscore to a hyphen
- hyphen_exceptions = ["health_monitors", "firewall_rules",
- "firewall_policies", "service_profiles"]
+ hyphen_exceptions = [
+ "firewall_rules", "firewall_policies", "service_profiles"]
# the following map is used to construct proper URI
# for the given neutron resource
service_resource_prefix_map = {
'subnets': '',
'subnetpools': '',
'ports': '',
- 'pools': 'lb',
- 'vips': 'lb',
- 'health_monitors': 'lb',
- 'members': 'lb',
'ipsecpolicies': 'vpn',
'vpnservices': 'vpn',
'ikepolicies': 'vpn',
body = json.loads(body)
return service_client.ResponseBody(resp, body)
- def associate_health_monitor_with_pool(self, health_monitor_id,
- pool_id):
- post_body = {
- "health_monitor": {
- "id": health_monitor_id,
- }
- }
- body = json.dumps(post_body)
- uri = '%s/lb/pools/%s/health_monitors' % (self.uri_prefix,
- pool_id)
- resp, body = self.post(uri, body)
- self.expected_success(201, resp.status)
- body = json.loads(body)
- return service_client.ResponseBody(resp, body)
-
- def disassociate_health_monitor_with_pool(self, health_monitor_id,
- pool_id):
- uri = '%s/lb/pools/%s/health_monitors/%s' % (self.uri_prefix, pool_id,
- health_monitor_id)
- resp, body = self.delete(uri)
- self.expected_success(204, resp.status)
- return service_client.ResponseBody(resp, body)
-
def list_router_interfaces(self, uuid):
uri = '%s/ports?device_id=%s' % (self.uri_prefix, uuid)
resp, body = self.get(uri)
body = json.loads(body)
return service_client.ResponseBody(resp, body)
- def list_pools_hosted_by_one_lbaas_agent(self, agent_id):
- uri = '%s/agents/%s/loadbalancer-pools' % (self.uri_prefix, agent_id)
- resp, body = self.get(uri)
- self.expected_success(200, resp.status)
- body = json.loads(body)
- return service_client.ResponseBody(resp, body)
-
- def show_lbaas_agent_hosting_pool(self, pool_id):
- uri = ('%s/lb/pools/%s/loadbalancer-agent' %
- (self.uri_prefix, pool_id))
- resp, body = self.get(uri)
- self.expected_success(200, resp.status)
- body = json.loads(body)
- return service_client.ResponseBody(resp, body)
-
def list_routers_on_l3_agent(self, agent_id):
uri = '%s/agents/%s/l3-routers' % (self.uri_prefix, agent_id)
resp, body = self.get(uri)
body = json.loads(body)
return service_client.ResponseBody(resp, body)
- def list_lb_pool_stats(self, pool_id):
- uri = '%s/lb/pools/%s/stats' % (self.uri_prefix, pool_id)
- resp, body = self.get(uri)
- self.expected_success(200, resp.status)
- body = json.loads(body)
- return service_client.ResponseBody(resp, body)
-
def add_dhcp_agent_to_network(self, agent_id, network_id):
post_body = {'network_id': network_id}
body = json.dumps(post_body)
def delete(self):
self.client.delete_security_group_rule(self.id)
-
-
-class DeletablePool(DeletableResource):
-
- def delete(self):
- self.client.delete_pool(self.id)
-
-
-class DeletableMember(DeletableResource):
-
- def delete(self):
- self.client.delete_member(self.id)
-
-
-class DeletableVip(DeletableResource):
-
- def delete(self):
- self.client.delete_vip(self.id)
-
- def refresh(self):
- result = self.client.show_vip(self.id)
- super(DeletableVip, self).update(**result['vip'])
Code Review / openstack-build / neutron-build.git / commitdiff