tests/test_mysql2postgres.py

   1 from sqlparse import parse
   2 from sqlobject.tests.dbtest import getConnection
   3 import pytest
   4
   5 from sqlconvert.print_tokens import tlist2str
   6 from sqlconvert.process_mysql import unescape_strings
   7
   8 connection = getConnection()
   9 pytestmark = pytest.mark.skipif(connection.dbName != "postgres",
  10                                 reason="This test requires PostgreSQL")
  11
  12 create_postgres_test_table = """
  13 CREATE TABLE test (
  14     id serial PRIMARY KEY,
  15     test_str VARCHAR(255) NOT NULL
  16 );
  17 """
  18
  19
  20 def test_mysql2postgres_string():
  21     connection.query(create_postgres_test_table)
  22     parsed = parse("insert into test (id, test_str) values "
  23                    "(1, '\"te\\'st\\\"\\n')")[0]
  24     unescape_strings(parsed)
  25     query = tlist2str(parsed)
  26     assert query == u"INSERT INTO test (id, test_str) VALUES " \
  27                     u"(1, '\"te''st\"\n')"
  28     connection.query(query)
  29     test_str = connection.queryOne("SELECT test_str FROM test WHERE id=1")[0]
  30     assert test_str == u"\"te'st\"\n"