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 from __future__ import absolute_import
  10 import unittest
  11 import mock
  12 import uuid
  13 import datetime
  14
  15 from api.utils import influx
  16
  17 import six.moves.configparser as ConfigParser
  18
  19
  20 class GetDataDbClientTestCase(unittest.TestCase):
  21
  22     @mock.patch('api.utils.influx.ConfigParser')
  23     def test_get_data_db_client_dispatcher_not_influxdb(self, mock_parser):
  24         mock_parser.ConfigParser().get.return_value = 'file'
  25         # reset exception to avoid
  26         # TypeError: catching classes that do not inherit from BaseException
  27         mock_parser.NoOptionError = ConfigParser.NoOptionError
  28         try:
  29             influx.get_data_db_client()
  30         except Exception as e:
  31             self.assertIsInstance(e, RuntimeError)
  32
  33
  34 class GetIpTestCase(unittest.TestCase):
  35
  36     def test_get_url(self):
  37         url = 'http://localhost:8086/hello'
  38         output = influx._get_ip(url)
  39
  40         result = 'localhost'
  41         self.assertEqual(result, output)
  42
  43
  44 class WriteDataTestCase(unittest.TestCase):
  45
  46     @mock.patch('api.utils.influx.get_data_db_client')
  47     def test_write_data(self, mock_get_client):
  48         measurement = 'tasklist'
  49         field = {'status': 1}
  50         timestamp = datetime.datetime.now()
  51         tags = {'task_id': str(uuid.uuid4())}
  52
  53         influx._write_data(measurement, field, timestamp, tags)
  54         mock_get_client.assert_called_with()
  55
  56
  57 class WriteDataTasklistTestCase(unittest.TestCase):
  58
  59     @mock.patch('api.utils.influx._write_data')
  60     def test_write_data_tasklist(self, mock_write_data):
  61         task_id = str(uuid.uuid4())
  62         timestamp = datetime.datetime.now()
  63         status = 1
  64         influx.write_data_tasklist(task_id, timestamp, status)
  65
  66         field = {'status': status, 'error': ''}
  67         tags = {'task_id': task_id}
  68         mock_write_data.assert_called_with('tasklist', field, timestamp, tags)
  69
  70
  71 class QueryTestCase(unittest.TestCase):
  72
  73     @mock.patch('api.utils.influx.ConfigParser')
  74     def test_query_dispatcher_not_influxdb(self, mock_parser):
  75         mock_parser.ConfigParser().get.return_value = 'file'
  76         # reset exception to avoid
  77         # TypeError: catching classes that do not inherit from BaseException
  78         mock_parser.NoOptionError = ConfigParser.NoOptionError
  79         try:
  80             sql = 'select * form tasklist'
  81             influx.query(sql)
  82         except Exception as e:
  83             self.assertIsInstance(e, RuntimeError)
  84
  85
  86 def main():
  87     unittest.main()
  88
  89
  90 if __name__ == '__main__':
  91     main()