salt/tests/unit/cloud/clouds/test_joyent.py

# -*- coding: utf-8 -*-
"""
    :codeauthor: Eric Radman <ericshane@eradman.com>
"""

# Import Salt Libs
from __future__ import absolute_import, print_function, unicode_literals

# Import Salt Libs
from salt.cloud.clouds import joyent

# Import Salt Testing Libs
from tests.support.mixins import LoaderModuleMockMixin
from tests.support.mock import MagicMock, patch
from tests.support.unit import TestCase, skipIf


# Stubs
def fake_wait_for_ip(
    check_for_ip_fn, interval=None, timeout=None, interval_multiplier=None
):
    """
    Callback that returns immediately instead of waiting
    """
    assert isinstance(interval, int)
    assert isinstance(timeout, int)
    assert isinstance(interval_multiplier, int)
    return check_for_ip_fn()


@skipIf(
    joyent.HAS_REQUIRED_CRYPTO is False, reason="PyCrypto or Cryptodome not installed"
)
class JoyentTestCase(TestCase, LoaderModuleMockMixin):
    """
    Unit TestCase for the salt.cloud.clouds.joyent module
    """

    def setup_loader_modules(self):
        patcher = patch("salt.utils.cloud.wait_for_ip", fake_wait_for_ip)
        patcher.start()
        self.addCleanup(patcher.stop)
        return {
            joyent: {
                "__utils__": {
                    "cloud.fire_event": MagicMock(),
                    "cloud.bootstrap": MagicMock(),
                },
                "__opts__": {
                    "sock_dir": True,
                    "transport": True,
                    "providers": {"my_joyent": {}},
                    "profiles": {"my_joyent": {}},
                },
                "__active_provider_name__": "my_joyent:joyent",
            }
        }

    def setUp(self):
        self.vm_ = {
            "profile": "my_joyent",
            "name": "vm3",
            "driver": "joyent",
            "size": "k4-highcpu-kvm-750M",
            "image": "freebsd10",
            "location": "us-east-1",
        }

    def tearDown(self):
        del self.vm_

    def test_query_instance_init(self):
        """
        Initial provisioning, no IP assigned
        """
        # Not yet reachable
        reply = (200, {"state": "provisioning"})
        with patch.object(joyent, "show_instance", return_value=reply):
            result = joyent.query_instance(self.vm_)
        self.assertTrue(joyent.__utils__["cloud.fire_event"].called_once())
        self.assertEqual(result, None)

    def test_query_instance_has_ip(self):
        """
        IP address assigned but not yet ready
        """
        reply = (200, {"primaryIp": "1.1.1.1", "state": "provisioning"})
        with patch.object(joyent, "show_instance", return_value=reply):
            result = joyent.query_instance(self.vm_)
        self.assertTrue(joyent.__utils__["cloud.fire_event"].called_once())
        self.assertEqual(result, None)

    def test_query_instance_ready(self):
        """
        IP address assigned, and VM is ready
        """
        reply = (200, {"primaryIp": "1.1.1.1", "state": "running"})
        with patch.object(joyent, "show_instance", return_value=reply):
            result = joyent.query_instance(self.vm_)
        self.assertTrue(joyent.__utils__["cloud.fire_event"].called_once())
        self.assertEqual(result, "1.1.1.1")