Infinidat/infi.clickhouse_orm
1
1
Fork 0
mirror of https://github.com/Infinidat/infi.clickhouse_orm.git synced 2024-11-28 19:53:44 +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:32:49 +03:00
parent 5d97b4c84a
commit 0c2d0f0ffd
4 changed files with 9 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/infi/clickhouse_orm/funcs.py
View File

@ -431,7 +431,7 @@ class F(Cond, FunctionOperatorsMixin, metaclass=FMeta):
@staticmethod @staticmethod
def toQuarter(d, timezone=''): def toQuarter(d, timezone=''):
return F('toQuarter', d, timezone) return F('toQuarter', d, timezone) if timezone else F('toQuarter', d)
@staticmethod @staticmethod
def toMonth(d): def toMonth(d):

3
tests/test_compressed_fields.py
View File

@ -89,8 +89,9 @@ class CompressedFieldsTestCase(unittest.TestCase):
"datetime_field": datetime.datetime(1970, 1, 1, 0, 0, 0, tzinfo=pytz.utc) "datetime_field": datetime.datetime(1970, 1, 1, 0, 0, 0, tzinfo=pytz.utc)
}) })
# This test will fail on clickhouse version < 19.1.16, use skip test
def test_confirm_compression_codec(self): def test_confirm_compression_codec(self):
if self.database.server_version < (19, 17):
raise unittest.SkipTest('ClickHouse version too old')
instance = CompressedModel(date_field='1973-12-06', int64_field='100', float_field='7', array_field='[a,b,c]') instance = CompressedModel(date_field='1973-12-06', int64_field='100', float_field='7', array_field='[a,b,c]')
self.database.insert([instance]) self.database.insert([instance])
r = self.database.raw("select name, compression_codec from system.columns where table = '{}' and database='{}' FORMAT TabSeparatedWithNamesAndTypes".format(instance.table_name(), self.database.db_name)) r = self.database.raw("select name, compression_codec from system.columns where table = '{}' and database='{}' FORMAT TabSeparatedWithNamesAndTypes".format(instance.table_name(), self.database.db_name))

4
tests/test_ip_fields.py
View File

@ -16,6 +16,8 @@ class IPFieldsTest(unittest.TestCase):
self.database.drop_database() self.database.drop_database()
def test_ipv4_field(self): def test_ipv4_field(self):
if self.database.server_version < (19, 17):
raise unittest.SkipTest('ClickHouse version too old')
# Create a model # Create a model
class TestModel(Model): class TestModel(Model):
i = Int16Field() i = Int16Field()
@ -40,6 +42,8 @@ class IPFieldsTest(unittest.TestCase):
TestModel(i=1, f=value) TestModel(i=1, f=value)
def test_ipv6_field(self): def test_ipv6_field(self):
if self.database.server_version < (19, 17):
raise unittest.SkipTest('ClickHouse version too old')
# Create a model # Create a model
class TestModel(Model): class TestModel(Model):
i = Int16Field() i = Int16Field()

2
tests/test_querysets.py
View File

@ -471,6 +471,8 @@ class AggregateTestCase(TestCaseWithData):
self.assertEqual(qs.conditions_as_sql(), 'the__next__number > 1') self.assertEqual(qs.conditions_as_sql(), 'the__next__number > 1')
def test_limit_by(self): def test_limit_by(self):
if self.database.server_version < (19, 17):
raise unittest.SkipTest('ClickHouse version too old')
# Test without offset # Test without offset
qs = Person.objects_in(self.database).aggregate('first_name', 'last_name', 'height', n='count()').\ qs = Person.objects_in(self.database).aggregate('first_name', 'last_name', 'height', n='count()').\
order_by('first_name', '-height').limit_by(1, 'first_name') order_by('first_name', '-height').limit_by(1, 'first_name')