Merge "Drop mox usage from FWaaS v2 panel tests"

This commit is contained in:
Zuul 2018-04-03 14:20:25 +00:00 committed by Gerrit Code Review
commit e8e6f4a5a1
1 changed files with 356 additions and 297 deletions

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)