Source

sqlalchemy / lib / sqlalchemy / testing / engines.py

Full commit
  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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
from __future__ import absolute_import

import types
import weakref
from collections import deque
from . import config
from .util import decorator
from .. import event, pool
import re
import warnings


class ConnectionKiller(object):

    def __init__(self):
        self.proxy_refs = weakref.WeakKeyDictionary()
        self.testing_engines = weakref.WeakKeyDictionary()
        self.conns = set()

    def add_engine(self, engine):
        self.testing_engines[engine] = True

    def connect(self, dbapi_conn, con_record):
        self.conns.add(dbapi_conn)

    def checkout(self, dbapi_con, con_record, con_proxy):
        self.proxy_refs[con_proxy] = True

    def _safe(self, fn):
        try:
            fn()
        except (SystemExit, KeyboardInterrupt):
            raise
        except Exception, e:
            warnings.warn(
                    "testing_reaper couldn't "
                    "rollback/close connection: %s" % e)

    def rollback_all(self):
        for rec in self.proxy_refs.keys():
            if rec is not None and rec.is_valid:
                self._safe(rec.rollback)

    def close_all(self):
        for rec in self.proxy_refs.keys():
            if rec is not None:
                self._safe(rec._close)

    def _after_test_ctx(self):
        pass
        # this can cause a deadlock with pg8000 - pg8000 acquires
        # prepared statment lock inside of rollback() - if async gc
        # is collecting in finalize_fairy, deadlock.
        # not sure if this should be if pypy/jython only
        #for conn in self.conns:
        #    self._safe(conn.rollback)

    def _stop_test_ctx(self):
        if config.options.low_connections:
            self._stop_test_ctx_minimal()
        else:
            self._stop_test_ctx_aggressive()

    def _stop_test_ctx_minimal(self):
        self.close_all()

        self.conns = set()

        for rec in self.testing_engines.keys():
            if rec is not config.db:
                rec.dispose()

    def _stop_test_ctx_aggressive(self):
        self.close_all()
        for conn in self.conns:
            self._safe(conn.close)
        self.conns = set()
        for rec in self.testing_engines.keys():
            rec.dispose()

    def assert_all_closed(self):
        for rec in self.proxy_refs:
            if rec.is_valid:
                assert False

testing_reaper = ConnectionKiller()


def drop_all_tables(metadata, bind):
    testing_reaper.close_all()
    if hasattr(bind, 'close'):
        bind.close()
    metadata.drop_all(bind)


@decorator
def assert_conns_closed(fn, *args, **kw):
    try:
        fn(*args, **kw)
    finally:
        testing_reaper.assert_all_closed()


@decorator
def rollback_open_connections(fn, *args, **kw):
    """Decorator that rolls back all open connections after fn execution."""

    try:
        fn(*args, **kw)
    finally:
        testing_reaper.rollback_all()


@decorator
def close_first(fn, *args, **kw):
    """Decorator that closes all connections before fn execution."""

    testing_reaper.close_all()
    fn(*args, **kw)


@decorator
def close_open_connections(fn, *args, **kw):
    """Decorator that closes all connections after fn execution."""
    try:
        fn(*args, **kw)
    finally:
        testing_reaper.close_all()


def all_dialects(exclude=None):
    import sqlalchemy.databases as d
    for name in d.__all__:
        # TEMPORARY
        if exclude and name in exclude:
            continue
        mod = getattr(d, name, None)
        if not mod:
            mod = getattr(__import__(
                'sqlalchemy.databases.%s' % name).databases, name)
        yield mod.dialect()


class ReconnectFixture(object):

    def __init__(self, dbapi):
        self.dbapi = dbapi
        self.connections = []

    def __getattr__(self, key):
        return getattr(self.dbapi, key)

    def connect(self, *args, **kwargs):
        conn = self.dbapi.connect(*args, **kwargs)
        self.connections.append(conn)
        return conn

    def _safe(self, fn):
        try:
            fn()
        except (SystemExit, KeyboardInterrupt):
            raise
        except Exception, e:
            warnings.warn(
                    "ReconnectFixture couldn't "
                    "close connection: %s" % e)

    def shutdown(self):
        # TODO: this doesn't cover all cases
        # as nicely as we'd like, namely MySQLdb.
        # would need to implement R. Brewer's
        # proxy server idea to get better
        # coverage.
        for c in list(self.connections):
            self._safe(c.close)
        self.connections = []


def reconnecting_engine(url=None, options=None):
    url = url or config.db_url
    dbapi = config.db.dialect.dbapi
    if not options:
        options = {}
    options['module'] = ReconnectFixture(dbapi)
    engine = testing_engine(url, options)
    _dispose = engine.dispose

    def dispose():
        engine.dialect.dbapi.shutdown()
        _dispose()

    engine.test_shutdown = engine.dialect.dbapi.shutdown
    engine.dispose = dispose
    return engine


def testing_engine(url=None, options=None):
    """Produce an engine configured by --options with optional overrides."""

    from sqlalchemy import create_engine
    from .assertsql import asserter

    if not options:
        use_reaper = True
    else:
        use_reaper = options.pop('use_reaper', True)

    url = url or config.db_url
    if options is None:
        options = config.db_opts

    engine = create_engine(url, **options)
    if isinstance(engine.pool, pool.QueuePool):
        engine.pool._timeout = 0
        engine.pool._max_overflow = 0
    event.listen(engine, 'after_execute', asserter.execute)
    event.listen(engine, 'after_cursor_execute', asserter.cursor_execute)
    if use_reaper:
        event.listen(engine.pool, 'connect', testing_reaper.connect)
        event.listen(engine.pool, 'checkout', testing_reaper.checkout)
        testing_reaper.add_engine(engine)

    return engine


def utf8_engine(url=None, options=None):
    """Hook for dialects or drivers that don't handle utf8 by default."""

    from sqlalchemy.engine import url as engine_url

    if config.db.dialect.name == 'mysql' and \
        config.db.driver in ['mysqldb', 'pymysql']:
        # note 1.2.1.gamma.6 or greater of MySQLdb
        # needed here
        url = url or config.db_url
        url = engine_url.make_url(url)
        url.query['charset'] = 'utf8'
        url.query['use_unicode'] = '0'
        url = str(url)

    return testing_engine(url, options)


def mock_engine(dialect_name=None):
    """Provides a mocking engine based on the current testing.db.

    This is normally used to test DDL generation flow as emitted
    by an Engine.

    It should not be used in other cases, as assert_compile() and
    assert_sql_execution() are much better choices with fewer
    moving parts.

    """

    from sqlalchemy import create_engine

    if not dialect_name:
        dialect_name = config.db.name

    buffer = []

    def executor(sql, *a, **kw):
        buffer.append(sql)

    def assert_sql(stmts):
        recv = [re.sub(r'[\n\t]', '', str(s)) for s in buffer]
        assert  recv == stmts, recv

    def print_sql():
        d = engine.dialect
        return "\n".join(
            str(s.compile(dialect=d))
            for s in engine.mock
        )

    engine = create_engine(dialect_name + '://',
                           strategy='mock', executor=executor)
    assert not hasattr(engine, 'mock')
    engine.mock = buffer
    engine.assert_sql = assert_sql
    engine.print_sql = print_sql
    return engine


class DBAPIProxyCursor(object):
    """Proxy a DBAPI cursor.

    Tests can provide subclasses of this to intercept
    DBAPI-level cursor operations.

    """
    def __init__(self, engine, conn):
        self.engine = engine
        self.connection = conn
        self.cursor = conn.cursor()

    def execute(self, stmt, parameters=None, **kw):
        if parameters:
            return self.cursor.execute(stmt, parameters, **kw)
        else:
            return self.cursor.execute(stmt, **kw)

    def executemany(self, stmt, params, **kw):
        return self.cursor.executemany(stmt, params, **kw)

    def __getattr__(self, key):
        return getattr(self.cursor, key)


class DBAPIProxyConnection(object):
    """Proxy a DBAPI connection.

    Tests can provide subclasses of this to intercept
    DBAPI-level connection operations.

    """
    def __init__(self, engine, cursor_cls):
        self.conn = self._sqla_unwrap = engine.pool._creator()
        self.engine = engine
        self.cursor_cls = cursor_cls

    def cursor(self):
        return self.cursor_cls(self.engine, self.conn)

    def close(self):
        self.conn.close()

    def __getattr__(self, key):
        return getattr(self.conn, key)


def proxying_engine(conn_cls=DBAPIProxyConnection,
                    cursor_cls=DBAPIProxyCursor):
    """Produce an engine that provides proxy hooks for
    common methods.

    """
    def mock_conn():
        return conn_cls(config.db, cursor_cls)
    return testing_engine(options={'creator': mock_conn})


class ReplayableSession(object):
    """A simple record/playback tool.

    This is *not* a mock testing class.  It only records a session for later
    playback and makes no assertions on call consistency whatsoever.  It's
    unlikely to be suitable for anything other than DB-API recording.

    """

    Callable = object()
    NoAttribute = object()

    # Py3K
    #Natives = set([getattr(types, t)
    #               for t in dir(types) if not t.startswith('_')]). \
    #               union([type(t) if not isinstance(t, type)
    #                        else t for t in __builtins__.values()]).\
    #               difference([getattr(types, t)
    #                        for t in ('FunctionType', 'BuiltinFunctionType',
    #                                  'MethodType', 'BuiltinMethodType',
    #                                  'LambdaType', )])
    # Py2K
    Natives = set([getattr(types, t)
                   for t in dir(types) if not t.startswith('_')]). \
                   difference([getattr(types, t)
                           for t in ('FunctionType', 'BuiltinFunctionType',
                                     'MethodType', 'BuiltinMethodType',
                                     'LambdaType', 'UnboundMethodType',)])
    # end Py2K

    def __init__(self):
        self.buffer = deque()

    def recorder(self, base):
        return self.Recorder(self.buffer, base)

    def player(self):
        return self.Player(self.buffer)

    class Recorder(object):
        def __init__(self, buffer, subject):
            self._buffer = buffer
            self._subject = subject

        def __call__(self, *args, **kw):
            subject, buffer = [object.__getattribute__(self, x)
                               for x in ('_subject', '_buffer')]

            result = subject(*args, **kw)
            if type(result) not in ReplayableSession.Natives:
                buffer.append(ReplayableSession.Callable)
                return type(self)(buffer, result)
            else:
                buffer.append(result)
                return result

        @property
        def _sqla_unwrap(self):
            return self._subject

        def __getattribute__(self, key):
            try:
                return object.__getattribute__(self, key)
            except AttributeError:
                pass

            subject, buffer = [object.__getattribute__(self, x)
                               for x in ('_subject', '_buffer')]
            try:
                result = type(subject).__getattribute__(subject, key)
            except AttributeError:
                buffer.append(ReplayableSession.NoAttribute)
                raise
            else:
                if type(result) not in ReplayableSession.Natives:
                    buffer.append(ReplayableSession.Callable)
                    return type(self)(buffer, result)
                else:
                    buffer.append(result)
                    return result

    class Player(object):
        def __init__(self, buffer):
            self._buffer = buffer

        def __call__(self, *args, **kw):
            buffer = object.__getattribute__(self, '_buffer')
            result = buffer.popleft()
            if result is ReplayableSession.Callable:
                return self
            else:
                return result

        @property
        def _sqla_unwrap(self):
            return None

        def __getattribute__(self, key):
            try:
                return object.__getattribute__(self, key)
            except AttributeError:
                pass
            buffer = object.__getattribute__(self, '_buffer')
            result = buffer.popleft()
            if result is ReplayableSession.Callable:
                return self
            elif result is ReplayableSession.NoAttribute:
                raise AttributeError(key)
            else:
                return result