view env/lib/python3.9/site-packages/bioblend/_tests/TestCloudmanMock.py @ 0:4f3585e2f14b draft default tip

"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author shellac
date Mon, 22 Mar 2021 18:12:50 +0000
parents
children
line wrap: on
line source

"""
Tests the functionality of the BioBlend CloudMan API, without actually making
calls to a remote CloudMan instance/server. These don't actually ensure
that BioBlend is integrated with CloudMan correctly. They only ensure that
if you refactor the BioBlend CloudMan API code, that it will maintain its
current behaviour.
"""
from unittest.mock import MagicMock

from bioblend import cloudman
from .test_util import unittest


class TestCloudmanMock(unittest.TestCase):

    def setUp(self):
        url = "http://127.0.0.1:42284"
        password = "password"
        self.cm = cloudman.CloudManInstance(url, password)

#    def test_initialize(self):
#        self.cm._make_get_request = MagicMock(return_value="{}")
#
#        ## Set cluster type
#        self.cm.initialize(type="Galaxy")
#
#        params = {'startup_opt': 'Galaxy'}
#        self.cm._make_get_request.assert_called_with("initialize_cluster", parameters=params)

    def test_get_status(self):
        # Set return value of call
        self.cm._make_get_request = MagicMock(return_value={})

        status = self.cm.get_status()
        self.assertNotEqual(status, None)
        self.assertEqual(status, {})

        # Check that the correct URL was called
        self.cm._make_get_request.assert_called_with("instance_state_json")

    def test_get_nodes(self):
        # Set return value of call
        self.cm._make_get_request = MagicMock(return_value={'instances': []})

        nodes = self.cm.get_nodes()
        self.assertIsNotNone(nodes)
        self.assertEqual(len(nodes), 0)

        # Check that the correct URL was called
        self.cm._make_get_request.assert_called_with("instance_feed_json")

    def test_add_nodes(self):
        self.cm._make_get_request = MagicMock(return_value="{}")
        num_nodes = 10
        status = self.cm.add_nodes(num_nodes)
        self.assertIsNotNone(status)

        # Check that the correct URL was called
        params = {'number_nodes': 10, 'instance_type': '', 'spot_price': ''}
        self.cm._make_get_request.assert_called_with("add_instances", parameters=params)

    def test_remove_nodes(self):
        self.cm._make_get_request = MagicMock(return_value="{}")
        num_nodes = 10
        status = self.cm.remove_nodes(num_nodes, force=True)
        self.assertIsNotNone(status)

        # Check that the correct URL was called
        params = {'number_nodes': 10, 'force_termination': True}
        self.cm._make_get_request.assert_called_with("remove_instances", parameters=params)

    def test_remove_node(self):
        self.cm._make_get_request = MagicMock(return_value="{}")
        instance_id = "abcdef"
        self.cm.remove_node(instance_id, force=True)

        # Check that the correct URL was called
        params = {'instance_id': "abcdef"}
        self.cm._make_get_request.assert_called_with("remove_instance", parameters=params)

    def test_reboot_node(self):
        self.cm._make_get_request = MagicMock(return_value="{}")
        instance_id = "abcdef"
        self.cm.reboot_node(instance_id)

        # Check that the correct URL was called
        params = {'instance_id': "abcdef"}
        self.cm._make_get_request.assert_called_with("reboot_instance", parameters=params)

    def test_autoscaling_enabled_true(self):
        return_json_string = {"autoscaling": {"use_autoscaling": True, "as_max": "3", "as_min": "1"}}
        self.cm._make_get_request = MagicMock(return_value=return_json_string)
        self.assertTrue(self.cm.autoscaling_enabled())

    def test_autoscaling_enabled_false(self):
        return_json_string = {"autoscaling": {"use_autoscaling": False, "as_max": "3", "as_min": "1"}}
        self.cm._make_get_request = MagicMock(return_value=return_json_string)
        self.assertFalse(self.cm.autoscaling_enabled())

    def test_enable_autoscaling(self):
        return_json_string = {"autoscaling": {"use_autoscaling": False, "as_max": "N/A", "as_min": "N/A"}}
        self.cm._make_get_request = MagicMock(return_value=return_json_string)
        self.assertFalse(self.cm.autoscaling_enabled())
        self.cm.enable_autoscaling(minimum_nodes=0, maximum_nodes=19)

        # Check that the correct URL was called
        params = {'as_min': 0, 'as_max': 19}
        self.cm._make_get_request.assert_called_with("toggle_autoscaling", parameters=params)

        return_json_string = {"autoscaling": {"use_autoscaling": True, "as_max": "19", "as_min": "0"}}
        self.cm.enable_autoscaling(minimum_nodes=0, maximum_nodes=19)

        # Check that the correct URL was called
        params = {'as_min': 0, 'as_max': 19}
        self.cm._make_get_request.assert_called_with("toggle_autoscaling", parameters=params)

    def test_disable_autoscaling(self):
        return_json_string = {"autoscaling": {"use_autoscaling": True, "as_max": "3", "as_min": "1"}}
        self.cm._make_get_request = MagicMock(return_value=return_json_string)
        self.cm.disable_autoscaling()

        self.cm._make_get_request.assert_called_with("toggle_autoscaling")

    def test_adjust_autoscaling(self):
        return_json_string = {"autoscaling": {"use_autoscaling": True, "as_max": "3", "as_min": "1"}}
        self.cm._make_get_request = MagicMock(return_value=return_json_string)
        self.cm.adjust_autoscaling(minimum_nodes=3, maximum_nodes=4)
        params = {'as_min_adj': 3, 'as_max_adj': 4}
        self.cm._make_get_request.assert_called_with("adjust_autoscaling", parameters=params)

    def test_get_galaxy_state_stopped(self):
        return_json = {"status": "'Galaxy' is not running", "srvc": "Galaxy"}
        self.cm._make_get_request = MagicMock(return_value=return_json)

        self.assertEqual(self.cm.get_galaxy_state()['status'], "'Galaxy' is not running")
        params = {'srvc': "Galaxy"}
        self.cm._make_get_request.assert_called_with("get_srvc_status", parameters=params)