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 from __future__ import absolute_import
   7 import unittest
   8 from third_party.influxdb.influxdb_line_protocol import make_lines
   9
  10
  11 class TestLineProtocol(unittest.TestCase):
  12
  13     def test_make_lines(self):
  14         data = {
  15             "tags": {
  16                 "empty_tag": "",
  17                 "none_tag": None,
  18                 "integer_tag": 2,
  19                 "string_tag": "hello"
  20             },
  21             "points": [
  22                 {
  23                     "measurement": "test",
  24                     "fields": {
  25                         "string_val": "hello!",
  26                         "int_val": 1,
  27                         "float_val": 1.1,
  28                         "none_field": None,
  29                         "bool_val": True,
  30                     }
  31                 }
  32             ]
  33         }
  34
  35         self.assertEqual(
  36             make_lines(data),
  37             'test,integer_tag=2,string_tag=hello '
  38             'bool_val=True,float_val=1.1,int_val=1i,string_val="hello!"\n'
  39         )
  40
  41     def test_string_val_newline(self):
  42         data = {
  43             "points": [
  44                 {
  45                     "measurement": "m1",
  46                     "fields": {
  47                         "multi_line": "line1\nline1\nline3"
  48                     }
  49                 }
  50             ]
  51         }
  52
  53         self.assertEqual(
  54             make_lines(data),
  55             'm1 multi_line="line1\\nline1\\nline3"\n'
  56         )