diff options
Diffstat (limited to 'lib/sqlalchemy/testing')
| -rw-r--r-- | lib/sqlalchemy/testing/replay_fixture.py | 209 |
1 files changed, 0 insertions, 209 deletions
diff --git a/lib/sqlalchemy/testing/replay_fixture.py b/lib/sqlalchemy/testing/replay_fixture.py deleted file mode 100644 index 613d7572b..000000000 --- a/lib/sqlalchemy/testing/replay_fixture.py +++ /dev/null @@ -1,209 +0,0 @@ -from collections import deque -import contextlib -import types - -from . import config -from . import fixtures -from . import profiling -from .. import create_engine -from .. import MetaData -from .. import util -from ..orm import Session - - -class ReplayFixtureTest(fixtures.TestBase): - @contextlib.contextmanager - def _dummy_ctx(self, *arg, **kw): - yield - - def test_invocation(self): - - dbapi_session = ReplayableSession() - creator = config.db.pool._creator - - def recorder(): - return dbapi_session.recorder(creator()) - - engine = create_engine( - config.db.url, creator=recorder, use_native_hstore=False - ) - self.metadata = MetaData(engine) - self.engine = engine - self.session = Session(engine) - - self.setup_engine() - try: - self._run_steps(ctx=self._dummy_ctx) - finally: - self.teardown_engine() - engine.dispose() - - def player(): - return dbapi_session.player() - - engine = create_engine( - config.db.url, creator=player, use_native_hstore=False - ) - - self.metadata = MetaData(engine) - self.engine = engine - self.session = Session(engine) - - self.setup_engine() - try: - self._run_steps(ctx=profiling.count_functions) - finally: - self.session.close() - engine.dispose() - - def setup_engine(self): - pass - - def teardown_engine(self): - pass - - def _run_steps(self, ctx): - raise NotImplementedError() - - -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() - - if util.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", - ) - ] - ) - else: - 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", - ) - ] - ) - ) - - 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 |