services/nodemanager/tests/test_nodelist.py

   1 #!/usr/bin/env python
   2
   3 from __future__ import absolute_import, print_function
   4
   5 import unittest
   6 import mock
   7
   8 import arvnodeman.nodelist as nodelist
   9 from . import testutil
  10
  11 class ArvadosNodeListMonitorActorTestCase(testutil.RemotePollLoopActorTestMixin,
  12                                           unittest.TestCase):
  13     TEST_CLASS = nodelist.ArvadosNodeListMonitorActor
  14
  15     def build_monitor(self, side_effect, *args, **kwargs):
  16         super(ArvadosNodeListMonitorActorTestCase, self).build_monitor(
  17             *args, **kwargs)
  18         self.client.nodes().list().execute.side_effect = side_effect
  19
  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()
  24         self.build_monitor([{
  25             'items': [node],
  26             'items_available': 1,
  27             'offset': 0
  28         }, {
  29             'items': [],
  30             'items_available': 1,
  31             'offset': 1
  32         }])
  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"])
  38
  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()
  43         self.build_monitor([{
  44             'items': [node],
  45             'items_available': 1,
  46             'offset': 0
  47         }, {
  48             'items': [],
  49             'items_available': 1,
  50             'offset': 1
  51         }])
  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"])
  57
  58
  59 class CloudNodeListMonitorActorTestCase(testutil.RemotePollLoopActorTestMixin,
  60                                         unittest.TestCase):
  61     TEST_CLASS = nodelist.CloudNodeListMonitorActor
  62
  63     class MockNode(object):
  64         def __init__(self, count):
  65             self.id = str(count)
  66             self.name = 'test{}.example.com'.format(count)
  67             self.private_ips = ['10.0.0.{}'.format(count)]
  68             self.public_ips = []
  69             self.size = None
  70             self.state = 0
  71
  72
  73     def build_monitor(self, side_effect, *args, **kwargs):
  74         super(CloudNodeListMonitorActorTestCase, self).build_monitor(
  75             *args, **kwargs)
  76         self.client.list_nodes.side_effect = side_effect
  77
  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)
  84
  85
  86 if __name__ == '__main__':
  87     unittest.main()