tests/unit/api/utils/test_influx.py

   1 import unittest
   2 import mock
   3 import uuid
   4 import datetime
   5
   6 from api.utils import influx
   7
   8
   9 class GetDataDbClientTestCase(unittest.TestCase):
  10
  11     @mock.patch('api.utils.influx.ConfigParser')
  12     def test_get_data_db_client_dispatcher_not_influxdb(self, mock_parser):
  13         mock_parser.ConfigParser().get.return_value = 'file'
  14         try:
  15             influx.get_data_db_client()
  16         except Exception, e:
  17             self.assertIsInstance(e, RuntimeError)
  18
  19
  20 class GetIpTestCase(unittest.TestCase):
  21
  22     def test_get_url(self):
  23         url = 'http://localhost:8086/hello'
  24         output = influx._get_ip(url)
  25
  26         result = 'localhost'
  27         self.assertEqual(result, output)
  28
  29
  30 class WriteDataTestCase(unittest.TestCase):
  31
  32     @mock.patch('api.utils.influx.get_data_db_client')
  33     def test_write_data(self, mock_get_client):
  34         measurement = 'tasklist'
  35         field = {'status': 1}
  36         timestamp = datetime.datetime.now()
  37         tags = {'task_id': str(uuid.uuid4())}
  38
  39         influx._write_data(measurement, field, timestamp, tags)
  40         mock_get_client.assert_called_with()
  41
  42
  43 class WriteDataTasklistTestCase(unittest.TestCase):
  44
  45     @mock.patch('api.utils.influx._write_data')
  46     def test_write_data_tasklist(self, mock_write_data):
  47         task_id = str(uuid.uuid4())
  48         timestamp = datetime.datetime.now()
  49         status = 1
  50         influx.write_data_tasklist(task_id, timestamp, status)
  51
  52         field = {'status': status, 'error': ''}
  53         tags = {'task_id': task_id}
  54         mock_write_data.assert_called_with('tasklist', field, timestamp, tags)
  55
  56
  57 def main():
  58     unittest.main()
  59
  60
  61 if __name__ == '__main__':
  62     main()