1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
|
# -*- coding: utf-8 -*-
"""Chunked JSON test."""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals
import unittest
from influxdb import chunked_json
class TestChunkJson(unittest.TestCase):
"""Set up the TestChunkJson object."""
@classmethod
def setUpClass(cls):
"""Initialize the TestChunkJson object."""
super(TestChunkJson, cls).setUpClass()
def test_load(self):
"""Test reading a sequence of JSON values from a string."""
example_response = \
'{"results": [{"series": [{"measurement": "sdfsdfsdf", ' \
'"columns": ["time", "value"], "values": ' \
'[["2009-11-10T23:00:00Z", 0.64]]}]}, {"series": ' \
'[{"measurement": "cpu_load_short", "columns": ["time", "value"],'\
'"values": [["2009-11-10T23:00:00Z", 0.64]]}]}]}'
res = list(chunked_json.loads(example_response))
# import ipdb; ipdb.set_trace()
self.assertListEqual(
[
{
'results': [
{'series': [{
'values': [['2009-11-10T23:00:00Z', 0.64]],
'measurement': 'sdfsdfsdf',
'columns':
['time', 'value']}]},
{'series': [{
'values': [['2009-11-10T23:00:00Z', 0.64]],
'measurement': 'cpu_load_short',
'columns': ['time', 'value']}]}
]
}
],
res
)
|
