functest/tests/unit/vnf/ims/test_orchestrator_cloudify.py

   1 #!/usr/bin/env python
   2
   3 # All rights reserved. This program and the accompanying materials
   4 # are made available under the terms of the Apache License, Version 2.0
   5 # which accompanies this distribution, and is available at
   6 # http://www.apache.org/licenses/LICENSE-2.0
   7
   8 import logging
   9 import subprocess32 as subprocess
  10 import unittest
  11
  12 import mock
  13
  14 from functest.opnfv_tests.vnf.ims import orchestrator_cloudify
  15
  16
  17 class ImsVnfTesting(unittest.TestCase):
  18
  19     def setUp(self):
  20         self.orchestrator = orchestrator_cloudify.Orchestrator('test_dir')
  21         self.bp = {'file_name': 'test_file',
  22                    'destination_folder': 'test_folder',
  23                    'url': 'test_url',
  24                    'branch': 'test_branch'}
  25
  26     def test_download_manager_blueprint_download_blueprint_failed(self):
  27         self.orchestrator.manager_blueprint = False
  28         with mock.patch.object(self.orchestrator, '_download_blueprints',
  29                                return_value=False), \
  30             mock.patch('functest.opnfv_tests.vnf.ims.orchestrator_cloudify.'
  31                        'exit') as mock_exit:
  32             self.orchestrator.download_manager_blueprint('test_url',
  33                                                          'test_branch')
  34             mock_exit.assert_any_call(-1)
  35
  36     def test_download_manager_blueprint_download_blueprint_passed(self):
  37         self.orchestrator.manager_blueprint = False
  38         with mock.patch.object(self.orchestrator, '_download_blueprints',
  39                                return_value=True):
  40             self.orchestrator.download_manager_blueprint('test_url',
  41                                                          'test_branch')
  42             self.assertEqual(self.orchestrator.manager_blueprint,
  43                              True)
  44
  45     def test_deploy_manager_failed(self):
  46         self.orchestrator.manager_blueprint = True
  47         with mock.patch('__builtin__.open', mock.mock_open()), \
  48             mock.patch('functest.opnfv_tests.vnf.ims.orchestrator_cloudify.'
  49                        'os.remove'), \
  50             mock.patch('functest.opnfv_tests.vnf.ims.orchestrator_cloudify.'
  51                        'execute_command', return_value='error'):
  52             self.assertEqual(self.orchestrator.deploy_manager(),
  53                              'error')
  54             self.assertEqual(self.orchestrator.manager_up,
  55                              False)
  56
  57     def test_deploy_manager_passed(self):
  58         self.orchestrator.manager_blueprint = True
  59         with mock.patch('__builtin__.open', mock.mock_open()), \
  60             mock.patch('functest.opnfv_tests.vnf.ims.orchestrator_cloudify.'
  61                        'os.remove'), \
  62             mock.patch('functest.opnfv_tests.vnf.ims.orchestrator_cloudify.'
  63                        'execute_command', return_value=''):
  64             self.orchestrator.deploy_manager()
  65             self.assertEqual(self.orchestrator.manager_up,
  66                              True)
  67
  68     def test_undeploy_manager_passed(self):
  69         with mock.patch('functest.opnfv_tests.vnf.ims.orchestrator_cloudify.'
  70                         'execute_command', return_value=''):
  71             self.orchestrator.deploy_manager()
  72             self.assertEqual(self.orchestrator.manager_up,
  73                              False)
  74
  75     def test_dwnld_upload_and_depl_blueprint_dwnld_blueprint_failed(self):
  76         with mock.patch.object(self.orchestrator, '_download_blueprints',
  77                                return_value=False), \
  78             mock.patch('functest.opnfv_tests.vnf.ims.orchestrator_cloudify.'
  79                        'exit', side_effect=Exception) as mock_exit, \
  80                 self.assertRaises(Exception):
  81             self.orchestrator.download_upload_and_deploy_blueprint(self.bp,
  82                                                                    'cfig',
  83                                                                    'bpn',
  84                                                                    'dpn')
  85             mock_exit.assert_any_call(-1)
  86
  87     def test_dwnld_upload_and_depl_blueprint_failed(self):
  88         with mock.patch.object(self.orchestrator, '_download_blueprints',
  89                                return_value=True), \
  90             mock.patch('__builtin__.open', mock.mock_open()), \
  91             mock.patch('functest.opnfv_tests.vnf.ims.orchestrator_cloudify.'
  92                        'execute_command', return_value='error'):
  93             r = self.orchestrator.download_upload_and_deploy_blueprint(self.bp,
  94                                                                        'cfig',
  95                                                                        'bpn',
  96                                                                        'dpn')
  97             self.assertEqual(r, 'error')
  98
  99     def test__download_blueprints_failed(self):
 100         with mock.patch('functest.opnfv_tests.vnf.ims.orchestrator_cloudify.'
 101                         'shutil.rmtree'), \
 102             mock.patch('functest.opnfv_tests.vnf.ims.orchestrator_cloudify.'
 103                        'Repo.clone_from', side_effect=Exception):
 104             self.assertEqual(self.orchestrator._download_blueprints('bp_url',
 105                                                                     'branch',
 106                                                                     'dest'),
 107                              False)
 108
 109     def test__download_blueprints_passed(self):
 110         with mock.patch('functest.opnfv_tests.vnf.ims.orchestrator_cloudify.'
 111                         'shutil.rmtree'), \
 112             mock.patch('functest.opnfv_tests.vnf.ims.orchestrator_cloudify.'
 113                        'Repo.clone_from'):
 114             self.assertEqual(self.orchestrator._download_blueprints('bp_url',
 115                                                                     'branch',
 116                                                                     'dest'),
 117                              True)
 118
 119     def test_execute_command_failed(self):
 120         with mock.patch('__builtin__.open',
 121                         mock.mock_open(read_data='test_data\n')):
 122             subprocess.call = mock.create_autospec(subprocess.call,
 123                                                    return_value=0)
 124             mock_log = mock.Mock()
 125             cmd = 'test_cmd -e test_env bash_script'
 126             ret = orchestrator_cloudify.execute_command(cmd, mock_log,
 127                                                         timeout=100)
 128             self.assertEqual(ret, False)
 129
 130     def test_execute_command_default(self):
 131         with mock.patch('__builtin__.open',
 132                         mock.mock_open(read_data='test_data\n')):
 133             subprocess.call = mock. \
 134                 create_autospec(subprocess.call,
 135                                 return_value=subprocess.TimeoutExpired)
 136             mock_log = mock.Mock()
 137             cmd = 'test_cmd -e test_env bash_script'
 138             ret = orchestrator_cloudify.execute_command(cmd, mock_log,
 139                                                         timeout=100)
 140             self.assertEqual(ret, ['test_data\n'])
 141
 142     def test_set_methods(self):
 143         self.orchestrator.set_credentials('test_username', 'test_password',
 144                                           'test_tenant_name', 'test_auth_url')
 145         self.assertTrue(self.orchestrator.config['keystone_username'],
 146                         'test_username')
 147         self.assertTrue(self.orchestrator.config['keystone_password'],
 148                         'test_password')
 149         self.assertTrue(self.orchestrator.config['keystone_url'],
 150                         'test_auth_url')
 151         self.assertTrue(self.orchestrator.config['keystone_tenant_name'],
 152                         'test_tenant_name')
 153         self.orchestrator.set_flavor_id('test_flavor_id')
 154         self.assertTrue(self.orchestrator.config['flavor_id'],
 155                         'test_flavor_id')
 156         self.orchestrator.set_image_id('test_image_id')
 157         self.assertTrue(self.orchestrator.config['image_id'], 'test_image_id')
 158         self.orchestrator.set_external_network_name('test_network')
 159         self.assertTrue(self.orchestrator.config['external_network_name'],
 160                         'test_network')
 161         self.orchestrator.set_ssh_user('test_user')
 162         self.assertTrue(self.orchestrator.config['ssh_user'],
 163                         'test_user')
 164         self.orchestrator.set_nova_url('test_nova_url')
 165         self.assertTrue(self.orchestrator.config['nova_url'],
 166                         'test_nova_url')
 167         self.orchestrator.set_neutron_url('test_neutron_url')
 168         self.assertTrue(self.orchestrator.config['neutron_url'],
 169                         'test_neutron_url')
 170         self.orchestrator.set_nameservers(['test_subnet'])
 171         self.assertTrue(self.orchestrator.config['dns_subnet_1'],
 172                         'test_subnet')
 173
 174 if __name__ == "__main__":
 175     logging.disable(logging.CRITICAL)
 176     unittest.main(verbosity=2)