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

Fix tests for ClickHouse v20.3

Browse Source
This commit is contained in:
Itai Shirav 2020-03-16 10:24:02 +02:00
parent 262ce13f4d
commit 54c67f2777
3 changed files with 15 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
tests/test_database.py
View File

@ -149,8 +149,12 @@ class DatabaseTestCase(TestCaseWithData):
Database(self.database.db_name, username='default', password='wrong')
exc = cm.exception
self.assertEqual(exc.code, 193)
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:
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(exc.message.startswith('default: Cannot execute query in readonly mode'))
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'))