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