File Manager

X7ROOT File Manager

Current Path: /opt/cloudlinux/venv/lib/python3.11/site-packages/sqlalchemy/testing

opt / cloudlinux / venv / lib / python3.11 / site-packages / sqlalchemy / testing /

?? ..
?? __init__.py(2.8 KB)
?? __pycache__
?? assertions.py(21.74 KB)
?? assertsql.py(13.27 KB)
?? config.py(5.53 KB)
?? engines.py(10.28 KB)
?? entities.py(3.13 KB)
?? exclusions.py(12.91 KB)
?? fixtures.py(14.57 KB)
?? mock.py(893 B)
?? pickleable.py(2.63 KB)
?? plugin
?? profiling.py(8.54 KB)
?? provision.py(5.44 KB)
?? replay_fixture.py(5.74 KB)
?? requirements.py(32.29 KB)
?? schema.py(4.65 KB)
?? suite
?? util.py(9.91 KB)
?? warnings.py(1.61 KB)

Editing: replay_fixture.py

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

X7ROOT File Manager

Editing: replay_fixture.py

Upload File

Create Folder