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

Compatibility: update tests for 21.3+

Browse Source
This commit is contained in:
olliemath 2021-08-14 12:24:33 +01:00
parent 1a4785d61e
commit a924f6c7e2
2 changed files with 14 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
tests/test_database.py
View File

@ -296,6 +296,10 @@ class DatabaseTestCase(TestCaseWithData):
self.assertTrue(model.is_system_model())
self.assertTrue(model.is_read_only())
self.assertEqual(model.table_name(), row.name)
if row.name == "distributed_ddl_queue":
continue # Since zookeeper is not set up in our tests
# Read a few records
try:
list(model.objects_in(self.database)[:10])

14
tests/test_dictionaries.py
View File

@ -14,14 +14,17 @@ class DictionaryTestMixin:
def tearDown(self):
self.database.drop_database()
def _test_func(self, func, expected_value):
def _call_func(self, func):
sql = "SELECT %s AS value" % func.to_sql()
logging.info(sql)
result = list(self.database.select(sql))
logging.info("\t==> %s", result[0].value if result else "<empty>")
return result
def _test_func(self, func, expected_value):
result = self._call_func(func)
print("Comparing %s to %s" % (result[0].value, expected_value))
self.assertEqual(result[0].value, expected_value)
return result[0].value if result else None
assert result[0].value == expected_value
class SimpleDictionaryTest(DictionaryTestMixin, unittest.TestCase):
@ -114,7 +117,10 @@ class HierarchicalDictionaryTest(DictionaryTestMixin, unittest.TestCase):
def test_dictgethierarchy(self):
self._test_func(F.dictGetHierarchy(self.dict_name, F.toUInt64(3)), [3, 2, 1])
self._test_func(F.dictGetHierarchy(self.dict_name, F.toUInt64(99)), [99])
# Default behaviour changed in CH, but we're not really testing that
default = self._call_func(F.dictGetHierarchy(self.dict_name, F.toUInt64(99)))
assert isinstance(default, list)
assert len(default) <= 1 # either [] or [99]
def test_dictisin(self):
self._test_func(F.dictIsIn(self.dict_name, F.toUInt64(3), F.toUInt64(1)), 1)