File Manager

Hackfut Security File Manager

Current Path: /opt/alt/python27/lib64/python2.7/site-packages/sqlalchemy/testing

opt / alt / python27 / lib64 / python2.7 / site-packages / sqlalchemy / testing /

📁 ..
📄 __init__.py(1.12 KB)
📄 __init__.pyc(1.84 KB)
📄 __init__.pyo(1.84 KB)
📄 assertions.py(16.64 KB)
📄 assertions.pyc(20.71 KB)
📄 assertions.pyo(19.03 KB)
📄 assertsql.py(12.29 KB)
📄 assertsql.pyc(14.05 KB)
📄 assertsql.pyo(13.78 KB)
📄 config.py(2.67 KB)
📄 config.pyc(4.1 KB)
📄 config.pyo(4.01 KB)
📄 engines.py(9.29 KB)
📄 engines.pyc(14.47 KB)
📄 engines.pyo(14.38 KB)
📄 entities.py(2.92 KB)
📄 entities.pyc(3.12 KB)
📄 entities.pyo(3.12 KB)
📄 exclusions.py(12.35 KB)
📄 exclusions.pyc(19.3 KB)
📄 exclusions.pyo(19.13 KB)
📄 fixtures.py(10.47 KB)
📄 fixtures.pyc(16.16 KB)
📄 fixtures.pyo(16.08 KB)
📄 mock.py(630 B)
📄 mock.pyc(664 B)
📄 mock.pyo(664 B)
📄 pickleable.py(2.58 KB)
📄 pickleable.pyc(7.7 KB)
📄 pickleable.pyo(7.7 KB)
📁 plugin
📄 profiling.py(8.2 KB)
📄 profiling.pyc(8.5 KB)
📄 profiling.pyo(8.5 KB)
📄 provision.py(10.13 KB)
📄 provision.pyc(13.72 KB)
📄 provision.pyo(13.72 KB)
📄 replay_fixture.py(5.3 KB)
📄 replay_fixture.pyc(7.5 KB)
📄 replay_fixture.pyo(7.5 KB)
📄 requirements.py(22.16 KB)
📄 requirements.pyc(37.59 KB)
📄 requirements.pyo(37.59 KB)
📄 runner.py(1.55 KB)
📄 runner.pyc(1.84 KB)
📄 runner.pyo(1.84 KB)
📄 schema.py(3.47 KB)
📄 schema.pyc(3.18 KB)
📄 schema.pyo(3.18 KB)
📁 suite
📄 util.py(7.36 KB)
📄 util.pyc(10.67 KB)
📄 util.pyo(10.63 KB)
📄 warnings.py(1.27 KB)
📄 warnings.pyc(1.37 KB)
📄 warnings.pyo(1.37 KB)

Editing: replay_fixture.py

from . import fixtures
from . import profiling
from .. import util
import types
from collections import deque
import contextlib
from . import config
from sqlalchemy import MetaData
from sqlalchemy import create_engine
from sqlalchemy.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
        recorder = lambda: 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()

player = lambda: 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

Hackfut Security File Manager

Editing: replay_fixture.py

Upload File

Create Folder