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 libcloud.compute.base import NodeSize
  10 from . import testutil
  11
  12 class ArvadosNodeListMonitorActorTestCase(testutil.RemotePollLoopActorTestMixin,
  13                                           unittest.TestCase):
  14     TEST_CLASS = nodelist.ArvadosNodeListMonitorActor
  15
  16     def build_monitor(self, side_effect, *args, **kwargs):
  17         super(ArvadosNodeListMonitorActorTestCase, self).build_monitor(
  18             *args, **kwargs)
  19         self.client.nodes().list().execute.side_effect = side_effect
  20
  21     @mock.patch("subprocess.check_output")
  22     def test_uuid_is_subscription_key(self, sinfo_mock):
  23         sinfo_mock.return_value = ""
  24         node = testutil.arvados_node_mock()
  25         self.build_monitor([{
  26             'items': [node],
  27             'items_available': 1,
  28             'offset': 0
  29         }, {
  30             'items': [],
  31             'items_available': 1,
  32             'offset': 1
  33         }])
  34         self.monitor.subscribe_to(node['uuid'],
  35                                   self.subscriber).get(self.TIMEOUT)
  36         self.stop_proxy(self.monitor)
  37         self.subscriber.assert_called_with(node)
  38         self.assertEqual("down", node["crunch_worker_state"])
  39
  40     @mock.patch("subprocess.check_output")
  41     def test_update_from_sinfo(self, sinfo_mock):
  42         sinfo_mock.return_value = "compute99 alloc"
  43         node = testutil.arvados_node_mock()
  44         self.build_monitor([{
  45             'items': [node],
  46             'items_available': 1,
  47             'offset': 0
  48         }, {
  49             'items': [],
  50             'items_available': 1,
  51             'offset': 1
  52         }])
  53         self.monitor.subscribe_to(node['uuid'],
  54                                   self.subscriber).get(self.TIMEOUT)
  55         self.stop_proxy(self.monitor)
  56         self.subscriber.assert_called_with(node)
  57         self.assertEqual("busy", node["crunch_worker_state"])
  58
  59
  60 class CloudNodeListMonitorActorTestCase(testutil.RemotePollLoopActorTestMixin,
  61                                         unittest.TestCase):
  62     TEST_CLASS = nodelist.CloudNodeListMonitorActor
  63
  64     class MockNode(object):
  65         def __init__(self, count):
  66             self.id = str(count)
  67             self.name = 'test{}.example.com'.format(count)
  68             self.private_ips = ['10.0.0.{}'.format(count)]
  69             self.public_ips = []
  70             self.size = testutil.MockSize(1)
  71             self.state = 0
  72
  73
  74     def build_monitor(self, side_effect, *args, **kwargs):
  75         super(CloudNodeListMonitorActorTestCase, self).build_monitor(
  76             *args, **kwargs)
  77         self.client.list_nodes.side_effect = side_effect
  78
  79     def test_id_is_subscription_key(self):
  80         node = self.MockNode(1)
  81         mock_calc = mock.MagicMock()
  82         mock_calc.find_size.return_value = testutil.MockSize(2)
  83         self.build_monitor([[node]], mock_calc)
  84         self.monitor.subscribe_to('1', self.subscriber).get(self.TIMEOUT)
  85         self.stop_proxy(self.monitor)
  86         self.subscriber.assert_called_with(node)
  87         self.assertEqual(testutil.MockSize(2), node.size)
  88
  89 if __name__ == '__main__':
  90     unittest.main()