tests/unit/dispatcher/test_influxdb_line_protocol.py

   1 # Unittest for yardstick.dispatcher.influxdb_line_protocol
   2
   3 # yardstick comment: this file is a modified copy of
   4 # influxdb-python/influxdb/tests/test_line_protocol.py
   5
   6 import unittest
   7 from yardstick.dispatcher.influxdb_line_protocol import make_lines
   8
   9
  10 class TestLineProtocol(unittest.TestCase):
  11
  12     def test_make_lines(self):
  13         data = {
  14             "tags": {
  15                 "empty_tag": "",
  16                 "none_tag": None,
  17                 "integer_tag": 2,
  18                 "string_tag": "hello"
  19             },
  20             "points": [
  21                 {
  22                     "measurement": "test",
  23                     "fields": {
  24                         "string_val": "hello!",
  25                         "int_val": 1,
  26                         "float_val": 1.1,
  27                         "none_field": None,
  28                         "bool_val": True,
  29                     }
  30                 }
  31             ]
  32         }
  33
  34         self.assertEqual(
  35             make_lines(data),
  36             'test,integer_tag=2,string_tag=hello '
  37             'bool_val=True,float_val=1.1,int_val=1i,string_val="hello!"\n'
  38         )
  39
  40     def test_string_val_newline(self):
  41         data = {
  42             "points": [
  43                 {
  44                     "measurement": "m1",
  45                     "fields": {
  46                         "multi_line": "line1\nline1\nline3"
  47                     }
  48                 }
  49             ]
  50         }
  51
  52         self.assertEqual(
  53             make_lines(data),
  54             'm1 multi_line="line1\\nline1\\nline3"\n'
  55         )