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