3 from __future__ import absolute_import, print_function
8 import arvnodeman.nodelist as nodelist
11 class ArvadosNodeListMonitorActorTestCase(testutil.RemotePollLoopActorTestMixin,
13 TEST_CLASS = nodelist.ArvadosNodeListMonitorActor
15 def build_monitor(self, side_effect, *args, **kwargs):
16 super(ArvadosNodeListMonitorActorTestCase, self).build_monitor(
18 self.client.nodes().list().execute.side_effect = side_effect
20 @mock.patch("subprocess.check_output")
21 def test_uuid_is_subscription_key(self, sinfo_mock):
22 sinfo_mock.return_value = ""
23 node = testutil.arvados_node_mock()
33 self.monitor.subscribe_to(node['uuid'],
34 self.subscriber).get(self.TIMEOUT)
35 self.stop_proxy(self.monitor)
36 self.subscriber.assert_called_with(node)
37 self.assertEqual("down", node["crunch_worker_state"])
39 @mock.patch("subprocess.check_output")
40 def test_update_from_sinfo(self, sinfo_mock):
41 sinfo_mock.return_value = "compute99 alloc"
42 node = testutil.arvados_node_mock()
52 self.monitor.subscribe_to(node['uuid'],
53 self.subscriber).get(self.TIMEOUT)
54 self.stop_proxy(self.monitor)
55 self.subscriber.assert_called_with(node)
56 self.assertEqual("busy", node["crunch_worker_state"])
59 class CloudNodeListMonitorActorTestCase(testutil.RemotePollLoopActorTestMixin,
61 TEST_CLASS = nodelist.CloudNodeListMonitorActor
63 class MockNode(object):
64 def __init__(self, count):
66 self.name = 'test{}.example.com'.format(count)
67 self.private_ips = ['10.0.0.{}'.format(count)]
73 def build_monitor(self, side_effect, *args, **kwargs):
74 super(CloudNodeListMonitorActorTestCase, self).build_monitor(
76 self.client.list_nodes.side_effect = side_effect
78 def test_id_is_subscription_key(self):
79 node = self.MockNode(1)
80 self.build_monitor([[node]])
81 self.monitor.subscribe_to('1', self.subscriber).get(self.TIMEOUT)
82 self.stop_proxy(self.monitor)
83 self.subscriber.assert_called_with(node)
86 if __name__ == '__main__':