infi.clickhouse_orm/tests/test_system_models.py
Itai Shirav ab0755ad90 - Rename Model.system to Model._system to prevent collision with a column that has the same name
- Rename `Model.readonly` to `Model._readonly` to prevent collision with a column that has the same name
- The `field_names` argument to `Model.to_tsv` is now mandatory
2018-05-14 07:09:57 -04:00

120 lines
3.9 KiB
Python

from __future__ import unicode_literals
import unittest
from datetime import date
import os
from infi.clickhouse_orm.database import Database, DatabaseException
from infi.clickhouse_orm.engines import *
from infi.clickhouse_orm.fields import *
from infi.clickhouse_orm.models import Model
from infi.clickhouse_orm.system_models import SystemPart
class SystemTest(unittest.TestCase):
def setUp(self):
self.database = Database('test-db')
def tearDown(self):
self.database.drop_database()
def test_insert_system(self):
m = SystemPart()
with self.assertRaises(DatabaseException):
self.database.insert([m])
def test_create_readonly_table(self):
with self.assertRaises(DatabaseException):
self.database.create_table(SystemTestModel)
def test_drop_readonly_table(self):
with self.assertRaises(DatabaseException):
self.database.drop_table(SystemTestModel)
class SystemPartTest(unittest.TestCase):
BACKUP_DIRS = ['/var/lib/clickhouse/shadow', '/opt/clickhouse/shadow/']
def setUp(self):
self.database = Database('test-db')
self.database.create_table(TestTable)
self.database.create_table(CustomPartitionedTable)
self.database.insert([TestTable(date_field=date.today())])
self.database.insert([CustomPartitionedTable(date_field=date.today(), group_field=13)])
def tearDown(self):
self.database.drop_database()
def _get_backups(self):
for dir in self.BACKUP_DIRS:
if os.path.exists(dir):
_, dirnames, _ = next(os.walk(dir))
return dirnames
raise unittest.SkipTest('Cannot find backups dir')
def test_get_all(self):
parts = SystemPart.get(self.database)
self.assertEqual(len(list(parts)), 2)
def test_get_active(self):
parts = list(SystemPart.get_active(self.database))
self.assertEqual(len(parts), 2)
parts[0].detach()
self.assertEqual(len(list(SystemPart.get_active(self.database))), 1)
def test_get_conditions(self):
parts = list(SystemPart.get(self.database, conditions="table='testtable'"))
self.assertEqual(len(parts), 1)
parts = list(SystemPart.get(self.database, conditions=u"table='custompartitionedtable'"))
self.assertEqual(len(parts), 1)
parts = list(SystemPart.get(self.database, conditions=u"table='invalidtable'"))
self.assertEqual(len(parts), 0)
def test_attach_detach(self):
parts = list(SystemPart.get_active(self.database))
self.assertEqual(len(parts), 2)
for p in parts:
p.detach()
self.assertEqual(len(list(SystemPart.get_active(self.database))), 0)
for p in parts:
p.attach()
self.assertEqual(len(list(SystemPart.get_active(self.database))), 2)
def test_drop(self):
parts = list(SystemPart.get_active(self.database))
for p in parts:
p.drop()
self.assertEqual(len(list(SystemPart.get_active(self.database))), 0)
def test_freeze(self):
parts = list(SystemPart.get(self.database))
# There can be other backups in the folder
prev_backups = set(self._get_backups())
for p in parts:
p.freeze()
backups = set(self._get_backups())
self.assertEqual(len(backups), len(prev_backups) + 2)
def test_fetch(self):
# TODO Not tested, as I have no replication set
pass
class TestTable(Model):
date_field = DateField()
engine = MergeTree('date_field', ('date_field',))
class CustomPartitionedTable(Model):
date_field = DateField()
group_field = UInt32Field()
engine = MergeTree(order_by=('date_field', 'group_field'), partition_key=('toYYYYMM(date_field)', 'group_field'))
class SystemTestModel(Model):
_system = True