Infinidat/infi.clickhouse_orm
1
1
Fork 0
mirror of https://github.com/Infinidat/infi.clickhouse_orm.git synced 2024-11-22 00:56:34 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'develop' into funcs

Browse Source 
# Conflicts:
#	tests/test_database.py
This commit is contained in:
Itai Shirav 2020-04-14 23:04:14 +03:00
commit 2e586fa61c
3 changed files with 14 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
tests/test_database.py
View File

@ -166,12 +166,12 @@ class DatabaseTestCase(TestCaseWithData):
Database(self.database.db_name, username='default', password='wrong')
exc = cm.exception
print(exc.code, exc.message)
self.assertIn(exc.code, (193, 516))
if exc.code == 193:
self.assertTrue('Wrong password for user default' in exc.message)
if exc.code == 193: # ClickHouse version < 20.3
self.assertTrue(exc.message.startswith('Wrong password for user default'))
elif exc.code == 516: # ClickHouse version >= 20.3
self.assertTrue(exc.message.startswith('default: Authentication failed'))
else:
self.assertTrue('default: Authentication failed: password is incorrect' in exc.message)
raise Exception('Unexpected error code - %s' % exc.code)
def test_nonexisting_db(self):
db = Database('db_not_here', autocreate=False)

4
tests/test_join.py
View File

@ -30,8 +30,8 @@ class JoinTest(unittest.TestCase):
self.print_res("SELECT b FROM $db.{} ALL LEFT JOIN $db.{} USING id".format(Foo.table_name(), Bar.table_name()))
def test_with_subquery(self):
self.print_res("SELECT b FROM {} ALL LEFT JOIN (SELECT * from {}) USING id".format(Foo.table_name(), Bar.table_name()))
self.print_res("SELECT b FROM $db.{} ALL LEFT JOIN (SELECT * from $db.{}) USING id".format(Foo.table_name(), Bar.table_name()))
self.print_res("SELECT b FROM {} ALL LEFT JOIN (SELECT * from {}) subquery USING id".format(Foo.table_name(), Bar.table_name()))
self.print_res("SELECT b FROM $db.{} ALL LEFT JOIN (SELECT * from $db.{}) subquery USING id".format(Foo.table_name(), Bar.table_name()))
class Foo(models.Model):

9
tests/test_readonly.py
View File

@ -26,7 +26,9 @@ class ReadonlyTestCase(TestCaseWithData):
self.database.drop_database()
self._check_db_readonly_err(cm.exception, drop_table=True)
except ServerError as e:
if e.code == 192 and e.message.startswith('Unknown user'):
if e.code == 192 and e.message.startswith('Unknown user'): # ClickHouse version < 20.3
raise unittest.SkipTest('Database user "%s" is not defined' % username)
elif e.code == 516 and e.message.startswith('readonly: Authentication failed'): # ClickHouse version >= 20.3
raise unittest.SkipTest('Database user "%s" is not defined' % username)
else:
raise
@ -35,7 +37,10 @@ class ReadonlyTestCase(TestCaseWithData):
def _check_db_readonly_err(self, exc, drop_table=None):
self.assertEqual(exc.code, 164)
if drop_table:
print(exc.message)
if self.database.server_version >= (20, 3):
self.assertTrue('Cannot execute query in readonly mode' in exc.message)
elif drop_table:
self.assertTrue(exc.message.startswith('Cannot drop table in readonly mode'))
else:
self.assertTrue(exc.message.startswith('Cannot insert into table in readonly mode'))