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
109
110
111
112
113
114
115
116
117
|
# -*- coding: utf-8 -*-
"""
Test queries.
"""
import pytest
from .helpers import pymssqlconn, drop_table
from pymssql._mssql import substitute_params
@pytest.mark.mssql_server_required
class Test_609:
table_name = 'testtab'
table_ddl = f'''
CREATE TABLE {table_name} (
int_col int,
int_col_none int,
text_col nvarchar(100)
)
'''
N = 10
table_data = tuple( (x, f"Column {x}") for x in range(N) )
@classmethod
def setup_class(cls):
cls.conn = pymssqlconn(encryption='require')
drop_table(cls.conn._conn, cls.table_name)
cls.create_test_table()
cls.fill_test_table()
@classmethod
def teardown_class(cls):
drop_table(cls.conn._conn, cls.table_name)
cls.conn.close()
@classmethod
def create_test_table(cls):
with cls.conn.cursor() as c:
c.execute(cls.table_ddl)
@classmethod
def fill_test_table(cls):
query = f"INSERT INTO {cls.table_name} (int_col, text_col) VALUES (%d, %s);"
with cls.conn.cursor() as c:
for vals in cls.table_data:
c.execute(query, vals)
def test_text0(self):
with self.conn.cursor() as c:
c.execute(f'SELECT * FROM {self.table_name} WHERE text_col=%s', "a")
rows = c.fetchall()
assert len(rows) == 0
@pytest.mark.parametrize('i', range(N))
def test_int(self, i):
int_col, text_col = self.table_data[i]
with self.conn.cursor() as c:
c.execute(f'SELECT * FROM {self.table_name} WHERE int_col=%s', (int_col, ))
rows = c.fetchall()
assert len(rows) == 1
assert rows[0] == (int_col, None, text_col)
@pytest.mark.parametrize('i', range(N))
def test_int_pymssql(self, i):
int_col, text_col = self.table_data[i]
with self.conn.cursor() as c:
c.execute(f'SELECT * FROM {self.table_name} WHERE int_col=%s', int_col)
rows = c.fetchall()
assert len(rows) == 1
assert rows[0] == (int_col, None, text_col)
@pytest.mark.parametrize('i', range(N))
def test_text(self, i):
int_col, text_col = self.table_data[i]
with self.conn.cursor() as c:
c.execute(f'SELECT * FROM {self.table_name} WHERE text_col=%s', (text_col, ))
rows = c.fetchall()
assert len(rows) == 1
assert rows[0] == (int_col, None, text_col)
@pytest.mark.parametrize('i', range(N))
def test_text_pymssql(self, i):
int_col, text_col = self.table_data[i]
with self.conn.cursor() as c:
c.execute(f'SELECT * FROM {self.table_name} WHERE text_col=%s', text_col)
rows = c.fetchall()
assert len(rows) == 1
assert rows[0] == (int_col, None, text_col)
@pytest.mark.xfail(reason="known parser issue")
def test_424(self):
query = "SELECT COUNT(*) FROM document WHERE title LIKE '%summary%' AND id < %s"
parms = (1500,)
with self.conn.cursor() as c:
c.execute(query, parms)
@pytest.mark.xfail(reason="known parser issue")
def test_276(self):
res = substitute_params('select %s;'.encode('utf-8'), tuple(["Фрязино".encode('utf-16-le')]))
assert res == b'select 0x240440044f04370438043d043e042000;'
#assert res == b"select '$\x04@\x04O\x047\x048\x04=\x04>\x04';"
res = substitute_params('select %s;'.encode('utf-8'), tuple(["Фрязино ".encode('utf-16-le')]))
assert res == b'select 0x240440044f04370438043d043e042000;'
"Фрязино".encode('utf-16-le')
'$\x04@\x04O\x047\x048\x04=\x04>\x04'
"Фрязино ".encode('utf-16-le')
'$\x04@\x04O\x047\x048\x04=\x04>\x04 \x00'
|
