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

Skip mutations test on old ClickHouse versions

Browse Source
This commit is contained in:
Itai Shirav 2020-06-27 00:26:21 +03:00
parent 436c296609
commit bc900c2ef1
1 changed files with 7 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
tests/test_mutations.py
View File

@ -1,3 +1,4 @@
import unittest
from infi.clickhouse_orm import F
from .base_test_with_data import *
from time import sleep
@ -5,19 +6,23 @@ from time import sleep
class MutationsTestCase(TestCaseWithData):
def setUp(self):
super().setUp()
if self.database.server_version < (18,):
raise unittest.SkipTest('ClickHouse version too old')
self._insert_all()
def _wait_for_mutations(self):
sql = 'SELECT * FROM system.mutations WHERE is_done = 0'
while list(self.database.raw(sql)):
sleep(0.25)
def test_delete_all(self):
self._insert_all()
Person.objects_in(self.database).delete()
self._wait_for_mutations()
self.assertFalse(Person.objects_in(self.database))
def test_delete_with_where_cond(self):
self._insert_all()
cond = Person.first_name == 'Cassady'
self.assertTrue(Person.objects_in(self.database).filter(cond))
Person.objects_in(self.database).filter(cond).delete()
@ -26,7 +31,6 @@ class MutationsTestCase(TestCaseWithData):
self.assertTrue(Person.objects_in(self.database).exclude(cond))
def test_delete_with_prewhere_cond(self):
self._insert_all()
cond = F.toYear(Person.birthday) == 1977
self.assertTrue(Person.objects_in(self.database).filter(cond))
Person.objects_in(self.database).filter(cond, prewhere=True).delete()
@ -35,35 +39,30 @@ class MutationsTestCase(TestCaseWithData):
self.assertTrue(Person.objects_in(self.database).exclude(cond))
def test_update_all(self):
self._insert_all()
Person.objects_in(self.database).update(height=0)
self._wait_for_mutations()
for p in Person.objects_in(self.database): print(p.height)
self.assertFalse(Person.objects_in(self.database).exclude(height=0))
def test_update_with_where_cond(self):
self._insert_all()
cond = Person.first_name == 'Cassady'
Person.objects_in(self.database).filter(cond).update(height=0)
self._wait_for_mutations()
self.assertFalse(Person.objects_in(self.database).filter(cond).exclude(height=0))
def test_update_with_prewhere_cond(self):
self._insert_all()
cond = F.toYear(Person.birthday) == 1977
Person.objects_in(self.database).filter(cond, prewhere=True).update(height=0)
self._wait_for_mutations()
self.assertFalse(Person.objects_in(self.database).filter(cond).exclude(height=0))
def test_update_multiple_fields(self):
self._insert_all()
Person.objects_in(self.database).update(height=0, passport=None)
self._wait_for_mutations()
self.assertFalse(Person.objects_in(self.database).exclude(height=0))
self.assertFalse(Person.objects_in(self.database).exclude(passport=None))
def test_chained_update(self):
self._insert_all()
Person.objects_in(self.database).update(height=F.rand()).update(passport=99999)
self._wait_for_mutations()
self.assertFalse(Person.objects_in(self.database).exclude(passport=99999))