infi.clickhouse_orm/tests/test_system_models.py

102 lines
3.3 KiB
Python
Raw Normal View History

2017-08-16 23:48:18 +03:00
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):
2017-08-20 09:46:00 +03:00
BACKUP_DIRS = ['/var/lib/clickhouse/shadow', '/opt/clickhouse/shadow/']
def setUp(self):
self.database = Database('test-db')
self.database.create_table(TestTable)
self.database.insert([TestTable(date_field=date.today())])
def tearDown(self):
self.database.drop_database()
def _get_backups(self):
2017-08-20 09:46:00 +03:00
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)), 1)
def test_get_active(self):
parts = list(SystemPart.get_active(self.database))
self.assertEqual(len(parts), 1)
parts[0].detach()
self.assertEqual(len(list(SystemPart.get_active(self.database))), 0)
def test_get_conditions(self):
parts = list(SystemPart.get(self.database, conditions="table='testtable'"))
self.assertEqual(len(parts), 1)
2017-06-06 18:00:15 +03:00
parts = list(SystemPart.get(self.database, conditions=u"table='othertable'"))
self.assertEqual(len(parts), 0)
def test_attach_detach(self):
parts = list(SystemPart.get_active(self.database))
self.assertEqual(len(parts), 1)
parts[0].detach()
self.assertEqual(len(list(SystemPart.get_active(self.database))), 0)
parts[0].attach()
self.assertEqual(len(list(SystemPart.get_active(self.database))), 1)
def test_drop(self):
parts = list(SystemPart.get_active(self.database))
parts[0].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())
parts[0].freeze()
backups = set(self._get_backups())
self.assertEqual(len(backups), len(prev_backups) + 1)
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 SystemTestModel(Model):
system = True