summaryrefslogtreecommitdiff
path: root/lib/sqlalchemy/testing/replay_fixture.py
diff options
context:
space:
mode:
authorFederico Caselli <cfederico87@gmail.com>2020-05-22 22:48:16 +0200
committerMike Bayer <mike_mp@zzzcomputing.com>2020-05-23 17:01:17 -0400
commit7bbf5f88629ff98715c66549321f1c70e6e13ffa (patch)
tree81612ad6e86c33bb31b04395bd0b6be730243b54 /lib/sqlalchemy/testing/replay_fixture.py
parent3fb0eb5a156ab7dbe19d82717dd197d9d22e64f1 (diff)
downloadsqlalchemy-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.py209
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