snowflake-connector-python/test/test_qmark.py at master · baxen/snowflake-connector-python · 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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright (c) 2012-2018 Snowflake Computing Inc. All right reserved.
#


import pytest

from snowflake.connector import errors


def test_qmark_paramstyle(conn_cnx, db_parameters):
    """
    Binding question marks is not supported by default
    """
    try:
        with conn_cnx() as cnx:
            cnx.cursor().execute(
                "CREATE OR REPLACE TABLE {name} "
                "(aa STRING, bb STRING)".format(
                    name=db_parameters['name']))
            cnx.cursor().execute(
                "INSERT INTO {name} VALUES('?', '?')".format(
                    name=db_parameters['name']))
            for rec in cnx.cursor().execute(
                    "SELECT * FROM {name}".format(name=db_parameters['name'])):
                assert rec[0] == "?", "First column value"
                with pytest.raises(errors.ProgrammingError):
                    cnx.cursor().execute(
                        "INSERT INTO {name} VALUES(?,?)".format(
                            name=db_parameters['name']))
    finally:
        with conn_cnx() as cnx:
            cnx.cursor().execute(
                "DROP TABLE IF EXISTS {name}".format(
                    name=db_parameters['name']))


def test_numeric_paramstyle(conn_cnx, db_parameters):
    """
    Binding numeric positional style is not supported
    """
    try:
        with conn_cnx() as cnx:
            cnx.cursor().execute(
                "CREATE OR REPLACE TABLE {name} "
                "(aa STRING, bb STRING)".format(
                    name=db_parameters['name']))
            cnx.cursor().execute(
                "INSERT INTO {name} VALUES(':1', ':2')".format(
                    name=db_parameters['name']))
            for rec in cnx.cursor().execute(
                    "SELECT * FROM {name}".format(name=db_parameters['name'])):
                assert rec[0] == ":1", "First column value"
                with pytest.raises(errors.ProgrammingError):
                    cnx.cursor().execute(
                        "INSERT INTO {name} VALUES(:1,:2)".format(
                            name=db_parameters['name']))
    finally:
        with conn_cnx() as cnx:
            cnx.cursor().execute(
                "DROP TABLE IF EXISTS {name}".format(
                    name=db_parameters['name']))


def test_qmark_paramstyle_enabled(conn_cnx, db_parameters):
    """
    Enable qmark binding
    """
    import snowflake.connector
    snowflake.connector.paramstyle = u'qmark'
    try:
        with conn_cnx() as cnx:
            cnx.cursor().execute(
                "CREATE OR REPLACE TABLE {name} "
                "(aa STRING, bb STRING)".format(
                    name=db_parameters['name']))
            cnx.cursor().execute(
                "INSERT INTO {name} VALUES(?, ?)".format(
                    name=db_parameters['name']), ('test11', 'test12'))
            ret = cnx.cursor().execute("select * from {name}".format(
                name=db_parameters['name'])).fetchone()
            assert ret[0] == 'test11'
            assert ret[1] == 'test12'
    finally:
        with conn_cnx() as cnx:
            cnx.cursor().execute(
                "DROP TABLE IF EXISTS {name}".format(
                    name=db_parameters['name']))
        snowflake.connector.paramstyle = u'pyformat'

    # After changing back to pyformat, binding qmark should fail.
    try:
        with conn_cnx() as cnx:
            cnx.cursor().execute(
                "CREATE OR REPLACE TABLE {name} "
                "(aa STRING, bb STRING)".format(
                    name=db_parameters['name']))
            with pytest.raises(TypeError):
                cnx.cursor().execute(
                    "INSERT INTO {name} VALUES(?, ?)".format(
                        name=db_parameters['name']), ('test11', 'test12'))
    finally:
        with conn_cnx() as cnx:
            cnx.cursor().execute(
                "DROP TABLE IF EXISTS {name}".format(
                    name=db_parameters['name']))