Infinidat/infi.clickhouse_orm
1
1
Fork 0
mirror of https://github.com/Infinidat/infi.clickhouse_orm.git synced 2025-10-31 07:47:31 +03:00
Code Issues Packages Projects Releases Wiki Activity

Fix tests that fail on older ClickHouse versions

Browse Source
This commit is contained in:
Itai Shirav 2020-05-29 11:08:30 +03:00
parent 3fb3936a8a
commit 5d97b4c84a
1 changed files with 14 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
tests/test_funcs.py
View File

@ -45,11 +45,17 @@ class FuncsTestCase(TestCaseWithData):
def _test_aggr(self, func, expected_value=NO_VALUE): def _test_aggr(self, func, expected_value=NO_VALUE):
qs = Person.objects_in(self.database).aggregate(value=func) qs = Person.objects_in(self.database).aggregate(value=func)
logging.info(qs.as_sql()) logging.info(qs.as_sql())
result = list(qs) try:
logging.info('\t==> %s', result[0].value if result else '<empty>') result = list(qs)
if expected_value != NO_VALUE: logging.info('\t==> %s', result[0].value if result else '<empty>')
self.assertEqual(result[0].value, expected_value) if expected_value != NO_VALUE:
return result[0].value if result else None self.assertEqual(result[0].value, expected_value)
return result[0].value if result else None
except ServerError as e:
if 'Unknown function' in e.message:
logging.warning(e.message)
return # ignore functions that don't exist in the used ClickHouse version
raise
def test_func_to_sql(self): def test_func_to_sql(self):
# No args # No args
@ -584,8 +590,9 @@ class FuncsTestCase(TestCaseWithData):
self._test_func(F.rand(17)) self._test_func(F.rand(17))
self._test_func(F.rand64()) self._test_func(F.rand64())
self._test_func(F.rand64(17)) self._test_func(F.rand64(17))
self._test_func(F.randConstant()) if self.database.server_version >= (19, 15): # buggy in older versions
self._test_func(F.randConstant(17)) self._test_func(F.randConstant())
self._test_func(F.randConstant(17))
def test_encoding_functions(self): def test_encoding_functions(self):
self._test_func(F.hex(F.unhex('0FA1')), '0FA1') self._test_func(F.hex(F.unhex('0FA1')), '0FA1')