services/nodemanager/tests/test_status.py

   1 #!/usr/bin/env python
   2
   3 from __future__ import absolute_import, print_function
   4 from future import standard_library
   5
   6 import requests
   7 import unittest
   8
   9 import arvnodeman.status as status
  10 import arvnodeman.config as config
  11
  12
  13 class TestServer(object):
  14     def __enter__(self):
  15         cfg = config.NodeManagerConfig()
  16         cfg.set('Manage', 'port', '0')
  17         cfg.set('Manage', 'address', '127.0.0.1')
  18         self.srv = status.Server(cfg)
  19         self.srv.start()
  20         addr, port = self.srv.server_address
  21         self.srv_base = 'http://127.0.0.1:'+str(port)
  22         return self
  23
  24     def __exit__(self, exc_type, exc_value, traceback):
  25         self.srv.shutdown()
  26
  27     def get_status_response(self):
  28         return requests.get(self.srv_base+'/status.json')
  29
  30     def get_status(self):
  31         return self.get_status_response().json()
  32
  33
  34 class StatusServerUpdates(unittest.TestCase):
  35     def test_updates(self):
  36         with TestServer() as srv:
  37             for n in [1, 2, 3]:
  38                 status.tracker.update({'nodes_'+str(n): n})
  39                 r = srv.get_status_response()
  40                 self.assertEqual(200, r.status_code)
  41                 self.assertEqual('application/json', r.headers['content-type'])
  42                 resp = r.json()
  43                 self.assertEqual(n, resp['nodes_'+str(n)])
  44             self.assertEqual(1, resp['nodes_1'])
  45
  46
  47 class StatusServerDisabled(unittest.TestCase):
  48     def test_config_disabled(self):
  49         cfg = config.NodeManagerConfig()
  50         cfg.set('Manage', 'port', '-1')
  51         cfg.set('Manage', 'address', '127.0.0.1')
  52         self.srv = status.Server(cfg)
  53         self.srv.start()
  54         self.assertFalse(self.srv.enabled)
  55         self.assertFalse(getattr(self.srv, '_thread', False))