1 # -*- coding: utf-8 -*-
4 from sqlparse import parse
6 from sqlconvert.print_tokens import tlist2str
7 from sqlconvert.process_mysql import remove_directive_tokens, \
8 is_directive_statement, requote_names, unescape_strings, \
9 get_DML_type, process_statement
10 from sqlconvert.process_tokens import escape_strings
14 parsed = parse("select /*! test */ * from /* test */ `T`")[0]
15 remove_directive_tokens(parsed)
16 query = tlist2str(parsed)
17 assert query == u'SELECT * FROM /* test */ `T`'
20 def test_directive_statement():
21 parsed = parse("/*! test */ test ;")[0]
22 assert not is_directive_statement(parsed)
23 parsed = parse("/*! test */ ;")[0]
24 assert is_directive_statement(parsed)
28 parsed = parse("select * from `T`")[0]
30 query = tlist2str(parsed)
31 assert query == u'SELECT * FROM "T"'
34 def test_unescape_string():
35 parsed = parse("insert into test values ('\"te\\'st\\\"\\n')")[0]
36 unescape_strings(parsed)
37 query = tlist2str(parsed)
38 assert query == u"INSERT INTO test VALUES ('\"te'st\"\n')"
41 def test_escape_string_postgres():
42 parsed = parse("insert into test values ('\"te\\'st\\\"\\n')")[0]
43 unescape_strings(parsed)
44 escape_strings(parsed, 'postgres')
45 query = tlist2str(parsed)
46 assert query == u"INSERT INTO test VALUES (E'\"te''st\"\\n')"
49 def test_escape_string_sqlite():
50 parsed = parse("insert into test values ('\"te\\'st\\\"\\n')")[0]
51 unescape_strings(parsed)
52 escape_strings(parsed, 'sqlite')
53 query = tlist2str(parsed)
54 assert query == u"INSERT INTO test VALUES ('\"te''st\"\n')"
58 parsed = parse("create table test ();")[0]
59 statement = next(process_statement(parsed))
60 with pytest.raises(ValueError):
61 get_DML_type(statement)
63 parsed = parse("select /*! test */ * from /* test */ `T`")[0]
64 statement = next(process_statement(parsed))
65 assert get_DML_type(statement) == "SELECT"
67 parsed = parse("insert into test values ('\"te\\'st\\\"\\n')")[0]
68 statement = next(process_statement(parsed))
69 assert get_DML_type(statement) == "INSERT"
72 def test_split_ext_insert():
73 parsed = parse("insert into test values (1, 2)")[0]
74 statement = next(process_statement(parsed))
75 query = tlist2str(statement)
76 assert query == u"INSERT INTO test VALUES (1, 2)"
78 parsed = parse("insert into test (age, salary) values (1, 2);")[0]
79 statement = next(process_statement(parsed))
80 query = tlist2str(statement)
81 assert query == u"INSERT INTO test (age, salary) VALUES (1, 2);"
83 parsed = parse("insert into test values (1, 2), (3, 4);")[0]
84 stiter = process_statement(parsed)
85 statement = next(stiter)
86 query = tlist2str(statement)
87 assert query == u"INSERT INTO test VALUES (1, 2);\n"
88 statement = next(stiter)
89 query = tlist2str(statement)
90 assert query == u"INSERT INTO test VALUES (3, 4);"
92 parsed = parse("insert into test (age, salary) values (1, 2), (3, 4)")[0]
93 stiter = process_statement(parsed)
94 statement = next(stiter)
95 query = tlist2str(statement)
96 assert query == u"INSERT INTO test (age, salary) VALUES (1, 2)\n"
97 statement = next(stiter)
98 query = tlist2str(statement)
99 assert query == u"INSERT INTO test (age, salary) VALUES (3, 4)"
103 parsed = parse("select /*! test */ * from /* test */ `T`")[0]
104 statement = next(process_statement(parsed))
105 query = tlist2str(statement)
106 assert query == u'SELECT * FROM /* test */ "T"'