parameters tests

Author: vgvoleg

examples/query-service/basic_example.py

@@ -94,9 +94,7 @@ def callee(session: ydb.QuerySessionSync):
 
         for value in values:
             print(f"value: {value}")
-            with session.transaction().execute(
-                query=query_print, parameters={'$a': value}, commit_tx=True
-            ) as results:
+            with session.transaction().execute(query=query_print, parameters={'$a': value}, commit_tx=True) as results:
                 for result_set in results:
                     print(f"rows: {str(result_set.rows)}")
 

tests/query/test_query_parameters.py

@@ -0,0 +1,56 @@
+import pytest
+import ydb
+
+
+query = """SELECT $a AS value"""
+
+
+def test_select_implicit_int(pool: ydb.QuerySessionPool):
+    expected_value = 111
+    res = pool.execute_with_retries(query, parameters={'$a': expected_value})
+    actual_value = res[0].rows[0]['value']
+    assert expected_value == actual_value
+
+
+def test_select_implicit_float(pool: ydb.QuerySessionPool):
+    expected_value = 11.1
+    res = pool.execute_with_retries(query, parameters={'$a': expected_value})
+    actual_value = res[0].rows[0]['value']
+    assert expected_value == pytest.approx(actual_value)
+
+
+def test_select_implicit_bool(pool: ydb.QuerySessionPool):
+    expected_value = False
+    res = pool.execute_with_retries(query, parameters={'$a': expected_value})
+    actual_value = res[0].rows[0]['value']
+    assert expected_value == actual_value
+
+
+def test_select_implicit_str(pool: ydb.QuerySessionPool):
+    expected_value = "text"
+    res = pool.execute_with_retries(query, parameters={'$a': expected_value})
+    actual_value = res[0].rows[0]['value']
+    assert expected_value == actual_value
+
+
+def test_select_explicit_primitive(pool: ydb.QuerySessionPool):
+    expected_value = 111
+    res = pool.execute_with_retries(query, parameters={'$a': (expected_value, ydb.PrimitiveType.Int64)})
+    actual_value = res[0].rows[0]['value']
+    assert expected_value == actual_value
+
+
+def test_select_explicit_list(pool: ydb.QuerySessionPool):
+    expected_value = [1, 2, 3]
+    type_ = ydb.ListType(ydb.PrimitiveType.Int64)
+    res = pool.execute_with_retries(query, parameters={'$a': (expected_value, type_)})
+    actual_value = res[0].rows[0]['value']
+    assert expected_value == actual_value
+
+
+def test_select_explicit_dict(pool: ydb.QuerySessionPool):
+    expected_value = {'key': 'value'}
+    type_ = ydb.DictType(ydb.PrimitiveType.Utf8, ydb.PrimitiveType.Utf8)
+    res = pool.execute_with_retries(query, parameters={'$a': (expected_value, type_)})
+    actual_value = res[0].rows[0]['value']
+    assert expected_value == actual_value