infi.clickhouse_orm/tests/test_system_models.py

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.insert([TestTable(date_field=date.today())])

    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)), 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)
        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