mirror of
https://github.com/carrotquest/django-clickhouse.git
synced 2024-11-22 17:16:37 +03:00
Added tests on database tuples operations
This commit is contained in:
parent
90ec66f380
commit
fc100fb066
|
@ -13,6 +13,7 @@ class ClickHouseTestModel(ClickHouseModel):
|
||||||
id = fields.Int32Field()
|
id = fields.Int32Field()
|
||||||
created_date = fields.DateField()
|
created_date = fields.DateField()
|
||||||
value = fields.Int32Field()
|
value = fields.Int32Field()
|
||||||
|
str_field = fields.StringField()
|
||||||
|
|
||||||
engine = ReplacingMergeTree('created_date', ('id',))
|
engine = ReplacingMergeTree('created_date', ('id',))
|
||||||
migrate_replicated_db_aliases = ('default', 'secondary')
|
migrate_replicated_db_aliases = ('default', 'secondary')
|
||||||
|
|
59
tests/test_database.py
Normal file
59
tests/test_database.py
Normal file
|
@ -0,0 +1,59 @@
|
||||||
|
from datetime import date
|
||||||
|
|
||||||
|
from django.test import TestCase
|
||||||
|
|
||||||
|
from django_clickhouse.migrations import migrate_app
|
||||||
|
from tests.clickhouse_models import ClickHouseTestModel
|
||||||
|
|
||||||
|
|
||||||
|
class CollapsingMergeTreeTest(TestCase):
|
||||||
|
maxDiff = None
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
self.db = ClickHouseTestModel.get_database(for_write=True)
|
||||||
|
self.db.drop_database()
|
||||||
|
self.db.create_database()
|
||||||
|
migrate_app('tests', 'default')
|
||||||
|
|
||||||
|
def test_insert_tuples(self):
|
||||||
|
tuple_class = ClickHouseTestModel.get_tuple_class()
|
||||||
|
data = [
|
||||||
|
tuple_class(id=i, created_date=date.today(), value=i, str_field=str(i))
|
||||||
|
for i in range(10)
|
||||||
|
]
|
||||||
|
self.db.insert_tuples(ClickHouseTestModel, data)
|
||||||
|
|
||||||
|
qs = ClickHouseTestModel.objects.order_by('id').all()
|
||||||
|
self.assertListEqual([{
|
||||||
|
'id': i,
|
||||||
|
'created_date': date.today(),
|
||||||
|
'value': i,
|
||||||
|
'str_field': str(i)
|
||||||
|
} for i in range(10)], [item.to_dict() for item in qs])
|
||||||
|
|
||||||
|
def test_insert_tuples_special_characters(self):
|
||||||
|
tuple_class = ClickHouseTestModel.get_tuple_class()
|
||||||
|
data = [tuple_class(id=1, created_date=date.today(), value=1, str_field='\t')]
|
||||||
|
self.db.insert_tuples(ClickHouseTestModel, data)
|
||||||
|
|
||||||
|
item = ClickHouseTestModel.objects.filter(id=1)[0]
|
||||||
|
self.assertEqual('\t', item.str_field)
|
||||||
|
|
||||||
|
data = [tuple_class(id=2, created_date=date.today(), value=2, str_field='\n')]
|
||||||
|
self.db.insert_tuples(ClickHouseTestModel, data)
|
||||||
|
|
||||||
|
item = ClickHouseTestModel.objects.filter(id=2)[0]
|
||||||
|
self.assertEqual('\n', item.str_field)
|
||||||
|
|
||||||
|
def test_select_tuples(self):
|
||||||
|
ClickHouseTestModel.objects.bulk_create([
|
||||||
|
ClickHouseTestModel(id=i, created_date=date.today(), value=i, str_field=str(i))
|
||||||
|
for i in range(10)
|
||||||
|
])
|
||||||
|
|
||||||
|
res = self.db.select_tuples('SELECT * FROM $table ORDER BY id', ClickHouseTestModel)
|
||||||
|
tuple_class = ClickHouseTestModel.get_tuple_class()
|
||||||
|
self.assertListEqual([
|
||||||
|
tuple_class(id=i, created_date=date.today(), value=i, str_field=str(i))
|
||||||
|
for i in range(10)
|
||||||
|
], list(res))
|
Loading…
Reference in New Issue
Block a user