diff --git a/sql/engines/clickhouse.py b/sql/engines/clickhouse.py index 064dcfbc53..c09b25889f 100644 --- a/sql/engines/clickhouse.py +++ b/sql/engines/clickhouse.py @@ -342,7 +342,7 @@ def execute_check(self, db_name=None, sql=""): table_engine = self.get_table_engine(table_name)["engine"] table_exist = self.get_table_engine(table_name)["status"] if table_exist == 1: - if table_engine in ("View", "File,", "URL", "Buffer", "Null"): + if table_engine in ("View", "File", "URL", "Buffer", "Null"): result = ReviewResult( id=line, errlevel=2, diff --git a/sql/engines/tests.py b/sql/engines/tests.py index 74654ac01b..5556829833 100644 --- a/sql/engines/tests.py +++ b/sql/engines/tests.py @@ -2106,13 +2106,43 @@ def test_execute_check_alter_sql(self, mock_query): mock_query.return_value = result new_engine = ClickHouseEngine(instance=self.ins1) table_engine = new_engine.get_table_engine(table_name) - alter_sql = "alter table default.tb_test add column remark String" + alter_sql = "alter table tb_test add column remark String" check_result = new_engine.execute_check(db_name="some_db", sql=alter_sql) self.assertEqual( check_result.rows[0].errormessage, "ALTER TABLE仅支持*MergeTree,Merge以及Distributed等引擎表!", ) + @patch.object(ClickHouseEngine, "query") + def test_execute_check_truncate_sql(self, mock_query): + table_name = "default.tb_test" + result = ResultSet() + result.rows = [("File",)] + mock_query.return_value = result + new_engine = ClickHouseEngine(instance=self.ins1) + table_engine = new_engine.get_table_engine(table_name) + alter_sql = "truncate table tb_test" + check_result = new_engine.execute_check(db_name="some_db", sql=alter_sql) + self.assertEqual( + check_result.rows[0].errormessage, + "TRUNCATE不支持View,File,URL,Buffer和Null表引擎!", + ) + + @patch.object(ClickHouseEngine, "query") + def test_execute_check_insert_sql(self, mock_query): + table_name = "default.tb_test" + result = ResultSet() + result.rows = [("Log",)] + mock_query.return_value = result + new_engine = ClickHouseEngine(instance=self.ins1) + table_engine = new_engine.get_table_engine(table_name) + alter_sql = "insert into tb_test(name) values('nick');" + check_result = new_engine.execute_check(db_name="some_db", sql=alter_sql) + self.assertEqual( + check_result.rows[0].errlevel, + 0, + ) + def test_filter_sql_with_delimiter(self): new_engine = ClickHouseEngine(instance=self.ins1) sql_without_limit = "select user from usertable;"