"Fossies" - the Fresh Open Source Software Archive

Member "salt-3002.2/tests/unit/cloud/clouds/test_openstack.py" (18 Nov 2020, 9945 Bytes) of package /linux/misc/salt-3002.2.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "test_openstack.py": 3002.1_vs_3002.2.

    1 """
    2     :codeauthor: `Tyler Johnson <tjohnson@saltstack.com>`
    3 
    4     tests.unit.cloud.clouds.openstack_test
    5     ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    6 """
    7 
    8 
    9 from salt.cloud.clouds import openstack
   10 from salt.utils import dictupdate
   11 from tests.support.mixins import LoaderModuleMockMixin
   12 from tests.support.mock import MagicMock, patch
   13 from tests.support.unit import TestCase
   14 
   15 # pylint: disable=confusing-with-statement
   16 
   17 
   18 class MockImage:
   19     name = "image name"
   20     id = "image id"
   21 
   22 
   23 class 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 
   36 class 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 
   53 class 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__, {"dictupdate.update": dictupdate.update},
  269         )
  270         with patch_utils:
  271             ret = openstack._clean_create_kwargs(**params)
  272             params.pop("thisgetsdropped")
  273             self.assertDictEqual(params, ret)