openstack
/
heat
Code Issues Proposed changes

Merge "Remove mox from test_neutron_metering"

This commit is contained in:
Zuul 2018-03-27 05:49:09 +00:00 committed by Gerrit Code Review
parent 9a09472f09 c5d9d6a104
commit dc161baf81
1 changed files with 164 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

256
heat/tests/openstack/neutron/test_neutron_metering.py
View File

@ -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')