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'], self.subscriber)
  23         self.wait_for_call(self.subscriber)
  24         self.subscriber.assert_called_with(node)
  25
  26
  27 class CloudNodeListMonitorActorTestCase(testutil.RemotePollLoopActorTestMixin,
  28                                         unittest.TestCase):
  29     TEST_CLASS = nodelist.CloudNodeListMonitorActor
  30
  31     class MockNode(object):
  32         def __init__(self, count):
  33             self.id = str(count)
  34             self.name = 'test{}.example.com'.format(count)
  35             self.private_ips = ['10.0.0.{}'.format(count)]
  36             self.public_ips = []
  37             self.size = None
  38             self.state = 0
  39
  40
  41     def build_monitor(self, side_effect, *args, **kwargs):
  42         super(CloudNodeListMonitorActorTestCase, self).build_monitor(
  43             *args, **kwargs)
  44         self.client.list_nodes.side_effect = side_effect
  45
  46     def test_id_is_subscription_key(self):
  47         node = self.MockNode(1)
  48         self.build_monitor([[node]])
  49         self.monitor.subscribe_to('1', self.subscriber)
  50         self.wait_for_call(self.subscriber)
  51         self.subscriber.assert_called_with(node)
  52
  53
  54 if __name__ == '__main__':
  55     unittest.main()
  56