mirror of
https://github.com/Infinidat/infi.clickhouse_orm.git
synced 2024-09-21 03:08:53 +03:00
de9f64cd3a
1) Divided readonly and system flags of Field model. Readonly flag only restricts insert operations, while system flag restricts also create and drop table operations 2) Added Merge engine and tests for it 3) Added docs for Merge engine 4) Added opportunity to make Field readonly. This is useful for "virtual" columns (https://clickhouse.yandex/docs/en/single/index.html#virtual-columns)
55 lines
1.8 KiB
Python
55 lines
1.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
from __future__ import unicode_literals
|
|
|
|
from infi.clickhouse_orm.database import DatabaseException
|
|
from .base_test_with_data import *
|
|
|
|
|
|
class ReadonlyTestCase(TestCaseWithData):
|
|
|
|
def _test_readonly_db(self, username):
|
|
self._insert_and_check(self._sample_data(), len(data))
|
|
orig_database = self.database
|
|
try:
|
|
self.database = Database(orig_database.db_name, username=username, readonly=True)
|
|
with self.assertRaises(DatabaseException):
|
|
self._insert_and_check(self._sample_data(), len(data))
|
|
self.assertEquals(self.database.count(Person), 100)
|
|
list(self.database.select('SELECT * from $table', Person))
|
|
with self.assertRaises(DatabaseException):
|
|
self.database.drop_table(Person)
|
|
with self.assertRaises(DatabaseException):
|
|
self.database.drop_database()
|
|
except DatabaseException as e:
|
|
if 'Unknown user' in six.text_type(e):
|
|
raise unittest.SkipTest('Database user "%s" is not defined' % username)
|
|
else:
|
|
raise
|
|
finally:
|
|
self.database = orig_database
|
|
|
|
def test_readonly_db_with_default_user(self):
|
|
self._test_readonly_db('default')
|
|
|
|
def test_readonly_db_with_readonly_user(self):
|
|
self._test_readonly_db('readonly')
|
|
|
|
def test_insert_readonly(self):
|
|
m = ReadOnlyModel(name='readonly')
|
|
with self.assertRaises(DatabaseException):
|
|
self.database.insert([m])
|
|
|
|
def test_create_readonly_table(self):
|
|
self.database.create_table(ReadOnlyModel)
|
|
|
|
def test_drop_readonly_table(self):
|
|
self.database.drop_table(ReadOnlyModel)
|
|
|
|
|
|
class ReadOnlyModel(Model):
|
|
readonly = True
|
|
|
|
name = StringField()
|
|
date = DateField()
|
|
engine = MergeTree('date', ('name',))
|