diff --git a/heat/tests/openstack/neutron/test_neutron_metering.py b/heat/tests/openstack/neutron/test_neutron_metering.py index e59a076be3..997c629a20 100644 --- a/heat/tests/openstack/neutron/test_neutron_metering.py +++ b/heat/tests/openstack/neutron/test_neutron_metering.py @@ -11,9 +11,11 @@ # License for the specific language governing permissions and limitations # under the License. +import mock +import six + from neutronclient.common import exceptions from neutronclient.v2_0 import client as neutronclient -import six from heat.common import exception from heat.common import template_format @@ -48,24 +50,13 @@ class MeteringLabelTest(common.HeatTestCase): def setUp(self): super(MeteringLabelTest, self).setUp() - self.m.StubOutWithMock(neutronclient.Client, 'create_metering_label') - self.m.StubOutWithMock(neutronclient.Client, 'delete_metering_label') - self.m.StubOutWithMock(neutronclient.Client, 'show_metering_label') - self.m.StubOutWithMock(neutronclient.Client, - 'create_metering_label_rule') - self.m.StubOutWithMock(neutronclient.Client, - 'delete_metering_label_rule') - self.m.StubOutWithMock(neutronclient.Client, - 'show_metering_label_rule') + self.mockclient = mock.Mock(spec=neutronclient.Client) + self.patchobject(neutronclient, 'Client', return_value=self.mockclient) def create_metering_label(self): - neutronclient.Client.create_metering_label({ - 'metering_label': { - 'name': 'TestLabel', - 'description': 'Description of TestLabel', - 'shared': True} - }).AndReturn({'metering_label': {'id': '1234'}}) - + self.mockclient.create_metering_label.return_value = { + 'metering_label': {'id': '1234'} + } snippet = template_format.parse(metering_template) self.stack = utils.parse_stack(snippet) resource_defns = self.stack.t.resource_definitions(self.stack) @@ -74,19 +65,21 @@ class MeteringLabelTest(common.HeatTestCase): def test_create(self): rsrc = self.create_metering_label() - self.m.ReplayAll() + scheduler.TaskRunner(rsrc.create)() self.assertEqual((rsrc.CREATE, rsrc.COMPLETE), rsrc.state) - self.m.VerifyAll() - def test_create_failed(self): - neutronclient.Client.create_metering_label({ + self.mockclient.create_metering_label.assert_called_once_with({ 'metering_label': { 'name': 'TestLabel', 'description': 'Description of TestLabel', - 'shared': True} - }).AndRaise(exceptions.NeutronClientException()) - self.m.ReplayAll() + 'shared': True + } + }) + + def test_create_failed(self): + self.mockclient.create_metering_label.side_effect = ( + exceptions.NeutronClientException()) snippet = template_format.parse(metering_template) stack = utils.parse_stack(snippet) @@ -100,37 +93,59 @@ class MeteringLabelTest(common.HeatTestCase): 'An unknown exception occurred.', six.text_type(error)) self.assertEqual((rsrc.CREATE, rsrc.FAILED), rsrc.state) - self.m.VerifyAll() + + self.mockclient.create_metering_label.assert_called_once_with({ + 'metering_label': { + 'name': 'TestLabel', + 'description': 'Description of TestLabel', + 'shared': True + } + }) def test_delete(self): - neutronclient.Client.delete_metering_label('1234') - neutronclient.Client.show_metering_label('1234').AndRaise( + rsrc = self.create_metering_label() + self.mockclient.delete_metering_label.return_value = None + self.mockclient.show_metering_label.side_effect = ( exceptions.NeutronClientException(status_code=404)) - rsrc = self.create_metering_label() - self.m.ReplayAll() scheduler.TaskRunner(rsrc.create)() scheduler.TaskRunner(rsrc.delete)() self.assertEqual((rsrc.DELETE, rsrc.COMPLETE), rsrc.state) - self.m.VerifyAll() + + self.mockclient.create_metering_label.assert_called_once_with({ + 'metering_label': { + 'name': 'TestLabel', + 'description': 'Description of TestLabel', + 'shared': True + } + }) + self.mockclient.delete_metering_label.assert_called_once_with('1234') + self.mockclient.show_metering_label.assert_called_once_with('1234') def test_delete_already_gone(self): - neutronclient.Client.delete_metering_label('1234').AndRaise( + rsrc = self.create_metering_label() + self.mockclient.delete_metering_label.side_effect = ( exceptions.NeutronClientException(status_code=404)) - rsrc = self.create_metering_label() - self.m.ReplayAll() scheduler.TaskRunner(rsrc.create)() scheduler.TaskRunner(rsrc.delete)() self.assertEqual((rsrc.DELETE, rsrc.COMPLETE), rsrc.state) - self.m.VerifyAll() + + self.mockclient.create_metering_label.assert_called_once_with({ + 'metering_label': { + 'name': 'TestLabel', + 'description': 'Description of TestLabel', + 'shared': True + } + }) + self.mockclient.delete_metering_label.assert_called_once_with('1234') + self.mockclient.show_metering_label.assert_not_called() def test_delete_failed(self): - neutronclient.Client.delete_metering_label('1234').AndRaise( + rsrc = self.create_metering_label() + self.mockclient.delete_metering_label.side_effect = ( exceptions.NeutronClientException(status_code=400)) - rsrc = self.create_metering_label() - self.m.ReplayAll() scheduler.TaskRunner(rsrc.create)() error = self.assertRaises(exception.ResourceFailure, scheduler.TaskRunner(rsrc.delete)) @@ -139,47 +154,53 @@ class MeteringLabelTest(common.HeatTestCase): 'An unknown exception occurred.', six.text_type(error)) self.assertEqual((rsrc.DELETE, rsrc.FAILED), rsrc.state) - self.m.VerifyAll() + + self.mockclient.create_metering_label.assert_called_once_with({ + 'metering_label': { + 'name': 'TestLabel', + 'description': 'Description of TestLabel', + 'shared': True + } + }) + self.mockclient.delete_metering_label.assert_called_once_with('1234') def test_attribute(self): rsrc = self.create_metering_label() - neutronclient.Client.show_metering_label('1234').MultipleTimes( - ).AndReturn( - {'metering_label': - {'name': 'TestLabel', - 'description': 'Description of TestLabel', - 'shared': True}}) - self.m.ReplayAll() + self.mockclient.show_metering_label.return_value = { + 'metering_label': { + 'name': 'TestLabel', + 'description': 'Description of TestLabel', + 'shared': True + } + } + scheduler.TaskRunner(rsrc.create)() self.assertEqual('TestLabel', rsrc.FnGetAtt('name')) self.assertEqual('Description of TestLabel', rsrc.FnGetAtt('description')) self.assertTrue(rsrc.FnGetAtt('shared')) - self.m.VerifyAll() + + self.mockclient.create_metering_label.assert_called_once_with({ + 'metering_label': { + 'name': 'TestLabel', + 'description': 'Description of TestLabel', + 'shared': True + } + }) + self.mockclient.show_metering_label.assert_called_with('1234') class MeteringRuleTest(common.HeatTestCase): def setUp(self): super(MeteringRuleTest, self).setUp() - self.m.StubOutWithMock(neutronclient.Client, 'create_metering_label') - self.m.StubOutWithMock(neutronclient.Client, 'delete_metering_label') - self.m.StubOutWithMock(neutronclient.Client, 'show_metering_label') - self.m.StubOutWithMock(neutronclient.Client, - 'create_metering_label_rule') - self.m.StubOutWithMock(neutronclient.Client, - 'delete_metering_label_rule') - self.m.StubOutWithMock(neutronclient.Client, - 'show_metering_label_rule') + self.mockclient = mock.Mock(spec=neutronclient.Client) + self.patchobject(neutronclient, 'Client', return_value=self.mockclient) def create_metering_label_rule(self): - neutronclient.Client.create_metering_label_rule({ - 'metering_label_rule': { - 'metering_label_id': '1234', - 'remote_ip_prefix': '10.0.3.0/24', - 'direction': 'ingress', - 'excluded': False} - }).AndReturn({'metering_label_rule': {'id': '5678'}}) + self.mockclient.create_metering_label_rule.return_value = { + 'metering_label_rule': {'id': '5678'} + } snippet = template_format.parse(metering_template) self.stack = utils.parse_stack(snippet) @@ -191,20 +212,22 @@ class MeteringRuleTest(common.HeatTestCase): def test_create(self): rsrc = self.create_metering_label_rule() - self.m.ReplayAll() + scheduler.TaskRunner(rsrc.create)() self.assertEqual((rsrc.CREATE, rsrc.COMPLETE), rsrc.state) - self.m.VerifyAll() - def test_create_failed(self): - neutronclient.Client.create_metering_label_rule({ + self.mockclient.create_metering_label_rule.assert_called_once_with({ 'metering_label_rule': { 'metering_label_id': '1234', 'remote_ip_prefix': '10.0.3.0/24', 'direction': 'ingress', - 'excluded': False} - }).AndRaise(exceptions.NeutronClientException()) - self.m.ReplayAll() + 'excluded': False + } + }) + + def test_create_failed(self): + self.mockclient.create_metering_label_rule.side_effect = ( + exceptions.NeutronClientException()) snippet = template_format.parse(metering_template) stack = utils.parse_stack(snippet) @@ -220,37 +243,65 @@ class MeteringRuleTest(common.HeatTestCase): 'An unknown exception occurred.', six.text_type(error)) self.assertEqual((rsrc.CREATE, rsrc.FAILED), rsrc.state) - self.m.VerifyAll() + + self.mockclient.create_metering_label_rule.assert_called_once_with({ + 'metering_label_rule': { + 'metering_label_id': '1234', + 'remote_ip_prefix': '10.0.3.0/24', + 'direction': 'ingress', + 'excluded': False + } + }) def test_delete(self): - neutronclient.Client.delete_metering_label_rule('5678') - neutronclient.Client.show_metering_label_rule('5678').AndRaise( + rsrc = self.create_metering_label_rule() + self.mockclient.delete_metering_label_rule.return_value = None + self.mockclient.show_metering_label_rule.side_effect = ( exceptions.NeutronClientException(status_code=404)) - rsrc = self.create_metering_label_rule() - self.m.ReplayAll() scheduler.TaskRunner(rsrc.create)() scheduler.TaskRunner(rsrc.delete)() self.assertEqual((rsrc.DELETE, rsrc.COMPLETE), rsrc.state) - self.m.VerifyAll() + + self.mockclient.create_metering_label_rule.assert_called_once_with({ + 'metering_label_rule': { + 'metering_label_id': '1234', + 'remote_ip_prefix': '10.0.3.0/24', + 'direction': 'ingress', + 'excluded': False + } + }) + self.mockclient.delete_metering_label_rule.assert_called_once_with( + '5678') + self.mockclient.show_metering_label_rule.assert_called_once_with( + '5678') def test_delete_already_gone(self): - neutronclient.Client.delete_metering_label_rule('5678').AndRaise( + rsrc = self.create_metering_label_rule() + self.mockclient.delete_metering_label_rule.side_effect = ( exceptions.NeutronClientException(status_code=404)) - rsrc = self.create_metering_label_rule() - self.m.ReplayAll() scheduler.TaskRunner(rsrc.create)() scheduler.TaskRunner(rsrc.delete)() self.assertEqual((rsrc.DELETE, rsrc.COMPLETE), rsrc.state) - self.m.VerifyAll() + + self.mockclient.create_metering_label_rule.assert_called_once_with({ + 'metering_label_rule': { + 'metering_label_id': '1234', + 'remote_ip_prefix': '10.0.3.0/24', + 'direction': 'ingress', + 'excluded': False + } + }) + self.mockclient.delete_metering_label_rule.assert_called_once_with( + '5678') + self.mockclient.show_metering_label_rule.assert_not_called() def test_delete_failed(self): - neutronclient.Client.delete_metering_label_rule('5678').AndRaise( + rsrc = self.create_metering_label_rule() + self.mockclient.delete_metering_label_rule.side_effect = ( exceptions.NeutronClientException(status_code=400)) - rsrc = self.create_metering_label_rule() - self.m.ReplayAll() scheduler.TaskRunner(rsrc.create)() error = self.assertRaises(exception.ResourceFailure, scheduler.TaskRunner(rsrc.delete)) @@ -259,20 +310,41 @@ class MeteringRuleTest(common.HeatTestCase): 'An unknown exception occurred.', six.text_type(error)) self.assertEqual((rsrc.DELETE, rsrc.FAILED), rsrc.state) - self.m.VerifyAll() + + self.mockclient.create_metering_label_rule.assert_called_once_with({ + 'metering_label_rule': { + 'metering_label_id': '1234', + 'remote_ip_prefix': '10.0.3.0/24', + 'direction': 'ingress', + 'excluded': False + } + }) + self.mockclient.delete_metering_label_rule.assert_called_once_with( + '5678') + self.mockclient.show_metering_label_rule.assert_not_called() def test_attribute(self): rsrc = self.create_metering_label_rule() - neutronclient.Client.show_metering_label_rule('5678').MultipleTimes( - ).AndReturn( - {'metering_label_rule': - {'metering_label_id': '1234', - 'remote_ip_prefix': '10.0.3.0/24', - 'direction': 'ingress', - 'excluded': False}}) - self.m.ReplayAll() + self.mockclient.show_metering_label_rule.return_value = { + 'metering_label_rule': { + 'metering_label_id': '1234', + 'remote_ip_prefix': '10.0.3.0/24', + 'direction': 'ingress', + 'excluded': False + } + } + scheduler.TaskRunner(rsrc.create)() self.assertEqual('10.0.3.0/24', rsrc.FnGetAtt('remote_ip_prefix')) self.assertEqual('ingress', rsrc.FnGetAtt('direction')) self.assertIs(False, rsrc.FnGetAtt('excluded')) - self.m.VerifyAll() + + self.mockclient.create_metering_label_rule.assert_called_once_with({ + 'metering_label_rule': { + 'metering_label_id': '1234', + 'remote_ip_prefix': '10.0.3.0/24', + 'direction': 'ingress', + 'excluded': False + } + }) + self.mockclient.show_metering_label_rule.assert_called_with('5678')