Mercurial > repos > shellac > sam_consensus_v3
comparison 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 |
comparison
equal
deleted
inserted
replaced
-1:000000000000 | 0:4f3585e2f14b |
---|---|
1 """ | |
2 Tests the functionality of the BioBlend CloudMan API, without actually making | |
3 calls to a remote CloudMan instance/server. These don't actually ensure | |
4 that BioBlend is integrated with CloudMan correctly. They only ensure that | |
5 if you refactor the BioBlend CloudMan API code, that it will maintain its | |
6 current behaviour. | |
7 """ | |
8 from unittest.mock import MagicMock | |
9 | |
10 from bioblend import cloudman | |
11 from .test_util import unittest | |
12 | |
13 | |
14 class TestCloudmanMock(unittest.TestCase): | |
15 | |
16 def setUp(self): | |
17 url = "http://127.0.0.1:42284" | |
18 password = "password" | |
19 self.cm = cloudman.CloudManInstance(url, password) | |
20 | |
21 # def test_initialize(self): | |
22 # self.cm._make_get_request = MagicMock(return_value="{}") | |
23 # | |
24 # ## Set cluster type | |
25 # self.cm.initialize(type="Galaxy") | |
26 # | |
27 # params = {'startup_opt': 'Galaxy'} | |
28 # self.cm._make_get_request.assert_called_with("initialize_cluster", parameters=params) | |
29 | |
30 def test_get_status(self): | |
31 # Set return value of call | |
32 self.cm._make_get_request = MagicMock(return_value={}) | |
33 | |
34 status = self.cm.get_status() | |
35 self.assertNotEqual(status, None) | |
36 self.assertEqual(status, {}) | |
37 | |
38 # Check that the correct URL was called | |
39 self.cm._make_get_request.assert_called_with("instance_state_json") | |
40 | |
41 def test_get_nodes(self): | |
42 # Set return value of call | |
43 self.cm._make_get_request = MagicMock(return_value={'instances': []}) | |
44 | |
45 nodes = self.cm.get_nodes() | |
46 self.assertIsNotNone(nodes) | |
47 self.assertEqual(len(nodes), 0) | |
48 | |
49 # Check that the correct URL was called | |
50 self.cm._make_get_request.assert_called_with("instance_feed_json") | |
51 | |
52 def test_add_nodes(self): | |
53 self.cm._make_get_request = MagicMock(return_value="{}") | |
54 num_nodes = 10 | |
55 status = self.cm.add_nodes(num_nodes) | |
56 self.assertIsNotNone(status) | |
57 | |
58 # Check that the correct URL was called | |
59 params = {'number_nodes': 10, 'instance_type': '', 'spot_price': ''} | |
60 self.cm._make_get_request.assert_called_with("add_instances", parameters=params) | |
61 | |
62 def test_remove_nodes(self): | |
63 self.cm._make_get_request = MagicMock(return_value="{}") | |
64 num_nodes = 10 | |
65 status = self.cm.remove_nodes(num_nodes, force=True) | |
66 self.assertIsNotNone(status) | |
67 | |
68 # Check that the correct URL was called | |
69 params = {'number_nodes': 10, 'force_termination': True} | |
70 self.cm._make_get_request.assert_called_with("remove_instances", parameters=params) | |
71 | |
72 def test_remove_node(self): | |
73 self.cm._make_get_request = MagicMock(return_value="{}") | |
74 instance_id = "abcdef" | |
75 self.cm.remove_node(instance_id, force=True) | |
76 | |
77 # Check that the correct URL was called | |
78 params = {'instance_id': "abcdef"} | |
79 self.cm._make_get_request.assert_called_with("remove_instance", parameters=params) | |
80 | |
81 def test_reboot_node(self): | |
82 self.cm._make_get_request = MagicMock(return_value="{}") | |
83 instance_id = "abcdef" | |
84 self.cm.reboot_node(instance_id) | |
85 | |
86 # Check that the correct URL was called | |
87 params = {'instance_id': "abcdef"} | |
88 self.cm._make_get_request.assert_called_with("reboot_instance", parameters=params) | |
89 | |
90 def test_autoscaling_enabled_true(self): | |
91 return_json_string = {"autoscaling": {"use_autoscaling": True, "as_max": "3", "as_min": "1"}} | |
92 self.cm._make_get_request = MagicMock(return_value=return_json_string) | |
93 self.assertTrue(self.cm.autoscaling_enabled()) | |
94 | |
95 def test_autoscaling_enabled_false(self): | |
96 return_json_string = {"autoscaling": {"use_autoscaling": False, "as_max": "3", "as_min": "1"}} | |
97 self.cm._make_get_request = MagicMock(return_value=return_json_string) | |
98 self.assertFalse(self.cm.autoscaling_enabled()) | |
99 | |
100 def test_enable_autoscaling(self): | |
101 return_json_string = {"autoscaling": {"use_autoscaling": False, "as_max": "N/A", "as_min": "N/A"}} | |
102 self.cm._make_get_request = MagicMock(return_value=return_json_string) | |
103 self.assertFalse(self.cm.autoscaling_enabled()) | |
104 self.cm.enable_autoscaling(minimum_nodes=0, maximum_nodes=19) | |
105 | |
106 # Check that the correct URL was called | |
107 params = {'as_min': 0, 'as_max': 19} | |
108 self.cm._make_get_request.assert_called_with("toggle_autoscaling", parameters=params) | |
109 | |
110 return_json_string = {"autoscaling": {"use_autoscaling": True, "as_max": "19", "as_min": "0"}} | |
111 self.cm.enable_autoscaling(minimum_nodes=0, maximum_nodes=19) | |
112 | |
113 # Check that the correct URL was called | |
114 params = {'as_min': 0, 'as_max': 19} | |
115 self.cm._make_get_request.assert_called_with("toggle_autoscaling", parameters=params) | |
116 | |
117 def test_disable_autoscaling(self): | |
118 return_json_string = {"autoscaling": {"use_autoscaling": True, "as_max": "3", "as_min": "1"}} | |
119 self.cm._make_get_request = MagicMock(return_value=return_json_string) | |
120 self.cm.disable_autoscaling() | |
121 | |
122 self.cm._make_get_request.assert_called_with("toggle_autoscaling") | |
123 | |
124 def test_adjust_autoscaling(self): | |
125 return_json_string = {"autoscaling": {"use_autoscaling": True, "as_max": "3", "as_min": "1"}} | |
126 self.cm._make_get_request = MagicMock(return_value=return_json_string) | |
127 self.cm.adjust_autoscaling(minimum_nodes=3, maximum_nodes=4) | |
128 params = {'as_min_adj': 3, 'as_max_adj': 4} | |
129 self.cm._make_get_request.assert_called_with("adjust_autoscaling", parameters=params) | |
130 | |
131 def test_get_galaxy_state_stopped(self): | |
132 return_json = {"status": "'Galaxy' is not running", "srvc": "Galaxy"} | |
133 self.cm._make_get_request = MagicMock(return_value=return_json) | |
134 | |
135 self.assertEqual(self.cm.get_galaxy_state()['status'], "'Galaxy' is not running") | |
136 params = {'srvc': "Galaxy"} | |
137 self.cm._make_get_request.assert_called_with("get_srvc_status", parameters=params) |