tests/test_process_mysql.py

   1 # -*- coding: utf-8 -*-
   2
   3 from sqlparse import parse
   4
   5 from sqlconvert.print_tokens import tlist2str
   6 from sqlconvert.process_mysql import remove_directive_tokens, \
   7         is_directive_statement, requote_names, unescape_strings, \
   8         is_insert, process_statement
   9 from sqlconvert.process_tokens import escape_strings
  10
  11
  12 def test_directive():
  13     parsed = parse("select /*! test */ * from /* test */ `T`")[0]
  14     remove_directive_tokens(parsed)
  15     query = tlist2str(parsed)
  16     assert query == u'SELECT * FROM /* test */ `T`'
  17
  18
  19 def test_directive_statement():
  20     parsed = parse("/*! test */ test ;")[0]
  21     assert not is_directive_statement(parsed)
  22     parsed = parse("/*! test */ ;")[0]
  23     assert is_directive_statement(parsed)
  24
  25
  26 def test_requote():
  27     parsed = parse("select * from `T`")[0]
  28     requote_names(parsed)
  29     query = tlist2str(parsed)
  30     assert query == u'SELECT * FROM "T"'
  31
  32
  33 def test_unescape_string():
  34     parsed = parse("insert into test values ('\"te\\'st\\\"\\n')")[0]
  35     unescape_strings(parsed)
  36     query = tlist2str(parsed)
  37     assert query == u"INSERT INTO test VALUES ('\"te'st\"\n')"
  38
  39
  40 def test_escape_string_postgres():
  41     parsed = parse("insert into test values ('\"te\\'st\\\"\\n')")[0]
  42     unescape_strings(parsed)
  43     escape_strings(parsed, 'postgres')
  44     query = tlist2str(parsed)
  45     assert query == u"INSERT INTO test VALUES (E'\"te''st\"\\n')"
  46
  47
  48 def test_escape_string_sqlite():
  49     parsed = parse("insert into test values ('\"te\\'st\\\"\\n')")[0]
  50     unescape_strings(parsed)
  51     escape_strings(parsed, 'sqlite')
  52     query = tlist2str(parsed)
  53     assert query == u"INSERT INTO test VALUES ('\"te''st\"\n')"
  54
  55
  56 def test_is_insert():
  57     parsed = parse("select /*! test */ * from /* test */ `T`")[0]
  58     statement = next(process_statement(parsed))
  59     assert not is_insert(statement)
  60
  61     parsed = parse("insert into test values ('\"te\\'st\\\"\\n')")[0]
  62     statement = next(process_statement(parsed))
  63     assert is_insert(statement)
  64
  65
  66 def test_split_ext_insert():
  67     parsed = parse("insert into test values (1, 2)")[0]
  68     statement = next(process_statement(parsed))
  69     query = tlist2str(statement)
  70     assert query == u"INSERT INTO test VALUES (1, 2)"
  71
  72     parsed = parse("insert into test (age, salary) values (1, 2);")[0]
  73     statement = next(process_statement(parsed))
  74     query = tlist2str(statement)
  75     assert query == u"INSERT INTO test (age, salary) VALUES (1, 2);"
  76
  77     parsed = parse("insert into test values (1, 2), (3, 4);")[0]
  78     stiter = process_statement(parsed)
  79     statement = next(stiter)
  80     query = tlist2str(statement)
  81     assert query == u"INSERT INTO test VALUES  (1, 2);\n"
  82     statement = next(stiter)
  83     query = tlist2str(statement)
  84     assert query == u"INSERT INTO test VALUES  (3, 4);"
  85
  86     parsed = parse("insert into test (age, salary) values (1, 2), (3, 4)")[0]
  87     stiter = process_statement(parsed)
  88     statement = next(stiter)
  89     query = tlist2str(statement)
  90     assert query == u"INSERT INTO test (age, salary) VALUES  (1, 2)\n"
  91     statement = next(stiter)
  92     query = tlist2str(statement)
  93     assert query == u"INSERT INTO test (age, salary) VALUES  (3, 4)"
  94
  95
  96 def test_process():
  97     parsed = parse("select /*! test */ * from /* test */ `T`")[0]
  98     statement = next(process_statement(parsed))
  99     query = tlist2str(statement)
 100     assert query == u'SELECT * FROM /* test */ "T"'