openstack
/
octavia
Code Issues Proposed changes

Merge "Transition l7rule flows to dicts"

This commit is contained in:
Zuul 2020-01-26 18:19:27 +00:00 committed by Gerrit Code Review
parent 789f5310ca ea6b065cb0
commit fb37005661
13 changed files with 176 additions and 177 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
octavia/api/drivers/amphora_driver/v2/driver.py
View File

@ -321,21 +321,20 @@ class AmphoraProviderDriver(driver_base.ProviderDriver):
# L7 Rule
def l7rule_create(self, l7rule):
payload = {consts.L7RULE_ID: l7rule.l7rule_id}
payload = {consts.L7RULE: l7rule.to_dict()}
self.client.cast({}, 'create_l7rule', **payload)
def l7rule_delete(self, l7rule):
l7rule_id = l7rule.l7rule_id
payload = {consts.L7RULE_ID: l7rule_id}
payload = {consts.L7RULE: l7rule.to_dict()}
self.client.cast({}, 'delete_l7rule', **payload)
def l7rule_update(self, old_l7rule, new_l7rule):
l7rule_dict = new_l7rule.to_dict()
if 'admin_state_up' in l7rule_dict:
l7rule_dict['enabled'] = l7rule_dict.pop('admin_state_up')
l7rule_id = l7rule_dict.pop('l7rule_id')
if consts.ADMIN_STATE_UP in l7rule_dict:
l7rule_dict['enabled'] = l7rule_dict.pop(consts.ADMIN_STATE_UP)
l7rule_dict.pop(consts.L7RULE_ID)
payload = {consts.L7RULE_ID: l7rule_id,
payload = {consts.ORIGINAL_L7RULE: old_l7rule.to_dict(),
consts.L7RULE_UPDATES: l7rule_dict}
self.client.cast({}, 'update_l7rule', **payload)

1
octavia/common/constants.py
View File

@ -367,6 +367,7 @@ NICS = 'nics'
OBJECT = 'object'
ORIGINAL_HEALTH_MONITOR = 'original_health_monitor'
ORIGINAL_L7POLICY = 'original_l7policy'
ORIGINAL_L7RULE = 'original_l7rule'
ORIGINAL_LISTENER = 'original_listener'
ORIGINAL_LOADBALANCER = 'original_load_balancer'
ORIGINAL_MEMBER = 'original_member'

19
octavia/controller/queue/v2/endpoints.py
View File

@ -151,17 +151,18 @@ class Endpoints(object):
constants.L7POLICY_ID))
self.worker.delete_l7policy(l7policy)
def create_l7rule(self, context, l7rule_id):
LOG.info('Creating l7rule \'%s\'...', l7rule_id)
self.worker.create_l7rule(l7rule_id)
def create_l7rule(self, context, l7rule):
LOG.info('Creating l7rule \'%s\'...', l7rule.get(constants.L7RULE_ID))
self.worker.create_l7rule(l7rule)
def update_l7rule(self, context, l7rule_id, l7rule_updates):
LOG.info('Updating l7rule \'%s\'...', l7rule_id)
self.worker.update_l7rule(l7rule_id, l7rule_updates)
def update_l7rule(self, context, original_l7rule, l7rule_updates):
LOG.info('Updating l7rule \'%s\'...', original_l7rule.get(
constants.L7RULE_ID))
self.worker.update_l7rule(original_l7rule, l7rule_updates)
def delete_l7rule(self, context, l7rule_id):
LOG.info('Deleting l7rule \'%s\'...', l7rule_id)
self.worker.delete_l7rule(l7rule_id)
def delete_l7rule(self, context, l7rule):
LOG.info('Deleting l7rule \'%s\'...', l7rule.get(constants.L7RULE_ID))
self.worker.delete_l7rule(l7rule)
def update_amphora_agent_config(self, context, amphora_id):
LOG.info('Updating amphora \'%s\' agent configuration...',

88
octavia/controller/worker/v2/controller_worker.py
View File

@ -730,116 +730,88 @@ class ControllerWorker(base_taskflow.BaseTaskFlowEngine):
log=LOG):
update_l7policy_tf.run()
@tenacity.retry(
retry=tenacity.retry_if_exception_type(db_exceptions.NoResultFound),
wait=tenacity.wait_incrementing(
RETRY_INITIAL_DELAY, RETRY_BACKOFF, RETRY_MAX),
stop=tenacity.stop_after_attempt(RETRY_ATTEMPTS))
def create_l7rule(self, l7rule_id):
def create_l7rule(self, l7rule):
"""Creates an L7 Rule.
:param l7rule_id: ID of the l7rule to create
:param l7rule: Provider dict l7rule
:returns: None
:raises NoResultFound: Unable to find the object
"""
l7rule = self._l7rule_repo.get(db_apis.get_session(),
id=l7rule_id)
if not l7rule:
LOG.warning('Failed to fetch %s %s from DB. Retrying for up to '
'60 seconds.', 'l7rule', l7rule_id)
raise db_exceptions.NoResultFound
db_l7policy = self._l7policy_repo.get(db_apis.get_session(),
id=l7rule[constants.L7POLICY_ID])
l7policy = l7rule.l7policy
load_balancer = l7policy.listener.load_balancer
provider_lb = provider_utils.db_loadbalancer_to_provider_loadbalancer(
load_balancer).to_dict()
load_balancer = db_l7policy.listener.load_balancer
listeners_dicts = (
provider_utils.db_listeners_to_provider_dicts_list_of_dicts(
[l7policy.listener]))
[db_l7policy.listener]))
l7policy_dict = provider_utils.db_l7policy_to_provider_l7policy(
l7policy)
db_l7policy)
create_l7rule_tf = self._taskflow_load(
self._l7rule_flows.get_create_l7rule_flow(),
store={constants.L7RULE: l7rule,
constants.L7POLICY: l7policy_dict.to_dict(),
constants.LISTENERS: listeners_dicts,
constants.LOADBALANCER_ID: load_balancer.id,
constants.LOADBALANCER: provider_lb})
constants.L7POLICY_ID: db_l7policy.id,
constants.LOADBALANCER_ID: load_balancer.id
})
with tf_logging.DynamicLoggingListener(create_l7rule_tf,
log=LOG):
create_l7rule_tf.run()
def delete_l7rule(self, l7rule_id):
def delete_l7rule(self, l7rule):
"""Deletes an L7 rule.
:param l7rule_id: ID of the l7rule to delete
:param l7rule: Provider dict of the l7rule to delete
:returns: None
:raises L7RuleNotFound: The referenced l7rule was not found
"""
l7rule = self._l7rule_repo.get(db_apis.get_session(),
id=l7rule_id)
l7policy = l7rule.l7policy
load_balancer = l7policy.listener.load_balancer
db_l7policy = self._l7policy_repo.get(db_apis.get_session(),
id=l7rule[constants.L7POLICY_ID])
l7policy = provider_utils.db_l7policy_to_provider_l7policy(db_l7policy)
load_balancer = db_l7policy.listener.load_balancer
listeners_dicts = (
provider_utils.db_listeners_to_provider_dicts_list_of_dicts(
[l7policy.listener]))
provider_lb = provider_utils.db_loadbalancer_to_provider_loadbalancer(
load_balancer).to_dict()
l7policy_dict = provider_utils.db_l7policy_to_provider_l7policy(
l7policy)
[db_l7policy.listener]))
delete_l7rule_tf = self._taskflow_load(
self._l7rule_flows.get_delete_l7rule_flow(),
store={constants.L7RULE: l7rule,
constants.L7POLICY: l7policy_dict.to_dict(),
constants.L7POLICY: l7policy.to_dict(),
constants.LISTENERS: listeners_dicts,
constants.LOADBALANCER_ID: load_balancer.id,
constants.LOADBALANCER: provider_lb})
constants.L7POLICY_ID: db_l7policy.id,
constants.LOADBALANCER_ID: load_balancer.id
})
with tf_logging.DynamicLoggingListener(delete_l7rule_tf,
log=LOG):
delete_l7rule_tf.run()
def update_l7rule(self, l7rule_id, l7rule_updates):
def update_l7rule(self, original_l7rule, l7rule_updates):
"""Updates an L7 rule.
:param l7rule_id: ID of the l7rule to update
:param l7rule: Origin dict of the l7rule to update
:param l7rule_updates: Dict containing updated l7rule attributes
:returns: None
:raises L7RuleNotFound: The referenced l7rule was not found
"""
l7rule = None
try:
l7rule = self._get_db_obj_until_pending_update(
self._l7rule_repo, l7rule_id)
except tenacity.RetryError as e:
LOG.warning('L7 rule did not go into %s in 60 seconds. '
'This either due to an in-progress Octavia upgrade '
'or an overloaded and failing database. Assuming '
'an upgrade is in progress and continuing.',
constants.PENDING_UPDATE)
l7rule = e.last_attempt.result()
l7policy = l7rule.l7policy
load_balancer = l7policy.listener.load_balancer
provider_lb = provider_utils.db_loadbalancer_to_provider_loadbalancer(
load_balancer).to_dict()
db_l7policy = self._l7policy_repo.get(
db_apis.get_session(), id=original_l7rule[constants.L7POLICY_ID])
load_balancer = db_l7policy.listener.load_balancer
listeners_dicts = (
provider_utils.db_listeners_to_provider_dicts_list_of_dicts(
[l7policy.listener]))
[db_l7policy.listener]))
l7policy_dict = provider_utils.db_l7policy_to_provider_l7policy(
l7policy)
db_l7policy)
update_l7rule_tf = self._taskflow_load(
self._l7rule_flows.get_update_l7rule_flow(),
store={constants.L7RULE: l7rule,
store={constants.L7RULE: original_l7rule,
constants.L7POLICY: l7policy_dict.to_dict(),
constants.LISTENERS: listeners_dicts,
constants.LOADBALANCER: provider_lb,
constants.L7POLICY_ID: db_l7policy.id,
constants.LOADBALANCER_ID: load_balancer.id,
constants.UPDATE_DICT: l7rule_updates})
with tf_logging.DynamicLoggingListener(update_l7rule_tf,

9
octavia/controller/worker/v2/flows/l7rule_flows.py
View File

@ -31,8 +31,9 @@ class L7RuleFlows(object):
create_l7rule_flow = linear_flow.Flow(constants.CREATE_L7RULE_FLOW)
create_l7rule_flow.add(lifecycle_tasks.L7RuleToErrorOnRevertTask(
requires=[constants.L7RULE,
constants.L7POLICY_ID,
constants.LISTENERS,
constants.LOADBALANCER]))
constants.LOADBALANCER_ID]))
create_l7rule_flow.add(database_tasks.MarkL7RulePendingCreateInDB(
requires=constants.L7RULE))
create_l7rule_flow.add(amphora_driver_tasks.ListenersUpdate(
@ -54,8 +55,9 @@ class L7RuleFlows(object):
delete_l7rule_flow = linear_flow.Flow(constants.DELETE_L7RULE_FLOW)
delete_l7rule_flow.add(lifecycle_tasks.L7RuleToErrorOnRevertTask(
requires=[constants.L7RULE,
constants.L7POLICY_ID,
constants.LISTENERS,
constants.LOADBALANCER]))
constants.LOADBALANCER_ID]))
delete_l7rule_flow.add(database_tasks.MarkL7RulePendingDeleteInDB(
requires=constants.L7RULE))
delete_l7rule_flow.add(amphora_driver_tasks.ListenersUpdate(
@ -77,8 +79,9 @@ class L7RuleFlows(object):
update_l7rule_flow = linear_flow.Flow(constants.UPDATE_L7RULE_FLOW)
update_l7rule_flow.add(lifecycle_tasks.L7RuleToErrorOnRevertTask(
requires=[constants.L7RULE,
constants.L7POLICY_ID,
constants.LISTENERS,
constants.LOADBALANCER]))
constants.LOADBALANCER_ID]))
update_l7rule_flow.add(database_tasks.MarkL7RulePendingUpdateInDB(
requires=constants.L7RULE))
update_l7rule_flow.add(amphora_driver_tasks.ListenersUpdate(

68
octavia/controller/worker/v2/tasks/database_tasks.py
View File

@ -368,8 +368,10 @@ class DeleteL7RuleInDB(BaseDatabaseTask):
:returns: None
"""
LOG.debug("Delete in DB for l7rule id: %s ", l7rule.id)
self.l7rule_repo.delete(db_apis.get_session(), id=l7rule.id)
LOG.debug("Delete in DB for l7rule id: %s",
l7rule[constants.L7RULE_ID])
self.l7rule_repo.delete(db_apis.get_session(),
id=l7rule[constants.L7RULE_ID])
def revert(self, l7rule, *args, **kwargs):
"""Mark the l7rule ERROR since the delete couldn't happen
@ -378,14 +380,16 @@ class DeleteL7RuleInDB(BaseDatabaseTask):
:returns: None
"""
LOG.warning("Reverting delete in DB for l7rule id %s", l7rule.id)
LOG.warning("Reverting delete in DB for l7rule id %s",
l7rule[constants.L7RULE_ID])
try:
self.l7rule_repo.update(db_apis.get_session(), l7rule.id,
self.l7rule_repo.update(db_apis.get_session(),
l7rule[constants.L7RULE_ID],
provisioning_status=constants.ERROR)
except Exception as e:
LOG.error("Failed to update l7rule %(l7rule)s "
"provisioning_status to ERROR due to: %(except)s",
{'l7rule': l7rule.id, 'except': e})
{'l7rule': l7rule[constants.L7RULE_ID], 'except': e})
class ReloadAmphora(BaseDatabaseTask):
@ -1638,8 +1642,9 @@ class UpdateL7RuleInDB(BaseDatabaseTask):
:returns: None
"""
LOG.debug("Update DB for l7rule id: %s ", l7rule.id)
self.l7rule_repo.update(db_apis.get_session(), l7rule.id,
LOG.debug("Update DB for l7rule id: %s", l7rule[constants.L7RULE_ID])
self.l7rule_repo.update(db_apis.get_session(),
l7rule[constants.L7RULE_ID],
**update_dict)
def revert(self, l7rule, *args, **kwargs):
@ -1650,15 +1655,15 @@ class UpdateL7RuleInDB(BaseDatabaseTask):
"""
LOG.warning("Reverting update l7rule in DB "
"for l7rule id %s", l7rule.id)
"for l7rule id %s", l7rule[constants.L7RULE_ID])
try:
self.l7policy_repo.update(db_apis.get_session(),
l7rule.l7policy.id,
l7rule[constants.L7POLICY_ID],
provisioning_status=constants.ERROR)
except Exception as e:
LOG.error("Failed to update L7rule %(l7r)s provisioning_status to "
"ERROR due to: %(except)s", {'l7r': l7rule.l7policy.id,
'except': e})
"ERROR due to: %(except)s",
{'l7r': l7rule[constants.L7POLICY_ID], 'except': e})
class GetAmphoraDetails(BaseDatabaseTask):
@ -2120,10 +2125,13 @@ class MarkL7RuleActiveInDB(BaseDatabaseTask):
"""
LOG.debug("Mark ACTIVE in DB for l7rule id: %s",
l7rule.id)
op_status = constants.ONLINE if l7rule.enabled else constants.OFFLINE
l7rule[constants.L7RULE_ID])
db_rule = self.l7rule_repo.get(db_apis.get_session(),
id=l7rule[constants.L7RULE_ID])
op_status = (constants.ONLINE if db_rule.enabled
else constants.OFFLINE)
self.l7rule_repo.update(db_apis.get_session(),
l7rule.id,
l7rule[constants.L7RULE_ID],
provisioning_status=constants.ACTIVE,
operating_status=op_status)
@ -2135,8 +2143,9 @@ class MarkL7RuleActiveInDB(BaseDatabaseTask):
"""
LOG.warning("Reverting mark l7rule ACTIVE in DB "
"for l7rule id %s", l7rule.id)
self.task_utils.mark_l7rule_prov_status_error(l7rule.id)
"for l7rule id %s", l7rule[constants.L7RULE_ID])
self.task_utils.mark_l7rule_prov_status_error(
l7rule[constants.L7RULE_ID])
class MarkL7RulePendingCreateInDB(BaseDatabaseTask):
@ -2153,9 +2162,9 @@ class MarkL7RulePendingCreateInDB(BaseDatabaseTask):
"""
LOG.debug("Mark PENDING CREATE in DB for l7rule id: %s",
l7rule.id)
l7rule[constants.L7RULE_ID])
self.l7rule_repo.update(db_apis.get_session(),
l7rule.id,
l7rule[constants.L7RULE_ID],
provisioning_status=constants.PENDING_CREATE)
def revert(self, l7rule, *args, **kwargs):
@ -2166,8 +2175,9 @@ class MarkL7RulePendingCreateInDB(BaseDatabaseTask):
"""
LOG.warning("Reverting mark l7rule pending create in DB "
"for l7rule id %s", l7rule.id)
self.task_utils.mark_l7rule_prov_status_error(l7rule.id)
"for l7rule id %s", l7rule[constants.L7RULE_ID])
self.task_utils.mark_l7rule_prov_status_error(
l7rule[constants.L7RULE_ID])
class MarkL7RulePendingDeleteInDB(BaseDatabaseTask):
@ -2184,9 +2194,9 @@ class MarkL7RulePendingDeleteInDB(BaseDatabaseTask):
"""
LOG.debug("Mark PENDING DELETE in DB for l7rule id: %s",
l7rule.id)
l7rule[constants.L7RULE_ID])
self.l7rule_repo.update(db_apis.get_session(),
l7rule.id,
l7rule[constants.L7RULE_ID],
provisioning_status=constants.PENDING_DELETE)
def revert(self, l7rule, *args, **kwargs):
@ -2197,8 +2207,9 @@ class MarkL7RulePendingDeleteInDB(BaseDatabaseTask):
"""
LOG.warning("Reverting mark l7rule pending delete in DB "
"for l7rule id %s", l7rule.id)
self.task_utils.mark_l7rule_prov_status_error(l7rule.id)
"for l7rule id %s", l7rule[constants.L7RULE_ID])
self.task_utils.mark_l7rule_prov_status_error(
l7rule[constants.L7RULE_ID])
class MarkL7RulePendingUpdateInDB(BaseDatabaseTask):
@ -2215,9 +2226,9 @@ class MarkL7RulePendingUpdateInDB(BaseDatabaseTask):
"""
LOG.debug("Mark PENDING UPDATE in DB for l7rule id: %s",
l7rule.id)
l7rule[constants.L7RULE_ID])
self.l7rule_repo.update(db_apis.get_session(),
l7rule.id,
l7rule[constants.L7RULE_ID],
provisioning_status=constants.PENDING_UPDATE)
def revert(self, l7rule, *args, **kwargs):
@ -2228,8 +2239,9 @@ class MarkL7RulePendingUpdateInDB(BaseDatabaseTask):
"""
LOG.warning("Reverting mark l7rule pending update in DB "
"for l7rule id %s", l7rule.id)
self.task_utils.mark_l7rule_prov_status_error(l7rule.id)
"for l7rule id %s", l7rule[constants.L7RULE_ID])
self.task_utils.mark_l7rule_prov_status_error(
l7rule[constants.L7RULE_ID])
class MarkMemberActiveInDB(BaseDatabaseTask):

12
octavia/controller/worker/v2/tasks/lifecycle_tasks.py
View File

@ -83,14 +83,16 @@ class L7PolicyToErrorOnRevertTask(BaseLifecycleTask):
class L7RuleToErrorOnRevertTask(BaseLifecycleTask):
"""Task to set a l7rule to ERROR on revert."""
def execute(self, l7rule, listeners, loadbalancer):
def execute(self, l7rule, l7policy_id, listeners, loadbalancer_id):
pass
def revert(self, l7rule, listeners, loadbalancer, *args, **kwargs):
self.task_utils.mark_l7rule_prov_status_error(l7rule.id)
self.task_utils.mark_l7policy_prov_status_active(l7rule.l7policy_id)
def revert(self, l7rule, l7policy_id, listeners, loadbalancer_id, *args,
**kwargs):
self.task_utils.mark_l7rule_prov_status_error(
l7rule[constants.L7RULE_ID])
self.task_utils.mark_l7policy_prov_status_active(l7policy_id)
self.task_utils.mark_loadbalancer_prov_status_active(
loadbalancer[constants.LOADBALANCER_ID])
loadbalancer_id)
for listener in listeners:
self.task_utils.mark_listener_prov_status_active(
listener[constants.LISTENER_ID])

11
octavia/tests/unit/api/drivers/amphora_driver/v2/test_amphora_driver.py
View File

@ -581,7 +581,7 @@ class TestAmphoraDriver(base.TestRpc):
provider_l7rule = driver_dm.L7Rule(
l7rule_id=self.sample_data.l7rule1_id)
self.amp_driver.l7rule_create(provider_l7rule)
payload = {consts.L7RULE_ID: self.sample_data.l7rule1_id}
payload = {consts.L7RULE: provider_l7rule.to_dict()}
mock_cast.assert_called_with({}, 'create_l7rule', **payload)
@mock.patch('oslo_messaging.RPCClient.cast')
@ -589,7 +589,7 @@ class TestAmphoraDriver(base.TestRpc):
provider_l7rule = driver_dm.L7Rule(
l7rule_id=self.sample_data.l7rule1_id)
self.amp_driver.l7rule_delete(provider_l7rule)
payload = {consts.L7RULE_ID: self.sample_data.l7rule1_id}
payload = {consts.L7RULE: provider_l7rule.to_dict()}
mock_cast.assert_called_with({}, 'delete_l7rule', **payload)
@mock.patch('oslo_messaging.RPCClient.cast')
@ -599,8 +599,9 @@ class TestAmphoraDriver(base.TestRpc):
provider_l7rule = driver_dm.L7Rule(
l7rule_id=self.sample_data.l7rule1_id, admin_state_up=True)
l7rule_dict = {'enabled': True}
self.amp_driver.l7rule_update(old_provider_l7rule, provider_l7rule)
payload = {consts.L7RULE_ID: self.sample_data.l7rule1_id,
self.amp_driver.l7rule_update(old_provider_l7rule,
provider_l7rule)
payload = {consts.ORIGINAL_L7RULE: old_provider_l7rule.to_dict(),
consts.L7RULE_UPDATES: l7rule_dict}
mock_cast.assert_called_with({}, 'update_l7rule', **payload)
@ -612,7 +613,7 @@ class TestAmphoraDriver(base.TestRpc):
l7rule_id=self.sample_data.l7rule1_id, invert=True)
l7rule_dict = {'invert': True}
self.amp_driver.l7rule_update(old_provider_l7rule, provider_l7rule)
payload = {consts.L7RULE_ID: self.sample_data.l7rule1_id,
payload = {consts.ORIGINAL_L7RULE: old_provider_l7rule.to_dict(),
consts.L7RULE_UPDATES: l7rule_dict}
mock_cast.assert_called_with({}, 'update_l7rule', **payload)

12
octavia/tests/unit/controller/queue/v2/test_endpoints.py
View File

@ -170,20 +170,20 @@ class TestEndpoints(base.TestCase):
self.resource)
def test_create_l7rule(self):
self.ep.create_l7rule(self.context, self.resource_id)
self.ep.create_l7rule(self.context, self.resource)
self.ep.worker.create_l7rule.assert_called_once_with(
self.resource_id)
self.resource)
def test_update_l7rule(self):
self.ep.update_l7rule(self.context, self.resource_id,
self.ep.update_l7rule(self.context, self.resource,
self.resource_updates)
self.ep.worker.update_l7rule.assert_called_once_with(
self.resource_id, self.resource_updates)
self.resource, self.resource_updates)
def test_delete_l7rule(self):
self.ep.delete_l7rule(self.context, self.resource_id)
self.ep.delete_l7rule(self.context, self.resource)
self.ep.worker.delete_l7rule.assert_called_once_with(
self.resource_id)
self.resource)
def test_update_amphora_agent_config(self):
self.ep.update_amphora_agent_config(self.context, self.resource)

10
octavia/tests/unit/controller/worker/v2/flows/test_l7rule_flows.py
View File

@ -34,8 +34,10 @@ class TestL7RuleFlows(base.TestCase):
self.assertIsInstance(l7rule_flow, flow.Flow)
self.assertIn(constants.LISTENERS, l7rule_flow.requires)
self.assertIn(constants.LOADBALANCER, l7rule_flow.requires)
self.assertIn(constants.LOADBALANCER_ID, l7rule_flow.requires)
self.assertIn(constants.L7RULE, l7rule_flow.requires)
self.assertIn(constants.L7POLICY, l7rule_flow.requires)
self.assertIn(constants.L7POLICY_ID, l7rule_flow.requires)
self.assertEqual(5, len(l7rule_flow.requires))
self.assertEqual(0, len(l7rule_flow.provides))
@ -47,9 +49,10 @@ class TestL7RuleFlows(base.TestCase):
self.assertIsInstance(l7rule_flow, flow.Flow)
self.assertIn(constants.LISTENERS, l7rule_flow.requires)
self.assertIn(constants.LOADBALANCER, l7rule_flow.requires)
self.assertIn(constants.LOADBALANCER_ID, l7rule_flow.requires)
self.assertIn(constants.L7RULE, l7rule_flow.requires)
self.assertIn(constants.L7POLICY, l7rule_flow.requires)
self.assertIn(constants.L7POLICY_ID, l7rule_flow.requires)
self.assertEqual(5, len(l7rule_flow.requires))
self.assertEqual(0, len(l7rule_flow.provides))
@ -62,9 +65,10 @@ class TestL7RuleFlows(base.TestCase):
self.assertIn(constants.L7RULE, l7rule_flow.requires)
self.assertIn(constants.LISTENERS, l7rule_flow.requires)
self.assertIn(constants.LOADBALANCER, l7rule_flow.requires)
self.assertIn(constants.LOADBALANCER_ID, l7rule_flow.requires)
self.assertIn(constants.UPDATE_DICT, l7rule_flow.requires)
self.assertIn(constants.L7POLICY, l7rule_flow.requires)
self.assertIn(constants.L7POLICY_ID, l7rule_flow.requires)
self.assertEqual(6, len(l7rule_flow.requires))
self.assertEqual(0, len(l7rule_flow.provides))

17
octavia/tests/unit/controller/worker/v2/tasks/test_database_tasks.py
View File

@ -152,10 +152,11 @@ class TestDatabaseTasks(base.TestCase):
constants.ADMIN_STATE_UP: True,
}
self.l7rule_mock = mock.MagicMock()
self.l7rule_mock.id = L7RULE_ID
self.l7rule_mock.l7policy = mock.MagicMock()
self.l7rule_mock.l7policy.id = L7POLICY_ID
self.l7rule_mock = {
constants.L7RULE_ID: L7RULE_ID,
constants.ADMIN_STATE_UP: True,
constants.L7POLICY_ID: L7POLICY_ID,
}
self.amphora = {
constants.ID: AMP_ID,
@ -406,7 +407,7 @@ class TestDatabaseTasks(base.TestCase):
mock_amphora_repo_delete):
delete_l7rule = database_tasks.DeleteL7RuleInDB()
delete_l7rule.execute(_l7rule_mock)
delete_l7rule.execute(self.l7rule_mock)
repo.L7RuleRepository.delete.assert_called_once_with(
'TEST',
@ -415,7 +416,7 @@ class TestDatabaseTasks(base.TestCase):
# Test the revert
mock_l7rule_repo_delete.reset_mock()
delete_l7rule.revert(_l7rule_mock)
delete_l7rule.revert(self.l7rule_mock)
# TODO(sbalukoff) Fix
# repo.ListenerRepository.update.assert_called_once_with(
@ -2394,7 +2395,9 @@ class TestDatabaseTasks(base.TestCase):
provisioning_status=constants.ERROR)
@mock.patch('octavia.db.repositories.L7RuleRepository.update')
@mock.patch('octavia.db.repositories.L7RuleRepository.get')
def test_mark_l7rule_active_in_db(self,
mock_l7rule_repo_get,
mock_l7rule_repo_update,
mock_generate_uuid,
mock_LOG,
@ -2403,7 +2406,7 @@ class TestDatabaseTasks(base.TestCase):
mock_listener_repo_update,
mock_amphora_repo_update,
mock_amphora_repo_delete):
mock_l7rule_repo_get.return_value = _l7rule_mock
mark_l7rule_active = (database_tasks.MarkL7RuleActiveInDB())
mark_l7rule_active.execute(self.l7rule_mock)

11
octavia/tests/unit/controller/worker/v2/tasks/test_lifecycle_tasks.py
View File

@ -30,9 +30,10 @@ class TestLifecycleTasks(base.TestCase):
self.L7POLICY = mock.MagicMock()
self.L7POLICY_ID = uuidutils.generate_uuid()
self.L7POLICY.id = self.L7POLICY_ID
self.L7RULE = mock.MagicMock()
self.L7RULE_ID = uuidutils.generate_uuid()
self.L7RULE.id = self.L7RULE_ID
self.L7RULE = {
constants.L7RULE_ID: self.L7RULE_ID
}
self.LISTENER = mock.MagicMock()
self.LISTENER_ID = uuidutils.generate_uuid()
self.LISTENER = {constants.LISTENER_ID: self.LISTENER_ID}
@ -184,15 +185,17 @@ class TestLifecycleTasks(base.TestCase):
# Execute
l7rule_to_error_on_revert.execute(self.L7RULE,
self.L7POLICY_ID,
self.LISTENERS,
self.LOADBALANCER)
self.LOADBALANCER_ID)
self.assertFalse(mock_l7rule_prov_status_error.called)
# Revert
l7rule_to_error_on_revert.revert(self.L7RULE,
self.L7POLICY_ID,
self.LISTENERS,
self.LOADBALANCER)
self.LOADBALANCER_ID)
mock_l7rule_prov_status_error.assert_called_once_with(
self.L7RULE_ID)

82
octavia/tests/unit/controller/worker/v2/test_controller_worker.py
View File

@ -1205,28 +1205,27 @@ class TestControllerWorker(base.TestCase):
mock_amp_repo_get):
_flow_mock.reset_mock()
mock_l7rule_repo_get.side_effect = [None, _l7rule_mock]
mock_l7policy_repo_get.return_value = _l7policy_mock
cw = controller_worker.ControllerWorker()
cw.create_l7rule(L7RULE_ID)
provider_lb = provider_utils.db_loadbalancer_to_provider_loadbalancer(
_db_load_balancer_mock).to_dict()
cw.create_l7rule(_l7rule_mock.to_dict())
l7_policy = provider_utils.db_l7policy_to_provider_l7policy(
_l7policy_mock)
(base_taskflow.BaseTaskFlowEngine._taskflow_load.
assert_called_once_with(_flow_mock,
store={constants.L7RULE: _l7rule_mock,
constants.L7POLICY: {
constants.L7POLICY_ID:
L7POLICY_ID},
constants.LOADBALANCER_ID:
LB_ID,
store={constants.L7RULE:
_l7rule_mock.to_dict(),
constants.L7POLICY:
l7_policy.to_dict(),
constants.L7POLICY_ID: L7POLICY_ID,
constants.LOADBALANCER_ID: LB_ID,
constants.LISTENERS:
[self.ref_listener_dict],
constants.LOADBALANCER:
provider_lb}))
[self.ref_listener_dict]
}))
_flow_mock.run.assert_called_once_with()
self.assertEqual(2, mock_l7rule_repo_get.call_count)
@mock.patch('octavia.controller.worker.v2.flows.'
'l7rule_flows.L7RuleFlows.get_delete_l7rule_flow',
@ -1248,22 +1247,22 @@ class TestControllerWorker(base.TestCase):
_flow_mock.reset_mock()
cw = controller_worker.ControllerWorker()
cw.delete_l7rule(L7RULE_ID)
provider_lb = provider_utils.db_loadbalancer_to_provider_loadbalancer(
_db_load_balancer_mock).to_dict()
cw.delete_l7rule(_l7rule_mock.to_dict())
l7_policy = provider_utils.db_l7policy_to_provider_l7policy(
_l7policy_mock)
(base_taskflow.BaseTaskFlowEngine._taskflow_load.
assert_called_once_with(_flow_mock,
store={constants.L7RULE: _l7rule_mock,
constants.LOADBALANCER_ID:
LB_ID,
constants.L7POLICY: {
constants.L7POLICY_ID:
L7POLICY_ID},
constants.LISTENERS:
[self.ref_listener_dict],
constants.LOADBALANCER:
provider_lb}))
store={
constants.L7RULE:
_l7rule_mock.to_dict(),
constants.L7POLICY:
l7_policy.to_dict(),
constants.L7POLICY_ID: L7POLICY_ID,
constants.LISTENERS:
[self.ref_listener_dict],
constants.LOADBALANCER_ID: LB_ID,
}))
_flow_mock.run.assert_called_once_with()
@ -1288,24 +1287,23 @@ class TestControllerWorker(base.TestCase):
_l7rule_mock.provisioning_status = constants.PENDING_UPDATE
cw = controller_worker.ControllerWorker()
cw.update_l7rule(L7RULE_ID, L7RULE_UPDATE_DICT)
provider_lb = provider_utils.db_loadbalancer_to_provider_loadbalancer(
_db_load_balancer_mock).to_dict()
cw.update_l7rule(_l7rule_mock.to_dict(), L7RULE_UPDATE_DICT)
l7_policy = provider_utils.db_l7policy_to_provider_l7policy(
_l7policy_mock)
(base_taskflow.BaseTaskFlowEngine._taskflow_load.
assert_called_once_with(_flow_mock,
store={constants.L7RULE: _l7rule_mock,
constants.L7POLICY: {
constants.L7POLICY_ID:
L7POLICY_ID},
constants.LOADBALANCER_ID:
LB_ID,
constants.LISTENERS:
[self.ref_listener_dict],
constants.LOADBALANCER:
provider_lb,
constants.UPDATE_DICT:
L7RULE_UPDATE_DICT}))
store={
constants.L7RULE:
_l7rule_mock.to_dict(),
constants.L7POLICY:
l7_policy.to_dict(),
constants.L7POLICY_ID: L7POLICY_ID,
constants.LOADBALANCER_ID: LB_ID,
constants.LISTENERS:
[self.ref_listener_dict],
constants.UPDATE_DICT:
L7RULE_UPDATE_DICT}))
_flow_mock.run.assert_called_once_with()