1
1
from sqlalchemy import *
2
from test.lib.testing import eq_
3
from test.lib import engines
4
from test.lib import fixtures
6
# TODO: we should probably build mock bases for
7
# these to share with test_reconnect, test_parseconnect
8
class MockDBAPI(object):
12
def connect(self, *args, **kwargs):
13
return MockConnection(self)
15
class MockConnection(object):
16
def __init__(self, parent):
19
return MockCursor(self)
27
class MockCursor(object):
30
def __init__(self, parent):
32
def execute(self, *args, **kwargs):
33
self.parent.parent.log.append('execute')
34
def executedirect(self, *args, **kwargs):
35
self.parent.parent.log.append('executedirect')
2
from sqlalchemy.testing import eq_
3
from sqlalchemy.testing import engines
4
from sqlalchemy.testing import fixtures
5
from sqlalchemy.testing.mock import Mock
8
return Mock(paramstyle='qmark',
39
20
class MxODBCTest(fixtures.TestBase):
41
22
def test_native_odbc_execute(self):
42
23
t1 = Table('t1', MetaData(), Column('c1', Integer))
44
26
engine = engines.testing_engine('mssql+mxodbc://localhost',
45
27
options={'module': dbapi, '_initialize': False})
46
28
conn = engine.connect()
48
30
# crud: uses execute
50
31
conn.execute(t1.insert().values(c1='foo'))
51
32
conn.execute(t1.delete().where(t1.c.c1 == 'foo'))
52
conn.execute(t1.update().where(t1.c.c1 == 'foo').values(c1='bar'
33
conn.execute(t1.update().where(t1.c.c1 == 'foo').values(c1='bar'))
55
35
# select: uses executedirect
57
36
conn.execute(t1.select())
61
39
conn.execution_options(native_odbc_execute=True).\
62
40
execute(t1.select())
63
41
conn.execution_options(native_odbc_execute=False).\
64
execute(t1.insert().values(c1='foo'
42
execute(t1.insert().values(c1='foo'))
46
dbapi.connect.return_value.cursor.return_value.execute.mock_calls],
47
[{'direct': True}, {'direct': True}, {'direct': True},
48
{'direct': True}, {'direct': False}, {'direct': True}]