Merge "Drop mox usage from FWaaS v2 panel tests"
This commit is contained in:
commit
e8e6f4a5a1
|
@ -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)
|
||||
|
|
Loading…
Reference in New Issue