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
   7 import arvnodeman.nodelist as nodelist
   8 from . import testutil
   9
  10 class ArvadosNodeListMonitorActorTestCase(testutil.RemotePollLoopActorTestMixin,
  11                                           unittest.TestCase):
  12     TEST_CLASS = nodelist.ArvadosNodeListMonitorActor
  13
  14     def build_monitor(self, side_effect, *args, **kwargs):
  15         super(ArvadosNodeListMonitorActorTestCase, self).build_monitor(
  16             *args, **kwargs)
  17         self.client.nodes().list().execute.side_effect = side_effect
  18
  19     def test_uuid_is_subscription_key(self):
  20         node = testutil.arvados_node_mock()
  21         self.build_monitor([{'items': [node]}])
  22         self.monitor.subscribe_to(node['uuid'],
  23                                   self.subscriber).get(self.TIMEOUT)
  24         self.stop_proxy(self.monitor)
  25         self.subscriber.assert_called_with(node)
  26
  27
  28 class CloudNodeListMonitorActorTestCase(testutil.RemotePollLoopActorTestMixin,
  29                                         unittest.TestCase):
  30     TEST_CLASS = nodelist.CloudNodeListMonitorActor
  31
  32     class MockNode(object):
  33         def __init__(self, count):
  34             self.id = str(count)
  35             self.name = 'test{}.example.com'.format(count)
  36             self.private_ips = ['10.0.0.{}'.format(count)]
  37             self.public_ips = []
  38             self.size = None
  39             self.state = 0
  40
  41
  42     def build_monitor(self, side_effect, *args, **kwargs):
  43         super(CloudNodeListMonitorActorTestCase, self).build_monitor(
  44             *args, **kwargs)
  45         self.client.list_nodes.side_effect = side_effect
  46
  47     def test_id_is_subscription_key(self):
  48         node = self.MockNode(1)
  49         self.build_monitor([[node]])
  50         self.monitor.subscribe_to('1', self.subscriber).get(self.TIMEOUT)
  51         self.stop_proxy(self.monitor)
  52         self.subscriber.assert_called_with(node)
  53
  54
  55 if __name__ == '__main__':
  56     unittest.main()
  57