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             f'{self.cinder.case_name}-vm2_{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             f'{self.cinder.case_name}-vm2_{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=f'{self.cinder.case_name}-volume_{self.cinder.guid}',
  90             size='2', timeout=self.cinder.volume_timeout, wait=True)
  91
  92     @mock.patch('scp.SCPClient.put')
  93     def test_write(self, *args):
  94         # pylint: disable=protected-access
  95         self.cinder.ssh = mock.Mock()
  96         self.cinder.sshvm = mock.Mock(id='foo')
  97         self.cinder.volume = mock.Mock(id='volume')
  98         stdout = mock.Mock()
  99         stdout.channel.recv_exit_status.return_value = 0
 100         self.cinder.ssh.exec_command.return_value = (None, stdout, mock.Mock())
 101         self.assertEqual(self.cinder._write_data(), 0)
 102         self.cinder.ssh.exec_command.assert_called_once_with(
 103             f"sh ~/write_data.sh {env.get('VOLUME_DEVICE_NAME')}")
 104         self.cinder.cloud.attach_volume.assert_called_once_with(
 105             self.cinder.sshvm, self.cinder.volume,
 106             timeout=self.cinder.volume_timeout)
 107         self.cinder.cloud.detach_volume.assert_called_once_with(
 108             self.cinder.sshvm, self.cinder.volume,
 109             timeout=self.cinder.volume_timeout)
 110         args[0].assert_called_once_with(
 111             pkg_resources.resource_filename(
 112                 'functest.opnfv_tests.openstack.cinder', 'write_data.sh'),
 113             remote_path="~/")
 114
 115     @mock.patch('scp.SCPClient.put', side_effect=Exception)
 116     def test_write_exc1(self, *args):
 117         # pylint: disable=protected-access
 118         self.cinder.ssh = mock.Mock()
 119         self.cinder.sshvm = mock.Mock(id='foo')
 120         self.cinder.cloud.attach_volume = mock.Mock()
 121         self.assertEqual(
 122             self.cinder._write_data(), self.cinder.EX_RUN_ERROR)
 123         args[0].assert_called_once_with(
 124             pkg_resources.resource_filename(
 125                 'functest.opnfv_tests.openstack.cinder', 'write_data.sh'),
 126             remote_path="~/")
 127
 128     @mock.patch('scp.SCPClient.put')
 129     def test_read(self, *args):
 130         # pylint: disable=protected-access
 131         self.cinder.ssh2 = mock.Mock()
 132         self.cinder.vm2 = mock.Mock(id='foo')
 133         self.cinder.volume = mock.Mock(id='volume')
 134         stdout = mock.Mock()
 135         self.cinder.ssh2.exec_command.return_value = (
 136             None, stdout, mock.Mock())
 137         stdout.channel.recv_exit_status.return_value = 0
 138         self.assertEqual(self.cinder._read_data(), 0)
 139         self.cinder.ssh2.exec_command.assert_called_once_with(
 140             f"sh ~/read_data.sh {env.get('VOLUME_DEVICE_NAME')}")
 141         self.cinder.cloud.attach_volume.assert_called_once_with(
 142             self.cinder.vm2, self.cinder.volume,
 143             timeout=self.cinder.volume_timeout)
 144         self.cinder.cloud.detach_volume.assert_called_once_with(
 145             self.cinder.vm2, self.cinder.volume,
 146             timeout=self.cinder.volume_timeout)
 147         args[0].assert_called_once_with(
 148             pkg_resources.resource_filename(
 149                 'functest.opnfv_tests.openstack.cinder', 'read_data.sh'),
 150             remote_path="~/")
 151
 152     @mock.patch('scp.SCPClient.put', side_effect=Exception)
 153     def test_read_exc1(self, *args):
 154         # pylint: disable=protected-access
 155         self.cinder.ssh = mock.Mock()
 156         self.cinder.ssh2 = mock.Mock()
 157         self.cinder.sshvm = mock.Mock(id='foo')
 158         self.cinder.cloud.attach_volume = mock.Mock()
 159         self.assertEqual(
 160             self.cinder._read_data(), self.cinder.EX_RUN_ERROR)
 161         args[0].assert_called_once_with(
 162             pkg_resources.resource_filename(
 163                 'functest.opnfv_tests.openstack.cinder', 'read_data.sh'),
 164             remote_path="~/")
 165
 166     def test_execute_exc1(self):
 167         # pylint: disable=protected-access
 168         self.cinder._write_data = mock.Mock(side_effect=Exception)
 169         self.cinder._read_data = mock.Mock()
 170         with self.assertRaises(Exception):
 171             self.cinder.execute()
 172         self.cinder._write_data.assert_called_once_with()
 173         self.cinder._read_data.assert_not_called()
 174
 175     def test_execute_exc2(self):
 176         # pylint: disable=protected-access
 177         self.cinder._write_data = mock.Mock(return_value=0)
 178         self.cinder._read_data = mock.Mock(side_effect=Exception)
 179         with self.assertRaises(Exception):
 180             self.cinder.execute()
 181         self.cinder._write_data.assert_called_once_with()
 182         self.cinder._read_data.assert_called_once_with()
 183
 184     def test_execute_res1(self):
 185         # pylint: disable=protected-access
 186         self.cinder._write_data = mock.Mock(return_value=1)
 187         self.cinder._read_data = mock.Mock()
 188         self.assertEqual(self.cinder.execute(), 1)
 189         self.cinder._write_data.assert_called_once_with()
 190         self.cinder._read_data.assert_not_called()
 191
 192     def test_execute_res2(self):
 193         # pylint: disable=protected-access
 194         self.cinder._write_data = mock.Mock(return_value=0)
 195         self.cinder._read_data = mock.Mock(return_value=1)
 196         self.assertEqual(self.cinder.execute(), 1)
 197         self.cinder._write_data.assert_called_once_with()
 198         self.cinder._read_data.assert_called_once_with()
 199
 200     def test_execute_res3(self):
 201         # pylint: disable=protected-access
 202         self.cinder._write_data = mock.Mock(return_value=0)
 203         self.cinder._read_data = mock.Mock(return_value=0)
 204         self.assertEqual(self.cinder.execute(), 0)
 205         self.cinder._write_data.assert_called_once_with()
 206         self.cinder._read_data.assert_called_once_with()
 207
 208     def test_clean_exc1(self):
 209         self.cinder.cloud = None
 210         with self.assertRaises(AssertionError):
 211             self.cinder.clean()
 212
 213     @mock.patch('functest.core.singlevm.SingleVm2.clean')
 214     def test_clean_exc2(self, *args):
 215         self.cinder.vm2 = munch.Munch(id='vm2')
 216         self.cinder.cloud.delete_server = mock.Mock(
 217             side_effect=shade.OpenStackCloudException("Foo"))
 218         with self.assertRaises(shade.OpenStackCloudException):
 219             self.cinder.clean()
 220         self.cinder.cloud.delete_server.assert_called_once_with(
 221             self.cinder.vm2, wait=True,
 222             timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
 223         self.cinder.cloud.delete_floating_ip.assert_not_called()
 224         self.cinder.cloud.delete_volume.assert_not_called()
 225         args[0].assert_not_called()
 226
 227     @mock.patch('functest.core.singlevm.SingleVm2.clean',
 228                 side_effect=Exception)
 229     def test_clean_exc3(self, mock_clean):
 230         self.cinder.vm2 = munch.Munch(id='vm2')
 231         self.cinder.volume = munch.Munch(id='volume')
 232         self.cinder.fip2 = munch.Munch(id='fip2')
 233         with self.assertRaises(Exception):
 234             self.cinder.clean()
 235         self.cinder.cloud.delete_server.assert_called_once_with(
 236             self.cinder.vm2, wait=True,
 237             timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
 238         self.cinder.cloud.delete_floating_ip.assert_called_once_with(
 239             self.cinder.fip2.id)
 240         self.cinder.cloud.delete_volume.assert_called_once_with(
 241             self.cinder.volume.id)
 242         mock_clean.prepare()
 243
 244     @mock.patch('functest.core.singlevm.SingleVm2.clean')
 245     def test_clean(self, *args):
 246         self.cinder.vm2 = munch.Munch(id='vm2')
 247         self.cinder.volume = munch.Munch(id='volume')
 248         self.cinder.fip2 = munch.Munch(id='fip2')
 249         self.cinder.clean()
 250         self.cinder.cloud.delete_server.assert_called_once_with(
 251             self.cinder.vm2, wait=True,
 252             timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
 253         self.cinder.cloud.delete_floating_ip.assert_called_once_with(
 254             self.cinder.fip2.id)
 255         self.cinder.cloud.delete_volume.assert_called_once_with(
 256             self.cinder.volume.id)
 257         args[0].assert_called_once_with()
 258
 259     @mock.patch('functest.core.singlevm.SingleVm2.clean')
 260     def test_clean2(self, *args):
 261         self.cinder.clean()
 262         self.cinder.cloud.delete_server.assert_not_called()
 263         self.cinder.cloud.delete_floating_ip.assert_not_called()
 264         self.cinder.cloud.delete_volume.assert_not_called()
 265         args[0].assert_called_once_with()
 266
 267
 268 if __name__ == '__main__':
 269     logging.disable(logging.CRITICAL)
 270     unittest.main(verbosity=2)