diff options
author | Federico Caselli <cfederico87@gmail.com> | 2020-05-22 22:48:16 +0200 |
---|---|---|
committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2020-05-23 17:01:17 -0400 |
commit | 7bbf5f88629ff98715c66549321f1c70e6e13ffa (patch) | |
tree | 81612ad6e86c33bb31b04395bd0b6be730243b54 /lib/sqlalchemy/testing/replay_fixture.py | |
parent | 3fb0eb5a156ab7dbe19d82717dd197d9d22e64f1 (diff) | |
download | sqlalchemy-7bbf5f88629ff98715c66549321f1c70e6e13ffa.tar.gz |
Add python 3.8 profiles; remove zoomark tests
The zoomark tests have served us well for many years.
At this point, they have been using a very antiquated
calling style for many years and are no longer where we catch
performance issues.
The performance suite now has a large number of individual
tests that catch issues very specifically and additionally
record just one performance count per test. This also
allows us to remove the "replay" fixtures that were not
used for anything else.
Fixes: #5347
Change-Id: I0a8d078e7b7240602f4f3f7068f231e98a40f17e
Diffstat (limited to 'lib/sqlalchemy/testing/replay_fixture.py')
-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 |