test: adds integration test for client.query_async()

Author: karel-rehor

tests/test_influxdb_client_3_integration.py

@@ -10,6 +10,7 @@
 from pyarrow._flight import FlightError
 
 from influxdb_client_3 import InfluxDBClient3, InfluxDBError, write_client_options, WriteOptions
+from tests.util import asyncio_run, lp_to_py_object
 
 
 def random_hex(len=6):
@@ -249,3 +250,27 @@ def test_verify_ssl_false(self):
                 assert len(list_results) > 0
             finally:
                 self.remove_test_cert(cert_file)
+
+    @asyncio_run
+    async def test_verify_query_async(self):
+        measurement = f'test{random_hex(6)}'
+        data = []
+        lp_template = "%s,location=%s val=%f,ival=%di,index=%di %d"
+        data_size = 10
+        interval = 1_000_000_000 * 10
+        ts = time.time_ns() - interval * data_size
+        locations = ['springfield', 'gotham', 'balbec', 'yonville']
+        for i in range(data_size):
+            data.append(lp_template % (measurement, locations[random.randint(0, len(locations) - 1)],
+                                       random.random() * 10,
+                                       random.randint(0, 6), i, ts))
+            ts = ts + interval
+
+        self.client.write(data)
+        query = f"SELECT * FROM \"{measurement}\" ORDER BY time DESC"
+
+        result = await self.client.query_async(query)
+
+        result_list = result.to_pylist()
+        for item in data:
+            assert lp_to_py_object(item) in result_list, f"original lp data \"{item}\" should be in result list"

tests/util/__init__.py

@@ -4,6 +4,8 @@
 import sys
 import traceback
 
+import pandas
+
 
 def asyncio_run(async_func):
     def wrapper(*args, **kwargs):
@@ -15,3 +17,44 @@ def wrapper(*args, **kwargs):
 
     wrapper.__signature__ = inspect.signature(async_func)
     return wrapper
+
+
+def lp_to_py_object(lp: str):
+    """
+    Result format matches the format of objects returned in pyarrow.Table.to_pylist.
+
+    For verifying test data returned from queries.
+
+    :param lp: a lineprotocol formatted string
+    :return: a list object
+    """
+    result = {}
+    groups = lp.split(' ')
+
+    tags = groups[0].split(',')
+    tags.remove(tags[0])
+    for tag in tags:
+        t_set = tag.split('=')
+        result[t_set[0]] = t_set[1]
+
+    fields = groups[1].split(',')
+    for field in fields:
+        f_set = field.split('=')
+        lastchar = f_set[1][len(f_set[1]) - 1]
+        match lastchar:
+            case 'i': # integer
+                result[f_set[0]] = int(f_set[1].replace('i',''))
+            case 'u': # unsigned integer
+                result[f_set[0]] = int(f_set[1].replace('u',''))
+            case '"': # string
+                result[f_set[0]] = f_set[1].replace('"',"")
+            case 'e' | 'E' | 't' | 'T' | 'f' | 'F':
+                if f_set[1][0].lower() == 't':
+                    result[f_set[0]] = True
+                else:
+                    result[f_set[0]] = False
+            case _: # assume float
+                result[f_set[0]] = float(f_set[1])
+
+    result['time'] = pandas.Timestamp(int(groups[2]))
+    return result