openstack/openstacksdk
Code Issues Proposed changes

Merge "Use mock context in test_fwaas"

Browse Source
This commit is contained in:
Zuul
2019-02-28 21:04:28 +00:00
committed by Gerrit Code Review
parent 30d8a15bcd d8fafcfe52
commit 9cd0ef2ac5
1 changed files with 100 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

233
openstack/tests/unit/cloud/test_fwaas.py
View File

@@ -10,7 +10,7 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
from copy import deepcopy from copy import deepcopy
from mock import Mock import mock
from openstack import exceptions from openstack import exceptions
from openstack.network.v2.firewall_group import FirewallGroup from openstack.network.v2.firewall_group import FirewallGroup
@@ -121,8 +121,6 @@ class TestFirewallRule(FirewallTestCase):
self.assert_calls() self.assert_calls()
def test_delete_firewall_rule_not_found(self): def test_delete_firewall_rule_not_found(self):
_delete = self.cloud.network.delete_firewall_rule
_log = self.cloud.log.debug
self.register_uris([ self.register_uris([
dict(method='GET', # short-circuit dict(method='GET', # short-circuit
uri=self._make_mock_url('firewall_rules', uri=self._make_mock_url('firewall_rules',
@@ -132,18 +130,14 @@ class TestFirewallRule(FirewallTestCase):
uri=self._make_mock_url('firewall_rules'), uri=self._make_mock_url('firewall_rules'),
json={'firewall_rules': []}) json={'firewall_rules': []})
]) ])
self.cloud.network.delete_firewall_rule = Mock()
self.cloud.log.debug = Mock()
self.assertFalse( with mock.patch.object(self.cloud.network, 'delete_firewall_rule'), \
self.cloud.delete_firewall_rule(self.firewall_rule_name)) mock.patch.object(self.cloud.log, 'debug'):
self.assertFalse(
self.cloud.delete_firewall_rule(self.firewall_rule_name))
self.cloud.network.delete_firewall_rule.assert_not_called() self.cloud.network.delete_firewall_rule.assert_not_called()
self.cloud.log.debug.assert_called_once() self.cloud.log.debug.assert_called_once()
# restore methods
self.cloud.network.delete_firewall_rule = _delete
self.cloud.log.debug = _log
def test_delete_firewall_multiple_matches(self): def test_delete_firewall_multiple_matches(self):
self.register_uris([ self.register_uris([
@@ -305,8 +299,6 @@ class TestFirewallPolicy(FirewallTestCase):
def test_create_firewall_policy_rule_not_found(self): def test_create_firewall_policy_rule_not_found(self):
posted_policy = deepcopy(self._mock_firewall_policy_attrs) posted_policy = deepcopy(self._mock_firewall_policy_attrs)
del posted_policy['id'] del posted_policy['id']
_create = self.cloud.network.create_firewall_policy
self.cloud.network.create_firewall_policy = Mock()
self.register_uris([ self.register_uris([
dict(method='GET', # short-circuit dict(method='GET', # short-circuit
uri=self._make_mock_url('firewall_rules', uri=self._make_mock_url('firewall_rules',
@@ -316,16 +308,15 @@ class TestFirewallPolicy(FirewallTestCase):
uri=self._make_mock_url('firewall_rules'), uri=self._make_mock_url('firewall_rules'),
json={'firewall_rules': []}) json={'firewall_rules': []})
]) ])
self.assertRaises(exceptions.ResourceNotFound,
self.cloud.create_firewall_policy, **posted_policy) with mock.patch.object(self.cloud.network, 'create_firewall_policy'):
self.cloud.network.create_firewall_policy.assert_not_called() self.assertRaises(exceptions.ResourceNotFound,
self.assert_calls() self.cloud.create_firewall_policy,
# restore **posted_policy)
self.cloud.network.create_firewall_policy = _create self.cloud.network.create_firewall_policy.assert_not_called()
self.assert_calls()
def test_delete_firewall_policy(self): def test_delete_firewall_policy(self):
_log = self.cloud.log.debug
self.cloud.log.debug = Mock()
self.register_uris([ self.register_uris([
dict(method='GET', # short-circuit dict(method='GET', # short-circuit
uri=self._make_mock_url('firewall_policies', uri=self._make_mock_url('firewall_policies',
@@ -339,37 +330,32 @@ class TestFirewallPolicy(FirewallTestCase):
self.firewall_policy_id), self.firewall_policy_id),
json={}, status_code=204) json={}, status_code=204)
]) ])
self.assertTrue(
self.cloud.delete_firewall_policy(self.firewall_policy_name))
self.assert_calls()
self.cloud.log.debug.assert_not_called() with mock.patch.object(self.cloud.log, 'debug'):
# restore self.assertTrue(
self.cloud.log.debug = _log self.cloud.delete_firewall_policy(self.firewall_policy_name))
self.assert_calls()
self.cloud.log.debug.assert_not_called()
def test_delete_firewall_policy_filters(self): def test_delete_firewall_policy_filters(self):
filters = {'project_id': self.mock_firewall_policy['project_id']} filters = {'project_id': self.mock_firewall_policy['project_id']}
_find = self.cloud.network.find_firewall_policy
_log = self.cloud.log.debug
self.cloud.log.debug = Mock()
self.cloud.network.find_firewall_policy = Mock(
return_value=self.mock_firewall_policy)
self.register_uris([ self.register_uris([
dict(method='DELETE', dict(method='DELETE',
uri=self._make_mock_url('firewall_policies', uri=self._make_mock_url('firewall_policies',
self.firewall_policy_id), self.firewall_policy_id),
json={}, status_code=204) json={}, status_code=204)
]) ])
self.assertTrue(
self.cloud.delete_firewall_policy(self.firewall_policy_name, with mock.patch.object(self.cloud.network, 'find_firewall_policy',
filters)) return_value=self.mock_firewall_policy), \
self.assert_calls() mock.patch.object(self.cloud.log, 'debug'):
self.cloud.network.find_firewall_policy.assert_called_once_with( self.assertTrue(
self.firewall_policy_name, ignore_missing=False, **filters) self.cloud.delete_firewall_policy(self.firewall_policy_name,
self.cloud.log.debug.assert_not_called() filters))
# restore self.assert_calls()
self.cloud.network.find_firewall_policy = _find self.cloud.network.find_firewall_policy.assert_called_once_with(
self.cloud.log.debug = _log self.firewall_policy_name, ignore_missing=False, **filters)
self.cloud.log.debug.assert_not_called()
def test_delete_firewall_policy_not_found(self): def test_delete_firewall_policy_not_found(self):
self.register_uris([ self.register_uris([
@@ -381,14 +367,12 @@ class TestFirewallPolicy(FirewallTestCase):
uri=self._make_mock_url('firewall_policies'), uri=self._make_mock_url('firewall_policies'),
json={'firewall_policies': []}) json={'firewall_policies': []})
]) ])
_log = self.cloud.log.debug
self.cloud.log.debug = Mock() with mock.patch.object(self.cloud.log, 'debug'):
self.assertFalse( self.assertFalse(
self.cloud.delete_firewall_policy(self.firewall_policy_name)) self.cloud.delete_firewall_policy(self.firewall_policy_name))
self.assert_calls() self.assert_calls()
self.cloud.log.debug.assert_called_once() self.cloud.log.debug.assert_called_once()
# restore
self.cloud.log.debug = _log
def test_get_firewall_policy(self): def test_get_firewall_policy(self):
self.register_uris([ self.register_uris([
@@ -510,10 +494,6 @@ class TestFirewallPolicy(FirewallTestCase):
updated_policy = deepcopy(self.mock_firewall_policy) updated_policy = deepcopy(self.mock_firewall_policy)
updated_policy.update(params) updated_policy.update(params)
_find = self.cloud.network.find_firewall_policy
self.cloud.network.find_firewall_policy = Mock(
return_value=deepcopy(self.mock_firewall_policy))
self.register_uris([ self.register_uris([
dict(method='PUT', dict(method='PUT',
uri=self._make_mock_url('firewall_policies', uri=self._make_mock_url('firewall_policies',
@@ -521,14 +501,17 @@ class TestFirewallPolicy(FirewallTestCase):
json={'firewall_policy': updated_policy}, json={'firewall_policy': updated_policy},
validate=dict(json={'firewall_policy': params})), validate=dict(json={'firewall_policy': params})),
]) ])
self.assertDictEqual(updated_policy,
self.cloud.update_firewall_policy( with mock.patch.object(self.cloud.network, 'find_firewall_policy',
self.firewall_policy_name, filters, **params)) return_value=deepcopy(
self.assert_calls() self.mock_firewall_policy)):
self.cloud.network.find_firewall_policy.assert_called_once_with( self.assertDictEqual(
self.firewall_policy_name, ignore_missing=False, **filters) updated_policy,
# restore self.cloud.update_firewall_policy(self.firewall_policy_name,
self.cloud.network.find_firewall_policy = _find filters, **params))
self.assert_calls()
self.cloud.network.find_firewall_policy.assert_called_once_with(
self.firewall_policy_name, ignore_missing=False, **filters)
def test_insert_rule_into_policy(self): def test_insert_rule_into_policy(self):
rule0 = FirewallRule( rule0 = FirewallRule(
@@ -638,8 +621,6 @@ class TestFirewallPolicy(FirewallTestCase):
def test_insert_rule_into_policy_not_found(self): def test_insert_rule_into_policy_not_found(self):
policy_name = 'bogus_policy' policy_name = 'bogus_policy'
_find_rule = self.cloud.network.find_firewall_rule
self.cloud.network.find_firewall_rule = Mock()
self.register_uris([ self.register_uris([
dict(method='GET', # short-circuit dict(method='GET', # short-circuit
uri=self._make_mock_url('firewall_policies', policy_name), uri=self._make_mock_url('firewall_policies', policy_name),
@@ -648,13 +629,13 @@ class TestFirewallPolicy(FirewallTestCase):
uri=self._make_mock_url('firewall_policies'), uri=self._make_mock_url('firewall_policies'),
json={'firewall_policies': []}) json={'firewall_policies': []})
]) ])
self.assertRaises(exceptions.ResourceNotFound,
self.cloud.insert_rule_into_policy, with mock.patch.object(self.cloud.network, 'find_firewall_rule'):
policy_name, 'bogus_rule') self.assertRaises(exceptions.ResourceNotFound,
self.assert_calls() self.cloud.insert_rule_into_policy,
self.cloud.network.find_firewall_rule.assert_not_called() policy_name, 'bogus_rule')
# restore self.assert_calls()
self.cloud.network.find_firewall_rule = _find_rule self.cloud.network.find_firewall_rule.assert_not_called()
def test_insert_rule_into_policy_rule_not_found(self): def test_insert_rule_into_policy_rule_not_found(self):
rule_name = 'unknown_rule' rule_name = 'unknown_rule'
@@ -676,8 +657,6 @@ class TestFirewallPolicy(FirewallTestCase):
self.assert_calls() self.assert_calls()
def test_insert_rule_into_policy_already_associated(self): def test_insert_rule_into_policy_already_associated(self):
_log = self.cloud.log.debug
self.cloud.log.debug = Mock()
rule = FirewallRule( rule = FirewallRule(
**TestFirewallRule._mock_firewall_rule_attrs).to_dict() **TestFirewallRule._mock_firewall_rule_attrs).to_dict()
policy = deepcopy(self.mock_firewall_policy) policy = deepcopy(self.mock_firewall_policy)
@@ -691,12 +670,12 @@ class TestFirewallPolicy(FirewallTestCase):
uri=self._make_mock_url('firewall_rules', rule['id']), uri=self._make_mock_url('firewall_rules', rule['id']),
json={'firewall_rule': rule}) json={'firewall_rule': rule})
]) ])
r = self.cloud.insert_rule_into_policy(policy['id'], rule['id'])
self.assertDictEqual(policy, r.to_dict()) with mock.patch.object(self.cloud.log, 'debug'):
self.assert_calls() r = self.cloud.insert_rule_into_policy(policy['id'], rule['id'])
self.cloud.log.debug.assert_called() self.assertDictEqual(policy, r.to_dict())
# restore self.assert_calls()
self.cloud.log.debug = _log self.cloud.log.debug.assert_called()
def test_remove_rule_from_policy(self): def test_remove_rule_from_policy(self):
policy_name = self.firewall_policy_name policy_name = self.firewall_policy_name
@@ -734,8 +713,6 @@ class TestFirewallPolicy(FirewallTestCase):
self.assert_calls() self.assert_calls()
def test_remove_rule_from_policy_not_found(self): def test_remove_rule_from_policy_not_found(self):
_find_rule = self.cloud.network.find_firewall_rule
self.cloud.network.find_firewall_rule = Mock()
self.register_uris([ self.register_uris([
dict(method='GET', # short-circuit dict(method='GET', # short-circuit
uri=self._make_mock_url('firewall_policies', uri=self._make_mock_url('firewall_policies',
@@ -745,14 +722,14 @@ class TestFirewallPolicy(FirewallTestCase):
uri=self._make_mock_url('firewall_policies'), uri=self._make_mock_url('firewall_policies'),
json={'firewall_policies': []}) json={'firewall_policies': []})
]) ])
self.assertRaises(exceptions.ResourceNotFound,
self.cloud.remove_rule_from_policy, with mock.patch.object(self.cloud.network, 'find_firewall_rule'):
self.firewall_policy_name, self.assertRaises(exceptions.ResourceNotFound,
TestFirewallRule.firewall_rule_name) self.cloud.remove_rule_from_policy,
self.assert_calls() self.firewall_policy_name,
self.cloud.network.find_firewall_rule.assert_not_called() TestFirewallRule.firewall_rule_name)
# restore self.assert_calls()
self.cloud.network.find_firewall_rule = _find_rule self.cloud.network.find_firewall_rule.assert_not_called()
def test_remove_rule_from_policy_rule_not_found(self): def test_remove_rule_from_policy_rule_not_found(self):
retrieved_policy = deepcopy(self.mock_firewall_policy) retrieved_policy = deepcopy(self.mock_firewall_policy)
@@ -782,10 +759,6 @@ class TestFirewallPolicy(FirewallTestCase):
policy = deepcopy(self.mock_firewall_policy) policy = deepcopy(self.mock_firewall_policy)
del policy['firewall_rules'][0] del policy['firewall_rules'][0]
_log = self.cloud.log.debug
_remove = self.cloud.network.remove_rule_from_policy
self.cloud.log.debug = Mock()
self.cloud.network.remove_rule_from_policy = Mock()
self.register_uris([ self.register_uris([
dict(method='GET', dict(method='GET',
uri=self._make_mock_url('firewall_policies', policy['id']), uri=self._make_mock_url('firewall_policies', policy['id']),
@@ -794,14 +767,14 @@ class TestFirewallPolicy(FirewallTestCase):
uri=self._make_mock_url('firewall_rules', rule['id']), uri=self._make_mock_url('firewall_rules', rule['id']),
json={'firewall_rule': rule}) json={'firewall_rule': rule})
]) ])
r = self.cloud.remove_rule_from_policy(policy['id'], rule['id'])
self.assertDictEqual(policy, r.to_dict()) with mock.patch.object(self.cloud.network, 'remove_rule_from_policy'),\
self.assert_calls() mock.patch.object(self.cloud.log, 'debug'):
self.cloud.log.debug.assert_called_once() r = self.cloud.remove_rule_from_policy(policy['id'], rule['id'])
self.cloud.network.remove_rule_from_policy.assert_not_called() self.assertDictEqual(policy, r.to_dict())
# restore self.assert_calls()
self.cloud.log.debug = _log self.cloud.log.debug.assert_called_once()
self.cloud.network.remove_rule_from_policy = _remove self.cloud.network.remove_rule_from_policy.assert_not_called()
class TestFirewallGroup(FirewallTestCase): class TestFirewallGroup(FirewallTestCase):
@@ -945,24 +918,22 @@ class TestFirewallGroup(FirewallTestCase):
def test_delete_firewall_group_filters(self): def test_delete_firewall_group_filters(self):
filters = {'project_id': self.mock_firewall_group['project_id']} filters = {'project_id': self.mock_firewall_group['project_id']}
_find = self.cloud.network.find_firewall_group
self.cloud.network.find_firewall_group = Mock(
return_value=deepcopy(self.mock_firewall_group))
self.register_uris([ self.register_uris([
dict(method='DELETE', dict(method='DELETE',
uri=self._make_mock_url('firewall_groups', uri=self._make_mock_url('firewall_groups',
self.firewall_group_id), self.firewall_group_id),
status_code=204) status_code=204)
]) ])
self.assertTrue(
self.cloud.delete_firewall_group(self.firewall_group_name,
filters))
self.assert_calls()
self.cloud.network.find_firewall_group.assert_called_once_with( with mock.patch.object(self.cloud.network, 'find_firewall_group',
self.firewall_group_name, ignore_missing=False, **filters) return_value=deepcopy(
# restore self.mock_firewall_group)):
self.cloud.network.find_firewall_group = _find self.assertTrue(
self.cloud.delete_firewall_group(self.firewall_group_name,
filters))
self.assert_calls()
self.cloud.network.find_firewall_group.assert_called_once_with(
self.firewall_group_name, ignore_missing=False, **filters)
def test_delete_firewall_group_not_found(self): def test_delete_firewall_group_not_found(self):
self.register_uris([ self.register_uris([
@@ -974,14 +945,12 @@ class TestFirewallGroup(FirewallTestCase):
uri=self._make_mock_url('firewall_groups'), uri=self._make_mock_url('firewall_groups'),
json={'firewall_groups': []}) json={'firewall_groups': []})
]) ])
_log = self.cloud.log.debug
self.cloud.log.debug = Mock() with mock.patch.object(self.cloud.log, 'debug'):
self.assertFalse( self.assertFalse(
self.cloud.delete_firewall_group(self.firewall_group_name)) self.cloud.delete_firewall_group(self.firewall_group_name))
self.assert_calls() self.assert_calls()
self.cloud.log.debug.assert_called_once() self.cloud.log.debug.assert_called_once()
# restore
self.cloud.log.debug = _log
def test_get_firewall_group(self): def test_get_firewall_group(self):
returned_group = deepcopy(self.mock_returned_firewall_group) returned_group = deepcopy(self.mock_returned_firewall_group)
@@ -1123,9 +1092,6 @@ class TestFirewallGroup(FirewallTestCase):
def test_update_firewall_group_filters(self): def test_update_firewall_group_filters(self):
filters = {'project_id': self.mock_firewall_group['project_id']} filters = {'project_id': self.mock_firewall_group['project_id']}
_find = self.cloud.network.find_firewall_group
self.cloud.network.find_firewall_group = Mock(
return_value=deepcopy(self.mock_firewall_group))
params = {'description': 'updated again!'} params = {'description': 'updated again!'}
updated_group = deepcopy(self.mock_returned_firewall_group) updated_group = deepcopy(self.mock_returned_firewall_group)
self.register_uris([ self.register_uris([
@@ -1135,15 +1101,16 @@ class TestFirewallGroup(FirewallTestCase):
json={'firewall_group': updated_group}, json={'firewall_group': updated_group},
validate=dict(json={'firewall_group': params})) validate=dict(json={'firewall_group': params}))
]) ])
r = self.cloud.update_firewall_group(self.firewall_group_name, filters,
**params)
self.assertDictEqual(updated_group, r.to_dict())
self.assert_calls()
self.cloud.network.find_firewall_group.assert_called_once_with( with mock.patch.object(self.cloud.network, 'find_firewall_group',
self.firewall_group_name, ignore_missing=False, **filters) return_value=deepcopy(
# restore self.mock_firewall_group)):
self.cloud.network.find_firewall_group = _find r = self.cloud.update_firewall_group(self.firewall_group_name,
filters, **params)
self.assertDictEqual(updated_group, r.to_dict())
self.assert_calls()
self.cloud.network.find_firewall_group.assert_called_once_with(
self.firewall_group_name, ignore_missing=False, **filters)
def test_update_firewall_group_unset_policies(self): def test_update_firewall_group_unset_policies(self):
transformed_params = {'ingress_firewall_policy_id': None, transformed_params = {'ingress_firewall_policy_id': None,