diff --git a/neutron_fwaas_dashboard/dashboards/project/firewalls_v2/tests.py b/neutron_fwaas_dashboard/dashboards/project/firewalls_v2/tests.py index 0b465da..23bc3ba 100644 --- a/neutron_fwaas_dashboard/dashboards/project/firewalls_v2/tests.py +++ b/neutron_fwaas_dashboard/dashboards/project/firewalls_v2/tests.py @@ -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)