forked from exasol/sqlalchemy-exasol
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest_update.py
126 lines (112 loc) · 4.55 KB
/
test_update.py
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
118
119
120
121
122
123
124
125
126
import unittest
from sqlalchemy import *
from sqlalchemy import testing
from sqlalchemy.dialects import mysql
from sqlalchemy.engine import default
from sqlalchemy.testing import AssertsCompiledSQL, eq_, fixtures
from sqlalchemy.testing.schema import Table, Column
class _UpdateTestBase(object):
@classmethod
def define_tables(cls, metadata):
Table('mytable', metadata,
Column('myid', Integer),
Column('name', String(30)),
Column('description', String(50)))
Table('myothertable', metadata,
Column('otherid', Integer),
Column('othername', String(30)))
Table('users', metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('name', String(30), nullable=False))
Table('addresses', metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('user_id', None, ForeignKey('users.id')),
Column('name', String(30), nullable=False),
Column('email_address', String(50), nullable=False))
Table('dingalings', metadata,
Column('id', Integer, primary_key=True,
test_needs_autoincrement=True),
Column('address_id', None, ForeignKey('addresses.id')),
Column('data', String(30)))
@classmethod
def fixtures(cls):
return dict(
users=(
('id', 'name'),
(7, 'jack'),
(8, 'ed'),
(9, 'fred'),
(10, 'chuck')
),
addresses = (
('id', 'user_id', 'name', 'email_address'),
(1, 7, 'x', '[email protected]'),
(2, 8, 'x', '[email protected]'),
(3, 8, 'x', '[email protected]'),
(4, 8, 'x', '[email protected]'),
(5, 9, 'x', '[email protected]')
),
dingalings = (
('id', 'address_id', 'data'),
(1, 2, 'ding 1/2'),
(2, 5, 'ding 2/5')
),
)
class UpdateTest(_UpdateTestBase, fixtures.TablesTest):
__backend__ = True
@unittest.skipIf(testing.db.dialect.driver == 'turbodbc',
'not supported by turbodbc')
def test_update_simple(self):
"""test simple update and assert that exasol returns the right rowcount"""
users = self.tables.users
result = testing.db.execute(users.update().values(name="peter").where(users.c.id == 10))
expected = [
(7, 'jack'),
(8, 'ed'),
(9, 'fred'),
(10, 'peter')
]
assert result.rowcount == 1
self._assert_users(users, expected)
@unittest.skipIf(testing.db.dialect.driver == 'turbodbc',
'not supported by turbodbc')
def test_update_simple_multiple_rows_rowcount(self):
"""test simple update and assert that exasol returns the right rowcount"""
users = self.tables.users
result = testing.db.execute(users.update().values(name="peter").where(users.c.id >= 9))
expected = [
(7, 'jack'),
(8, 'ed'),
(9, 'peter'),
(10, 'peter')
]
assert result.rowcount == 2
self._assert_users(users, expected)
@unittest.skipIf(testing.db.dialect.driver == 'turbodbc',
'not supported by turbodbc')
def test_update_executemany(self):
"""test that update with executemany work as well, but rowcount
is undefined for executemany updates"""
users = self.tables.users
stmt = users.update().\
where(users.c.name == bindparam('oldname')).\
values(name=bindparam('newname'))
values = [{'oldname':'jack', 'newname':'jack2'},
{'oldname':'fred', 'newname':'fred2'}]
result = testing.db.execute(stmt, values)
expected = [
(7, 'jack2'),
(8, 'ed'),
(9, 'fred2'),
(10, 'chuck')
]
assert result.rowcount == -1
self._assert_users(users, expected)
def _assert_addresses(self, addresses, expected):
stmt = addresses.select().order_by(addresses.c.id)
eq_(testing.db.execute(stmt).fetchall(), expected)
def _assert_users(self, users, expected):
stmt = users.select().order_by(users.c.id)
eq_(testing.db.execute(stmt).fetchall(), expected)