tests/unit/api/utils/test_influx.py

   1 ##############################################################################
   2 # Copyright (c) 2016 Huawei Technologies Co.,Ltd and others.
   3 #
   4 # All rights reserved. This program and the accompanying materials
   5 # are made available under the terms of the Apache License, Version 2.0
   6 # which accompanies this distribution, and is available at
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8 ##############################################################################
   9 import unittest
  10 import mock
  11 import uuid
  12 import datetime
  13
  14 from api.utils import influx
  15
  16
  17 class GetDataDbClientTestCase(unittest.TestCase):
  18
  19     @mock.patch('api.utils.influx.ConfigParser')
  20     def test_get_data_db_client_dispatcher_not_influxdb(self, mock_parser):
  21         mock_parser.ConfigParser().get.return_value = 'file'
  22         try:
  23             influx.get_data_db_client()
  24         except Exception as e:
  25             self.assertIsInstance(e, RuntimeError)
  26
  27
  28 class GetIpTestCase(unittest.TestCase):
  29
  30     def test_get_url(self):
  31         url = 'http://localhost:8086/hello'
  32         output = influx._get_ip(url)
  33
  34         result = 'localhost'
  35         self.assertEqual(result, output)
  36
  37
  38 class WriteDataTestCase(unittest.TestCase):
  39
  40     @mock.patch('api.utils.influx.get_data_db_client')
  41     def test_write_data(self, mock_get_client):
  42         measurement = 'tasklist'
  43         field = {'status': 1}
  44         timestamp = datetime.datetime.now()
  45         tags = {'task_id': str(uuid.uuid4())}
  46
  47         influx._write_data(measurement, field, timestamp, tags)
  48         mock_get_client.assert_called_with()
  49
  50
  51 class WriteDataTasklistTestCase(unittest.TestCase):
  52
  53     @mock.patch('api.utils.influx._write_data')
  54     def test_write_data_tasklist(self, mock_write_data):
  55         task_id = str(uuid.uuid4())
  56         timestamp = datetime.datetime.now()
  57         status = 1
  58         influx.write_data_tasklist(task_id, timestamp, status)
  59
  60         field = {'status': status, 'error': ''}
  61         tags = {'task_id': task_id}
  62         mock_write_data.assert_called_with('tasklist', field, timestamp, tags)
  63
  64
  65 class QueryTestCase(unittest.TestCase):
  66
  67     @mock.patch('api.utils.influx.ConfigParser')
  68     def test_query_dispatcher_not_influxdb(self, mock_parser):
  69         mock_parser.ConfigParser().get.return_value = 'file'
  70         try:
  71             sql = 'select * form tasklist'
  72             influx.query(sql)
  73         except Exception as e:
  74             self.assertIsInstance(e, RuntimeError)
  75
  76
  77 def main():
  78     unittest.main()
  79
  80
  81 if __name__ == '__main__':
  82     main()