1""" 2 :codeauthor: `Tyler Johnson <tjohnson@saltstack.com>` 3 4 tests.unit.cloud.clouds.openstack_test 5 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 6""" 7 8 9from salt.cloud.clouds import openstack 10from salt.utils import dictupdate 11from tests.support.mixins import LoaderModuleMockMixin 12from tests.support.mock import MagicMock, patch 13from tests.support.unit import TestCase 14 15# pylint: disable=confusing-with-statement 16 17 18class MockImage: 19 name = "image name" 20 id = "image id" 21 22 23class MockNode: 24 name = "node name" 25 id = "node id" 26 flavor = MockImage() 27 status = "node status" 28 29 def __init__(self, image): 30 self.image = image 31 32 def __iter__(self): 33 return iter(()) 34 35 36class MockConn: 37 def __init__(self, image): 38 self.node = MockNode(image) 39 40 def get_image(self, *args, **kwargs): 41 return self.node.image 42 43 def get_flavor(self, *args, **kwargs): 44 return self.node.flavor 45 46 def get_server(self, *args, **kwargs): 47 return self.node 48 49 def list_servers(self, *args, **kwargs): 50 return [self.node] 51 52 53class OpenstackTestCase(TestCase, LoaderModuleMockMixin): 54 """ 55 Unit TestCase for salt.cloud.clouds.openstack module. 56 """ 57 58 def setup_loader_modules(self): 59 return { 60 openstack: { 61 "__active_provider_name__": "", 62 "__opts__": { 63 "providers": { 64 "my-openstack-cloud": { 65 "openstack": { 66 "auth": "daenerys", 67 "region_name": "westeros", 68 "cloud": "openstack", 69 } 70 } 71 } 72 }, 73 } 74 } 75 76 def test_get_configured_provider_bad(self): 77 with patch.dict(openstack.__opts__, {"providers": {}}): 78 result = openstack.get_configured_provider() 79 self.assertEqual(result, False) 80 81 def test_get_configured_provider_auth(self): 82 config = { 83 "region_name": "westeros", 84 "auth": "daenerys", 85 } 86 with patch.dict( 87 openstack.__opts__, 88 {"providers": {"my-openstack-cloud": {"openstack": config}}}, 89 ): 90 result = openstack.get_configured_provider() 91 self.assertEqual(config, result) 92 93 def test_get_configured_provider_cloud(self): 94 config = { 95 "region_name": "westeros", 96 "cloud": "foo", 97 } 98 with patch.dict( 99 openstack.__opts__, 100 {"providers": {"my-openstack-cloud": {"openstack": config}}}, 101 ): 102 result = openstack.get_configured_provider() 103 self.assertEqual(config, result) 104 105 def test_get_dependencies(self): 106 HAS_SHADE = (True, "Please install newer version of shade: >= 1.19.0") 107 with patch("salt.cloud.clouds.openstack.HAS_SHADE", HAS_SHADE): 108 result = openstack.get_dependencies() 109 self.assertEqual(result, True) 110 111 def test_get_dependencies_no_shade(self): 112 HAS_SHADE = (False, "Install pypi module shade >= 1.19.0") 113 with patch("salt.cloud.clouds.openstack.HAS_SHADE", HAS_SHADE): 114 result = openstack.get_dependencies() 115 self.assertEqual(result, False) 116 117 def test_list_nodes_full_image_str(self): 118 node_image = "node image" 119 conn = MockConn(node_image) 120 with patch("salt.cloud.clouds.openstack._get_ips", return_value=[]): 121 ret = openstack.list_nodes_full(conn=conn) 122 self.assertEqual(ret[conn.node.name]["image"], node_image) 123 124 def test_list_nodes_full_image_obj(self): 125 conn = MockConn(MockImage()) 126 with patch("salt.cloud.clouds.openstack._get_ips", return_value=[]): 127 ret = openstack.list_nodes_full(conn=conn) 128 self.assertEqual(ret[conn.node.name]["image"], MockImage.name) 129 130 def test_show_instance(self): 131 conn = MockConn(MockImage()) 132 with patch("salt.cloud.clouds.openstack._get_ips", return_value=[]): 133 ret = openstack.show_instance(conn.node.name, conn=conn, call="action") 134 self.assertEqual(ret["image"], MockImage.name) 135 136 def test_request_instance_should_use_provided_connection_if_not_None(self): 137 fake_conn = MagicMock() 138 139 patch_get_conn = patch("salt.cloud.clouds.openstack.get_conn", autospec=True) 140 patch_utils = patch.dict( 141 openstack.__utils__, 142 {"cloud.check_name": MagicMock(), "dictupdate.update": dictupdate.update}, 143 ) 144 patch_shade = patch.object( 145 openstack, "shade.exc.OpenStackCloudException", Exception, create=True 146 ) 147 148 with patch_get_conn as fake_get_conn, patch_utils, patch_shade: 149 openstack.request_instance( 150 vm_={"name": "fnord", "driver": "fnord"}, conn=fake_conn 151 ) 152 153 fake_get_conn.assert_not_called() 154 155 def test_request_instance_should_create_conn_if_provided_is_None(self): 156 none_conn = None 157 158 patch_get_conn = patch("salt.cloud.clouds.openstack.get_conn", autospec=True) 159 patch_utils = patch.dict( 160 openstack.__utils__, 161 {"cloud.check_name": MagicMock(), "dictupdate.update": dictupdate.update}, 162 ) 163 patch_shade = patch.object( 164 openstack, "shade.exc.OpenStackCloudException", Exception, create=True 165 ) 166 167 with patch_get_conn as fake_get_conn, patch_utils, patch_shade: 168 openstack.request_instance( 169 vm_={"name": "fnord", "driver": "fnord"}, conn=none_conn 170 ) 171 172 fake_get_conn.assert_called_once_with() 173 174 # According to 175 # https://docs.openstack.org/shade/latest/user/usage.html#shade.OpenStackCloud.create_server 176 # the `network` parameter can be: 177 # (optional) Network dict or name or ID to attach the server to. 178 # Mutually exclusive with the nics parameter. Can also be be a list of 179 # network names or IDs or network dicts. 180 # 181 # Here we're testing a normal dictionary 182 def test_request_instance_should_be_able_to_provide_a_dictionary_for_network(self): 183 fake_conn = MagicMock() 184 expected_network = {"foo": "bar"} 185 vm_ = {"name": "fnord", "driver": "fnord", "network": expected_network} 186 patch_utils = patch.dict( 187 openstack.__utils__, 188 {"cloud.check_name": MagicMock(), "dictupdate.update": dictupdate.update}, 189 ) 190 with patch_utils: 191 openstack.request_instance(vm_=vm_, conn=fake_conn) 192 193 call_kwargs = fake_conn.create_server.mock_calls[0][-1] 194 self.assertDictEqual(call_kwargs["network"], expected_network) 195 196 # Here we're testing the list of dictionaries 197 def test_request_instance_should_be_able_to_provide_a_list_of_dictionaries_for_network( 198 self, 199 ): 200 fake_conn = MagicMock() 201 expected_network = [{"foo": "bar"}, {"bang": "quux"}] 202 vm_ = {"name": "fnord", "driver": "fnord", "network": expected_network} 203 patch_utils = patch.dict( 204 openstack.__utils__, 205 {"cloud.check_name": MagicMock(), "dictupdate.update": dictupdate.update}, 206 ) 207 with patch_utils: 208 openstack.request_instance(vm_=vm_, conn=fake_conn) 209 210 call_kwargs = fake_conn.create_server.mock_calls[0][-1] 211 assert call_kwargs["network"] == expected_network 212 213 # Here we're testing for names/IDs 214 def test_request_instance_should_be_able_to_provide_a_list_of_single_ids_or_names_for_network( 215 self, 216 ): 217 fake_conn = MagicMock() 218 expected_network = ["foo", "bar", "bang", "fnord1", "fnord2"] 219 vm_ = {"name": "fnord", "driver": "fnord", "network": expected_network} 220 patch_utils = patch.dict( 221 openstack.__utils__, 222 {"cloud.check_name": MagicMock(), "dictupdate.update": dictupdate.update}, 223 ) 224 with patch_utils: 225 openstack.request_instance(vm_=vm_, conn=fake_conn) 226 227 call_kwargs = fake_conn.create_server.mock_calls[0][-1] 228 assert call_kwargs["network"] == expected_network 229 230 # Testing that we get a dict that we expect for create_server 231 def test__clean_create_kwargs(self): 232 params = { 233 "name": "elmer", 234 "image": "mirrormirror", 235 "flavor": "chocolate", 236 "auto_ip": True, 237 "ips": ["hihicats"], 238 "ip_pool": "olympic", 239 "root_volume": "iamgroot", 240 "boot_volume": "pussnboots", 241 "terminate_volume": False, 242 "volumes": ["lots", "of", "books"], 243 "meta": {"full": "meta"}, 244 "files": {"shred": "this"}, 245 "reservation_id": "licenseandregistration", 246 "security_groups": ["wanna", "play", "repeat"], 247 "key_name": "clortho", 248 "availability_zone": "callmemaybe", 249 "block_device_mapping": [{"listof": "dicts"}], 250 "block_device_mapping_v2": [{"listof": "dicts"}], 251 "nics": ["thats", "me"], 252 "scheduler_hints": {"so": "many"}, 253 "config_drive": True, 254 "disk_config": "donkey", 255 "admin_pass": "password", 256 "wait": False, 257 "timeout": 30, 258 "reuse_ips": True, 259 "network": ["also", "a", "dict"], 260 "boot_from_volume": True, 261 "volume_size": 30, 262 "nat_destination": "albuquerque", 263 "group": "ledzeppelin", 264 "userdata": "needmoreinput", 265 "thisgetsdropped": "yup", 266 } 267 patch_utils = patch.dict( 268 openstack.__utils__, 269 {"dictupdate.update": dictupdate.update}, 270 ) 271 with patch_utils: 272 ret = openstack._clean_create_kwargs(**params) 273 params.pop("thisgetsdropped") 274 self.assertDictEqual(params, ret) 275