openstack
/
neutron-fwaas-dashboard
Code Issues Proposed changes

Drop mox usage from FWaaS v2 panel tests 

Previously expected PUT data for update tests were completely wrong.
I don't know why the previous mox unit tests pass. After switching to
mock, they no longer can pass and this commit fixes them.

This patch also clean up unnecessary mocked methods.

This is part of mox-removal community goal in Rocky.
Partial-Bug: #1753504

Change-Id: I135b1745c933deca37d17e6af250e16aabf912e7
This commit is contained in:
Akihiro Motoki 2018-03-11 09:28:40 +09:00 committed by Yushiro FURUKAWA
parent 58804c1e99
commit 6cd9d487ed
1 changed files with 356 additions and 297 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

653
neutron_fwaas_dashboard/dashboards/project/firewalls_v2/tests.py
View File

@ -13,22 +13,17 @@
# under the License.
from django.core.urlresolvers import reverse
from django import http
from mox3.mox import IsA
import mock
from openstack_dashboard.api import neutron as api_neutron
from openstack_dashboard.test import helpers
from neutron_fwaas_dashboard.api import fwaas_v2 as api_fwaas_v2
from neutron_fwaas_dashboard.test import helpers as test
class FirewallTests(test.TestCase):
class AttributeDict(dict):
def __getattr__(self, attr):
return self[attr]
def __setattr__(self, attr, value):
self[attr] = value
use_mox = False
INDEX_URL = reverse('horizon:project:firewalls_v2:index')
@ -51,48 +46,46 @@ class FirewallTests(test.TestCase):
ADDPORT_PATH = 'horizon:project:firewalls_v2:addport'
REMOVEPORT_PATH = 'horizon:project:firewalls_v2:removeport'
def set_up_expect(self):
def setup_mocks(self):
firewallgroups = self.firewall_groups_v2.list()
self.mock_firewall_list_for_tenant.return_value = firewallgroups
policies = self.fw_policies_v2.list()
self.mock_policy_list_for_tenant.return_value = policies
self.mock_rule_list_for_tenant.return_value = self.fw_rules_v2.list()
def check_mocks(self):
tenant_id = self.tenant.id
# retrieves firewallgroups
firewallgroups = self.firewall_groups_v2.list()
api_fwaas_v2.firewall_list_for_tenant(
IsA(http.HttpRequest), tenant_id).AndReturn(firewallgroups)
# retrieves policies
self.mock_firewall_list_for_tenant.assert_called_once_with(
helpers.IsHttpRequest(), tenant_id)
# TODO(amotoki): get_firewallgroupstable_data() also calls
# policy_list_for_tenant(). This needs to be clean up.
policies = self.fw_policies_v2.list()
api_fwaas_v2.policy_list_for_tenant(
IsA(http.HttpRequest), tenant_id).AndReturn(policies)
api_fwaas_v2.policy_list_for_tenant(
IsA(http.HttpRequest), tenant_id).AndReturn(policies)
self.assert_mock_multiple_calls_with_same_arguments(
self.mock_policy_list_for_tenant, 2,
mock.call(helpers.IsHttpRequest(), tenant_id))
self.mock_rule_list_for_tenant.assert_called_once_with(
helpers.IsHttpRequest(), tenant_id)
# retrieve rules
api_fwaas_v2.rule_list_for_tenant(
IsA(http.HttpRequest),
tenant_id).AndReturn(self.fw_rules_v2.list())
def setup_mocks_with_exception(self):
self.mock_rule_list_for_tenant.side_effect = self.exceptions.neutron
self.mock_policy_list_for_tenant.side_effect = self.exceptions.neutron
self.mock_firewall_list_for_tenant.side_effect = \
self.exceptions.neutron
def set_up_expect_with_exception(self):
def check_mocks_with_exception(self):
tenant_id = self.tenant.id
api_fwaas_v2.rule_list_for_tenant(
IsA(http.HttpRequest),
tenant_id).AndRaise(self.exceptions.neutron)
api_fwaas_v2.policy_list_for_tenant(
IsA(http.HttpRequest),
tenant_id).AndRaise(self.exceptions.neutron)
api_fwaas_v2.firewall_list_for_tenant(
IsA(http.HttpRequest),
tenant_id).AndRaise(self.exceptions.neutron)
self.mock_rule_list_for_tenant.assert_called_once_with(
helpers.IsHttpRequest(), tenant_id)
self.mock_policy_list_for_tenant.assert_called_once_with(
helpers.IsHttpRequest(), tenant_id)
self.mock_firewall_list_for_tenant.assert_called_once_with(
helpers.IsHttpRequest(), tenant_id)
@test.create_stubs({api_fwaas_v2: ('firewall_list_for_tenant',
'policy_list_for_tenant',
'rule_list_for_tenant',),
api_neutron: ('is_extension_supported',), })
@helpers.create_mocks({api_fwaas_v2: ('firewall_list_for_tenant',
'policy_list_for_tenant',
'rule_list_for_tenant',)})
def test_index_firewallgroups(self):
self.set_up_expect()
self.mox.ReplayAll()
self.setup_mocks()
tenant_id = self.tenant.id
@ -102,15 +95,13 @@ class FirewallTests(test.TestCase):
self.assertTemplateUsed(res, 'horizon/common/_detail_table.html')
self.assertEqual(len(res.context['table'].data),
len(self.firewall_groups_v2.list()))
self.check_mocks()
@test.create_stubs({api_fwaas_v2: ('firewall_list_for_tenant',
'policy_list_for_tenant',
'rule_list_for_tenant',),
api_neutron: ('is_extension_supported',), })
@helpers.create_mocks({api_fwaas_v2: ('firewall_list_for_tenant',
'policy_list_for_tenant',
'rule_list_for_tenant',)})
def test_index_policies(self):
self.set_up_expect()
self.mox.ReplayAll()
self.setup_mocks()
tenant_id = self.tenant.id
@ -121,15 +112,13 @@ class FirewallTests(test.TestCase):
self.assertTemplateUsed(res, 'horizon/common/_detail_table.html')
self.assertEqual(len(res.context['policiestable_table'].data),
len(self.fw_policies_v2.list()))
self.check_mocks()
@test.create_stubs({api_fwaas_v2: ('firewall_list_for_tenant',
'policy_list_for_tenant',
'rule_list_for_tenant',),
api_neutron: ('is_extension_supported',), })
@helpers.create_mocks({api_fwaas_v2: ('firewall_list_for_tenant',
'policy_list_for_tenant',
'rule_list_for_tenant',)})
def test_index_rules(self):
self.set_up_expect()
self.mox.ReplayAll()
self.setup_mocks()
tenant_id = self.tenant.id
@ -140,15 +129,13 @@ class FirewallTests(test.TestCase):
self.assertTemplateUsed(res, 'horizon/common/_detail_table.html')
self.assertEqual(len(res.context['rulestable_table'].data),
len(self.fw_rules_v2.list()))
self.check_mocks()
@test.create_stubs({api_fwaas_v2: ('firewall_list_for_tenant',
'policy_list_for_tenant',
'rule_list_for_tenant'),
api_neutron: ('is_extension_supported',), })
@helpers.create_mocks({api_fwaas_v2: ('firewall_list_for_tenant',
'policy_list_for_tenant',
'rule_list_for_tenant')})
def test_index_exception_firewallgroups(self):
self.set_up_expect_with_exception()
self.mox.ReplayAll()
self.setup_mocks_with_exception()
tenant_id = self.tenant.id
@ -158,14 +145,13 @@ class FirewallTests(test.TestCase):
self.assertTemplateUsed(res, 'horizon/common/_detail_table.html')
self.assertEqual(len(res.context['table'].data), 0)
@test.create_stubs({api_fwaas_v2: ('firewall_list_for_tenant',
'policy_list_for_tenant',
'rule_list_for_tenant'),
api_neutron: ('is_extension_supported',), })
def test_index_exception_policies(self):
self.set_up_expect_with_exception()
self.check_mocks_with_exception()
self.mox.ReplayAll()
@helpers.create_mocks({api_fwaas_v2: ('firewall_list_for_tenant',
'policy_list_for_tenant',
'rule_list_for_tenant')})
def test_index_exception_policies(self):
self.setup_mocks_with_exception()
tenant_id = self.tenant.id
@ -177,14 +163,13 @@ class FirewallTests(test.TestCase):
'horizon/common/_detail_table.html')
self.assertEqual(len(res.context['policiestable_table'].data), 0)
@test.create_stubs({api_fwaas_v2: ('firewall_list_for_tenant',
'policy_list_for_tenant',
'rule_list_for_tenant'),
api_neutron: ('is_extension_supported',), })
def test_index_exception_rules(self):
self.set_up_expect_with_exception()
self.check_mocks_with_exception()
self.mox.ReplayAll()
@helpers.create_mocks({api_fwaas_v2: ('firewall_list_for_tenant',
'policy_list_for_tenant',
'rule_list_for_tenant')})
def test_index_exception_rules(self):
self.setup_mocks_with_exception()
tenant_id = self.tenant.id
@ -195,7 +180,9 @@ class FirewallTests(test.TestCase):
self.assertTemplateUsed(res, 'horizon/common/_detail_table.html')
self.assertEqual(len(res.context['rulestable_table'].data), 0)
@test.create_stubs({api_fwaas_v2: ('rule_create',), })
self.check_mocks_with_exception()
@helpers.create_mocks({api_fwaas_v2: ('rule_create',), })
def test_add_rule_post(self):
rule1 = self.fw_rules_v2.first()
@ -212,41 +199,46 @@ class FirewallTests(test.TestCase):
'ip_version': rule1.ip_version
}
api_fwaas_v2.rule_create(
IsA(http.HttpRequest), **form_data).AndReturn(rule1)
self.mox.ReplayAll()
self.mock_rule_create.return_value = rule1
res = self.client.post(reverse(self.ADDRULE_PATH), form_data)
self.assertNoFormErrors(res)
self.assertRedirectsNoFollow(res, str(self.INDEX_URL))
@test.create_stubs({api_fwaas_v2: ('rule_create',), })
self.mock_rule_create.assert_called_once_with(
helpers.IsHttpRequest(), **form_data)
@helpers.create_mocks({api_fwaas_v2: ('rule_create',), })
def test_add_rule_post_src_None(self):
rule1 = self.fw_rules_v2.first()
form_data = {'name': rule1.name,
'description': rule1.description,
'protocol': rule1.protocol,
'action': rule1.action,
'source_ip_address': '',
'destination_ip_address': rule1.destination_ip_address,
'source_port': '',
'destination_port': rule1.destination_port,
'shared': rule1.shared,
'enabled': rule1.enabled,
'ip_version': rule1.ip_version
}
api_fwaas_v2.rule_create(
IsA(http.HttpRequest), **form_data).AndReturn(rule1)
self.mox.ReplayAll()
self.mock_rule_create.return_value = rule1
res = self.client.post(reverse(self.ADDRULE_PATH), form_data)
self.assertNoFormErrors(res)
self.assertRedirectsNoFollow(res, str(self.INDEX_URL))
@test.create_stubs({api_fwaas_v2: ('rule_create',), })
data = form_data.copy()
data['source_ip_address'] = None
data['source_port'] = None
self.mock_rule_create.assert_called_once_with(
helpers.IsHttpRequest(), **data)
@helpers.create_mocks({api_fwaas_v2: ('rule_create',), })
def test_add_rule_post_dest_None(self):
rule1 = self.fw_rules_v2.first()
form_data = {'name': rule1.name,
@ -254,22 +246,27 @@ class FirewallTests(test.TestCase):
'protocol': rule1.protocol,
'action': rule1.action,
'source_ip_address': rule1.source_ip_address,
'destination_ip_address': '',
'source_port': rule1.source_port,
'destination_port': '',
'shared': rule1.shared,
'enabled': rule1.enabled,
'ip_version': rule1.ip_version
}
api_fwaas_v2.rule_create(
IsA(http.HttpRequest), **form_data).AndReturn(rule1)
self.mox.ReplayAll()
self.mock_rule_create.return_value = rule1
res = self.client.post(reverse(self.ADDRULE_PATH), form_data)
self.assertNoFormErrors(res)
self.assertRedirectsNoFollow(res, str(self.INDEX_URL))
data = form_data.copy()
data['destination_ip_address'] = None
data['destination_port'] = None
self.mock_rule_create.assert_called_once_with(
helpers.IsHttpRequest(), **data)
def test_add_rule_post_with_error(self):
rule1 = self.fw_rules_v2.first()
@ -286,14 +283,12 @@ class FirewallTests(test.TestCase):
'ip_version': 6
}
self.mox.ReplayAll()
res = self.client.post(reverse(self.ADDRULE_PATH), form_data)
self.assertFormErrors(res, 3)
@test.create_stubs({api_fwaas_v2: ('policy_create',
'rule_list_for_tenant'), })
@helpers.create_mocks({api_fwaas_v2: ('policy_create',
'rule_list_for_tenant'), })
def test_add_policy_post(self):
policy = self.fw_policies_v2.first()
rules = self.fw_rules_v2.list()
@ -319,20 +314,20 @@ class FirewallTests(test.TestCase):
for rule in rules:
if rule.id in policy.firewall_rules:
rule.firewall_policy_id = rule.policy = None
api_fwaas_v2.rule_list_for_tenant(
IsA(http.HttpRequest), tenant_id).AndReturn(rules)
api_fwaas_v2.policy_create(
IsA(http.HttpRequest), **form_data).AndReturn(policy)
self.mox.ReplayAll()
self.mock_rule_list_for_tenant.return_value = rules
self.mock_policy_create.return_value = policy
res = self.client.post(reverse(self.ADDPOLICY_PATH), post_data)
self.assertNoFormErrors(res)
self.assertRedirectsNoFollow(res, str(self.INDEX_URL))
@test.create_stubs({api_fwaas_v2: ('policy_create',
'rule_list_for_tenant'), })
self.mock_rule_list_for_tenant.assert_called_once_with(
helpers.IsHttpRequest(), tenant_id)
self.mock_policy_create.assert_called_once_with(
helpers.IsHttpRequest(), **form_data)
@helpers.create_mocks({api_fwaas_v2: ('rule_list_for_tenant',)})
def test_add_policy_post_with_error(self):
policy = self.fw_policies_v2.first()
rules = self.fw_rules_v2.list()
@ -342,49 +337,54 @@ class FirewallTests(test.TestCase):
'shared': policy.shared,
'audited': policy.audited
}
api_fwaas_v2.rule_list_for_tenant(
IsA(http.HttpRequest), tenant_id).AndReturn(rules)
self.mox.ReplayAll()
self.mock_rule_list_for_tenant.return_value = rules
res = self.client.post(reverse(self.ADDPOLICY_PATH), form_data)
self.assertFormErrors(res, 1)
@test.create_stubs({api_fwaas_v2: ('firewall_group_create',
'policy_list_for_tenant',
'fwg_port_list_for_tenant',),
api_neutron: ('is_extension_supported',), })
self.mock_rule_list_for_tenant.assert_called_once_with(
helpers.IsHttpRequest(), tenant_id)
@helpers.create_mocks({api_fwaas_v2: ('firewall_group_create',
'policy_list_for_tenant',
'fwg_port_list_for_tenant',)})
def test_add_firewall_group_post(self):
firewall_group = self.firewall_groups_v2.first()
fwg = self.firewall_groups_v2.first()
policies = self.fw_policies_v2.list()
tenant_id = self.tenant.id
form_data = {'name': firewall_group.name,
'description': firewall_group.description,
'ingress_firewall_policy_id':
firewall_group.ingress_firewall_policy_id,
'egress_firewall_policy_id':
firewall_group.egress_firewall_policy_id,
'admin_state_up': firewall_group.admin_state_up
}
api_fwaas_v2.policy_list_for_tenant(
IsA(http.HttpRequest), tenant_id).AndReturn(policies)
api_fwaas_v2.fwg_port_list_for_tenant(
IsA(http.HttpRequest), tenant_id).AndReturn([])
api_fwaas_v2.firewall_group_create(
IsA(http.HttpRequest), **form_data).AndReturn(firewall_group)
self.mox.ReplayAll()
form_data = {
'name': fwg.name,
'description': fwg.description,
'ingress_firewall_policy_id': fwg.ingress_firewall_policy_id,
'egress_firewall_policy_id': fwg.egress_firewall_policy_id,
'admin_state_up': fwg.admin_state_up,
'shared': False,
'port': [],
}
self.mock_policy_list_for_tenant.return_value = policies
self.mock_fwg_port_list_for_tenant.return_value = []
self.mock_firewall_group_create.return_value = fwg
res = self.client.post(reverse(self.ADDFIREWALLGROUP_PATH), form_data)
self.assertNoFormErrors(res)
self.assertRedirectsNoFollow(res, str(self.INDEX_URL))
self.mock_policy_list_for_tenant.assert_called_once_with(
helpers.IsHttpRequest(), tenant_id)
self.mock_fwg_port_list_for_tenant.assert_called_once_with(
helpers.IsHttpRequest(), tenant_id)
data = form_data.copy()
data['ports'] = data.pop('port')
self.mock_firewall_group_create.assert_called_once_with(
helpers.IsHttpRequest(), **data)
# TODO(SarathMekala) : Fix this test.
# @test.create_stubs({api_fwaas_v2: ('firewall_group_create',
# 'policy_list_for_tenant',
# 'fwg_port_list_for_tenant',),
# api_neutron: ('is_extension_supported',), })
# @helpers.create_mocks({api_fwaas_v2: ('firewall_group_create',
# 'policy_list_for_tenant',
# 'fwg_port_list_for_tenant',)})
# def test_add_firewall_post_with_error(self):
# firewall_groups = self.firewall_groups_v2.first()
# policies = self.fw_policies_v2.list()
@ -393,49 +393,55 @@ class FirewallTests(test.TestCase):
# 'description': firewall_groups.description,
# 'admin_state_up': False
# }
# api_fwaas_v2.policy_list_for_tenant(
# IsA(http.HttpRequest), tenant_id).AndReturn(policies)
# self.mock_policy_list_for_tenant(
# helpers.IsHttpRequest(), tenant_id).AndReturn(policies)
#
# self.mox.ReplayAll()
# res = self.client.post(reverse(self.ADDFIREWALLGROUP_PATH), form_data)
#
# self.assertFormErrors(res, 1)
@test.create_stubs({api_fwaas_v2: ('rule_get',)})
@helpers.create_mocks({api_fwaas_v2: ('rule_get',)})
def test_update_rule_get(self):
rule = self.fw_rules_v2.first()
api_fwaas_v2.rule_get(IsA(http.HttpRequest), rule.id).AndReturn(rule)
self.mox.ReplayAll()
self.mock_rule_get.return_value = rule
res = self.client.get(reverse(self.UPDATERULE_PATH, args=(rule.id,)))
self.assertTemplateUsed(res, 'project/firewalls_v2/updaterule.html')
self.mock_rule_get.assert_called_once_with(helpers.IsHttpRequest(),
rule.id)
@test.create_stubs({api_fwaas_v2: ('rule_get', 'rule_update')})
@helpers.create_mocks({api_fwaas_v2: ('rule_get', 'rule_update')})
def test_update_rule_post(self):
rule = self.fw_rules_v2.first()
api_fwaas_v2.rule_get(IsA(http.HttpRequest), rule.id).AndReturn(rule)
self.mock_rule_get.return_value = rule
self.mock_rule_update.return_value = rule
data = {'name': 'new name',
'description': 'new desc',
'protocol': 'icmp',
'action': 'allow',
'shared': False,
'enabled': True,
'ip_version': rule.ip_version,
'source_ip_address': rule.source_ip_address,
'destination_ip_address': None,
'source_port': None,
'destination_port': rule.destination_port,
}
api_fwaas_v2.rule_update(IsA(http.HttpRequest), rule.id, **data)\
.AndReturn(rule)
self.mox.ReplayAll()
data = {
'name': 'new name',
'description': 'new desc',
'protocol': 'icmp',
'action': 'allow',
'shared': False,
'enabled': True,
'ip_version': rule.ip_version,
'source_ip_address': rule.source_ip_address,
'destination_ip_address': None,
'source_port': None,
'destination_port': rule.destination_port,
}
expected_put_data = {
'name': 'new name',
'description': 'new desc',
'protocol': 'icmp',
'action': 'allow',
'shared': False,
'destination_ip_address': None,
'source_port': None,
}
form_data = data.copy()
form_data['destination_ip_address'] = ''
@ -447,30 +453,41 @@ class FirewallTests(test.TestCase):
self.assertNoFormErrors(res)
self.assertRedirectsNoFollow(res, str(self.INDEX_URL))
@test.create_stubs({api_fwaas_v2: ('rule_get', 'rule_update')})
self.mock_rule_get.assert_called_once_with(helpers.IsHttpRequest(),
rule.id)
self.mock_rule_update.assert_called_once_with(
helpers.IsHttpRequest(), rule.id, **expected_put_data)
@helpers.create_mocks({api_fwaas_v2: ('rule_get', 'rule_update')})
def test_update_protocol_any_rule_post(self):
# protocol any means protocol == None in neutron context.
rule = self.fw_rules_v2.get(protocol=None)
api_fwaas_v2.rule_get(IsA(http.HttpRequest), rule.id).AndReturn(rule)
self.mock_rule_get.return_value = rule
self.mock_rule_update.return_value = rule
data = {'name': 'new name',
'description': 'new desc',
'protocol': 'icmp',
'action': 'allow',
'shared': False,
'enabled': True,
'ip_version': rule.ip_version,
'source_ip_address': rule.source_ip_address,
'destination_ip_address': None,
'source_port': None,
'destination_port': rule.destination_port,
}
api_fwaas_v2.rule_update(IsA(http.HttpRequest), rule.id, **data)\
.AndReturn(rule)
self.mox.ReplayAll()
data = {
'name': 'new name',
'description': 'new desc',
'protocol': 'icmp',
'action': 'allow',
'shared': False,
'enabled': True,
'ip_version': rule.ip_version,
'source_ip_address': rule.source_ip_address,
'destination_ip_address': None,
'source_port': None,
'destination_port': rule.destination_port,
}
expected_put_data = {
'name': 'new name',
'description': 'new desc',
'protocol': 'icmp',
'action': 'allow',
'shared': False,
'destination_ip_address': None,
'source_port': None,
}
form_data = data.copy()
form_data['destination_ip_address'] = ''
@ -482,33 +499,41 @@ class FirewallTests(test.TestCase):
self.assertNoFormErrors(res)
self.assertRedirectsNoFollow(res, str(self.INDEX_URL))
@test.create_stubs({api_fwaas_v2: ('rule_get', 'rule_update')})
self.mock_rule_get.assert_called_once_with(helpers.IsHttpRequest(),
rule.id)
self.mock_rule_update.assert_called_once_with(
helpers.IsHttpRequest(), rule.id, **expected_put_data)
@helpers.create_mocks({api_fwaas_v2: ('rule_get', 'rule_update')})
def test_update_rule_protocol_to_ANY_post(self):
rule = self.fw_rules_v2.first()
api_fwaas_v2.rule_get(IsA(http.HttpRequest), rule.id).AndReturn(rule)
self.mock_rule_get.return_value = rule
self.mock_rule_update.return_value = rule
data = {'name': 'new name',
'description': 'new desc',
'protocol': None,
'action': 'allow',
'shared': False,
'enabled': True,
'ip_version': rule.ip_version,
'source_ip_address': rule.source_ip_address,
'destination_ip_address': None,
'source_port': None,
'destination_port': rule.destination_port,
}
api_fwaas_v2.rule_update(IsA(http.HttpRequest), rule.id, **data)\
.AndReturn(rule)
form_data = {
'name': 'new name',
'description': 'new desc',
'protocol': 'any',
'action': 'allow',
'shared': False,
'enabled': True,
'ip_version': rule.ip_version,
'source_ip_address': rule.source_ip_address,
'destination_ip_address': '',
'source_port': '',
'destination_port': rule.destination_port,
}
self.mox.ReplayAll()
form_data = data.copy()
form_data['destination_ip_address'] = ''
form_data['source_port'] = ''
form_data['protocol'] = 'any'
expected_put_data = {
'name': 'new name',
'description': 'new desc',
'protocol': None,
'action': 'allow',
'shared': False,
'destination_ip_address': None,
'source_port': None,
}
res = self.client.post(
reverse(self.UPDATERULE_PATH, args=(rule.id,)), form_data)
@ -516,38 +541,43 @@ class FirewallTests(test.TestCase):
self.assertNoFormErrors(res)
self.assertRedirectsNoFollow(res, str(self.INDEX_URL))
@test.create_stubs({api_fwaas_v2: ('policy_get',)})
self.mock_rule_get.assert_called_once_with(helpers.IsHttpRequest(),
rule.id)
self.mock_rule_update.assert_called_once_with(
helpers.IsHttpRequest(), rule.id, **expected_put_data)
@helpers.create_mocks({api_fwaas_v2: ('policy_get',)})
def test_update_policy_get(self):
policy = self.fw_policies_v2.first()
api_fwaas_v2.policy_get(IsA(http.HttpRequest),
policy.id).AndReturn(policy)
self.mox.ReplayAll()
self.mock_policy_get.return_value = policy
res = self.client.get(
reverse(self.UPDATEPOLICY_PATH, args=(policy.id,)))
self.assertTemplateUsed(res, 'project/firewalls_v2/updatepolicy.html')
@test.create_stubs({api_fwaas_v2: ('policy_get', 'policy_update',
'rule_list_for_tenant')})
self.mock_policy_get.assert_called_once_with(helpers.IsHttpRequest(),
policy.id)
@helpers.create_mocks({api_fwaas_v2: ('policy_get', 'policy_update')})
def test_update_policy_post(self):
policy = self.fw_policies_v2.first()
api_fwaas_v2.policy_get(IsA(http.HttpRequest),
policy.id).AndReturn(policy)
self.mock_policy_get.return_value = policy
self.mock_policy_update.return_value = policy
data = {'name': 'new name',
'description': 'new desc',
'shared': True,
'audited': False
}
api_fwaas_v2.policy_update(IsA(http.HttpRequest), policy.id, **data)\
.AndReturn(policy)
self.mox.ReplayAll()
data = {
'name': 'new name',
'description': 'new desc',
'shared': True,
'audited': False
}
expected_put_data = {
'name': 'new name',
'description': 'new desc',
'audited': False,
}
res = self.client.post(
reverse(self.UPDATEPOLICY_PATH, args=(policy.id,)), data)
@ -555,20 +585,20 @@ class FirewallTests(test.TestCase):
self.assertNoFormErrors(res)
self.assertRedirectsNoFollow(res, str(self.INDEX_URL))
@test.create_stubs({api_fwaas_v2: ('firewall_get',
'policy_list_for_tenant')})
self.mock_policy_get.assert_called_once_with(helpers.IsHttpRequest(),
policy.id)
self.mock_policy_update.assert_called_once_with(
helpers.IsHttpRequest(), policy.id, **expected_put_data)
@helpers.create_mocks({api_fwaas_v2: ('firewall_get',
'policy_list_for_tenant')})
def test_update_firewall_group_get(self):
firewall_group = self.firewall_groups_v2.first()
policies = self.fw_policies_v2.list()
tenant_id = self.tenant.id
api_fwaas_v2.policy_list_for_tenant(
IsA(http.HttpRequest), tenant_id).AndReturn(policies)
api_fwaas_v2.firewall_get(IsA(http.HttpRequest),
firewall_group.id).AndReturn(firewall_group)
self.mox.ReplayAll()
self.mock_policy_list_for_tenant.return_value = policies
self.mock_firewall_get.return_value = firewall_group
res = self.client.get(
reverse(self.UPDATEFIREWALLGROUP_PATH, args=(firewall_group.id,)))
@ -576,45 +606,59 @@ class FirewallTests(test.TestCase):
self.assertTemplateUsed(res,
'project/firewalls_v2/updatefirewall.html')
@test.create_stubs({api_fwaas_v2: ('firewall_get',
'policy_list_for_tenant',
'firewall_update')})
self.mock_policy_list_for_tenant.assert_called_once_with(
helpers.IsHttpRequest(), tenant_id)
self.mock_firewall_get.assert_called_once_with(
helpers.IsHttpRequest(), firewall_group.id)
@helpers.create_mocks({api_fwaas_v2: ('firewall_get',
'policy_list_for_tenant',
'firewall_update')})
def test_update_firewall_post(self):
firewall_group = self.firewall_groups_v2.first()
fwg = self.firewall_groups_v2.first()
tenant_id = self.tenant.id
api_fwaas_v2.firewall_get(IsA(http.HttpRequest),
firewall_group.id).AndReturn(firewall_group)
data = {'name': 'new name',
'description': 'new desc',
'ingress_firewall_policy_id':
firewall_group.ingress_firewall_policy_id,
'admin_state_up': False
}
policies = self.fw_policies_v2.list()
api_fwaas_v2.policy_list_for_tenant(
IsA(http.HttpRequest), tenant_id).AndReturn(policies)
api_fwaas_v2.firewall_update(
IsA(http.HttpRequest), firewall_group.id, **data)\
.AndReturn(firewall_group)
data = {
'name': 'new name',
'description': 'new desc',
'ingress_firewall_policy_id': fwg.ingress_firewall_policy_id,
'egress_firewall_policy_id': '',
'admin_state_up': False,
'shared': False,
}
self.mox.ReplayAll()
expected_put_data = {
'name': 'new name',
'description': 'new desc',
'egress_firewall_policy_id': None,
'admin_state_up': False,
}
self.mock_firewall_get.return_value = fwg
self.mock_policy_list_for_tenant.return_value = policies
self.mock_firewall_update.return_value = fwg
res = self.client.post(
reverse(
self.UPDATEFIREWALLGROUP_PATH,
args=(
firewall_group.id,
fwg.id,
)),
data)
self.assertNoFormErrors(res)
self.assertRedirectsNoFollow(res, str(self.INDEX_URL))
@test.create_stubs({api_fwaas_v2: ('policy_get', 'policy_insert_rule',
'rule_list_for_tenant', 'rule_get')})
self.mock_firewall_get.assert_called_once_with(
helpers.IsHttpRequest(), fwg.id)
self.mock_policy_list_for_tenant.assert_called_once_with(
helpers.IsHttpRequest(), tenant_id)
self.mock_firewall_update.assert_called_once_with(
helpers.IsHttpRequest(), fwg.id, **expected_put_data)
@helpers.create_mocks({api_fwaas_v2: ('policy_get', 'policy_insert_rule',
'rule_list_for_tenant', 'rule_get')})
def test_policy_insert_rule(self):
policy = self.fw_policies_v2.first()
tenant_id = self.tenant.id
@ -626,21 +670,15 @@ class FirewallTests(test.TestCase):
'insert_before': rules[1].id,
'insert_after': rules[0].id}
api_fwaas_v2.policy_get(IsA(http.HttpRequest),
policy.id).AndReturn(policy)
self.mock_policy_get.return_value = policy
policy.firewall_rules = [rules[0].id,
new_rule_id,
rules[1].id]
api_fwaas_v2.rule_list_for_tenant(
IsA(http.HttpRequest), tenant_id).AndReturn(rules)
api_fwaas_v2.rule_get(
IsA(http.HttpRequest), new_rule_id).AndReturn(rules[2])
api_fwaas_v2.policy_insert_rule(
IsA(http.HttpRequest), policy.id, **data) .AndReturn(policy)
self.mox.ReplayAll()
self.mock_rule_list_for_tenant.return_value = rules
self.mock_rule_get.return_value = rules[2]
self.mock_policy_insert_rule.return_value = policy
res = self.client.post(
reverse(self.INSERTRULE_PATH, args=(policy.id,)), data)
@ -648,8 +686,17 @@ class FirewallTests(test.TestCase):
self.assertNoFormErrors(res)
self.assertRedirectsNoFollow(res, str(self.INDEX_URL))
@test.create_stubs({api_fwaas_v2: ('policy_get', 'policy_remove_rule',
'rule_list_for_tenant', 'rule_get')})
self.mock_policy_get.assert_called_once_with(helpers.IsHttpRequest(),
policy.id)
self.mock_rule_list_for_tenant.assert_called_once_with(
helpers.IsHttpRequest(), tenant_id)
self.mock_rule_get.assert_called_once_with(
helpers.IsHttpRequest(), new_rule_id)
self.mock_policy_insert_rule.assert_called_once_with(
helpers.IsHttpRequest(), policy.id, **data)
@helpers.create_mocks({api_fwaas_v2: ('policy_get', 'policy_remove_rule',
'rule_list_for_tenant', 'rule_get')})
def test_policy_remove_rule(self):
policy = self.fw_policies_v2.first()
tenant_id = self.tenant.id
@ -669,16 +716,10 @@ class FirewallTests(test.TestCase):
'shared': True}
after_remove_policy = api_fwaas_v2.Policy(after_remove_policy_dict)
api_fwaas_v2.policy_get(IsA(http.HttpRequest),
policy.id).AndReturn(policy)
api_fwaas_v2.rule_list_for_tenant(
IsA(http.HttpRequest), tenant_id).AndReturn(rules)
api_fwaas_v2.rule_get(
IsA(http.HttpRequest), remove_rule_id).AndReturn(rules[0])
api_fwaas_v2.policy_remove_rule(IsA(http.HttpRequest), policy.id, **data)\
.AndReturn(after_remove_policy)
self.mox.ReplayAll()
self.mock_policy_get.return_value = policy
self.mock_rule_list_for_tenant.return_value = rules
self.mock_rule_get.return_value = rules[0]
self.mock_policy_remove_rule.return_value = after_remove_policy
res = self.client.post(
reverse(self.REMOVERULE_PATH, args=(policy.id,)), data)
@ -686,51 +727,62 @@ class FirewallTests(test.TestCase):
self.assertNoFormErrors(res)
self.assertRedirectsNoFollow(res, str(self.INDEX_URL))
@test.create_stubs({api_fwaas_v2: ('rule_list_for_tenant',
'rule_delete'),
api_neutron: ('is_extension_supported',)})
self.mock_policy_get.assert_called_once_with(helpers.IsHttpRequest(),
policy.id)
self.mock_rule_list_for_tenant.assert_called_once_with(
helpers.IsHttpRequest(), tenant_id)
self.mock_rule_get.assert_called_once_with(
helpers.IsHttpRequest(), remove_rule_id)
self.mock_policy_remove_rule.assert_called_once_with(
helpers.IsHttpRequest(), policy.id, **data)
@helpers.create_mocks({api_fwaas_v2: ('rule_list_for_tenant',
'rule_delete')})
def test_delete_rule(self):
rule = self.fw_rules_v2.list()[2]
api_fwaas_v2.rule_list_for_tenant(
IsA(http.HttpRequest),
self.tenant.id).AndReturn(self.fw_rules_v2.list())
api_fwaas_v2.rule_delete(IsA(http.HttpRequest), rule.id)
self.mox.ReplayAll()
self.mock_rule_list_for_tenant.return_value = self.fw_rules_v2.list()
self.mock_rule_delete.return_value = None
form_data = {"action": "rulestable__deleterule__%s" % rule.id}
res = self.client.post(self.INDEX_URL, form_data)
self.assertNoFormErrors(res)
@test.create_stubs({api_fwaas_v2: ('policy_list_for_tenant',
'policy_delete'),
api_neutron: ('is_extension_supported',)})
self.mock_rule_list_for_tenant.assert_called_once_with(
helpers.IsHttpRequest(), self.tenant.id)
self.mock_rule_delete.assert_called_once_with(
helpers.IsHttpRequest(), rule.id)
@helpers.create_mocks({api_fwaas_v2: ('policy_list_for_tenant',
'policy_delete')})
def test_delete_policy(self):
policy = self.fw_policies_v2.first()
api_fwaas_v2.policy_list_for_tenant(
IsA(http.HttpRequest),
self.tenant.id).AndReturn(self.fw_policies_v2.list())
api_fwaas_v2.policy_delete(IsA(http.HttpRequest), policy.id)
self.mox.ReplayAll()
self.mock_policy_list_for_tenant.return_value = \
self.fw_policies_v2.list()
self.mock_policy_delete.return_value = None
form_data = {"action": "policiestable__deletepolicy__%s" % policy.id}
res = self.client.post(self.INDEX_URL, form_data)
self.assertNoFormErrors(res)
@test.create_stubs({api_fwaas_v2: ('firewall_list_for_tenant',
'policy_list_for_tenant',
'firewall_delete',),
api_neutron: ('is_extension_supported',)})
self.mock_policy_list_for_tenant.assert_called_once_with(
helpers.IsHttpRequest(), self.tenant.id)
self.mock_policy_delete.assert_called_once_with(
helpers.IsHttpRequest(), policy.id)
@helpers.create_mocks({api_fwaas_v2: ('firewall_list_for_tenant',
'policy_list_for_tenant',
'firewall_delete',)})
def test_delete_firewall_group(self):
fwl = self.firewall_groups_v2.first()
api_fwaas_v2.firewall_list_for_tenant(
IsA(http.HttpRequest), self.tenant.id).AndReturn([fwl])
api_fwaas_v2.policy_list_for_tenant(
IsA(http.HttpRequest),
self.tenant.id).AndReturn(self.fw_policies_v2.list())
api_fwaas_v2.firewall_delete(IsA(http.HttpRequest), fwl.id)
self.mox.ReplayAll()
self.mock_firewall_list_for_tenant.return_value = [fwl]
self.mock_policy_list_for_tenant.return_value = \
self.fw_policies_v2.list()
self.mock_firewall_delete.return_value = None
form_data = {
"action": "FirewallGroupsTable__deletefirewallgroup__%s" %
@ -738,3 +790,10 @@ class FirewallTests(test.TestCase):
res = self.client.post(self.INDEX_URL, form_data)
self.assertNoFormErrors(res)
self.mock_firewall_list_for_tenant.assert_called_once_with(
helpers.IsHttpRequest(), self.tenant.id)
self.mock_policy_list_for_tenant.assert_called_once_with(
helpers.IsHttpRequest(), self.tenant.id)
self.mock_firewall_delete.assert_called_once_with(
helpers.IsHttpRequest(), fwl.id)