openstack/octavia
Code Issues Proposed changes

Transition l7rule flows to dicts

Browse Source 
Change-Id: Id94a293a88d27546c0fe7b45367922fa4cca260a
Story: 2005072
Task: 30813
This commit is contained in:
Ann Kamyshnikova 2019-06-28 17:13:13 +04:00
parent 7d310c3985
commit ea6b065cb0
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()