Server IP : 85.214.239.14 / Your IP : 3.144.119.149 Web Server : Apache/2.4.62 (Debian) System : Linux h2886529.stratoserver.net 4.9.0 #1 SMP Tue Jan 9 19:45:01 MSK 2024 x86_64 User : www-data ( 33) PHP Version : 7.4.18 Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare, MySQL : OFF | cURL : OFF | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : OFF Directory : /proc/self/root/srv/automx/env/lib/python3.5/site-packages/sqlalchemy/testing/ |
Upload File : |
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