functest/tests/unit/openstack/cinder/test_cinder.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2018 Enea AB and others.
   4 #
   5 # All rights reserved. This program and the accompanying materials
   6 # are made available under the terms of the Apache License, Version 2.0
   7 # which accompanies this distribution, and is available at
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9
  10 # pylint: disable=missing-docstring
  11
  12
  13 import logging
  14 import unittest
  15
  16 import pkg_resources
  17 import mock
  18 import munch
  19 import shade
  20
  21 from functest.opnfv_tests.openstack.cinder import cinder_test
  22 from functest.utils import config
  23 from functest.utils import env
  24
  25
  26 class CinderTesting(unittest.TestCase):
  27
  28     def setUp(self):
  29         with mock.patch('functest.core.singlevm.SingleVm2.__init__'):
  30             self.cinder = cinder_test.CinderCheck()
  31             self.cinder.cloud = mock.Mock()
  32             self.cinder.case_name = 'cinder'
  33             self.cinder.guid = '1'
  34
  35     @mock.patch('functest.opnfv_tests.openstack.cinder.cinder_test.'
  36                 'CinderCheck.connect')
  37     @mock.patch('functest.core.singlevm.SingleVm2.prepare',
  38                 side_effect=Exception)
  39     def test_prepare_exc1(self, *args):
  40         self.cinder.cloud.boot_vm = mock.Mock()
  41         with self.assertRaises(Exception):
  42             self.cinder.prepare()
  43         args[0].assert_called_once_with()
  44         args[1].assert_not_called()
  45         self.cinder.cloud.boot_vm.assert_not_called()
  46         self.cinder.cloud.create_volume.assert_not_called()
  47
  48     @mock.patch('functest.opnfv_tests.openstack.cinder.cinder_test.'
  49                 'CinderCheck.connect')
  50     @mock.patch('functest.opnfv_tests.openstack.cinder.cinder_test.'
  51                 'CinderCheck.boot_vm',
  52                 side_effect=Exception)
  53     @mock.patch('functest.core.singlevm.SingleVm2.prepare')
  54     def test_prepare_exc2(self, *args):
  55         self.cinder.sec = munch.Munch(id='foo')
  56         self.cinder.keypair = munch.Munch(id='foo')
  57         self.cinder.volume_timeout = munch.Munch(id='foo')
  58         with self.assertRaises(Exception):
  59             self.cinder.prepare()
  60         args[0].assert_called_with()
  61         args[1].assert_called_once_with(
  62             '{}-vm2_{}'.format(self.cinder.case_name, self.cinder.guid),
  63             security_groups=[self.cinder.sec.id],
  64             key_name=self.cinder.keypair.id)
  65         self.cinder.cloud.create_volume.assert_not_called()
  66         args[2].assert_not_called()
  67
  68     @mock.patch('functest.opnfv_tests.openstack.cinder.cinder_test.'
  69                 'CinderCheck.boot_vm', return_value=munch.Munch(id='vm2'))
  70     @mock.patch('functest.core.singlevm.SingleVm2.prepare')
  71     def test_prepare(self, *args):
  72         self.cinder.sec = munch.Munch(id='foo')
  73         self.cinder.keypair = munch.Munch(id='foo')
  74         self.cinder.ext_net = mock.Mock(id='foo')
  75         self.cinder.ssh2 = mock.Mock()
  76         self.cinder.fip2 = munch.Munch(id='fip2')
  77         self.cinder.connect = mock.Mock(
  78             return_value=(self.cinder.fip2, self.cinder.ssh2))
  79         self.cinder.cloud.create_volume = mock.Mock(
  80             return_value=munch.Munch())
  81         self.cinder.prepare()
  82         args[0].assert_called_once_with()
  83         args[1].assert_called_once_with(
  84             '{}-vm2_{}'.format(self.cinder.case_name, self.cinder.guid),
  85             security_groups=[self.cinder.sec.id],
  86             key_name=self.cinder.keypair.id)
  87         self.cinder.connect.assert_called_once_with(args[1].return_value)
  88         self.cinder.cloud.create_volume.assert_called_once_with(
  89             name='{}-volume_{}'.format(
  90                 self.cinder.case_name, self.cinder.guid),
  91             size='2', timeout=self.cinder.volume_timeout, wait=True)
  92
  93     @mock.patch('scp.SCPClient.put')
  94     def test_write(self, *args):
  95         # pylint: disable=protected-access
  96         self.cinder.ssh = mock.Mock()
  97         self.cinder.sshvm = mock.Mock(id='foo')
  98         self.cinder.volume = mock.Mock(id='volume')
  99         stdout = mock.Mock()
 100         stdout.channel.recv_exit_status.return_value = 0
 101         self.cinder.ssh.exec_command.return_value = (None, stdout, mock.Mock())
 102         self.assertEqual(self.cinder._write_data(), 0)
 103         self.cinder.ssh.exec_command.assert_called_once_with(
 104             "sh ~/write_data.sh {}".format(env.get('VOLUME_DEVICE_NAME')))
 105         self.cinder.cloud.attach_volume.assert_called_once_with(
 106             self.cinder.sshvm, self.cinder.volume,
 107             timeout=self.cinder.volume_timeout)
 108         self.cinder.cloud.detach_volume.assert_called_once_with(
 109             self.cinder.sshvm, self.cinder.volume,
 110             timeout=self.cinder.volume_timeout)
 111         args[0].assert_called_once_with(
 112             pkg_resources.resource_filename(
 113                 'functest.opnfv_tests.openstack.cinder', 'write_data.sh'),
 114             remote_path="~/")
 115
 116     @mock.patch('scp.SCPClient.put', side_effect=Exception)
 117     def test_write_exc1(self, *args):
 118         # pylint: disable=protected-access
 119         self.cinder.ssh = mock.Mock()
 120         self.cinder.sshvm = mock.Mock(id='foo')
 121         self.cinder.cloud.attach_volume = mock.Mock()
 122         self.assertEqual(
 123             self.cinder._write_data(), self.cinder.EX_RUN_ERROR)
 124         args[0].assert_called_once_with(
 125             pkg_resources.resource_filename(
 126                 'functest.opnfv_tests.openstack.cinder', 'write_data.sh'),
 127             remote_path="~/")
 128
 129     @mock.patch('scp.SCPClient.put')
 130     def test_read(self, *args):
 131         # pylint: disable=protected-access
 132         self.cinder.ssh2 = mock.Mock()
 133         self.cinder.vm2 = mock.Mock(id='foo')
 134         self.cinder.volume = mock.Mock(id='volume')
 135         stdout = mock.Mock()
 136         self.cinder.ssh2.exec_command.return_value = (
 137             None, stdout, mock.Mock())
 138         stdout.channel.recv_exit_status.return_value = 0
 139         self.assertEqual(self.cinder._read_data(), 0)
 140         self.cinder.ssh2.exec_command.assert_called_once_with(
 141             "sh ~/read_data.sh {}".format(env.get('VOLUME_DEVICE_NAME')))
 142         self.cinder.cloud.attach_volume.assert_called_once_with(
 143             self.cinder.vm2, self.cinder.volume,
 144             timeout=self.cinder.volume_timeout)
 145         self.cinder.cloud.detach_volume.assert_called_once_with(
 146             self.cinder.vm2, self.cinder.volume,
 147             timeout=self.cinder.volume_timeout)
 148         args[0].assert_called_once_with(
 149             pkg_resources.resource_filename(
 150                 'functest.opnfv_tests.openstack.cinder', 'read_data.sh'),
 151             remote_path="~/")
 152
 153     @mock.patch('scp.SCPClient.put', side_effect=Exception)
 154     def test_read_exc1(self, *args):
 155         # pylint: disable=protected-access
 156         self.cinder.ssh = mock.Mock()
 157         self.cinder.ssh2 = mock.Mock()
 158         self.cinder.sshvm = mock.Mock(id='foo')
 159         self.cinder.cloud.attach_volume = mock.Mock()
 160         self.assertEqual(
 161             self.cinder._read_data(), self.cinder.EX_RUN_ERROR)
 162         args[0].assert_called_once_with(
 163             pkg_resources.resource_filename(
 164                 'functest.opnfv_tests.openstack.cinder', 'read_data.sh'),
 165             remote_path="~/")
 166
 167     def test_execute_exc1(self):
 168         # pylint: disable=protected-access
 169         self.cinder._write_data = mock.Mock(side_effect=Exception)
 170         self.cinder._read_data = mock.Mock()
 171         with self.assertRaises(Exception):
 172             self.cinder.execute()
 173         self.cinder._write_data.assert_called_once_with()
 174         self.cinder._read_data.assert_not_called()
 175
 176     def test_execute_exc2(self):
 177         # pylint: disable=protected-access
 178         self.cinder._write_data = mock.Mock(return_value=0)
 179         self.cinder._read_data = mock.Mock(side_effect=Exception)
 180         with self.assertRaises(Exception):
 181             self.cinder.execute()
 182         self.cinder._write_data.assert_called_once_with()
 183         self.cinder._read_data.assert_called_once_with()
 184
 185     def test_execute_res1(self):
 186         # pylint: disable=protected-access
 187         self.cinder._write_data = mock.Mock(return_value=1)
 188         self.cinder._read_data = mock.Mock()
 189         self.assertEqual(self.cinder.execute(), 1)
 190         self.cinder._write_data.assert_called_once_with()
 191         self.cinder._read_data.assert_not_called()
 192
 193     def test_execute_res2(self):
 194         # pylint: disable=protected-access
 195         self.cinder._write_data = mock.Mock(return_value=0)
 196         self.cinder._read_data = mock.Mock(return_value=1)
 197         self.assertEqual(self.cinder.execute(), 1)
 198         self.cinder._write_data.assert_called_once_with()
 199         self.cinder._read_data.assert_called_once_with()
 200
 201     def test_execute_res3(self):
 202         # pylint: disable=protected-access
 203         self.cinder._write_data = mock.Mock(return_value=0)
 204         self.cinder._read_data = mock.Mock(return_value=0)
 205         self.assertEqual(self.cinder.execute(), 0)
 206         self.cinder._write_data.assert_called_once_with()
 207         self.cinder._read_data.assert_called_once_with()
 208
 209     def test_clean_exc1(self):
 210         self.cinder.cloud = None
 211         with self.assertRaises(AssertionError):
 212             self.cinder.clean()
 213
 214     @mock.patch('functest.core.singlevm.SingleVm2.clean')
 215     def test_clean_exc2(self, *args):
 216         self.cinder.vm2 = munch.Munch(id='vm2')
 217         self.cinder.cloud.delete_server = mock.Mock(
 218             side_effect=shade.OpenStackCloudException("Foo"))
 219         with self.assertRaises(shade.OpenStackCloudException):
 220             self.cinder.clean()
 221         self.cinder.cloud.delete_server.assert_called_once_with(
 222             self.cinder.vm2, wait=True,
 223             timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
 224         self.cinder.cloud.delete_floating_ip.assert_not_called()
 225         self.cinder.cloud.delete_volume.assert_not_called()
 226         args[0].assert_not_called()
 227
 228     @mock.patch('functest.core.singlevm.SingleVm2.clean',
 229                 side_effect=Exception)
 230     def test_clean_exc3(self, mock_clean):
 231         self.cinder.vm2 = munch.Munch(id='vm2')
 232         self.cinder.volume = munch.Munch(id='volume')
 233         self.cinder.fip2 = munch.Munch(id='fip2')
 234         with self.assertRaises(Exception):
 235             self.cinder.clean()
 236         self.cinder.cloud.delete_server.assert_called_once_with(
 237             self.cinder.vm2, wait=True,
 238             timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
 239         self.cinder.cloud.delete_floating_ip.assert_called_once_with(
 240             self.cinder.fip2.id)
 241         self.cinder.cloud.delete_volume.assert_called_once_with(
 242             self.cinder.volume.id)
 243         mock_clean.prepare()
 244
 245     @mock.patch('functest.core.singlevm.SingleVm2.clean')
 246     def test_clean(self, *args):
 247         self.cinder.vm2 = munch.Munch(id='vm2')
 248         self.cinder.volume = munch.Munch(id='volume')
 249         self.cinder.fip2 = munch.Munch(id='fip2')
 250         self.cinder.clean()
 251         self.cinder.cloud.delete_server.assert_called_once_with(
 252             self.cinder.vm2, wait=True,
 253             timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
 254         self.cinder.cloud.delete_floating_ip.assert_called_once_with(
 255             self.cinder.fip2.id)
 256         self.cinder.cloud.delete_volume.assert_called_once_with(
 257             self.cinder.volume.id)
 258         args[0].assert_called_once_with()
 259
 260     @mock.patch('functest.core.singlevm.SingleVm2.clean')
 261     def test_clean2(self, *args):
 262         self.cinder.clean()
 263         self.cinder.cloud.delete_server.assert_not_called()
 264         self.cinder.cloud.delete_floating_ip.assert_not_called()
 265         self.cinder.cloud.delete_volume.assert_not_called()
 266         args[0].assert_called_once_with()
 267
 268
 269 if __name__ == '__main__':
 270     logging.disable(logging.CRITICAL)
 271     unittest.main(verbosity=2)