mindsdb_sql_parser/tests/test_mindsdb/test_variables.py at 34b13bef93d5a89f8283626f174ecee0d99d153c · mindsdb/mindsdb_sql_parser · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
from mindsdb_sql_parser import parse_sql
from mindsdb_sql_parser.ast import *
from mindsdb_sql_parser.ast import Variable

class TestMDBParser:
    def test_select_variable(self):
        sql = 'SELECT @version'
        ast = parse_sql(sql)
        expected_ast = Select(targets=[Variable('version')])
        assert str(ast).lower() == sql.lower()
        assert str(ast) == str(expected_ast)

        sql = 'SELECT @@version'
        ast = parse_sql(sql)
        expected_ast = Select(targets=[Variable('version', is_system_var=True)])
        assert str(ast).lower() == sql.lower()
        assert str(ast) == str(expected_ast)

        sql = "set autocommit=1, global sql_mode=concat(@@sql_mode, ',STRICT_TRANS_TABLES'), NAMES utf8mb4 COLLATE utf8mb4_unicode_ci"
        ast = parse_sql(sql)
        expected_ast = Set(
            set_list=[
                Set(name=Identifier('autocommit'), value=Constant(1)),
                Set(name=Identifier('sql_mode'),
                    scope='global',
                    value=Function(op='concat', args=[
                        Variable('sql_mode', is_system_var=True),
                        Constant(',STRICT_TRANS_TABLES')
                    ])
                ),
                Set(category="NAMES",
                    value=Constant('utf8mb4', with_quotes=False),
                    params={'COLLATE': Constant('utf8mb4_unicode_ci', with_quotes=False)})
            ]
        )

        assert str(ast).lower() == sql.lower()
        assert str(ast) == str(expected_ast)

    def test_set(self):
        for value in (0, 1, 'TRUE', 'FALSE', 'ON', 'OFF'):
            sql = f"set @@session.autocommit={value}"
            ast = parse_sql(sql)
            expected_ast = Set(
                name=Variable('session.autocommit', is_system_var=True),
                value=Constant(value, with_quotes=False)
            )
            assert str(ast).lower() == sql.lower()
            assert str(ast) == str(expected_ast)

    def test_mysql(self):
        sql = 'select @@session.auto_increment_increment, @@character_set_client'
        ast = parse_sql(sql)
        expected_ast = Select(
            targets=[
                Variable('session.auto_increment_increment', is_system_var=True),
                Variable('character_set_client', is_system_var=True),
            ]
        )

        assert str(ast).lower() == sql.lower()
        assert str(ast) == str(expected_ast)