openstack/octavia
Code Issues Proposed changes

Merge "Fix calls to "DELETED" items"

Browse Source
This commit is contained in:
Zuul 2018-04-26 01:50:55 +00:00 committed by Gerrit Code Review
commit d25b2cc619
22 changed files with 426 additions and 174 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
octavia/api/root_controller.py
View File

@ -49,8 +49,8 @@ class RootController(rest.RestController):
self._versions.append(
{
'status': 'CURRENT',
'updated': '2018-03-23T00:00:00Z',
'id': 'v2.0'
'updated': '2018-04-20T00:00:00Z',
'id': 'v2.1'
})
if not (v1_enabled or v2_enabled):
LOG.warning("Both v1 and v2.0 API endpoints are disabled -- is "

5
octavia/api/v2/controllers/amphora.py
View File

@ -42,7 +42,7 @@ class AmphoraController(base.BaseController):
def get_one(self, id):
"""Gets a single amphora's details."""
context = pecan.request.context.get('octavia_context')
db_amp = self._get_db_amp(context.session, id)
db_amp = self._get_db_amp(context.session, id, show_deleted=False)
self._auth_validate_action(context, context.project_id,
constants.RBAC_GET_ONE)
@ -98,7 +98,8 @@ class FailoverController(base.BaseController):
"""Fails over an amphora"""
pcontext = pecan.request.context
context = pcontext.get('octavia_context')
db_amp = self._get_db_amp(context.session, self.amp_id)
db_amp = self._get_db_amp(context.session, self.amp_id,
show_deleted=False)
self._auth_validate_action(
context, db_amp.load_balancer.project_id,

55
octavia/api/v2/controllers/base.py
View File

@ -59,9 +59,9 @@ class BaseController(rest.RestController):
return converted
@staticmethod
def _get_db_obj(session, repo, data_model, id):
def _get_db_obj(session, repo, data_model, id, show_deleted=True):
"""Gets an object from the database and returns it."""
db_obj = repo.get(session, id=id)
db_obj = repo.get(session, id=id, show_deleted=show_deleted)
if not db_obj:
LOG.exception('%(name)s %(id)s not found',
{'name': data_model._name(), 'id': id})
@ -69,51 +69,66 @@ class BaseController(rest.RestController):
resource=data_model._name(), id=id)
return db_obj
def _get_db_lb(self, session, id):
def _get_db_lb(self, session, id, show_deleted=True):
"""Get a load balancer from the database."""
return self._get_db_obj(session, self.repositories.load_balancer,
data_models.LoadBalancer, id)
data_models.LoadBalancer, id,
show_deleted=show_deleted)
def _get_db_listener(self, session, id):
def _get_db_listener(self, session, id, show_deleted=True):
"""Get a listener from the database."""
return self._get_db_obj(session, self.repositories.listener,
data_models.Listener, id)
data_models.Listener, id,
show_deleted=show_deleted)
def _get_db_pool(self, session, id):
def _get_db_pool(self, session, id, show_deleted=True):
"""Get a pool from the database."""
return self._get_db_obj(session, self.repositories.pool,
data_models.Pool, id)
data_models.Pool, id,
show_deleted=show_deleted)
def _get_db_member(self, session, id):
def _get_db_member(self, session, id, show_deleted=True):
"""Get a member from the database."""
return self._get_db_obj(session, self.repositories.member,
data_models.Member, id)
data_models.Member, id,
show_deleted=show_deleted)
def _get_db_l7policy(self, session, id):
def _get_db_hm(self, session, id, show_deleted=True):
"""Get a health monitor from the database."""
return self._get_db_obj(session, self.repositories.health_monitor,
data_models.HealthMonitor, id,
show_deleted=show_deleted)
def _get_db_l7policy(self, session, id, show_deleted=True):
"""Get a L7 Policy from the database."""
return self._get_db_obj(session, self.repositories.l7policy,
data_models.L7Policy, id)
data_models.L7Policy, id,
show_deleted=show_deleted)
def _get_db_l7rule(self, session, id):
def _get_db_l7rule(self, session, id, show_deleted=True):
"""Get a L7 Rule from the database."""
return self._get_db_obj(session, self.repositories.l7rule,
data_models.L7Rule, id)
data_models.L7Rule, id,
show_deleted=show_deleted)
def _get_db_amp(self, session, id):
def _get_db_amp(self, session, id, show_deleted=True):
"""Gets an Amphora from the database."""
return self._get_db_obj(session, self.repositories.amphora,
data_models.Amphora, id)
data_models.Amphora, id,
show_deleted=show_deleted)
def _get_lb_project_id(self, session, id):
def _get_lb_project_id(self, session, id, show_deleted=True):
"""Get the project_id of the load balancer from the database."""
lb = self._get_db_obj(session, self.repositories.load_balancer,
data_models.LoadBalancer, id)
data_models.LoadBalancer, id,
show_deleted=show_deleted)
return lb.project_id
def _get_l7policy_project_id(self, session, id):
def _get_l7policy_project_id(self, session, id, show_deleted=True):
"""Get the project_id of the load balancer from the database."""
l7policy = self._get_db_obj(session, self.repositories.l7policy,
data_models.LoadBalancer, id)
data_models.LoadBalancer, id,
show_deleted=show_deleted)
return l7policy.project_id
def _get_default_quotas(self, project_id):

17
octavia/api/v2/controllers/health_monitor.py
View File

@ -42,23 +42,12 @@ class HealthMonitorController(base.BaseController):
super(HealthMonitorController, self).__init__()
self.handler = self.handler.health_monitor
def _get_db_hm(self, session, hm_id):
"""Gets the current health monitor object from the database."""
db_hm = self.repositories.health_monitor.get(
session, id=hm_id)
if not db_hm:
LOG.info("Health Monitor %s was not found", hm_id)
raise exceptions.NotFound(
resource=data_models.HealthMonitor._name(),
id=hm_id)
return db_hm
@wsme_pecan.wsexpose(hm_types.HealthMonitorRootResponse, wtypes.text,
wtypes.text)
def get_one(self, id):
"""Gets a single healthmonitor's details."""
context = pecan.request.context.get('octavia_context')
db_hm = self._get_db_hm(context.session, id)
db_hm = self._get_db_hm(context.session, id, show_deleted=False)
self._auth_validate_action(context, db_hm.project_id,
constants.RBAC_GET_ONE)
@ -206,7 +195,7 @@ class HealthMonitorController(base.BaseController):
"""Updates a health monitor."""
context = pecan.request.context.get('octavia_context')
health_monitor = health_monitor_.healthmonitor
db_hm = self._get_db_hm(context.session, id)
db_hm = self._get_db_hm(context.session, id, show_deleted=False)
self._auth_validate_action(context, db_hm.project_id,
constants.RBAC_PUT)
@ -239,7 +228,7 @@ class HealthMonitorController(base.BaseController):
def delete(self, id):
"""Deletes a health monitor."""
context = pecan.request.context.get('octavia_context')
db_hm = self._get_db_hm(context.session, id)
db_hm = self._get_db_hm(context.session, id, show_deleted=False)
self._auth_validate_action(context, db_hm.project_id,
constants.RBAC_DELETE)

9
octavia/api/v2/controllers/l7policy.py
View File

@ -47,7 +47,8 @@ class L7PolicyController(base.BaseController):
def get(self, id):
"""Gets a single l7policy's details."""
context = pecan.request.context.get('octavia_context')
db_l7policy = self._get_db_l7policy(context.session, id)
db_l7policy = self._get_db_l7policy(context.session, id,
show_deleted=False)
self._auth_validate_action(context, db_l7policy.project_id,
constants.RBAC_GET_ONE)
@ -216,7 +217,8 @@ class L7PolicyController(base.BaseController):
if l7policy_dict.get('redirect_pool_id'):
self._get_db_pool(
context.session, l7policy_dict['redirect_pool_id'])
db_l7policy = self._get_db_l7policy(context.session, id)
db_l7policy = self._get_db_l7policy(context.session, id,
show_deleted=False)
load_balancer_id, listener_id = self._get_listener_and_loadbalancer_id(
db_l7policy)
@ -253,7 +255,8 @@ class L7PolicyController(base.BaseController):
def delete(self, id):
"""Deletes a l7policy."""
context = pecan.request.context.get('octavia_context')
db_l7policy = self._get_db_l7policy(context.session, id)
db_l7policy = self._get_db_l7policy(context.session, id,
show_deleted=False)
load_balancer_id, listener_id = self._get_listener_and_loadbalancer_id(
db_l7policy)

12
octavia/api/v2/controllers/l7rule.py
View File

@ -45,7 +45,8 @@ class L7RuleController(base.BaseController):
def get(self, id):
"""Gets a single l7rule's details."""
context = pecan.request.context.get('octavia_context')
db_l7rule = self._get_db_l7rule(context.session, id)
db_l7rule = self._get_db_l7rule(context.session, id,
show_deleted=False)
self._auth_validate_action(context, db_l7rule.project_id,
constants.RBAC_GET_ONE)
@ -61,7 +62,8 @@ class L7RuleController(base.BaseController):
pcontext = pecan.request.context
context = pcontext.get('octavia_context')
l7policy = self._get_db_l7policy(context.session, self.l7policy_id)
l7policy = self._get_db_l7policy(context.session, self.l7policy_id,
show_deleted=False)
self._auth_validate_action(context, l7policy.project_id,
constants.RBAC_GET_ALL)
@ -190,7 +192,8 @@ class L7RuleController(base.BaseController):
"""Updates a l7rule."""
l7rule = l7rule_.rule
context = pecan.request.context.get('octavia_context')
db_l7rule = self._get_db_l7rule(context.session, id)
db_l7rule = self._get_db_l7rule(context.session, id,
show_deleted=False)
new_l7rule = db_l7rule.to_dict()
new_l7rule.update(l7rule.to_dict())
new_l7rule = data_models.L7Rule.from_dict(new_l7rule)
@ -228,7 +231,8 @@ class L7RuleController(base.BaseController):
def delete(self, id):
"""Deletes a l7rule."""
context = pecan.request.context.get('octavia_context')
db_l7rule = self._get_db_l7rule(context.session, id)
db_l7rule = self._get_db_l7rule(context.session, id,
show_deleted=False)
self._auth_validate_action(context, db_l7rule.project_id,
constants.RBAC_DELETE)

32
octavia/api/v2/controllers/listener.py
View File

@ -50,24 +50,16 @@ class ListenersController(base.BaseController):
invoke_on_load=True,
).driver
def _get_db_listener(self, session, id):
"""Gets a listener object from the database."""
listener = super(ListenersController, self)._get_db_listener(
session, id)
load_balancer_id = listener.load_balancer_id
db_listener = self.repositories.listener.get(
session, load_balancer_id=load_balancer_id, id=id)
if not db_listener:
LOG.info("Listener %s not found.", id)
raise exceptions.NotFound(
resource=data_models.Listener._name(), id=id)
return db_listener
@wsme_pecan.wsexpose(listener_types.ListenerRootResponse, wtypes.text)
def get_one(self, id):
"""Gets a single listener's details."""
context = pecan.request.context.get('octavia_context')
db_listener = self._get_db_listener(context.session, id)
db_listener = self._get_db_listener(context.session, id,
show_deleted=False)
if not db_listener:
raise exceptions.NotFound(resource=data_models.Listener._name(),
id=id)
self._auth_validate_action(context, db_listener.project_id,
constants.RBAC_GET_ONE)
@ -288,7 +280,8 @@ class ListenersController(base.BaseController):
"""Updates a listener on a load balancer."""
listener = listener_.listener
context = pecan.request.context.get('octavia_context')
db_listener = self._get_db_listener(context.session, id)
db_listener = self._get_db_listener(context.session, id,
show_deleted=False)
load_balancer_id = db_listener.load_balancer_id
self._auth_validate_action(context, db_listener.project_id,
@ -333,15 +326,13 @@ class ListenersController(base.BaseController):
def delete(self, id):
"""Deletes a listener from a load balancer."""
context = pecan.request.context.get('octavia_context')
db_listener = self._get_db_listener(context.session, id)
db_listener = self._get_db_listener(context.session, id,
show_deleted=False)
load_balancer_id = db_listener.load_balancer_id
self._auth_validate_action(context, db_listener.project_id,
constants.RBAC_DELETE)
if db_listener.provisioning_status == constants.DELETED:
return
self._test_lb_and_listener_statuses(
context.session, load_balancer_id,
id=id, listener_status=constants.PENDING_DELETE)
@ -382,7 +373,8 @@ class StatisticsController(base.BaseController, stats.StatsMixin):
status_code=200)
def get(self):
context = pecan.request.context.get('octavia_context')
db_listener = self._get_db_listener(context.session, self.id)
db_listener = self._get_db_listener(context.session, self.id,
show_deleted=False)
if not db_listener:
LOG.info("Listener %s not found.", id)
raise exceptions.NotFound(

24
octavia/api/v2/controllers/load_balancer.py
View File

@ -53,7 +53,13 @@ class LoadBalancersController(base.BaseController):
def get_one(self, id):
"""Gets a single load balancer's details."""
context = pecan.request.context.get('octavia_context')
load_balancer = self._get_db_lb(context.session, id)
load_balancer = self._get_db_lb(context.session, id,
show_deleted=False)
if not load_balancer:
raise exceptions.NotFound(
resource=data_models.LoadBalancer._name(),
id=id)
self._auth_validate_action(context, load_balancer.project_id,
constants.RBAC_GET_ONE)
@ -426,7 +432,7 @@ class LoadBalancersController(base.BaseController):
"""Updates a load balancer."""
load_balancer = load_balancer.loadbalancer
context = pecan.request.context.get('octavia_context')
db_lb = self._get_db_lb(context.session, id)
db_lb = self._get_db_lb(context.session, id, show_deleted=False)
self._auth_validate_action(context, db_lb.project_id,
constants.RBAC_PUT)
@ -453,14 +459,11 @@ class LoadBalancersController(base.BaseController):
"""Deletes a load balancer."""
context = pecan.request.context.get('octavia_context')
cascade = strutils.bool_from_string(cascade)
db_lb = self._get_db_lb(context.session, id)
db_lb = self._get_db_lb(context.session, id, show_deleted=False)
self._auth_validate_action(context, db_lb.project_id,
constants.RBAC_DELETE)
if db_lb.provisioning_status == constants.DELETED:
return
with db_api.get_lock_session() as lock_session:
if (db_lb.listeners or db_lb.pools) and not cascade:
msg = _("Cannot delete Load Balancer %s - "
@ -514,7 +517,8 @@ class StatusController(base.BaseController):
status_code=200)
def get(self):
context = pecan.request.context.get('octavia_context')
load_balancer = self._get_db_lb(context.session, self.id)
load_balancer = self._get_db_lb(context.session, self.id,
show_deleted=False)
if not load_balancer:
LOG.info("Load balancer %s not found.", id)
raise exceptions.NotFound(
@ -541,7 +545,8 @@ class StatisticsController(base.BaseController, stats.StatsMixin):
status_code=200)
def get(self):
context = pecan.request.context.get('octavia_context')
load_balancer = self._get_db_lb(context.session, self.id)
load_balancer = self._get_db_lb(context.session, self.id,
show_deleted=False)
if not load_balancer:
LOG.info("Load balancer %s not found.", id)
raise exceptions.NotFound(
@ -568,7 +573,8 @@ class FailoverController(LoadBalancersController):
def put(self, **kwargs):
"""Fails over a loadbalancer"""
context = pecan.request.context.get('octavia_context')
db_lb = self._get_db_lb(context.session, self.lb_id)
db_lb = self._get_db_lb(context.session, self.lb_id,
show_deleted=False)
self._auth_validate_action(context, db_lb.project_id,
constants.RBAC_PUT_FAILOVER)

15
octavia/api/v2/controllers/member.py
View File

@ -46,7 +46,8 @@ class MemberController(base.BaseController):
def get(self, id):
"""Gets a single pool member's details."""
context = pecan.request.context.get('octavia_context')
db_member = self._get_db_member(context.session, id)
db_member = self._get_db_member(context.session, id,
show_deleted=False)
self._auth_validate_action(context, db_member.project_id,
constants.RBAC_GET_ONE)
@ -62,7 +63,8 @@ class MemberController(base.BaseController):
pcontext = pecan.request.context
context = pcontext.get('octavia_context')
pool = self._get_db_pool(context.session, self.pool_id)
pool = self._get_db_pool(context.session, self.pool_id,
show_deleted=False)
self._auth_validate_action(context, pool.project_id,
constants.RBAC_GET_ALL)
@ -210,7 +212,8 @@ class MemberController(base.BaseController):
"""Updates a pool member."""
member = member_.member
context = pecan.request.context.get('octavia_context')
db_member = self._get_db_member(context.session, id)
db_member = self._get_db_member(context.session, id,
show_deleted=False)
self._auth_validate_action(context, db_member.project_id,
constants.RBAC_PUT)
@ -242,14 +245,12 @@ class MemberController(base.BaseController):
def delete(self, id):
"""Deletes a pool member."""
context = pecan.request.context.get('octavia_context')
db_member = self._get_db_member(context.session, id)
db_member = self._get_db_member(context.session, id,
show_deleted=False)
self._auth_validate_action(context, db_member.project_id,
constants.RBAC_DELETE)
if db_member.provisioning_status == constants.DELETED:
return
self._test_lb_and_listener_and_pool_statuses(context.session,
member=db_member)
self.repositories.member.update(

9
octavia/api/v2/controllers/pool.py
View File

@ -50,7 +50,7 @@ class PoolsController(base.BaseController):
def get(self, id):
"""Gets a pool's details."""
context = pecan.request.context.get('octavia_context')
db_pool = self._get_db_pool(context.session, id)
db_pool = self._get_db_pool(context.session, id, show_deleted=False)
self._auth_validate_action(context, db_pool.project_id,
constants.RBAC_GET_ONE)
@ -253,7 +253,7 @@ class PoolsController(base.BaseController):
"""Updates a pool on a load balancer."""
pool = pool_.pool
context = pecan.request.context.get('octavia_context')
db_pool = self._get_db_pool(context.session, id)
db_pool = self._get_db_pool(context.session, id, show_deleted=False)
self._auth_validate_action(context, db_pool.project_id,
constants.RBAC_PUT)
@ -290,7 +290,7 @@ class PoolsController(base.BaseController):
def delete(self, id):
"""Deletes a pool from a load balancer."""
context = pecan.request.context.get('octavia_context')
db_pool = self._get_db_pool(context.session, id)
db_pool = self._get_db_pool(context.session, id, show_deleted=False)
if len(db_pool.l7policies) > 0:
raise exceptions.PoolInUseByL7Policy(
id=db_pool.id, l7policy_id=db_pool.l7policies[0].id)
@ -298,9 +298,6 @@ class PoolsController(base.BaseController):
self._auth_validate_action(context, db_pool.project_id,
constants.RBAC_DELETE)
if db_pool.provisioning_status == constants.DELETED:
return
self._test_lb_and_listener_statuses(
context.session, lb_id=db_pool.load_balancer_id,
listener_ids=self._get_affected_listener_ids(db_pool))

15
octavia/db/repositories.py
View File

@ -100,9 +100,22 @@ class BaseRepository(object):
:param filters: Filters to decide which entity should be retrieved.
:returns: octavia.common.data_model
"""
model = session.query(self.model_class).filter_by(**filters).first()
deleted = filters.pop('show_deleted', True)
model = session.query(self.model_class).filter_by(**filters)
if not deleted:
if hasattr(self.model_class, 'status'):
model = model.filter(
self.model_class.status != consts.DELETED)
else:
model = model.filter(
self.model_class.provisioning_status != consts.DELETED)
model = model.first()
if not model:
return
return model.to_data_model()
def get_all(self, session, pagination_helper=None, **filters):

4
octavia/tests/functional/api/test_root_controller.py
View File

@ -48,13 +48,13 @@ class TestRootController(base_db_test.OctaviaDBTestBase):
version_ids = tuple(v.get('id') for v in versions)
self.assertEqual(2, len(version_ids))
self.assertIn('v1', version_ids)
self.assertIn('v2.0', version_ids)
self.assertIn('v2.1', version_ids)
def test_api_v1_disabled(self):
versions = self._get_versions_with_config(
api_v1_enabled=False, api_v2_enabled=True)
self.assertEqual(1, len(versions))
self.assertEqual('v2.0', versions[0].get('id'))
self.assertEqual('v2.1', versions[0].get('id'))
def test_api_v2_disabled(self):
versions = self._get_versions_with_config(

9
octavia/tests/functional/api/v2/base.py
View File

@ -21,6 +21,7 @@ import pecan.testing
from octavia.api import config as pconfig
from octavia.common import constants
from octavia.common import exceptions
from octavia.db import api as db_api
from octavia.db import repositories
from octavia.tests.functional.db import base as base_db_test
@ -387,6 +388,7 @@ class BaseAPITest(base_db_test.OctaviaDBTestBase):
op_status = db_lb.operating_status
self._set_lb_and_children_statuses(lb_id, status, op_status,
autodetect=not explicit_status)
if status != constants.DELETED:
return self.get(self.LB_PATH.format(lb_id=lb_id)).json
@staticmethod
@ -399,13 +401,14 @@ class BaseAPITest(base_db_test.OctaviaDBTestBase):
def assert_final_listener_statuses(self, lb_id, listener_id, delete=False):
expected_prov_status = constants.ACTIVE
expected_op_status = constants.ONLINE
if delete:
expected_prov_status = constants.DELETED
expected_op_status = constants.OFFLINE
self.set_lb_status(lb_id, status=expected_prov_status)
try:
self.assert_correct_listener_status(expected_prov_status,
expected_op_status,
listener_id)
except exceptions.NotFound:
if not delete:
raise
def assert_correct_lb_status(self, lb_id,
operating_status, provisioning_status):

28
octavia/tests/functional/api/v2/test_amphora.py
View File

@ -103,6 +103,13 @@ class TestAmphora(base.BaseAPITest):
[mock.call(self.amp)]
)
def test_failover_deleted(self):
new_amp = self._create_additional_amp()
self.amphora_repo.update(self.session, new_amp.id,
status=constants.DELETED)
self.put(self.AMPHORA_FAILOVER_PATH.format(
amphora_id=new_amp.id), body={}, status=404)
def test_failover_bad_amp_id(self):
self.put(self.AMPHORA_FAILOVER_PATH.format(
amphora_id='asdf'), body={}, status=404)
@ -149,17 +156,12 @@ class TestAmphora(base.BaseAPITest):
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, response.json)
def test_get_hides_deleted(self):
def test_get_deleted_gives_404(self):
new_amp = self._create_additional_amp()
response = self.get(self.AMPHORAE_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 2)
self.amphora_repo.update(self.session, new_amp.id,
status=constants.DELETED)
response = self.get(self.AMPHORAE_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 1)
self.get(self.AMPHORA_PATH.format(amphora_id=new_amp.id), status=404)
def test_bad_get(self):
self.get(self.AMPHORA_PATH.format(
@ -210,6 +212,18 @@ class TestAmphora(base.BaseAPITest):
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, amps)
def test_get_all_hides_deleted(self):
new_amp = self._create_additional_amp()
response = self.get(self.AMPHORAE_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 2)
self.amphora_repo.update(self.session, new_amp.id,
status=constants.DELETED)
response = self.get(self.AMPHORAE_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 1)
def test_get_by_loadbalancer_id(self):
amps = self.get(
self.AMPHORAE_PATH,

35
octavia/tests/functional/api/v2/test_health_monitor.py
View File

@ -126,19 +126,15 @@ class TestHealthMonitor(base.BaseAPITest):
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, response.json)
def test_get_hides_deleted(self):
def test_get_deleted_gives_404(self):
api_hm = self.create_health_monitor(
self.pool_id, constants.HEALTH_MONITOR_HTTP,
1, 1, 1, 1).get(self.root_tag)
response = self.get(self.HMS_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 1)
self.set_object_status(self.health_monitor_repo, api_hm.get('id'),
provisioning_status=constants.DELETED)
response = self.get(self.HMS_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 0)
self.get(self.HM_PATH.format(healthmonitor_id=api_hm.get('id')),
status=404)
def test_bad_get(self):
self.get(self.HM_PATH.format(
@ -323,6 +319,20 @@ class TestHealthMonitor(base.BaseAPITest):
self.assertIn((hm2.get('id'), hm2.get('type')), hm_id_protocols)
self.assertIn((hm3.get('id'), hm3.get('type')), hm_id_protocols)
def test_get_all_hides_deleted(self):
api_hm = self.create_health_monitor(
self.pool_id, constants.HEALTH_MONITOR_HTTP,
1, 1, 1, 1).get(self.root_tag)
response = self.get(self.HMS_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 1)
self.set_object_status(self.health_monitor_repo, api_hm.get('id'),
provisioning_status=constants.DELETED)
response = self.get(self.HMS_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 0)
def test_get_by_project_id(self):
project1_id = uuidutils.generate_uuid()
project2_id = uuidutils.generate_uuid()
@ -1035,6 +1045,15 @@ class TestHealthMonitor(base.BaseAPITest):
self.put(self.HM_PATH.format(healthmonitor_id=api_hm.get('id')),
body=self._build_body(new_hm), status=409)
def test_update_already_deleted(self):
api_hm = self.create_health_monitor(
self.pool_id, constants.HEALTH_MONITOR_HTTP,
1, 1, 1, 1).get(self.root_tag)
# This updates the child objects
self.set_lb_status(self.lb_id, status=constants.DELETED)
self.put(self.HM_PATH.format(healthmonitor_id=api_hm.get('id')),
body=self._build_body({'max_retries': 2}), status=404)
def test_delete_when_lb_pending_delete(self):
api_hm = self.create_health_monitor(
self.pool_id, constants.HEALTH_MONITOR_HTTP,
@ -1052,4 +1071,4 @@ class TestHealthMonitor(base.BaseAPITest):
# This updates the child objects
self.set_lb_status(self.lb_id, status=constants.DELETED)
self.delete(self.HM_PATH.format(healthmonitor_id=api_hm.get('id')),
status=204)
status=404)

38
octavia/tests/functional/api/v2/test_l7policy.py
View File

@ -102,19 +102,15 @@ class TestL7Policy(base.BaseAPITest):
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, response.json)
def test_get_hides_deleted(self):
def test_get_deleted_gives_404(self):
api_l7policy = self.create_l7policy(
self.listener_id,
constants.L7POLICY_ACTION_REJECT).get(self.root_tag)
response = self.get(self.L7POLICIES_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 1)
self.set_object_status(self.l7policy_repo, api_l7policy.get('id'),
provisioning_status=constants.DELETED)
response = self.get(self.L7POLICIES_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 0)
self.get(self.L7POLICY_PATH.format(l7policy_id=api_l7policy.get('id')),
status=404)
def test_bad_get(self):
self.get(self.L7POLICY_PATH.format(
@ -299,6 +295,20 @@ class TestL7Policy(base.BaseAPITest):
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, policies)
def test_get_all_hides_deleted(self):
api_l7policy = self.create_l7policy(
self.listener_id,
constants.L7POLICY_ACTION_REJECT).get(self.root_tag)
response = self.get(self.L7POLICIES_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 1)
self.set_object_status(self.l7policy_repo, api_l7policy.get('id'),
provisioning_status=constants.DELETED)
response = self.get(self.L7POLICIES_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 0)
def test_get_by_project_id(self):
project1_id = uuidutils.generate_uuid()
project2_id = uuidutils.generate_uuid()
@ -1007,6 +1017,18 @@ class TestL7Policy(base.BaseAPITest):
l7policy_id=l7policy.get('id')),
status=409)
def test_update_already_deleted(self):
l7policy = self.create_l7policy(self.listener_id,
constants.L7POLICY_ACTION_REJECT,
).get(self.root_tag)
# This updates the child objects
self.set_lb_status(self.lb_id, status=constants.DELETED)
new_l7policy = {
'action': constants.L7POLICY_ACTION_REDIRECT_TO_URL,
'redirect_url': 'http://www.example.com'}
self.put(self.L7POLICY_PATH.format(l7policy_id=l7policy.get('id')),
body=self._build_body(new_l7policy), status=404)
def test_delete_already_deleted(self):
l7policy = self.create_l7policy(self.listener_id,
constants.L7POLICY_ACTION_REJECT,
@ -1015,4 +1037,4 @@ class TestL7Policy(base.BaseAPITest):
self.set_lb_status(self.lb_id, status=constants.DELETED)
self.delete(self.L7POLICY_PATH.format(
l7policy_id=l7policy.get('id')),
status=204)
status=404)

42
octavia/tests/functional/api/v2/test_l7rule.py
View File

@ -101,20 +101,16 @@ class TestL7Rule(base.BaseAPITest):
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, response)
def test_get_hides_deleted(self):
def test_get_deleted_gives_404(self):
api_l7rule = self.create_l7rule(
self.l7policy_id, constants.L7RULE_TYPE_PATH,
constants.L7RULE_COMPARE_TYPE_STARTS_WITH,
'/api').get(self.root_tag)
response = self.get(self.l7rules_path)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 1)
self.set_object_status(self.l7rule_repo, api_l7rule.get('id'),
provisioning_status=constants.DELETED)
response = self.get(self.l7rules_path)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 0)
self.get(self.l7rule_path.format(l7rule_id=api_l7rule.get('id')),
status=404)
def test_get_bad_parent_policy(self):
bad_path = (self.L7RULES_PATH.format(
@ -344,6 +340,21 @@ class TestL7Rule(base.BaseAPITest):
self.assertIsInstance(response, list)
self.assertEqual(0, len(response))
def test_get_all_hides_deleted(self):
api_l7rule = self.create_l7rule(
self.l7policy_id, constants.L7RULE_TYPE_PATH,
constants.L7RULE_COMPARE_TYPE_STARTS_WITH,
'/api').get(self.root_tag)
response = self.get(self.l7rules_path)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 1)
self.set_object_status(self.l7rule_repo, api_l7rule.get('id'),
provisioning_status=constants.DELETED)
response = self.get(self.l7rules_path)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 0)
def test_create_host_name_rule(self):
api_l7rule = self.create_l7rule(
self.l7policy_id, constants.L7RULE_TYPE_HOST_NAME,
@ -907,6 +918,21 @@ class TestL7Rule(base.BaseAPITest):
self.delete(self.l7rule_path.format(l7rule_id=l7rule.get('id')),
status=409)
def test_update_already_deleted(self):
l7rule = self.create_l7rule(
self.l7policy_id, constants.L7RULE_TYPE_PATH,
constants.L7RULE_COMPARE_TYPE_STARTS_WITH,
'/api').get(self.root_tag)
# This updates the child objects
self.set_lb_status(self.lb_id, status=constants.DELETED)
new_l7rule = {'type': constants.L7RULE_TYPE_COOKIE,
'compare_type':
constants.L7RULE_COMPARE_TYPE_ENDS_WITH,
'value': 'some-string',
'key': 'some-cookie'}
self.put(self.l7rule_path.format(l7rule_id=l7rule.get('id')),
body=self._build_body(new_l7rule), status=404)
def test_delete_already_deleted(self):
l7rule = self.create_l7rule(
self.l7policy_id, constants.L7RULE_TYPE_PATH,
@ -915,4 +941,4 @@ class TestL7Rule(base.BaseAPITest):
# This updates the child objects
self.set_lb_status(self.lb_id, status=constants.DELETED)
self.delete(self.l7rule_path.format(l7rule_id=l7rule.get('id')),
status=204)
status=404)

65
octavia/tests/functional/api/v2/test_listener.py
View File

@ -351,6 +351,19 @@ class TestListener(base.BaseAPITest):
self.assertEqual(li1['id'],
lis['listeners'][0]['id'])
def test_get_all_hides_deleted(self):
api_listener = self.create_listener(
constants.PROTOCOL_HTTP, 80, self.lb_id).get(self.root_tag)
response = self.get(self.LISTENERS_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 1)
self.set_object_status(self.listener_repo, api_listener.get('id'),
provisioning_status=constants.DELETED)
response = self.get(self.LISTENERS_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 0)
def test_get(self):
listener = self.create_listener(
constants.PROTOCOL_HTTP, 80, self.lb_id).get(self.root_tag)
@ -407,18 +420,14 @@ class TestListener(base.BaseAPITest):
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, response.json)
def test_get_hides_deleted(self):
def test_get_deleted_gives_404(self):
api_listener = self.create_listener(
constants.PROTOCOL_HTTP, 80, self.lb_id).get(self.root_tag)
response = self.get(self.LISTENERS_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 1)
self.set_object_status(self.listener_repo, api_listener.get('id'),
provisioning_status=constants.DELETED)
response = self.get(self.LISTENERS_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 0)
self.get(self.LISTENER_PATH.format(listener_id=api_listener.get('id')),
status=404)
def test_get_bad_listener_id(self):
listener_path = self.listener_path
@ -1130,6 +1139,28 @@ class TestListener(base.BaseAPITest):
listener_id=api_listener['id'])
self.put(listener_path, body, status=409)
def test_update_deleted(self):
lb = self.create_load_balancer(uuidutils.generate_uuid(),
name='lb1', description='desc1',
admin_state_up=False)
lb_id = lb['loadbalancer'].get('id')
self.set_lb_status(lb_id)
lb_listener = {'name': 'listener1', 'description': 'desc1',
'admin_state_up': False,
'protocol': constants.PROTOCOL_HTTP,
'protocol_port': 80, 'connection_limit': 10,
'loadbalancer_id': lb_id}
body = self._build_body(lb_listener)
api_listener = self.post(
self.LISTENERS_PATH, body).json.get(self.root_tag)
# This updates the child objects
self.set_lb_status(lb_id, status=constants.DELETED)
lb_listener_put = {'name': 'listener1_updated'}
body = self._build_body(lb_listener_put)
listener_path = self.LISTENER_PATH.format(
listener_id=api_listener['id'])
self.put(listener_path, body, status=404)
def test_delete_pending_delete(self):
lb = self.create_load_balancer(uuidutils.generate_uuid(),
name='lb1', description='desc1',
@ -1169,7 +1200,7 @@ class TestListener(base.BaseAPITest):
self.set_lb_status(lb_id, status=constants.DELETED)
listener_path = self.LISTENER_PATH.format(
listener_id=api_listener['id'])
self.delete(listener_path, status=204)
self.delete(listener_path, status=404)
def test_create_with_tls_termination_data(self):
cert_id = uuidutils.generate_uuid()
@ -1354,3 +1385,21 @@ class TestListener(base.BaseAPITest):
listener_id=li['id'] + "/stats"), status=403)
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, res.json)
def test_statistics_get_deleted(self):
lb = self.create_load_balancer(
uuidutils.generate_uuid()).get('loadbalancer')
self.set_lb_status(lb['id'])
li = self.create_listener(
constants.PROTOCOL_HTTP, 80, lb.get('id')).get('listener')
amphora = self.create_amphora(uuidutils.generate_uuid(), lb['id'])
self.create_listener_stats_dynamic(
listener_id=li.get('id'),
amphora_id=amphora.id,
bytes_in=random.randint(1, 9),
bytes_out=random.randint(1, 9),
total_connections=random.randint(1, 9),
request_errors=random.randint(1, 9))
self.set_lb_status(lb['id'], status=constants.DELETED)
self.get(self.LISTENER_PATH.format(
listener_id=li.get('id') + "/stats"), status=404)

77
octavia/tests/functional/api/v2/test_load_balancer.py
View File

@ -1072,6 +1072,19 @@ class TestLoadBalancer(base.BaseAPITest):
self.assertEqual(lb1['id'],
lbs['loadbalancers'][0]['id'])
def test_get_all_hides_deleted(self):
api_lb = self.create_load_balancer(
uuidutils.generate_uuid()).get(self.root_tag)
response = self.get(self.LBS_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 1)
self.set_object_status(self.lb_repo, api_lb.get('id'),
provisioning_status=constants.DELETED)
response = self.get(self.LBS_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 0)
def test_get(self):
project_id = uuidutils.generate_uuid()
subnet = network_models.Subnet(id=uuidutils.generate_uuid())
@ -1108,18 +1121,14 @@ class TestLoadBalancer(base.BaseAPITest):
self.assertEqual(network.id, response.get('vip_network_id'))
self.assertEqual(port.id, response.get('vip_port_id'))
def test_get_hides_deleted(self):
def test_get_deleted_gives_404(self):
api_lb = self.create_load_balancer(
uuidutils.generate_uuid()).get(self.root_tag)
response = self.get(self.LBS_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 1)
self.set_object_status(self.lb_repo, api_lb.get('id'),
provisioning_status=constants.DELETED)
response = self.get(self.LBS_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 0)
self.get(self.LB_PATH.format(lb_id=api_lb.get('id')), status=404)
def test_get_bad_lb_id(self):
path = self.LB_PATH.format(lb_id='SEAN-CONNERY')
@ -1447,6 +1456,19 @@ class TestLoadBalancer(base.BaseAPITest):
self.delete(self.LB_PATH.format(lb_id=lb_dict.get('id')))
self.delete(self.LB_PATH.format(lb_id=lb_dict.get('id')), status=409)
def test_update_already_deleted(self):
project_id = uuidutils.generate_uuid()
lb = self.create_load_balancer(uuidutils.generate_uuid(),
name='lb1',
project_id=project_id,
description='desc1',
admin_state_up=False)
lb_dict = lb.get(self.root_tag)
lb = self.set_lb_status(lb_dict.get('id'), status=constants.DELETED)
lb_json = self._build_body({'name': 'John'})
self.put(self.LB_PATH.format(lb_id=lb_dict.get('id')),
lb_json, status=404)
def test_delete_already_deleted(self):
project_id = uuidutils.generate_uuid()
lb = self.create_load_balancer(uuidutils.generate_uuid(),
@ -1456,7 +1478,7 @@ class TestLoadBalancer(base.BaseAPITest):
admin_state_up=False)
lb_dict = lb.get(self.root_tag)
lb = self.set_lb_status(lb_dict.get('id'), status=constants.DELETED)
self.delete(self.LB_PATH.format(lb_id=lb_dict.get('id')), status=204)
self.delete(self.LB_PATH.format(lb_id=lb_dict.get('id')), status=404)
def test_delete(self):
project_id = uuidutils.generate_uuid()
@ -1773,6 +1795,20 @@ class TestLoadBalancer(base.BaseAPITest):
self.app.put(path, status=202)
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
def test_failover_deleted(self):
project_id = uuidutils.generate_uuid()
lb = self.create_load_balancer(uuidutils.generate_uuid(),
name='lb1',
project_id=project_id,
description='desc1',
admin_state_up=False)
lb_dict = lb.get(self.root_tag)
lb = self.set_lb_status(lb_dict.get('id'), status=constants.DELETED)
path = self._get_full_path(self.LB_PATH.format(
lb_id=lb_dict.get('id')) + "/failover")
self.app.put(path, status=404)
def test_create_with_bad_handler(self):
self.handler_mock().load_balancer.create.side_effect = Exception()
api_lb = self.create_load_balancer(
@ -2748,6 +2784,15 @@ class TestLoadBalancerGraph(base.BaseAPITest):
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, res.json)
def test_statuses_get_deleted(self):
project_id = uuidutils.generate_uuid()
lb = self.create_load_balancer(
uuidutils.generate_uuid(),
project_id=project_id).get('loadbalancer')
self.set_lb_status(lb['id'], status=constants.DELETED)
self.get(self.LB_PATH.format(lb_id=lb['id'] + "/status"),
status=404)
def _getStats(self, lb_id):
res = self.get(self.LB_PATH.format(lb_id=lb_id + "/stats"))
return res.json.get('stats')
@ -2851,3 +2896,19 @@ class TestLoadBalancerGraph(base.BaseAPITest):
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, res.json)
def test_statistics_get_deleted(self):
lb = self.create_load_balancer(
uuidutils.generate_uuid()).get('loadbalancer')
self.set_lb_status(lb['id'])
li = self.create_listener(
constants.PROTOCOL_HTTP, 80, lb.get('id')).get('listener')
amphora = self.create_amphora(uuidutils.generate_uuid(), lb['id'])
self.create_listener_stats_dynamic(
listener_id=li.get('id'),
amphora_id=amphora.id,
bytes_in=random.randint(1, 9),
bytes_out=random.randint(1, 9),
total_connections=random.randint(1, 9))
self.set_lb_status(lb['id'], status=constants.DELETED)
self.get(self.LB_PATH.format(lb_id=lb['id'] + "/stats"), status=404)

51
octavia/tests/functional/api/v2/test_member.py
View File

@ -110,18 +110,14 @@ class TestMember(base.BaseAPITest):
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, response)
def test_get_hides_deleted(self):
def test_get_deleted_gives_404(self):
api_member = self.create_member(
self.pool_id, '192.0.2.1', 80).get(self.root_tag)
response = self.get(self.members_path)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 1)
self.set_object_status(self.member_repo, api_member.get('id'),
provisioning_status=constants.DELETED)
response = self.get(self.members_path)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 0)
self.get(self.member_path.format(member_id=api_member.get('id')),
status=404)
def test_bad_get(self):
self.get(self.member_path.format(member_id=uuidutils.generate_uuid()),
@ -148,6 +144,19 @@ class TestMember(base.BaseAPITest):
for m in [api_m_1, api_m_2]:
self.assertIn(m, response)
def test_get_all_hides_deleted(self):
api_member = self.create_member(
self.pool_id, '10.0.0.1', 80).get(self.root_tag)
response = self.get(self.members_path)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 1)
self.set_object_status(self.member_repo, api_member.get('id'),
provisioning_status=constants.DELETED)
response = self.get(self.members_path)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 0)
def test_get_all_authorized(self):
api_m_1 = self.create_member(
self.pool_id, '192.0.2.1', 80).get(self.root_tag)
@ -890,13 +899,8 @@ class TestMember(base.BaseAPITest):
member_prov_status=constants.PENDING_DELETE)
self.set_lb_status(self.lb_id)
self.assert_correct_status(
lb_id=self.lb_id, listener_id=self.listener_id,
pool_id=self.pool_with_listener_id, member_id=member.get('id'),
lb_prov_status=constants.ACTIVE,
listener_prov_status=constants.ACTIVE,
pool_prov_status=constants.ACTIVE,
member_prov_status=constants.DELETED)
member = self.get(self.member_path_listener.format(
member_id=api_member.get('id')), status=404)
def test_delete_authorized(self):
api_member = self.create_member(
@ -944,13 +948,8 @@ class TestMember(base.BaseAPITest):
member_prov_status=constants.PENDING_DELETE)
self.set_lb_status(self.lb_id)
self.assert_correct_status(
lb_id=self.lb_id, listener_id=self.listener_id,
pool_id=self.pool_with_listener_id, member_id=member.get('id'),
lb_prov_status=constants.ACTIVE,
listener_prov_status=constants.ACTIVE,
pool_prov_status=constants.ACTIVE,
member_prov_status=constants.DELETED)
member = self.get(self.member_path_listener.format(
member_id=api_member.get('id')), status=404)
def test_delete_not_authorized(self):
api_member = self.create_member(
@ -1061,6 +1060,14 @@ class TestMember(base.BaseAPITest):
self.put(self.member_path.format(member_id=member.get('id')),
body=self._build_body({'name': "member2"}), status=409)
def test_update_when_deleted(self):
member = self.create_member(
self.pool_id, address="10.0.0.1",
protocol_port=80).get(self.root_tag)
self.set_lb_status(self.lb_id, status=constants.DELETED)
self.put(self.member_path.format(member_id=member.get('id')),
body=self._build_body({'name': "member2"}), status=404)
def test_delete_when_lb_pending_delete(self):
member = self.create_member(
self.pool_id, address="192.0.2.1",
@ -1077,4 +1084,4 @@ class TestMember(base.BaseAPITest):
protocol_port=80).get(self.root_tag)
self.set_lb_status(self.lb_id, status=constants.DELETED)
self.delete(self.member_path.format(
member_id=member.get('id')), status=204)
member_id=member.get('id')), status=404)

39
octavia/tests/functional/api/v2/test_pool.py
View File

@ -124,21 +124,16 @@ class TestPool(base.BaseAPITest):
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, response.json)
def test_get_hides_deleted(self):
def test_get_deleted_gives_404(self):
api_pool = self.create_pool(
self.lb_id,
constants.PROTOCOL_HTTP,
constants.LB_ALGORITHM_ROUND_ROBIN,
listener_id=self.listener_id).get(self.root_tag)
response = self.get(self.POOLS_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 1)
self.set_object_status(self.pool_repo, api_pool.get('id'),
provisioning_status=constants.DELETED)
response = self.get(self.POOLS_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 0)
self.get(self.POOL_PATH.format(pool_id=api_pool.get('id')), status=404)
def test_bad_get(self):
self.get(self.POOL_PATH.format(pool_id=uuidutils.generate_uuid()),
@ -156,6 +151,22 @@ class TestPool(base.BaseAPITest):
self.assertEqual(1, len(pools))
self.assertEqual(api_pool.get('id'), pools[0].get('id'))
def test_get_all_hides_deleted(self):
api_pool = self.create_pool(
self.lb_id,
constants.PROTOCOL_HTTP,
constants.LB_ALGORITHM_ROUND_ROBIN,
listener_id=self.listener_id).get(self.root_tag)
response = self.get(self.POOLS_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 1)
self.set_object_status(self.pool_repo, api_pool.get('id'),
provisioning_status=constants.DELETED)
response = self.get(self.POOLS_PATH)
objects = response.json.get(self.root_tag_list)
self.assertEqual(len(objects), 0)
def test_get_all_admin(self):
project_id = uuidutils.generate_uuid()
lb1 = self.create_load_balancer(uuidutils.generate_uuid(), name='lb1',
@ -1382,6 +1393,18 @@ class TestPool(base.BaseAPITest):
self.delete(self.POOL_PATH.format(pool_id=api_pool.get('id')),
status=409)
def test_update_already_deleted(self):
api_pool = self.create_pool(
self.lb_id,
constants.PROTOCOL_HTTP,
constants.LB_ALGORITHM_ROUND_ROBIN,
listener_id=self.listener_id).get(self.root_tag)
# This updates the child objects
self.set_lb_status(self.lb_id, status=constants.DELETED)
new_pool = {'admin_state_up': False}
self.put(self.POOL_PATH.format(pool_id=api_pool.get('id')),
self._build_body(new_pool), status=404)
def test_delete_already_deleted(self):
api_pool = self.create_pool(
self.lb_id,
@ -1391,4 +1414,4 @@ class TestPool(base.BaseAPITest):
# This updates the child objects
self.set_lb_status(self.lb_id, status=constants.DELETED)
self.delete(self.POOL_PATH.format(pool_id=api_pool.get('id')),
status=204)
status=404)

7
releasenotes/notes/deleted-404-2cdd751e7afbe036.yaml Normal file
View File

@ -0,0 +1,7 @@
---
fixes:
- |
Fixes the v2 API returning "DELETED" records until the amphora_expiry_age
timeout expired. The API will now immediately return a 404 HTTP status
code when deleted objects are requested. The API version has been raised
to v2.1 to reflect this change.