Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified
📁 ..	-	2026-03-21 15:45
📄 __init__.py # testing/__init__.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php from . import config # noqa from . import mock # noqa from .assertions import assert_raises # noqa from .assertions import assert_raises_context_ok # noqa from .assertions import assert_raises_message # noqa from .assertions import assert_raises_message_context_ok # noqa from .assertions import assert_raises_return # noqa from .assertions import AssertsCompiledSQL # noqa from .assertions import AssertsExecutionResults # noqa from .assertions import ComparesTables # noqa from .assertions import emits_warning # noqa from .assertions import emits_warning_on # noqa from .assertions import eq_ # noqa from .assertions import eq_ignore_whitespace # noqa from .assertions import eq_regex # noqa from .assertions import expect_deprecated # noqa from .assertions import expect_warnings # noqa from .assertions import in_ # noqa from .assertions import is_ # noqa from .assertions import is_false # noqa from .assertions import is_instance_of # noqa from .assertions import is_not # noqa from .assertions import is_not_ # noqa from .assertions import is_true # noqa from .assertions import le_ # noqa from .assertions import ne_ # noqa from .assertions import not_in # noqa from .assertions import not_in_ # noqa from .assertions import startswith_ # noqa from .assertions import uses_deprecated # noqa from .config import combinations # noqa from .config import db # noqa from .config import fixture # noqa from .config import requirements as requires # noqa from .exclusions import _is_excluded # noqa from .exclusions import _server_version # noqa from .exclusions import against as _against # noqa from .exclusions import db_spec # noqa from .exclusions import exclude # noqa from .exclusions import fails # noqa from .exclusions import fails_if # noqa from .exclusions import fails_on # noqa from .exclusions import fails_on_everything_except # noqa from .exclusions import future # noqa from .exclusions import only_if # noqa from .exclusions import only_on # noqa from .exclusions import skip # noqa from .exclusions import skip_if # noqa from .util import adict # noqa from .util import fail # noqa from .util import flag_combinations # noqa from .util import force_drop_names # noqa from .util import metadata_fixture # noqa from .util import provide_metadata # noqa from .util import resolve_lambda # noqa from .util import rowset # noqa from .util import run_as_contextmanager # noqa from .util import teardown_events # noqa from .warnings import assert_warnings # noqa def against(queries): return _against(config._current, queries) crashes = skip	2.8 KB	2026-01-20 12:58
📁 __pycache__	-	2026-03-21 15:45
📄 assertions.py # testing/assertions.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php from __future__ import absolute_import import contextlib import re import sys import warnings from . import assertsql from . import config from . import mock from . import util as testutil from .exclusions import db_spec from .util import fail from .. import exc as sa_exc from .. import pool from .. import schema from .. import sql from .. import types as sqltypes from .. import util from ..engine import default from ..engine import url from ..util import compat from ..util import decorator def expect_warnings(messages, kw): """Context manager which expects one or more warnings. With no arguments, squelches all SAWarnings emitted via sqlalchemy.util.warn and sqlalchemy.util.warn_limited. Otherwise pass string expressions that will match selected warnings via regex; all non-matching warnings are sent through. The expect version asserts* that the warnings were in fact seen. Note that the test suite sets SAWarning warnings to raise exceptions. """ return _expect_warnings(sa_exc.SAWarning, messages, *kw) @contextlib.contextmanager def expect_warnings_on(db, messages, kw): """Context manager which expects one or more warnings on specific dialects. The expect version asserts** that the warnings were in fact seen. """ spec = db_spec(db) if isinstance(db, util.string_types) and not spec(config._current): yield else: with expect_warnings(messages, kw): yield def emits_warning(messages): """Decorator form of expect_warnings(). Note that emits_warning does not assert that the warnings were in fact seen. """ @decorator def decorate(fn, args, kw): with expect_warnings(assert_=False, messages): return fn(args, kw) return decorate def expect_deprecated(messages, kw): return _expect_warnings(sa_exc.SADeprecationWarning, messages, kw) def emits_warning_on(db, messages): """Mark a test as emitting a warning on a specific dialect. With no arguments, squelches all SAWarning failures. Or pass one or more strings; these will be matched to the root of the warning description by warnings.filterwarnings(). Note that emits_warning_on does not* assert that the warnings were in fact seen. """ @decorator def decorate(fn, args, kw): with expect_warnings_on(db, assert_=False, messages): return fn(args, kw) return decorate def uses_deprecated(messages): """Mark a test as immune from fatal deprecation warnings. With no arguments, squelches all SADeprecationWarning failures. Or pass one or more strings; these will be matched to the root of the warning description by warnings.filterwarnings(). As a special case, you may pass a function name prefixed with // and it will be re-written as needed to match the standard warning verbiage emitted by the sqlalchemy.util.deprecated decorator. Note that uses_deprecated does not assert that the warnings were in fact seen. """ @decorator def decorate(fn, args, kw): with expect_deprecated(messages, assert_=False): return fn(args, kw) return decorate @contextlib.contextmanager def _expect_warnings( exc_cls, messages, regex=True, assert_=True, py2konly=False ): if regex: filters = [re.compile(msg, re.I \| re.S) for msg in messages] else: filters = messages seen = set(filters) real_warn = warnings.warn def our_warn(msg, arg, *kw): if isinstance(msg, exc_cls): exception = msg msg = str(exception) elif arg: exception = arg[0] else: exception = None if not exception or not issubclass(exception, exc_cls): return real_warn(msg, arg, *kw) if not filters: return for filter_ in filters: if (regex and filter_.match(msg)) or ( not regex and filter_ == msg ): seen.discard(filter_) break else: real_warn(msg, arg, *kw) with mock.patch("warnings.warn", our_warn): yield if assert_ and (not py2konly or not compat.py3k): assert not seen, "Warnings were not seen: %s" % ", ".join( "%r" % (s.pattern if regex else s) for s in seen ) def global_cleanup_assertions(): """Check things that have to be finalized at the end of a test suite. Hardcoded at the moment, a modular system can be built here to support things like PG prepared transactions, tables all dropped, etc. """ _assert_no_stray_pool_connections() _STRAY_CONNECTION_FAILURES = 0 def _assert_no_stray_pool_connections(): global _STRAY_CONNECTION_FAILURES # lazy gc on cPython means "do nothing." pool connections # shouldn't be in cycles, should go away. testutil.lazy_gc() # however, once in awhile, on an EC2 machine usually, # there's a ref in there. usually just one. if pool._refs: # OK, let's be somewhat forgiving. _STRAY_CONNECTION_FAILURES += 1 print( "Encountered a stray connection in test cleanup: %s" % str(pool._refs) ) # then do a real GC sweep. We shouldn't even be here # so a single sweep should really be doing it, otherwise # there's probably a real unreachable cycle somewhere. testutil.gc_collect() # if we've already had two of these occurrences, or # after a hard gc sweep we still have pool._refs?! # now we have to raise. if pool._refs: err = str(pool._refs) # but clean out the pool refs collection directly, # reset the counter, # so the error doesn't at least keep happening. pool._refs.clear() _STRAY_CONNECTION_FAILURES = 0 warnings.warn( "Stray connection refused to leave " "after gc.collect(): %s" % err ) elif _STRAY_CONNECTION_FAILURES > 10: assert False, "Encountered more than 10 stray connections" _STRAY_CONNECTION_FAILURES = 0 def eq_regex(a, b, msg=None): assert re.match(b, a), msg or "%r !~ %r" % (a, b) def eq_(a, b, msg=None): """Assert a == b, with repr messaging on failure.""" assert a == b, msg or "%r != %r" % (a, b) def ne_(a, b, msg=None): """Assert a != b, with repr messaging on failure.""" assert a != b, msg or "%r == %r" % (a, b) def le_(a, b, msg=None): """Assert a <= b, with repr messaging on failure.""" assert a <= b, msg or "%r != %r" % (a, b) def is_instance_of(a, b, msg=None): assert isinstance(a, b), msg or "%r is not an instance of %r" % (a, b) def is_true(a, msg=None): is_(a, True, msg=msg) def is_false(a, msg=None): is_(a, False, msg=msg) def is_(a, b, msg=None): """Assert a is b, with repr messaging on failure.""" assert a is b, msg or "%r is not %r" % (a, b) def is_not(a, b, msg=None): """Assert a is not b, with repr messaging on failure.""" assert a is not b, msg or "%r is %r" % (a, b) # deprecated. See #5429 is_not_ = is_not def in_(a, b, msg=None): """Assert a in b, with repr messaging on failure.""" assert a in b, msg or "%r not in %r" % (a, b) def not_in(a, b, msg=None): """Assert a in not b, with repr messaging on failure.""" assert a not in b, msg or "%r is in %r" % (a, b) # deprecated. See #5429 not_in_ = not_in def startswith_(a, fragment, msg=None): """Assert a.startswith(fragment), with repr messaging on failure.""" assert a.startswith(fragment), msg or "%r does not start with %r" % ( a, fragment, ) def eq_ignore_whitespace(a, b, msg=None): a = re.sub(r"^\s+?\|\n", "", a) a = re.sub(r" {2,}", " ", a) b = re.sub(r"^\s+?\|\n", "", b) b = re.sub(r" {2,}", " ", b) assert a == b, msg or "%r != %r" % (a, b) def _assert_proper_exception_context(exception): """assert that any exception we're catching does not have a __context__ without a __cause__, and that __suppress_context__ is never set. Python 3 will report nested as exceptions as "during the handling of error X, error Y occurred". That's not what we want to do. we want these exceptions in a cause chain. """ if not util.py3k: return if ( exception.__context__ is not exception.__cause__ and not exception.__suppress_context__ ): assert False, ( "Exception %r was correctly raised but did not set a cause, " "within context %r as its cause." % (exception, exception.__context__) ) def assert_raises(except_cls, callable_, args, *kw): _assert_raises(except_cls, callable_, args, kw, check_context=True) def assert_raises_context_ok(except_cls, callable_, args, *kw): _assert_raises( except_cls, callable_, args, kw, ) def assert_raises_return(except_cls, callable_, args, *kw): return _assert_raises(except_cls, callable_, args, kw, check_context=True) def assert_raises_message(except_cls, msg, callable_, args, *kwargs): _assert_raises( except_cls, callable_, args, kwargs, msg=msg, check_context=True ) def assert_raises_message_context_ok( except_cls, msg, callable_, args, *kwargs ): _assert_raises(except_cls, callable_, args, kwargs, msg=msg) def _assert_raises( except_cls, callable_, args, kwargs, msg=None, check_context=False ): ret_err = None if check_context: are_we_already_in_a_traceback = sys.exc_info()[0] try: callable_(args, kwargs) success = False except except_cls as err: ret_err = err success = True if msg is not None: assert re.search( msg, util.text_type(err), re.UNICODE ), "%r !~ %s" % ( msg, err, ) if check_context and not are_we_already_in_a_traceback: _assert_proper_exception_context(err) print(util.text_type(err).encode("utf-8")) # assert outside the block so it works for AssertionError too ! assert success, "Callable did not raise an exception" return ret_err class AssertsCompiledSQL(object): def assert_compile( self, clause, result, params=None, checkparams=None, dialect=None, checkpositional=None, check_prefetch=None, use_default_dialect=False, allow_dialect_select=False, literal_binds=False, schema_translate_map=None, ): if use_default_dialect: dialect = default.DefaultDialect() elif allow_dialect_select: dialect = None else: if dialect is None: dialect = getattr(self, "__dialect__", None) if dialect is None: dialect = config.db.dialect elif dialect == "default": dialect = default.DefaultDialect() elif dialect == "default_enhanced": dialect = default.StrCompileDialect() elif isinstance(dialect, util.string_types): dialect = url.URL(dialect).get_dialect()() kw = {} compile_kwargs = {} if schema_translate_map: kw["schema_translate_map"] = schema_translate_map if params is not None: kw["column_keys"] = list(params) if literal_binds: compile_kwargs["literal_binds"] = True from sqlalchemy import orm if isinstance(clause, orm.Query): context = clause._compile_context() context.statement.use_labels = True clause = context.statement elif isinstance(clause, orm.persistence.BulkUD): with mock.patch.object(clause, "_execute_stmt") as stmt_mock: clause.exec_() clause = stmt_mock.mock_calls[0][1][0] if compile_kwargs: kw["compile_kwargs"] = compile_kwargs class DontAccess(object): def __getattribute__(self, key): raise NotImplementedError( "compiler accessed .statement; use " "compiler.current_executable" ) class CheckCompilerAccess(object): def __init__(self, test_statement): self.test_statement = test_statement self.supports_execution = getattr( test_statement, "supports_execution", False ) if self.supports_execution: self._execution_options = test_statement._execution_options if isinstance( test_statement, (sql.Insert, sql.Update, sql.Delete) ): self._returning = test_statement._returning if isinstance(test_statement, (sql.Insert, sql.Update)): self.inline = test_statement.inline self._return_defaults = test_statement._return_defaults def _default_dialect(self): return self.test_statement._default_dialect() def compile(self, dialect, kw): return self.test_statement.compile.__func__( self, dialect=dialect, kw ) def _compiler(self, dialect, kw): return self.test_statement._compiler.__func__( self, dialect, kw ) def _compiler_dispatch(self, compiler, kwargs): if hasattr(compiler, "statement"): with mock.patch.object( compiler, "statement", DontAccess() ): return self.test_statement._compiler_dispatch( compiler, kwargs ) else: return self.test_statement._compiler_dispatch( compiler, kwargs ) # no construct can assume it's the "top level" construct in all cases # as anything can be nested. ensure constructs don't assume they # are the "self.statement" element c = CheckCompilerAccess(clause).compile(dialect=dialect, *kw) param_str = repr(getattr(c, "params", {})) if util.py3k: param_str = param_str.encode("utf-8").decode("ascii", "ignore") print( ("\nSQL String:\n" + util.text_type(c) + param_str).encode( "utf-8" ) ) else: print( "\nSQL String:\n" + util.text_type(c).encode("utf-8") + param_str ) cc = re.sub(r"[\n\t]", "", util.text_type(c)) eq_(cc, result, "%r != %r on dialect %r" % (cc, result, dialect)) if checkparams is not None: eq_(c.construct_params(params), checkparams) if checkpositional is not None: p = c.construct_params(params) eq_(tuple([p[x] for x in c.positiontup]), checkpositional) if check_prefetch is not None: eq_(c.prefetch, check_prefetch) class ComparesTables(object): def assert_tables_equal(self, table, reflected_table, strict_types=False): assert len(table.c) == len(reflected_table.c) for c, reflected_c in zip(table.c, reflected_table.c): eq_(c.name, reflected_c.name) assert reflected_c is reflected_table.c[c.name] eq_(c.primary_key, reflected_c.primary_key) eq_(c.nullable, reflected_c.nullable) if strict_types: msg = "Type '%s' doesn't correspond to type '%s'" assert isinstance(reflected_c.type, type(c.type)), msg % ( reflected_c.type, c.type, ) else: self.assert_types_base(reflected_c, c) if isinstance(c.type, sqltypes.String): eq_(c.type.length, reflected_c.type.length) eq_( {f.column.name for f in c.foreign_keys}, {f.column.name for f in reflected_c.foreign_keys}, ) if c.server_default: assert isinstance( reflected_c.server_default, schema.FetchedValue ) assert len(table.primary_key) == len(reflected_table.primary_key) for c in table.primary_key: assert reflected_table.primary_key.columns[c.name] is not None def assert_types_base(self, c1, c2): assert c1.type._compare_type_affinity( c2.type ), "On column %r, type '%s' doesn't correspond to type '%s'" % ( c1.name, c1.type, c2.type, ) class AssertsExecutionResults(object): def assert_result(self, result, class_, objects): result = list(result) print(repr(result)) self.assert_list(result, class_, objects) def assert_list(self, result, class_, list_): self.assert_( len(result) == len(list_), "result list is not the same size as test list, " + "for class " + class_.__name__, ) for i in range(0, len(list_)): self.assert_row(class_, result[i], list_[i]) def assert_row(self, class_, rowobj, desc): self.assert_( rowobj.__class__ is class_, "item class is not " + repr(class_) ) for key, value in desc.items(): if isinstance(value, tuple): if isinstance(value[1], list): self.assert_list(getattr(rowobj, key), value[0], value[1]) else: self.assert_row(value[0], getattr(rowobj, key), value[1]) else: self.assert_( getattr(rowobj, key) == value, "attribute %s value %s does not match %s" % (key, getattr(rowobj, key), value), ) def assert_unordered_result(self, result, cls, expected): """As assert_result, but the order of objects is not considered. The algorithm is very expensive but not a big deal for the small numbers of rows that the test suite manipulates. """ class immutabledict(dict): def __hash__(self): return id(self) found = util.IdentitySet(result) expected = {immutabledict(e) for e in expected} for wrong in util.itertools_filterfalse( lambda o: isinstance(o, cls), found ): fail( 'Unexpected type "%s", expected "%s"' % (type(wrong).__name__, cls.__name__) ) if len(found) != len(expected): fail( 'Unexpected object count "%s", expected "%s"' % (len(found), len(expected)) ) NOVALUE = object() def _compare_item(obj, spec): for key, value in spec.items(): if isinstance(value, tuple): try: self.assert_unordered_result( getattr(obj, key), value[0], value[1] ) except AssertionError: return False else: if getattr(obj, key, NOVALUE) != value: return False return True for expected_item in expected: for found_item in found: if _compare_item(found_item, expected_item): found.remove(found_item) break else: fail( "Expected %s instance with attributes %s not found." % (cls.__name__, repr(expected_item)) ) return True def sql_execution_asserter(self, db=None): if db is None: from . import db as db return assertsql.assert_engine(db) def assert_sql_execution(self, db, callable_, rules): with self.sql_execution_asserter(db) as asserter: result = callable_() asserter.assert_(rules) return result def assert_sql(self, db, callable_, rules): newrules = [] for rule in rules: if isinstance(rule, dict): newrule = assertsql.AllOf( [assertsql.CompiledSQL(k, v) for k, v in rule.items()] ) else: newrule = assertsql.CompiledSQL(rule) newrules.append(newrule) return self.assert_sql_execution(db, callable_, newrules) def assert_sql_count(self, db, callable_, count): self.assert_sql_execution( db, callable_, assertsql.CountStatements(count) ) def assert_multiple_sql_count(self, dbs, callable_, counts): recs = [ (self.sql_execution_asserter(db), db, count) for (db, count) in zip(dbs, counts) ] asserters = [] for ctx, db, count in recs: asserters.append(ctx.__enter__()) try: return callable_() finally: for asserter, (ctx, db, count) in zip(asserters, recs): ctx.__exit__(None, None, None) asserter.assert_(assertsql.CountStatements(count)) @contextlib.contextmanager def assert_execution(self, db, rules): with self.sql_execution_asserter(db) as asserter: yield asserter.assert_(*rules) def assert_statement_count(self, db, count): return self.assert_execution(db, assertsql.CountStatements(count))	21.74 KB	2026-01-20 12:58
📄 assertsql.py # testing/assertsql.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php import collections import contextlib import re from .. import event from .. import util from ..engine import url from ..engine.default import DefaultDialect from ..engine.util import _distill_params from ..schema import _DDLCompiles class AssertRule(object): is_consumed = False errormessage = None consume_statement = True def process_statement(self, execute_observed): pass def no_more_statements(self): assert False, ( "All statements are complete, but pending " "assertion rules remain" ) class SQLMatchRule(AssertRule): pass class CursorSQL(SQLMatchRule): consume_statement = False def __init__(self, statement, params=None): self.statement = statement self.params = params def process_statement(self, execute_observed): stmt = execute_observed.statements[0] if self.statement != stmt.statement or ( self.params is not None and self.params != stmt.parameters ): self.errormessage = ( "Testing for exact SQL %s parameters %s received %s %s" % ( self.statement, self.params, stmt.statement, stmt.parameters, ) ) else: execute_observed.statements.pop(0) self.is_consumed = True if not execute_observed.statements: self.consume_statement = True class CompiledSQL(SQLMatchRule): def __init__(self, statement, params=None, dialect="default"): self.statement = statement self.params = params self.dialect = dialect def _compare_sql(self, execute_observed, received_statement): stmt = re.sub(r"[\n\t]", "", self.statement) return received_statement == stmt def _compile_dialect(self, execute_observed): if self.dialect == "default": return DefaultDialect() else: # ugh if self.dialect == "postgresql": params = {"implicit_returning": True} else: params = {} return url.URL(self.dialect).get_dialect()(*params) def _received_statement(self, execute_observed): """reconstruct the statement and params in terms of a target dialect, which for CompiledSQL is just DefaultDialect.""" context = execute_observed.context compare_dialect = self._compile_dialect(execute_observed) if isinstance(context.compiled.statement, _DDLCompiles): compiled = context.compiled.statement.compile( dialect=compare_dialect, schema_translate_map=context.execution_options.get( "schema_translate_map" ), ) else: compiled = context.compiled.statement.compile( dialect=compare_dialect, column_keys=context.compiled.column_keys, inline=context.compiled.inline, schema_translate_map=context.execution_options.get( "schema_translate_map" ), ) _received_statement = re.sub(r"[\n\t]", "", util.text_type(compiled)) parameters = execute_observed.parameters if not parameters: _received_parameters = [compiled.construct_params()] else: _received_parameters = [ compiled.construct_params(m) for m in parameters ] return _received_statement, _received_parameters def process_statement(self, execute_observed): context = execute_observed.context _received_statement, _received_parameters = self._received_statement( execute_observed ) params = self._all_params(context) equivalent = self._compare_sql(execute_observed, _received_statement) if equivalent: if params is not None: all_params = list(params) all_received = list(_received_parameters) while all_params and all_received: param = dict(all_params.pop(0)) for idx, received in enumerate(list(all_received)): # do a positive compare only for param_key in param: # a key in param did not match current # 'received' if ( param_key not in received or received[param_key] != param[param_key] ): break else: # all keys in param matched 'received'; # onto next param del all_received[idx] break else: # param did not match any entry # in all_received equivalent = False break if all_params or all_received: equivalent = False if equivalent: self.is_consumed = True self.errormessage = None else: self.errormessage = self._failure_message(params) % { "received_statement": _received_statement, "received_parameters": _received_parameters, } def _all_params(self, context): if self.params: if util.callable(self.params): params = self.params(context) else: params = self.params if not isinstance(params, list): params = [params] return params else: return None def _failure_message(self, expected_params): return ( "Testing for compiled statement %r partial params %s, " "received %%(received_statement)r with params " "%%(received_parameters)r" % ( self.statement.replace("%", "%%"), repr(expected_params).replace("%", "%%"), ) ) class RegexSQL(CompiledSQL): def __init__(self, regex, params=None, dialect="default"): SQLMatchRule.__init__(self) self.regex = re.compile(regex) self.orig_regex = regex self.params = params self.dialect = dialect def _failure_message(self, expected_params): return ( "Testing for compiled statement ~%r partial params %s, " "received %%(received_statement)r with params " "%%(received_parameters)r" % ( self.orig_regex.replace("%", "%%"), repr(expected_params).replace("%", "%%"), ) ) def _compare_sql(self, execute_observed, received_statement): return bool(self.regex.match(received_statement)) class DialectSQL(CompiledSQL): def _compile_dialect(self, execute_observed): return execute_observed.context.dialect def _compare_no_space(self, real_stmt, received_stmt): stmt = re.sub(r"[\n\t]", "", real_stmt) return received_stmt == stmt def _received_statement(self, execute_observed): received_stmt, received_params = super( DialectSQL, self )._received_statement(execute_observed) # TODO: why do we need this part? for real_stmt in execute_observed.statements: if self._compare_no_space(real_stmt.statement, received_stmt): break else: raise AssertionError( "Can't locate compiled statement %r in list of " "statements actually invoked" % received_stmt ) return received_stmt, execute_observed.context.compiled_parameters def _compare_sql(self, execute_observed, received_statement): stmt = re.sub(r"[\n\t]", "", self.statement) # convert our comparison statement to have the # paramstyle of the received paramstyle = execute_observed.context.dialect.paramstyle if paramstyle == "pyformat": stmt = re.sub(r":([\w_]+)", r"%(\1)s", stmt) else: # positional params repl = None if paramstyle == "qmark": repl = "?" elif paramstyle == "format": repl = r"%s" elif paramstyle == "numeric": repl = None stmt = re.sub(r":([\w_]+)", repl, stmt) return received_statement == stmt class CountStatements(AssertRule): def __init__(self, count): self.count = count self._statement_count = 0 def process_statement(self, execute_observed): self._statement_count += 1 def no_more_statements(self): if self.count != self._statement_count: assert False, "desired statement count %d does not match %d" % ( self.count, self._statement_count, ) class AllOf(AssertRule): def __init__(self, rules): self.rules = set(rules) def process_statement(self, execute_observed): for rule in list(self.rules): rule.errormessage = None rule.process_statement(execute_observed) if rule.is_consumed: self.rules.discard(rule) if not self.rules: self.is_consumed = True break elif not rule.errormessage: # rule is not done yet self.errormessage = None break else: self.errormessage = list(self.rules)[0].errormessage class EachOf(AssertRule): def __init__(self, rules): self.rules = list(rules) def process_statement(self, execute_observed): while self.rules: rule = self.rules[0] rule.process_statement(execute_observed) if rule.is_consumed: self.rules.pop(0) elif rule.errormessage: self.errormessage = rule.errormessage if rule.consume_statement: break if not self.rules: self.is_consumed = True def no_more_statements(self): if self.rules and not self.rules[0].is_consumed: self.rules[0].no_more_statements() elif self.rules: super(EachOf, self).no_more_statements() class Or(AllOf): def process_statement(self, execute_observed): for rule in self.rules: rule.process_statement(execute_observed) if rule.is_consumed: self.is_consumed = True break else: self.errormessage = list(self.rules)[0].errormessage class SQLExecuteObserved(object): def __init__(self, context, clauseelement, multiparams, params): self.context = context self.clauseelement = clauseelement self.parameters = _distill_params(multiparams, params) self.statements = [] class SQLCursorExecuteObserved( collections.namedtuple( "SQLCursorExecuteObserved", ["statement", "parameters", "context", "executemany"], ) ): pass class SQLAsserter(object): def __init__(self): self.accumulated = [] def _close(self): self._final = self.accumulated del self.accumulated def assert_(self, rules): rule = EachOf(*rules) observed = list(self._final) while observed: statement = observed.pop(0) rule.process_statement(statement) if rule.is_consumed: break elif rule.errormessage: assert False, rule.errormessage if observed: assert False, "Additional SQL statements remain" elif not rule.is_consumed: rule.no_more_statements() @contextlib.contextmanager def assert_engine(engine): asserter = SQLAsserter() orig = [] @event.listens_for(engine, "before_execute") def connection_execute(conn, clauseelement, multiparams, params): # grab the original statement + params before any cursor # execution orig[:] = clauseelement, multiparams, params @event.listens_for(engine, "after_cursor_execute") def cursor_execute( conn, cursor, statement, parameters, context, executemany ): if not context: return # then grab real cursor statements and associate them all # around a single context if ( asserter.accumulated and asserter.accumulated[-1].context is context ): obs = asserter.accumulated[-1] else: obs = SQLExecuteObserved(context, orig[0], orig[1], orig[2]) asserter.accumulated.append(obs) obs.statements.append( SQLCursorExecuteObserved( statement, parameters, context, executemany ) ) try: yield asserter finally: event.remove(engine, "after_cursor_execute", cursor_execute) event.remove(engine, "before_execute", connection_execute) asserter._close()	13.27 KB	2026-01-20 12:58
📄 config.py # testing/config.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php import collections requirements = None db = None db_url = None db_opts = None file_config = None test_schema = None test_schema_2 = None _current = None _fixture_functions = None # installed by plugin_base def combinations(comb, kw): r"""Deliver multiple versions of a test based on positional combinations. This is a facade over pytest.mark.parametrize. :param \comb: argument combinations. These are tuples that will be passed positionally to the decorated function. :param argnames: optional list of argument names. These are the names of the arguments in the test function that correspond to the entries in each argument tuple. pytest.mark.parametrize requires this, however the combinations function will derive it automatically if not present by using ``inspect.getfullargspec(fn).args[1:]``. Note this assumes the first argument is "self" which is discarded. :param id\_: optional id template. This is a string template that describes how the "id" for each parameter set should be defined, if any. The number of characters in the template should match the number of entries in each argument tuple. Each character describes how the corresponding entry in the argument tuple should be handled, as far as whether or not it is included in the arguments passed to the function, as well as if it is included in the tokens used to create the id of the parameter set. If omitted, the argument combinations are passed to parametrize as is. If passed, each argument combination is turned into a pytest.param() object, mapping the elements of the argument tuple to produce an id based on a character value in the same position within the string template using the following scheme:: i - the given argument is a string that is part of the id only, don't pass it as an argument n - the given argument should be passed and it should be added to the id by calling the .__name__ attribute r - the given argument should be passed and it should be added to the id by calling repr() s - the given argument should be passed and it should be added to the id by calling str() a - (argument) the given argument should be passed and it should not be used to generated the id e.g.:: @testing.combinations( (operator.eq, "eq"), (operator.ne, "ne"), (operator.gt, "gt"), (operator.lt, "lt"), id_="na" ) def test_operator(self, opfunc, name): pass The above combination will call ``.__name__`` on the first member of each tuple and use that as the "id" to pytest.param(). """ return _fixture_functions.combinations(comb, kw) def combinations_list(arg_iterable, kw): "As combination, but takes a single iterable" return combinations(arg_iterable, *kw) def fixture(arg, *kw): return _fixture_functions.fixture(arg, **kw) def get_current_test_name(): return _fixture_functions.get_current_test_name() class Config(object): def __init__(self, db, db_opts, options, file_config): self._set_name(db) self.db = db self.db_opts = db_opts self.options = options self.file_config = file_config self.test_schema = "test_schema" self.test_schema_2 = "test_schema_2" _stack = collections.deque() _configs = set() def _set_name(self, db): if db.dialect.server_version_info: svi = ".".join(str(tok) for tok in db.dialect.server_version_info) self.name = "%s+%s_[%s]" % (db.name, db.driver, svi) else: self.name = "%s+%s" % (db.name, db.driver) @classmethod def register(cls, db, db_opts, options, file_config): """add a config as one of the global configs. If there are no configs set up yet, this config also gets set as the "_current". """ cfg = Config(db, db_opts, options, file_config) cls._configs.add(cfg) return cfg @classmethod def set_as_current(cls, config, namespace): global db, _current, db_url, test_schema, test_schema_2, db_opts _current = config db_url = config.db.url db_opts = config.db_opts test_schema = config.test_schema test_schema_2 = config.test_schema_2 namespace.db = db = config.db @classmethod def push_engine(cls, db, namespace): assert _current, "Can't push without a default Config set up" cls.push( Config( db, _current.db_opts, _current.options, _current.file_config ), namespace, ) @classmethod def push(cls, config, namespace): cls._stack.append(_current) cls.set_as_current(config, namespace) @classmethod def reset(cls, namespace): if cls._stack: cls.set_as_current(cls._stack[0], namespace) cls._stack.clear() @classmethod def all_configs(cls): return cls._configs @classmethod def all_dbs(cls): for cfg in cls.all_configs(): yield cfg.db def skip_test(self, msg): skip_test(msg) def skip_test(msg): raise _fixture_functions.skip_test_exception(msg)	5.53 KB	2026-01-20 12:58
📄 engines.py # testing/engines.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php from __future__ import absolute_import import re import warnings import weakref from . import config from . import uses_deprecated from .util import decorator from .. import event from .. import pool class ConnectionKiller(object): def __init__(self): self.proxy_refs = weakref.WeakKeyDictionary() self.testing_engines = weakref.WeakKeyDictionary() self.conns = set() def add_engine(self, engine): self.testing_engines[engine] = True def connect(self, dbapi_conn, con_record): self.conns.add((dbapi_conn, con_record)) def checkout(self, dbapi_con, con_record, con_proxy): self.proxy_refs[con_proxy] = True def invalidate(self, dbapi_con, con_record, exception): self.conns.discard((dbapi_con, con_record)) def _safe(self, fn): try: fn() except Exception as e: warnings.warn( "testing_reaper couldn't " "rollback/close connection: %s" % e ) def rollback_all(self): for rec in list(self.proxy_refs): if rec is not None and rec.is_valid: self._safe(rec.rollback) def close_all(self): for rec in list(self.proxy_refs): if rec is not None and rec.is_valid: self._safe(rec._close) def _after_test_ctx(self): # this can cause a deadlock with pg8000 - pg8000 acquires # prepared statement lock inside of rollback() - if async gc # is collecting in finalize_fairy, deadlock. # not sure if this should be if pypy/jython only. # note that firebird/fdb definitely needs this though for conn, rec in list(self.conns): if rec.connection is None: # this is a hint that the connection is closed, which # is causing segfaults on mysqlclient due to # https://github.com/PyMySQL/mysqlclient-python/issues/270; # try to work around here continue self._safe(conn.rollback) def _stop_test_ctx(self): if config.options.low_connections: self._stop_test_ctx_minimal() else: self._stop_test_ctx_aggressive() @uses_deprecated() def _stop_test_ctx_minimal(self): self.close_all() self.conns = set() for rec in list(self.testing_engines): if rec is not config.db: rec.dispose() @uses_deprecated() def _stop_test_ctx_aggressive(self): self.close_all() for conn, rec in list(self.conns): self._safe(conn.close) rec.connection = None self.conns = set() for rec in list(self.testing_engines): rec.dispose() def assert_all_closed(self): for rec in self.proxy_refs: if rec.is_valid: assert False testing_reaper = ConnectionKiller() def drop_all_tables(metadata, bind): testing_reaper.close_all() if hasattr(bind, "close"): bind.close() if not config.db.dialect.supports_alter: from . import assertions with assertions.expect_warnings("Can't sort tables", assert_=False): metadata.drop_all(bind) else: metadata.drop_all(bind) @decorator def assert_conns_closed(fn, args, kw): try: fn(args, *kw) finally: testing_reaper.assert_all_closed() @decorator def rollback_open_connections(fn, args, *kw): """Decorator that rolls back all open connections after fn execution.""" try: fn(args, *kw) finally: testing_reaper.rollback_all() @decorator def close_first(fn, args, *kw): """Decorator that closes all connections before fn execution.""" testing_reaper.close_all() fn(args, *kw) @decorator def close_open_connections(fn, args, *kw): """Decorator that closes all connections after fn execution.""" try: fn(args, *kw) finally: testing_reaper.close_all() def all_dialects(exclude=None): import sqlalchemy.databases as d for name in d.__all__: # TEMPORARY if exclude and name in exclude: continue mod = getattr(d, name, None) if not mod: mod = getattr( __import__("sqlalchemy.databases.%s" % name).databases, name ) yield mod.dialect() class ReconnectFixture(object): def __init__(self, dbapi): self.dbapi = dbapi self.connections = [] self.is_stopped = False def __getattr__(self, key): return getattr(self.dbapi, key) def connect(self, args, *kwargs): conn = self.dbapi.connect(args, kwargs) if self.is_stopped: self._safe(conn.close) curs = conn.cursor() # should fail on Oracle etc. # should fail for everything that didn't fail # above, connection is closed curs.execute("select 1") assert False, "simulated connect failure didn't work" else: self.connections.append(conn) return conn def _safe(self, fn): try: fn() except Exception as e: warnings.warn( "ReconnectFixture couldn't " "close connection: %s" % e ) def shutdown(self, stop=False): # TODO: this doesn't cover all cases # as nicely as we'd like, namely MySQLdb. # would need to implement R. Brewer's # proxy server idea to get better # coverage. self.is_stopped = stop for c in list(self.connections): self._safe(c.close) self.connections = [] def restart(self): self.is_stopped = False def reconnecting_engine(url=None, options=None): url = url or config.db.url dbapi = config.db.dialect.dbapi if not options: options = {} options["module"] = ReconnectFixture(dbapi) engine = testing_engine(url, options) _dispose = engine.dispose def dispose(): engine.dialect.dbapi.shutdown() engine.dialect.dbapi.is_stopped = False _dispose() engine.test_shutdown = engine.dialect.dbapi.shutdown engine.test_restart = engine.dialect.dbapi.restart engine.dispose = dispose return engine def testing_engine(url=None, options=None): """Produce an engine configured by --options with optional overrides.""" from sqlalchemy import create_engine from sqlalchemy.engine.url import make_url if not options: use_reaper = True else: use_reaper = options.pop("use_reaper", True) url = url or config.db.url url = make_url(url) if options is None: if config.db is None or url.drivername == config.db.url.drivername: options = config.db_opts else: options = {} elif config.db is not None and url.drivername == config.db.url.drivername: default_opt = config.db_opts.copy() default_opt.update(options) engine = create_engine(url, options) engine._has_events = True # enable event blocks, helps with profiling if isinstance(engine.pool, pool.QueuePool): engine.pool._timeout = 0 engine.pool._max_overflow = 0 if use_reaper: event.listen(engine.pool, "connect", testing_reaper.connect) event.listen(engine.pool, "checkout", testing_reaper.checkout) event.listen(engine.pool, "invalidate", testing_reaper.invalidate) testing_reaper.add_engine(engine) return engine def mock_engine(dialect_name=None): """Provides a mocking engine based on the current testing.db. This is normally used to test DDL generation flow as emitted by an Engine. It should not be used in other cases, as assert_compile() and assert_sql_execution() are much better choices with fewer moving parts. """ from sqlalchemy import create_engine if not dialect_name: dialect_name = config.db.name buffer = [] def executor(sql, a, kw): buffer.append(sql) def assert_sql(stmts): recv = [re.sub(r"[\n\t]", "", str(s)) for s in buffer] assert recv == stmts, recv def print_sql(): d = engine.dialect return "\n".join(str(s.compile(dialect=d)) for s in engine.mock) engine = create_engine( dialect_name + "://", strategy="mock", executor=executor ) assert not hasattr(engine, "mock") engine.mock = buffer engine.assert_sql = assert_sql engine.print_sql = print_sql return engine class DBAPIProxyCursor(object): """Proxy a DBAPI cursor. Tests can provide subclasses of this to intercept DBAPI-level cursor operations. """ def __init__(self, engine, conn, args, *kwargs): self.engine = engine self.connection = conn self.cursor = conn.cursor(args, kwargs) def execute(self, stmt, parameters=None, kw): if parameters: return self.cursor.execute(stmt, parameters, kw) else: return self.cursor.execute(stmt, kw) def executemany(self, stmt, params, kw): return self.cursor.executemany(stmt, params, kw) def __getattr__(self, key): return getattr(self.cursor, key) def __enter__(self): return self def __exit__(self, args): pass class DBAPIProxyConnection(object): """Proxy a DBAPI connection. Tests can provide subclasses of this to intercept DBAPI-level connection operations. """ def __init__(self, engine, cursor_cls): self.conn = self._sqla_unwrap = engine.pool._creator() self.engine = engine self.cursor_cls = cursor_cls def cursor(self, args, *kwargs): return self.cursor_cls(self.engine, self.conn, args, **kwargs) def close(self): self.conn.close() def __getattr__(self, key): return getattr(self.conn, key) def proxying_engine( conn_cls=DBAPIProxyConnection, cursor_cls=DBAPIProxyCursor ): """Produce an engine that provides proxy hooks for common methods. """ def mock_conn(): return conn_cls(config.db, cursor_cls) return testing_engine(options={"creator": mock_conn})	10.28 KB	2026-01-20 12:58
📄 entities.py # testing/entities.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php import sqlalchemy as sa from .. import exc as sa_exc from ..util import compat _repr_stack = set() class BasicEntity(object): def __init__(self, **kw): for key, value in kw.items(): setattr(self, key, value) def __repr__(self): if id(self) in _repr_stack: return object.__repr__(self) _repr_stack.add(id(self)) try: return "%s(%s)" % ( (self.__class__.__name__), ", ".join( [ "%s=%r" % (key, getattr(self, key)) for key in sorted(self.__dict__.keys()) if not key.startswith("_") ] ), ) finally: _repr_stack.remove(id(self)) _recursion_stack = set() class ComparableEntity(BasicEntity): def __hash__(self): return hash(self.__class__) def __ne__(self, other): return not self.__eq__(other) def __eq__(self, other): """'Deep, sparse compare. Deeply compare two entities, following the non-None attributes of the non-persisted object, if possible. """ if other is self: return True elif not self.__class__ == other.__class__: return False if id(self) in _recursion_stack: return True _recursion_stack.add(id(self)) try: # pick the entity that's not SA persisted as the source try: self_key = sa.orm.attributes.instance_state(self).key except sa.orm.exc.NO_STATE: self_key = None if other is None: a = self b = other elif self_key is not None: a = other b = self else: a = self b = other for attr in list(a.__dict__): if attr.startswith("_"): continue value = getattr(a, attr) try: # handle lazy loader errors battr = getattr(b, attr) except (AttributeError, sa_exc.UnboundExecutionError): return False if hasattr(value, "__iter__") and not isinstance( value, compat.string_types ): if hasattr(value, "__getitem__") and not hasattr( value, "keys" ): if list(value) != list(battr): return False else: if set(value) != set(battr): return False else: if value is not None and value != battr: return False return True finally: _recursion_stack.remove(id(self))	3.13 KB	2026-01-20 12:58
📄 exclusions.py # testing/exclusions.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php import contextlib import operator import re import sys from . import config from .. import util from ..util import decorator from ..util.compat import inspect_getfullargspec def skip_if(predicate, reason=None): rule = compound() pred = _as_predicate(predicate, reason) rule.skips.add(pred) return rule def fails_if(predicate, reason=None): rule = compound() pred = _as_predicate(predicate, reason) rule.fails.add(pred) return rule class compound(object): def __init__(self): self.fails = set() self.skips = set() self.tags = set() def __add__(self, other): return self.add(other) def as_skips(self): rule = compound() rule.skips.update(self.skips) rule.skips.update(self.fails) rule.tags.update(self.tags) return rule def add(self, others): copy = compound() copy.fails.update(self.fails) copy.skips.update(self.skips) copy.tags.update(self.tags) for other in others: copy.fails.update(other.fails) copy.skips.update(other.skips) copy.tags.update(other.tags) return copy def not_(self): copy = compound() copy.fails.update(NotPredicate(fail) for fail in self.fails) copy.skips.update(NotPredicate(skip) for skip in self.skips) copy.tags.update(self.tags) return copy @property def enabled(self): return self.enabled_for_config(config._current) def enabled_for_config(self, config): for predicate in self.skips.union(self.fails): if predicate(config): return False else: return True def matching_config_reasons(self, config): return [ predicate._as_string(config) for predicate in self.skips.union(self.fails) if predicate(config) ] def include_test(self, include_tags, exclude_tags): return bool( not self.tags.intersection(exclude_tags) and (not include_tags or self.tags.intersection(include_tags)) ) def _extend(self, other): self.skips.update(other.skips) self.fails.update(other.fails) self.tags.update(other.tags) def __call__(self, fn): if hasattr(fn, "_sa_exclusion_extend"): fn._sa_exclusion_extend._extend(self) return fn @decorator def decorate(fn, args, *kw): return self._do(config._current, fn, args, *kw) decorated = decorate(fn) decorated._sa_exclusion_extend = self return decorated @contextlib.contextmanager def fail_if(self): all_fails = compound() all_fails.fails.update(self.skips.union(self.fails)) try: yield except Exception as ex: all_fails._expect_failure(config._current, ex) else: all_fails._expect_success(config._current) def _do(self, cfg, fn, args, *kw): for skip in self.skips: if skip(cfg): msg = "'%s' : %s" % ( config.get_current_test_name(), skip._as_string(cfg), ) config.skip_test(msg) try: return_value = fn(args, *kw) except Exception as ex: self._expect_failure(cfg, ex, name=fn.__name__) else: self._expect_success(cfg, name=fn.__name__) return return_value def _expect_failure(self, config, ex, name="block"): for fail in self.fails: if fail(config): if util.py2k: str_ex = unicode(ex).encode("utf-8", errors="ignore") else: str_ex = str(ex) print( ( "%s failed as expected (%s): %s " % (name, fail._as_string(config), str_ex) ) ) break else: util.raise_(ex, with_traceback=sys.exc_info()[2]) def _expect_success(self, config, name="block"): if not self.fails: return for fail in self.fails: if fail(config): raise AssertionError( "Unexpected success for '%s' (%s)" % ( name, " and ".join( fail._as_string(config) for fail in self.fails ), ) ) def requires_tag(tagname): return tags([tagname]) def tags(tagnames): comp = compound() comp.tags.update(tagnames) return comp def only_if(predicate, reason=None): predicate = _as_predicate(predicate) return skip_if(NotPredicate(predicate), reason) def succeeds_if(predicate, reason=None): predicate = _as_predicate(predicate) return fails_if(NotPredicate(predicate), reason) class Predicate(object): @classmethod def as_predicate(cls, predicate, description=None): if isinstance(predicate, compound): return cls.as_predicate(predicate.enabled_for_config, description) elif isinstance(predicate, Predicate): if description and predicate.description is None: predicate.description = description return predicate elif isinstance(predicate, (list, set)): return OrPredicate( [cls.as_predicate(pred) for pred in predicate], description ) elif isinstance(predicate, tuple): return SpecPredicate(predicate) elif isinstance(predicate, util.string_types): tokens = re.match( r"([\+\w]+)\s(?:(>=\|==\|!=\|<=\|<\|>)\s([\d\.]+))?", predicate ) if not tokens: raise ValueError( "Couldn't locate DB name in predicate: %r" % predicate ) db = tokens.group(1) op = tokens.group(2) spec = ( tuple(int(d) for d in tokens.group(3).split(".")) if tokens.group(3) else None ) return SpecPredicate(db, op, spec, description=description) elif util.callable(predicate): return LambdaPredicate(predicate, description) else: assert False, "unknown predicate type: %s" % predicate def _format_description(self, config, negate=False): bool_ = self(config) if negate: bool_ = not negate return self.description % { "driver": config.db.url.get_driver_name() if config else "<no driver>", "database": config.db.url.get_backend_name() if config else "<no database>", "doesnt_support": "doesn't support" if bool_ else "does support", "does_support": "does support" if bool_ else "doesn't support", } def _as_string(self, config=None, negate=False): raise NotImplementedError() class BooleanPredicate(Predicate): def __init__(self, value, description=None): self.value = value self.description = description or "boolean %s" % value def __call__(self, config): return self.value def _as_string(self, config, negate=False): return self._format_description(config, negate=negate) class SpecPredicate(Predicate): def __init__(self, db, op=None, spec=None, description=None): self.db = db self.op = op self.spec = spec self.description = description _ops = { "<": operator.lt, ">": operator.gt, "==": operator.eq, "!=": operator.ne, "<=": operator.le, ">=": operator.ge, "in": operator.contains, "between": lambda val, pair: val >= pair[0] and val <= pair[1], } def __call__(self, config): engine = config.db if "+" in self.db: dialect, driver = self.db.split("+") else: dialect, driver = self.db, None if dialect and engine.name != dialect: return False if driver is not None and engine.driver != driver: return False if self.op is not None: assert driver is None, "DBAPI version specs not supported yet" version = _server_version(engine) oper = ( hasattr(self.op, "__call__") and self.op or self._ops[self.op] ) return oper(version, self.spec) else: return True def _as_string(self, config, negate=False): if self.description is not None: return self._format_description(config) elif self.op is None: if negate: return "not %s" % self.db else: return "%s" % self.db else: if negate: return "not %s %s %s" % (self.db, self.op, self.spec) else: return "%s %s %s" % (self.db, self.op, self.spec) class LambdaPredicate(Predicate): def __init__(self, lambda_, description=None, args=None, kw=None): spec = inspect_getfullargspec(lambda_) if not spec[0]: self.lambda_ = lambda db: lambda_() else: self.lambda_ = lambda_ self.args = args or () self.kw = kw or {} if description: self.description = description elif lambda_.__doc__: self.description = lambda_.__doc__ else: self.description = "custom function" def __call__(self, config): return self.lambda_(config) def _as_string(self, config, negate=False): return self._format_description(config) class NotPredicate(Predicate): def __init__(self, predicate, description=None): self.predicate = predicate self.description = description def __call__(self, config): return not self.predicate(config) def _as_string(self, config, negate=False): if self.description: return self._format_description(config, not negate) else: return self.predicate._as_string(config, not negate) class OrPredicate(Predicate): def __init__(self, predicates, description=None): self.predicates = predicates self.description = description def __call__(self, config): for pred in self.predicates: if pred(config): return True return False def _eval_str(self, config, negate=False): if negate: conjunction = " and " else: conjunction = " or " return conjunction.join( p._as_string(config, negate=negate) for p in self.predicates ) def _negation_str(self, config): if self.description is not None: return "Not " + self._format_description(config) else: return self._eval_str(config, negate=True) def _as_string(self, config, negate=False): if negate: return self._negation_str(config) else: if self.description is not None: return self._format_description(config) else: return self._eval_str(config) _as_predicate = Predicate.as_predicate def _is_excluded(db, op, spec): return SpecPredicate(db, op, spec)(config._current) def _server_version(engine): """Return a server_version_info tuple.""" # force metadata to be retrieved conn = engine.connect() version = getattr(engine.dialect, "server_version_info", None) if version is None: version = () conn.close() return version def db_spec(dbs): return OrPredicate([Predicate.as_predicate(db) for db in dbs]) def open(): # noqa return skip_if(BooleanPredicate(False, "mark as execute")) def closed(): return skip_if(BooleanPredicate(True, "marked as skip")) def fails(reason=None): return fails_if(BooleanPredicate(True, reason or "expected to fail")) @decorator def future(fn, arg): return fails_if(LambdaPredicate(fn), "Future feature") def fails_on(db, reason=None): return fails_if(db, reason) def fails_on_everything_except(dbs): return succeeds_if(OrPredicate([Predicate.as_predicate(db) for db in dbs])) def skip(db, reason=None): return skip_if(db, reason) def only_on(dbs, reason=None): return only_if( OrPredicate( [Predicate.as_predicate(db, reason) for db in util.to_list(dbs)] ) ) def exclude(db, op, spec, reason=None): return skip_if(SpecPredicate(db, op, spec), reason) def against(config, queries): assert queries, "no queries sent!" return OrPredicate([Predicate.as_predicate(query) for query in queries])( config )	12.91 KB	2026-01-20 12:58
📄 fixtures.py # testing/fixtures.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php import re import sys import sqlalchemy as sa from . import assertions from . import config from . import schema from .engines import drop_all_tables from .entities import BasicEntity from .entities import ComparableEntity from .util import adict from .. import event from .. import util from ..ext.declarative import declarative_base from ..ext.declarative import DeclarativeMeta from ..schema import sort_tables_and_constraints # whether or not we use unittest changes things dramatically, # as far as how pytest collection works. class TestBase(object): # A sequence of requirement names matching testing.requires decorators __requires__ = () # A sequence of dialect names to exclude from the test class. __unsupported_on__ = () # If present, test class is only runnable for the single specified # dialect. If you need multiple, use __unsupported_on__ and invert. __only_on__ = None # A sequence of no-arg callables. If any are True, the entire testcase is # skipped. __skip_if__ = None def assert_(self, val, msg=None): assert val, msg # apparently a handful of tests are doing this....OK def setup(self): if hasattr(self, "setUp"): self.setUp() def teardown(self): if hasattr(self, "tearDown"): self.tearDown() @config.fixture() def connection(self): conn = config.db.connect() trans = conn.begin() try: yield conn finally: trans.rollback() conn.close() # propose a replacement for @testing.provide_metadata. # the problem with this is that TablesTest below has a ".metadata" # attribute already which is accessed directly as part of the # @testing.provide_metadata pattern. Might need to call this _metadata # for it to be useful. # @config.fixture() # def metadata(self): # """Provide bound MetaData for a single test, dropping afterwards.""" # # from . import engines # metadata = schema.MetaData(config.db) # try: # yield metadata # finally: # engines.drop_all_tables(metadata, config.db) class TablesTest(TestBase): # 'once', None run_setup_bind = "once" # 'once', 'each', None run_define_tables = "once" # 'once', 'each', None run_create_tables = "once" # 'once', 'each', None run_inserts = "each" # 'each', None run_deletes = "each" # 'once', None run_dispose_bind = None bind = None metadata = None tables = None other = None @classmethod def setup_class(cls): cls._init_class() cls._setup_once_tables() cls._setup_once_inserts() @classmethod def _init_class(cls): if cls.run_define_tables == "each": if cls.run_create_tables == "once": cls.run_create_tables = "each" assert cls.run_inserts in ("each", None) cls.other = adict() cls.tables = adict() cls.bind = cls.setup_bind() cls.metadata = sa.MetaData() cls.metadata.bind = cls.bind @classmethod def _setup_once_inserts(cls): if cls.run_inserts == "once": cls._load_fixtures() with cls.bind.begin() as conn: cls.insert_data(conn) @classmethod def _setup_once_tables(cls): if cls.run_define_tables == "once": cls.define_tables(cls.metadata) if cls.run_create_tables == "once": cls.metadata.create_all(cls.bind) cls.tables.update(cls.metadata.tables) def _setup_each_tables(self): if self.run_define_tables == "each": self.define_tables(self.metadata) if self.run_create_tables == "each": self.metadata.create_all(self.bind) self.tables.update(self.metadata.tables) elif self.run_create_tables == "each": self.metadata.create_all(self.bind) def _setup_each_inserts(self): if self.run_inserts == "each": self._load_fixtures() with self.bind.begin() as conn: self.insert_data(conn) def _teardown_each_tables(self): if self.run_define_tables == "each": self.tables.clear() if self.run_create_tables == "each": drop_all_tables(self.metadata, self.bind) self.metadata.clear() elif self.run_create_tables == "each": drop_all_tables(self.metadata, self.bind) # no need to run deletes if tables are recreated on setup if self.run_define_tables != "each" and self.run_deletes == "each": with self.bind.connect() as conn: for table in reversed( [ t for (t, fks) in sort_tables_and_constraints( self.metadata.tables.values() ) if t is not None ] ): try: conn.execute(table.delete()) except sa.exc.DBAPIError as ex: util.print_( ("Error emptying table %s: %r" % (table, ex)), file=sys.stderr, ) def setup(self): self._setup_each_tables() self._setup_each_inserts() def teardown(self): self._teardown_each_tables() @classmethod def _teardown_once_metadata_bind(cls): if cls.run_create_tables: drop_all_tables(cls.metadata, cls.bind) if cls.run_dispose_bind == "once": cls.dispose_bind(cls.bind) cls.metadata.bind = None if cls.run_setup_bind is not None: cls.bind = None @classmethod def teardown_class(cls): cls._teardown_once_metadata_bind() @classmethod def setup_bind(cls): return config.db @classmethod def dispose_bind(cls, bind): if hasattr(bind, "dispose"): bind.dispose() elif hasattr(bind, "close"): bind.close() @classmethod def define_tables(cls, metadata): pass @classmethod def fixtures(cls): return {} @classmethod def insert_data(cls, connection): pass def sql_count_(self, count, fn): self.assert_sql_count(self.bind, fn, count) def sql_eq_(self, callable_, statements): self.assert_sql(self.bind, callable_, statements) @classmethod def _load_fixtures(cls): """Insert rows as represented by the fixtures() method.""" headers, rows = {}, {} for table, data in cls.fixtures().items(): if len(data) < 2: continue if isinstance(table, util.string_types): table = cls.tables[table] headers[table] = data[0] rows[table] = data[1:] for table, fks in sort_tables_and_constraints( cls.metadata.tables.values() ): if table is None: continue if table not in headers: continue cls.bind.execute( table.insert(), [ dict(zip(headers[table], column_values)) for column_values in rows[table] ], ) class RemovesEvents(object): @util.memoized_property def _event_fns(self): return set() def event_listen(self, target, name, fn, kw): self._event_fns.add((target, name, fn)) event.listen(target, name, fn, kw) def teardown(self): for key in self._event_fns: event.remove(key) super_ = super(RemovesEvents, self) if hasattr(super_, "teardown"): super_.teardown() class _ORMTest(object): @classmethod def teardown_class(cls): sa.orm.session.close_all_sessions() sa.orm.clear_mappers() class ORMTest(_ORMTest, TestBase): pass class MappedTest(_ORMTest, TablesTest, assertions.AssertsExecutionResults): # 'once', 'each', None run_setup_classes = "once" # 'once', 'each', None run_setup_mappers = "each" classes = None @classmethod def setup_class(cls): cls._init_class() if cls.classes is None: cls.classes = adict() cls._setup_once_tables() cls._setup_once_classes() cls._setup_once_mappers() cls._setup_once_inserts() @classmethod def teardown_class(cls): cls._teardown_once_class() cls._teardown_once_metadata_bind() def setup(self): self._setup_each_tables() self._setup_each_classes() self._setup_each_mappers() self._setup_each_inserts() def teardown(self): sa.orm.session.close_all_sessions() self._teardown_each_mappers() self._teardown_each_classes() self._teardown_each_tables() @classmethod def _teardown_once_class(cls): cls.classes.clear() _ORMTest.teardown_class() @classmethod def _setup_once_classes(cls): if cls.run_setup_classes == "once": cls._with_register_classes(cls.setup_classes) @classmethod def _setup_once_mappers(cls): if cls.run_setup_mappers == "once": cls._with_register_classes(cls.setup_mappers) def _setup_each_mappers(self): if self.run_setup_mappers == "each": self._with_register_classes(self.setup_mappers) def _setup_each_classes(self): if self.run_setup_classes == "each": self._with_register_classes(self.setup_classes) @classmethod def _with_register_classes(cls, fn): """Run a setup method, framing the operation with a Base class that will catch new subclasses to be established within the "classes" registry. """ cls_registry = cls.classes class FindFixture(type): def __init__(cls, classname, bases, dict_): cls_registry[classname] = cls type.__init__(cls, classname, bases, dict_) class _Base(util.with_metaclass(FindFixture, object)): pass class Basic(BasicEntity, _Base): pass class Comparable(ComparableEntity, _Base): pass cls.Basic = Basic cls.Comparable = Comparable fn() def _teardown_each_mappers(self): # some tests create mappers in the test bodies # and will define setup_mappers as None - # clear mappers in any case if self.run_setup_mappers != "once": sa.orm.clear_mappers() def _teardown_each_classes(self): if self.run_setup_classes != "once": self.classes.clear() @classmethod def setup_classes(cls): pass @classmethod def setup_mappers(cls): pass class DeclarativeMappedTest(MappedTest): run_setup_classes = "once" run_setup_mappers = "once" @classmethod def _setup_once_tables(cls): pass @classmethod def _with_register_classes(cls, fn): cls_registry = cls.classes class FindFixtureDeclarative(DeclarativeMeta): def __init__(cls, classname, bases, dict_): cls_registry[classname] = cls DeclarativeMeta.__init__(cls, classname, bases, dict_) class DeclarativeBasic(object): __table_cls__ = schema.Table _DeclBase = declarative_base( metadata=cls.metadata, metaclass=FindFixtureDeclarative, cls=DeclarativeBasic, ) cls.DeclarativeBasic = _DeclBase # sets up cls.Basic which is helpful for things like composite # classes super(DeclarativeMappedTest, cls)._with_register_classes(fn) if cls.metadata.tables and cls.run_create_tables: cls.metadata.create_all(config.db) class ComputedReflectionFixtureTest(TablesTest): run_inserts = run_deletes = None __backend__ = True __requires__ = ("computed_columns", "table_reflection") regexp = re.compile(r"[\[\]\(\)\s`'\"]") def normalize(self, text): return self.regexp.sub("", text).lower() @classmethod def define_tables(cls, metadata): from .. import Integer from .. import testing from ..schema import Column from ..schema import Computed from ..schema import Table Table( "computed_default_table", metadata, Column("id", Integer, primary_key=True), Column("normal", Integer), Column("computed_col", Integer, Computed("normal + 42")), Column("with_default", Integer, server_default="42"), ) t = Table( "computed_column_table", metadata, Column("id", Integer, primary_key=True), Column("normal", Integer), Column("computed_no_flag", Integer, Computed("normal + 42")), ) if testing.requires.schemas.enabled: t2 = Table( "computed_column_table", metadata, Column("id", Integer, primary_key=True), Column("normal", Integer), Column("computed_no_flag", Integer, Computed("normal / 42")), schema=config.test_schema, ) if testing.requires.computed_columns_virtual.enabled: t.append_column( Column( "computed_virtual", Integer, Computed("normal + 2", persisted=False), ) ) if testing.requires.schemas.enabled: t2.append_column( Column( "computed_virtual", Integer, Computed("normal / 2", persisted=False), ) ) if testing.requires.computed_columns_stored.enabled: t.append_column( Column( "computed_stored", Integer, Computed("normal - 42", persisted=True), ) ) if testing.requires.schemas.enabled: t2.append_column( Column( "computed_stored", Integer, Computed("normal * 42", persisted=True), ) )	14.57 KB	2026-01-20 12:58
📄 mock.py # testing/mock.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php """Import stub for mock library. """ from __future__ import absolute_import from ..util import py33 if py33: from unittest.mock import MagicMock from unittest.mock import Mock from unittest.mock import call from unittest.mock import patch from unittest.mock import ANY else: try: from mock import MagicMock # noqa from mock import Mock # noqa from mock import call # noqa from mock import patch # noqa from mock import ANY # noqa except ImportError: raise ImportError( "SQLAlchemy's test suite requires the " "'mock' library as of 0.8.2." )	893 B	2026-01-20 12:58
📄 pickleable.py # testing/pickleable.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php """Classes used in pickling tests, need to be at the module level for unpickling. """ from . import fixtures class User(fixtures.ComparableEntity): pass class Order(fixtures.ComparableEntity): pass class Dingaling(fixtures.ComparableEntity): pass class EmailUser(User): pass class Address(fixtures.ComparableEntity): pass # TODO: these are kind of arbitrary.... class Child1(fixtures.ComparableEntity): pass class Child2(fixtures.ComparableEntity): pass class Parent(fixtures.ComparableEntity): pass class Screen(object): def __init__(self, obj, parent=None): self.obj = obj self.parent = parent class Foo(object): def __init__(self, moredata): self.data = "im data" self.stuff = "im stuff" self.moredata = moredata __hash__ = object.__hash__ def __eq__(self, other): return ( other.data == self.data and other.stuff == self.stuff and other.moredata == self.moredata ) class Bar(object): def __init__(self, x, y): self.x = x self.y = y __hash__ = object.__hash__ def __eq__(self, other): return ( other.__class__ is self.__class__ and other.x == self.x and other.y == self.y ) def __str__(self): return "Bar(%d, %d)" % (self.x, self.y) class OldSchool: def __init__(self, x, y): self.x = x self.y = y def __eq__(self, other): return ( other.__class__ is self.__class__ and other.x == self.x and other.y == self.y ) class OldSchoolWithoutCompare: def __init__(self, x, y): self.x = x self.y = y class BarWithoutCompare(object): def __init__(self, x, y): self.x = x self.y = y def __str__(self): return "Bar(%d, %d)" % (self.x, self.y) class NotComparable(object): def __init__(self, data): self.data = data def __hash__(self): return id(self) def __eq__(self, other): return NotImplemented def __ne__(self, other): return NotImplemented class BrokenComparable(object): def __init__(self, data): self.data = data def __hash__(self): return id(self) def __eq__(self, other): raise NotImplementedError def __ne__(self, other): raise NotImplementedError	2.63 KB	2026-01-20 12:58
📁 plugin	-	2026-03-21 15:45
📄 profiling.py # testing/profiling.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php """Profiling support for unit and performance tests. These are special purpose profiling methods which operate in a more fine-grained way than nose's profiling plugin. """ import collections import contextlib import os import platform import pstats import sys from . import config from .util import gc_collect from ..util import update_wrapper try: import cProfile except ImportError: cProfile = None _current_test = None # ProfileStatsFile instance, set up in plugin_base _profile_stats = None class ProfileStatsFile(object): """Store per-platform/fn profiling results in a file. We're still targeting Py2.5, 2.4 on 0.7 with no dependencies, so no json lib :( need to roll something silly """ def __init__(self, filename): self.force_write = ( config.options is not None and config.options.force_write_profiles ) self.write = self.force_write or ( config.options is not None and config.options.write_profiles ) self.fname = os.path.abspath(filename) self.short_fname = os.path.split(self.fname)[-1] self.data = collections.defaultdict( lambda: collections.defaultdict(dict) ) self._read() if self.write: # rewrite for the case where features changed, # etc. self._write() @property def platform_key(self): dbapi_key = config.db.name + "_" + config.db.driver if config.db.name == "sqlite" and config.db.dialect._is_url_file_db( config.db.url ): dbapi_key += "_file" # keep it at 2.7, 3.1, 3.2, etc. for now. py_version = ".".join([str(v) for v in sys.version_info[0:2]]) platform_tokens = [ platform.machine(), platform.system().lower(), platform.python_implementation().lower(), py_version, dbapi_key, ] platform_tokens.append( "nativeunicode" if config.db.dialect.convert_unicode else "dbapiunicode" ) _has_cext = config.requirements._has_cextensions() platform_tokens.append(_has_cext and "cextensions" or "nocextensions") return "_".join(platform_tokens) def has_stats(self): test_key = _current_test return ( test_key in self.data and self.platform_key in self.data[test_key] ) def result(self, callcount): test_key = _current_test per_fn = self.data[test_key] per_platform = per_fn[self.platform_key] if "counts" not in per_platform: per_platform["counts"] = counts = [] else: counts = per_platform["counts"] if "current_count" not in per_platform: per_platform["current_count"] = current_count = 0 else: current_count = per_platform["current_count"] has_count = len(counts) > current_count if not has_count: counts.append(callcount) if self.write: self._write() result = None else: result = per_platform["lineno"], counts[current_count] per_platform["current_count"] += 1 return result def replace(self, callcount): test_key = _current_test per_fn = self.data[test_key] per_platform = per_fn[self.platform_key] counts = per_platform["counts"] current_count = per_platform["current_count"] if current_count < len(counts): counts[current_count - 1] = callcount else: counts[-1] = callcount if self.write: self._write() def _header(self): return ( "# %s\n" "# This file is written out on a per-environment basis.\n" "# For each test in aaa_profiling, the corresponding " "function and \n" "# environment is located within this file. " "If it doesn't exist,\n" "# the test is skipped.\n" "# If a callcount does exist, it is compared " "to what we received. \n" "# assertions are raised if the counts do not match.\n" "# \n" "# To add a new callcount test, apply the function_call_count \n" "# decorator and re-run the tests using the --write-profiles \n" "# option - this file will be rewritten including the new count.\n" "# \n" ) % (self.fname) def _read(self): try: profile_f = open(self.fname) except IOError: return for lineno, line in enumerate(profile_f): line = line.strip() if not line or line.startswith("#"): continue test_key, platform_key, counts = line.split() per_fn = self.data[test_key] per_platform = per_fn[platform_key] c = [int(count) for count in counts.split(",")] per_platform["counts"] = c per_platform["lineno"] = lineno + 1 per_platform["current_count"] = 0 profile_f.close() def _write(self): print(("Writing profile file %s" % self.fname)) profile_f = open(self.fname, "w") profile_f.write(self._header()) for test_key in sorted(self.data): per_fn = self.data[test_key] profile_f.write("\n# TEST: %s\n\n" % test_key) for platform_key in sorted(per_fn): per_platform = per_fn[platform_key] c = ",".join(str(count) for count in per_platform["counts"]) profile_f.write("%s %s %s\n" % (test_key, platform_key, c)) profile_f.close() def function_call_count(variance=0.05, times=1, warmup=0): """Assert a target for a test case's function call count. The main purpose of this assertion is to detect changes in callcounts for various functions - the actual number is not as important. Callcounts are stored in a file keyed to Python version and OS platform information. This file is generated automatically for new tests, and versioned so that unexpected changes in callcounts will be detected. """ def decorate(fn): def wrap(args, kw): for warm in range(warmup): fn(args, *kw) timerange = range(times) with count_functions(variance=variance): for time in timerange: rv = fn(args, *kw) return rv return update_wrapper(wrap, fn) return decorate @contextlib.contextmanager def count_functions(variance=0.05): if cProfile is None: raise config._skip_test_exception("cProfile is not installed") if not _profile_stats.has_stats() and not _profile_stats.write: config.skip_test( "No profiling stats available on this " "platform for this function. Run tests with " "--write-profiles to add statistics to %s for " "this platform." % _profile_stats.short_fname ) gc_collect() pr = cProfile.Profile() pr.enable() # began = time.time() yield # ended = time.time() pr.disable() # s = compat.StringIO() stats = pstats.Stats(pr, stream=sys.stdout) # timespent = ended - began callcount = stats.total_calls expected = _profile_stats.result(callcount) if expected is None: expected_count = None else: line_no, expected_count = expected print(("Pstats calls: %d Expected %s" % (callcount, expected_count))) stats.sort_stats("cumulative") stats.print_stats() if expected_count: deviance = int(callcount variance) failed = abs(callcount - expected_count) > deviance if failed or _profile_stats.force_write: if _profile_stats.write: _profile_stats.replace(callcount) else: raise AssertionError( "Adjusted function call count %s not within %s%% " "of expected %s, platform %s. Rerun with " "--write-profiles to " "regenerate this callcount." % ( callcount, (variance * 100), expected_count, _profile_stats.platform_key, ) )	8.54 KB	2026-01-20 12:58
📄 provision.py import collections import logging from . import config from . import engines from ..engine import url as sa_url from ..util import compat log = logging.getLogger(__name__) FOLLOWER_IDENT = None class register(object): def __init__(self): self.fns = {} @classmethod def init(cls, fn): return register().for_db("")(fn) def for_db(self, dbname): def decorate(fn): self.fns[dbname] = fn return self return decorate def __call__(self, cfg, arg): if isinstance(cfg, compat.string_types): url = sa_url.make_url(cfg) elif isinstance(cfg, sa_url.URL): url = cfg else: url = cfg.db.url backend = url.get_backend_name() if backend in self.fns: return self.fns[backend](cfg, arg) else: return self.fns[""](cfg, *arg) def create_follower_db(follower_ident): for cfg in _configs_for_db_operation(): log.info("CREATE database %s, URI %r", follower_ident, cfg.db.url) create_db(cfg, cfg.db, follower_ident) def setup_config(db_url, options, file_config, follower_ident): # load the dialect, which should also have it set up its provision # hooks dialect = sa_url.make_url(db_url).get_dialect() dialect.load_provisioning() if follower_ident: db_url = follower_url_from_main(db_url, follower_ident) db_opts = {} update_db_opts(db_url, db_opts) eng = engines.testing_engine(db_url, db_opts) post_configure_engine(db_url, eng, follower_ident) eng.connect().close() cfg = config.Config.register(eng, db_opts, options, file_config) if follower_ident: configure_follower(cfg, follower_ident) return cfg def drop_follower_db(follower_ident): for cfg in _configs_for_db_operation(): log.info("DROP database %s, URI %r", follower_ident, cfg.db.url) drop_db(cfg, cfg.db, follower_ident) def _configs_for_db_operation(): hosts = set() for cfg in config.Config.all_configs(): cfg.db.dispose() for cfg in config.Config.all_configs(): url = cfg.db.url backend = url.get_backend_name() host_conf = (backend, url.username, url.host, url.database) if host_conf not in hosts: yield cfg hosts.add(host_conf) for cfg in config.Config.all_configs(): cfg.db.dispose() @register.init def create_db(cfg, eng, ident): """Dynamically create a database for testing. Used when a test run will employ multiple processes, e.g., when run via `tox` or `pytest -n4`. """ raise NotImplementedError("no DB creation routine for cfg: %s" % eng.url) @register.init def drop_db(cfg, eng, ident): """Drop a database that we dynamically created for testing.""" raise NotImplementedError("no DB drop routine for cfg: %s" % eng.url) @register.init def update_db_opts(db_url, db_opts): """Set database options (db_opts) for a test database that we created.""" pass @register.init def post_configure_engine(url, engine, follower_ident): """Perform extra steps after configuring an engine for testing. (For the internal dialects, currently only used by sqlite.) """ pass @register.init def follower_url_from_main(url, ident): """Create a connection URL for a dynamically-created test database. :param url: the connection URL specified when the test run was invoked :param ident: the pytest-xdist "worker identifier" to be used as the database name """ url = sa_url.make_url(url) url.database = ident return url @register.init def configure_follower(cfg, ident): """Create dialect-specific config settings for a follower database.""" pass @register.init def run_reap_dbs(url, ident): """Remove databases that were created during the test process, after the process has ended. This is an optional step that is invoked for certain backends that do not reliably release locks on the database as long as a process is still in use. For the internal dialects, this is currently only necessary for mssql and oracle. """ pass def reap_dbs(idents_file): log.info("Reaping databases...") urls = collections.defaultdict(set) idents = collections.defaultdict(set) dialects = {} with open(idents_file) as file_: for line in file_: line = line.strip() db_name, db_url = line.split(" ") url_obj = sa_url.make_url(db_url) if db_name not in dialects: dialects[db_name] = url_obj.get_dialect() dialects[db_name].load_provisioning() url_key = (url_obj.get_backend_name(), url_obj.host) urls[url_key].add(db_url) idents[url_key].add(db_name) for url_key in urls: url = list(urls[url_key])[0] ident = idents[url_key] run_reap_dbs(url, ident) @register.init def temp_table_keyword_args(cfg, eng): """Specify keyword arguments for creating a temporary Table. Dialect-specific implementations of this method will return the kwargs that are passed to the Table method when creating a temporary table for testing, e.g., in the define_temp_tables method of the ComponentReflectionTest class in suite/test_reflection.py """ raise NotImplementedError( "no temp table keyword args routine for cfg: %s" % eng.url ) @register.init def stop_test_class(config, db, testcls): pass	5.44 KB	2026-01-20 12:58
📄 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	5.74 KB	2026-01-20 12:58
📄 requirements.py # testing/requirements.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php """Global database feature support policy. Provides decorators to mark tests requiring specific feature support from the target database. External dialect test suites should subclass SuiteRequirements to provide specific inclusion/exclusions. """ import sys from . import exclusions from .. import util class Requirements(object): pass class SuiteRequirements(Requirements): @property def create_table(self): """target platform can emit basic CreateTable DDL.""" return exclusions.open() @property def drop_table(self): """target platform can emit basic DropTable DDL.""" return exclusions.open() @property def foreign_keys(self): """Target database must support foreign keys.""" return exclusions.open() @property def on_update_cascade(self): """target database must support ON UPDATE..CASCADE behavior in foreign keys.""" return exclusions.open() @property def non_updating_cascade(self): """target database must not support ON UPDATE..CASCADE behavior in foreign keys.""" return exclusions.closed() @property def deferrable_fks(self): return exclusions.closed() @property def on_update_or_deferrable_fks(self): # TODO: exclusions should be composable, # somehow only_if([x, y]) isn't working here, negation/conjunctions # getting confused. return exclusions.only_if( lambda: self.on_update_cascade.enabled or self.deferrable_fks.enabled ) @property def self_referential_foreign_keys(self): """Target database must support self-referential foreign keys.""" return exclusions.open() @property def foreign_key_ddl(self): """Target database must support the DDL phrases for FOREIGN KEY.""" return exclusions.open() @property def named_constraints(self): """target database must support names for constraints.""" return exclusions.open() @property def subqueries(self): """Target database must support subqueries.""" return exclusions.open() @property def offset(self): """target database can render OFFSET, or an equivalent, in a SELECT. """ return exclusions.open() @property def bound_limit_offset(self): """target database can render LIMIT and/or OFFSET using a bound parameter """ return exclusions.open() @property def sql_expression_limit_offset(self): """target database can render LIMIT and/or OFFSET with a complete SQL expression, such as one that uses the addition operator. parameter """ return exclusions.open() @property def parens_in_union_contained_select_w_limit_offset(self): """Target database must support parenthesized SELECT in UNION when LIMIT/OFFSET is specifically present. E.g. (SELECT ...) UNION (SELECT ..) This is known to fail on SQLite. """ return exclusions.open() @property def parens_in_union_contained_select_wo_limit_offset(self): """Target database must support parenthesized SELECT in UNION when OFFSET/LIMIT is specifically not present. E.g. (SELECT ... LIMIT ..) UNION (SELECT .. OFFSET ..) This is known to fail on SQLite. It also fails on Oracle because without LIMIT/OFFSET, there is currently no step that creates an additional subquery. """ return exclusions.open() @property def boolean_col_expressions(self): """Target database must support boolean expressions as columns""" return exclusions.closed() @property def nullable_booleans(self): """Target database allows boolean columns to store NULL.""" return exclusions.open() @property def nullsordering(self): """Target backends that support nulls ordering.""" return exclusions.closed() @property def standalone_binds(self): """target database/driver supports bound parameters as column expressions without being in the context of a typed column. """ return exclusions.closed() @property def standalone_null_binds_whereclause(self): """target database/driver supports bound parameters with NULL in the WHERE clause, in situations where it has to be typed. """ return exclusions.open() @property def intersect(self): """Target database must support INTERSECT or equivalent.""" return exclusions.closed() @property def except_(self): """Target database must support EXCEPT or equivalent (i.e. MINUS).""" return exclusions.closed() @property def window_functions(self): """Target database must support window functions.""" return exclusions.closed() @property def ctes(self): """Target database supports CTEs""" return exclusions.closed() @property def ctes_with_update_delete(self): """target database supports CTES that ride on top of a normal UPDATE or DELETE statement which refers to the CTE in a correlated subquery. """ return exclusions.closed() @property def ctes_on_dml(self): """target database supports CTES which consist of INSERT, UPDATE or DELETE within the CTE, e.g. WITH x AS (UPDATE....)""" return exclusions.closed() @property def autoincrement_insert(self): """target platform generates new surrogate integer primary key values when insert() is executed, excluding the pk column.""" return exclusions.open() @property def fetch_rows_post_commit(self): """target platform will allow cursor.fetchone() to proceed after a COMMIT. Typically this refers to an INSERT statement with RETURNING which is invoked within "autocommit". If the row can be returned after the autocommit, then this rule can be open. """ return exclusions.open() @property def group_by_complex_expression(self): """target platform supports SQL expressions in GROUP BY e.g. SELECT x + y AS somelabel FROM table GROUP BY x + y """ return exclusions.open() @property def sane_rowcount(self): return exclusions.skip_if( lambda config: not config.db.dialect.supports_sane_rowcount, "driver doesn't support 'sane' rowcount", ) @property def sane_multi_rowcount(self): return exclusions.fails_if( lambda config: not config.db.dialect.supports_sane_multi_rowcount, "driver %(driver)s %(doesnt_support)s 'sane' multi row count", ) @property def sane_rowcount_w_returning(self): return exclusions.fails_if( lambda config: not ( config.db.dialect.supports_sane_rowcount_returning ), "driver doesn't support 'sane' rowcount when returning is on", ) @property def empty_inserts(self): """target platform supports INSERT with no values, i.e. INSERT DEFAULT VALUES or equivalent.""" return exclusions.only_if( lambda config: config.db.dialect.supports_empty_insert or config.db.dialect.supports_default_values, "empty inserts not supported", ) @property def insert_from_select(self): """target platform supports INSERT from a SELECT.""" return exclusions.open() @property def returning(self): """target platform supports RETURNING.""" return exclusions.only_if( lambda config: config.db.dialect.implicit_returning, "%(database)s %(does_support)s 'returning'", ) @property def tuple_in(self): """Target platform supports the syntax "(x, y) IN ((x1, y1), (x2, y2), ...)" """ return exclusions.closed() @property def tuple_in_w_empty(self): """Target platform tuple IN w/ empty set""" return self.tuple_in @property def duplicate_names_in_cursor_description(self): """target platform supports a SELECT statement that has the same name repeated more than once in the columns list.""" return exclusions.open() @property def denormalized_names(self): """Target database must have 'denormalized', i.e. UPPERCASE as case insensitive names.""" return exclusions.skip_if( lambda config: not config.db.dialect.requires_name_normalize, "Backend does not require denormalized names.", ) @property def multivalues_inserts(self): """target database must support multiple VALUES clauses in an INSERT statement.""" return exclusions.skip_if( lambda config: not config.db.dialect.supports_multivalues_insert, "Backend does not support multirow inserts.", ) @property def implements_get_lastrowid(self): """target dialect implements the executioncontext.get_lastrowid() method without reliance on RETURNING. """ return exclusions.open() @property def emulated_lastrowid(self): """target dialect retrieves cursor.lastrowid, or fetches from a database-side function after an insert() construct executes, within the get_lastrowid() method. Only dialects that "pre-execute", or need RETURNING to get last inserted id, would return closed/fail/skip for this. """ return exclusions.closed() @property def dbapi_lastrowid(self): """target platform includes a 'lastrowid' accessor on the DBAPI cursor object. """ return exclusions.closed() @property def views(self): """Target database must support VIEWs.""" return exclusions.closed() @property def schemas(self): """Target database must support external schemas, and have one named 'test_schema'.""" return exclusions.closed() @property def cross_schema_fk_reflection(self): """target system must support reflection of inter-schema foreign keys""" return exclusions.closed() @property def foreign_key_constraint_name_reflection(self): """Target supports refleciton of FOREIGN KEY constraints and will return the name of the constraint that was used in the "CONSTRANT <name> FOREIGN KEY" DDL. MySQL prior to version 8 and MariaDB prior to version 10.5 don't support this. """ return exclusions.closed() @property def implicit_default_schema(self): """target system has a strong concept of 'default' schema that can be referred to implicitly. basically, PostgreSQL. """ return exclusions.closed() @property def server_side_cursors(self): """Target dialect must support server side cursors.""" return exclusions.only_if( [lambda config: config.db.dialect.supports_server_side_cursors], "no server side cursors support", ) @property def sequences(self): """Target database must support SEQUENCEs.""" return exclusions.only_if( [lambda config: config.db.dialect.supports_sequences], "no sequence support", ) @property def sequences_optional(self): """Target database supports sequences, but also optionally as a means of generating new PK values.""" return exclusions.only_if( [ lambda config: config.db.dialect.supports_sequences and config.db.dialect.sequences_optional ], "no sequence support, or sequences not optional", ) @property def supports_lastrowid(self): """target database / driver supports cursor.lastrowid as a means of retrieving the last inserted primary key value. note that if the target DB supports sequences also, this is still assumed to work. This is a new use case brought on by MariaDB 10.3. """ return exclusions.only_if( [lambda config: config.db.dialect.postfetch_lastrowid] ) @property def no_lastrowid_support(self): """the opposite of supports_lastrowid""" return exclusions.NotPredicate(self.supports_lastrowid) @property def reflects_pk_names(self): return exclusions.closed() @property def table_reflection(self): return exclusions.open() @property def comment_reflection(self): return exclusions.closed() @property def view_column_reflection(self): """target database must support retrieval of the columns in a view, similarly to how a table is inspected. This does not include the full CREATE VIEW definition. """ return self.views @property def view_reflection(self): """target database must support inspection of the full CREATE VIEW definition.""" return self.views @property def schema_reflection(self): return self.schemas @property def primary_key_constraint_reflection(self): return exclusions.open() @property def foreign_key_constraint_reflection(self): return exclusions.open() @property def foreign_key_constraint_option_reflection_ondelete(self): return exclusions.closed() @property def foreign_key_constraint_option_reflection_onupdate(self): return exclusions.closed() @property def temp_table_reflection(self): return exclusions.open() @property def temp_table_names(self): """target dialect supports listing of temporary table names""" return exclusions.closed() @property def temporary_tables(self): """target database supports temporary tables""" return exclusions.open() @property def temporary_views(self): """target database supports temporary views""" return exclusions.closed() @property def index_reflection(self): return exclusions.open() @property def indexes_with_ascdesc(self): """target database supports CREATE INDEX with per-column ASC/DESC.""" return exclusions.open() @property def indexes_with_expressions(self): """target database supports CREATE INDEX against SQL expressions.""" return exclusions.closed() @property def unique_constraint_reflection(self): """target dialect supports reflection of unique constraints""" return exclusions.open() @property def check_constraint_reflection(self): """target dialect supports reflection of check constraints""" return exclusions.closed() @property def duplicate_key_raises_integrity_error(self): """target dialect raises IntegrityError when reporting an INSERT with a primary key violation. (hint: it should) """ return exclusions.open() @property def unbounded_varchar(self): """Target database must support VARCHAR with no length""" return exclusions.open() @property def unicode_data(self): """Target database/dialect must support Python unicode objects with non-ASCII characters represented, delivered as bound parameters as well as in result rows. """ return exclusions.open() @property def unicode_ddl(self): """Target driver must support some degree of non-ascii symbol names. """ return exclusions.closed() @property def symbol_names_w_double_quote(self): """Target driver can create tables with a name like 'some " table'""" return exclusions.open() @property def datetime_literals(self): """target dialect supports rendering of a date, time, or datetime as a literal string, e.g. via the TypeEngine.literal_processor() method. """ return exclusions.closed() @property def datetime(self): """target dialect supports representation of Python datetime.datetime() objects.""" return exclusions.open() @property def datetime_microseconds(self): """target dialect supports representation of Python datetime.datetime() with microsecond objects.""" return exclusions.open() @property def timestamp_microseconds(self): """target dialect supports representation of Python datetime.datetime() with microsecond objects but only if TIMESTAMP is used.""" return exclusions.closed() @property def datetime_historic(self): """target dialect supports representation of Python datetime.datetime() objects with historic (pre 1970) values.""" return exclusions.closed() @property def date(self): """target dialect supports representation of Python datetime.date() objects.""" return exclusions.open() @property def date_coerces_from_datetime(self): """target dialect accepts a datetime object as the target of a date column.""" return exclusions.open() @property def date_historic(self): """target dialect supports representation of Python datetime.datetime() objects with historic (pre 1970) values.""" return exclusions.closed() @property def time(self): """target dialect supports representation of Python datetime.time() objects.""" return exclusions.open() @property def time_microseconds(self): """target dialect supports representation of Python datetime.time() with microsecond objects.""" return exclusions.open() @property def binary_comparisons(self): """target database/driver can allow BLOB/BINARY fields to be compared against a bound parameter value. """ return exclusions.open() @property def binary_literals(self): """target backend supports simple binary literals, e.g. an expression like:: SELECT CAST('foo' AS BINARY) Where ``BINARY`` is the type emitted from :class:`.LargeBinary`, e.g. it could be ``BLOB`` or similar. Basically fails on Oracle. """ return exclusions.open() @property def autocommit(self): """target dialect supports 'AUTOCOMMIT' as an isolation_level""" return exclusions.closed() @property def isolation_level(self): """target dialect supports general isolation level settings. Note that this requirement, when enabled, also requires that the get_isolation_levels() method be implemented. """ return exclusions.closed() def get_isolation_levels(self, config): """Return a structure of supported isolation levels for the current testing dialect. The structure indicates to the testing suite what the expected "default" isolation should be, as well as the other values that are accepted. The dictionary has two keys, "default" and "supported". The "supported" key refers to a list of all supported levels and it should include AUTOCOMMIT if the dialect supports it. If the :meth:`.DefaultRequirements.isolation_level` requirement is not open, then this method has no return value. E.g.:: >>> testing.requirements.get_isolation_levels() { "default": "READ_COMMITED", "supported": [ "SERIALIZABLE", "READ UNCOMMITTED", "READ COMMITTED", "REPEATABLE READ", "AUTOCOMMIT" ] } """ @property def json_type(self): """target platform implements a native JSON type.""" return exclusions.closed() @property def json_array_indexes(self): """target platform supports numeric array indexes within a JSON structure""" return self.json_type @property def json_index_supplementary_unicode_element(self): return exclusions.open() @property def precision_numerics_general(self): """target backend has general support for moderately high-precision numerics.""" return exclusions.open() @property def precision_numerics_enotation_small(self): """target backend supports Decimal() objects using E notation to represent very small values.""" return exclusions.closed() @property def precision_numerics_enotation_large(self): """target backend supports Decimal() objects using E notation to represent very large values.""" return exclusions.closed() @property def precision_numerics_many_significant_digits(self): """target backend supports values with many digits on both sides, such as 319438950232418390.273596, 87673.594069654243 """ return exclusions.closed() @property def implicit_decimal_binds(self): """target backend will return a selected Decimal as a Decimal, not a string. e.g.:: expr = decimal.Decimal("15.7563") value = e.scalar( select([literal(expr)]) ) assert value == expr See :ticket:`4036` """ return exclusions.open() @property def nested_aggregates(self): """target database can select an aggregate from a subquery that's also using an aggregate """ return exclusions.open() @property def recursive_fk_cascade(self): """target database must support ON DELETE CASCADE on a self-referential foreign key """ return exclusions.open() @property def precision_numerics_retains_significant_digits(self): """A precision numeric type will return empty significant digits, i.e. a value such as 10.000 will come back in Decimal form with the .000 maintained.""" return exclusions.closed() @property def precision_generic_float_type(self): """target backend will return native floating point numbers with at least seven decimal places when using the generic Float type. """ return exclusions.open() @property def floats_to_four_decimals(self): """target backend can return a floating-point number with four significant digits (such as 15.7563) accurately (i.e. without FP inaccuracies, such as 15.75629997253418). """ return exclusions.open() @property def fetch_null_from_numeric(self): """target backend doesn't crash when you try to select a NUMERIC value that has a value of NULL. Added to support Pyodbc bug #351. """ return exclusions.open() @property def text_type(self): """Target database must support an unbounded Text() " "type such as TEXT or CLOB""" return exclusions.open() @property def empty_strings_varchar(self): """target database can persist/return an empty string with a varchar. """ return exclusions.open() @property def empty_strings_text(self): """target database can persist/return an empty string with an unbounded text.""" return exclusions.open() @property def expressions_against_unbounded_text(self): """target database supports use of an unbounded textual field in a WHERE clause.""" return exclusions.open() @property def selectone(self): """target driver must support the literal statement 'select 1'""" return exclusions.open() @property def savepoints(self): """Target database must support savepoints.""" return exclusions.closed() @property def two_phase_transactions(self): """Target database must support two-phase transactions.""" return exclusions.closed() @property def update_from(self): """Target must support UPDATE..FROM syntax""" return exclusions.closed() @property def delete_from(self): """Target must support DELETE FROM..FROM or DELETE..USING syntax""" return exclusions.closed() @property def update_where_target_in_subquery(self): """Target must support UPDATE (or DELETE) where the same table is present in a subquery in the WHERE clause. This is an ANSI-standard syntax that apparently MySQL can't handle, such as:: UPDATE documents SET flag=1 WHERE documents.title IN (SELECT max(documents.title) AS title FROM documents GROUP BY documents.user_id ) """ return exclusions.open() @property def mod_operator_as_percent_sign(self): """target database must use a plain percent '%' as the 'modulus' operator.""" return exclusions.closed() @property def percent_schema_names(self): """target backend supports weird identifiers with percent signs in them, e.g. 'some % column'. this is a very weird use case but often has problems because of DBAPIs that use python formatting. It's not a critical use case either. """ return exclusions.closed() @property def order_by_col_from_union(self): """target database supports ordering by a column from a SELECT inside of a UNION E.g. (SELECT id, ...) UNION (SELECT id, ...) ORDER BY id """ return exclusions.open() @property def order_by_label_with_expression(self): """target backend supports ORDER BY a column label within an expression. Basically this:: select data as foo from test order by foo \|\| 'bar' Lots of databases including PostgreSQL don't support this, so this is off by default. """ return exclusions.closed() @property def order_by_collation(self): def check(config): try: self.get_order_by_collation(config) return False except NotImplementedError: return True return exclusions.skip_if(check) def get_order_by_collation(self, config): raise NotImplementedError() @property def unicode_connections(self): """Target driver must support non-ASCII characters being passed at all. """ return exclusions.open() @property def graceful_disconnects(self): """Target driver must raise a DBAPI-level exception, such as InterfaceError, when the underlying connection has been closed and the execute() method is called. """ return exclusions.open() @property def independent_connections(self): """ Target must support simultaneous, independent database connections. """ return exclusions.open() @property def skip_mysql_on_windows(self): """Catchall for a large variety of MySQL on Windows failures""" return exclusions.open() @property def ad_hoc_engines(self): """Test environment must allow ad-hoc engine/connection creation. DBs that scale poorly for many connections, even when closed, i.e. Oracle, may use the "--low-connections" option which flags this requirement as not present. """ return exclusions.skip_if( lambda config: config.options.low_connections ) @property def timing_intensive(self): return exclusions.requires_tag("timing_intensive") @property def memory_intensive(self): return exclusions.requires_tag("memory_intensive") @property def threading_with_mock(self): """Mark tests that use threading and mock at the same time - stability issues have been observed with coverage + python 3.3 """ return exclusions.skip_if( lambda config: util.py3k and config.options.has_coverage, "Stability issues with coverage + py3k", ) @property def python2(self): return exclusions.skip_if( lambda: sys.version_info >= (3,), "Python version 2.xx is required.", ) @property def python3(self): return exclusions.skip_if( lambda: sys.version_info < (3,), "Python version 3.xx is required." ) @property def pep520(self): return self.python36 @property def python36(self): return exclusions.skip_if( lambda: sys.version_info < (3, 6), "Python version 3.6 or greater is required.", ) @property def python37(self): return exclusions.skip_if( lambda: sys.version_info < (3, 7), "Python version 3.7 or greater is required.", ) @property def cpython(self): return exclusions.only_if( lambda: util.cpython, "cPython interpreter needed" ) @property def non_broken_pickle(self): from sqlalchemy.util import pickle return exclusions.only_if( lambda: not util.pypy and pickle.__name__ == "cPickle" or sys.version_info >= (3, 2), "Needs cPickle+cPython or newer Python 3 pickle", ) @property def predictable_gc(self): """target platform must remove all cycles unconditionally when gc.collect() is called, as well as clean out unreferenced subclasses. """ return self.cpython @property def no_coverage(self): """Test should be skipped if coverage is enabled. This is to block tests that exercise libraries that seem to be sensitive to coverage, such as PostgreSQL notice logging. """ return exclusions.skip_if( lambda config: config.options.has_coverage, "Issues observed when coverage is enabled", ) def _has_mysql_on_windows(self, config): return False def _has_mysql_fully_case_sensitive(self, config): return False @property def sqlite(self): return exclusions.skip_if(lambda: not self._has_sqlite()) @property def cextensions(self): return exclusions.skip_if( lambda: not self._has_cextensions(), "C extensions not installed" ) def _has_sqlite(self): from sqlalchemy import create_engine try: create_engine("sqlite://") return True except ImportError: return False def _has_cextensions(self): try: from sqlalchemy import cresultproxy, cprocessors # noqa return True except ImportError: return False @property def computed_columns(self): "Supports computed columns" return exclusions.closed() @property def computed_columns_stored(self): "Supports computed columns with `persisted=True`" return exclusions.closed() @property def computed_columns_virtual(self): "Supports computed columns with `persisted=False`" return exclusions.closed() @property def computed_columns_default_persisted(self): """If the default persistence is virtual or stored when `persisted` is omitted""" return exclusions.closed() @property def computed_columns_reflect_persisted(self): """If persistence information is returned by the reflection of computed columns""" return exclusions.closed() @property def supports_is_distinct_from(self): """Supports some form of "x IS [NOT] DISTINCT FROM y" construct. Different dialects will implement their own flavour, e.g., sqlite will emit "x IS NOT y" instead of "x IS DISTINCT FROM y". .. seealso:: :meth:`.ColumnOperators.is_distinct_from` """ return exclusions.skip_if( lambda config: not config.db.dialect.supports_is_distinct_from, "driver doesn't support an IS DISTINCT FROM construct", )	32.29 KB	2026-01-20 12:58
📄 schema.py # testing/schema.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php import sys from . import config from . import exclusions from .. import event from .. import schema from ..util import OrderedDict __all__ = ["Table", "Column"] table_options = {} def Table(args, kw): """A schema.Table wrapper/hook for dialect-specific tweaks.""" test_opts = {k: kw.pop(k) for k in list(kw) if k.startswith("test_")} kw.update(table_options) if exclusions.against(config._current, "mysql"): if ( "mysql_engine" not in kw and "mysql_type" not in kw and "autoload_with" not in kw ): if "test_needs_fk" in test_opts or "test_needs_acid" in test_opts: kw["mysql_engine"] = "InnoDB" else: kw["mysql_engine"] = "MyISAM" # Apply some default cascading rules for self-referential foreign keys. # MySQL InnoDB has some issues around selecting self-refs too. if exclusions.against(config._current, "firebird"): table_name = args[0] unpack = config.db.dialect.identifier_preparer.unformat_identifiers # Only going after ForeignKeys in Columns. May need to # expand to ForeignKeyConstraint too. fks = [ fk for col in args if isinstance(col, schema.Column) for fk in col.foreign_keys ] for fk in fks: # root around in raw spec ref = fk._colspec if isinstance(ref, schema.Column): name = ref.table.name else: # take just the table name: on FB there cannot be # a schema, so the first element is always the # table name, possibly followed by the field name name = unpack(ref)[0] if name == table_name: if fk.ondelete is None: fk.ondelete = "CASCADE" if fk.onupdate is None: fk.onupdate = "CASCADE" return schema.Table(args, *kw) def Column(args, *kw): """A schema.Column wrapper/hook for dialect-specific tweaks.""" test_opts = {k: kw.pop(k) for k in list(kw) if k.startswith("test_")} if not config.requirements.foreign_key_ddl.enabled_for_config(config): args = [arg for arg in args if not isinstance(arg, schema.ForeignKey)] col = schema.Column(args, **kw) if test_opts.get("test_needs_autoincrement", False) and kw.get( "primary_key", False ): if col.default is None and col.server_default is None: col.autoincrement = True # allow any test suite to pick up on this col.info["test_needs_autoincrement"] = True # hardcoded rule for firebird, oracle; this should # be moved out if exclusions.against(config._current, "firebird", "oracle"): def add_seq(c, tbl): c._init_items( schema.Sequence( _truncate_name( config.db.dialect, tbl.name + "_" + c.name + "_seq" ), optional=True, ) ) event.listen(col, "after_parent_attach", add_seq, propagate=True) return col def _truncate_name(dialect, name): if len(name) > dialect.max_identifier_length: return ( name[0 : max(dialect.max_identifier_length - 6, 0)] + "_" + hex(hash(name) % 64)[2:] ) else: return name def pep435_enum(name): # Implements PEP 435 in the minimal fashion needed by SQLAlchemy __members__ = OrderedDict() def __init__(self, name, value, alias=None): self.name = name self.value = value self.__members__[name] = self value_to_member[value] = self setattr(self.__class__, name, self) if alias: self.__members__[alias] = self setattr(self.__class__, alias, self) value_to_member = {} @classmethod def get(cls, value): return value_to_member[value] someenum = type( name, (object,), {"__members__": __members__, "__init__": __init__, "get": get}, ) # getframe() trick for pickling I don't understand courtesy # Python namedtuple() try: module = sys._getframe(1).f_globals.get("__name__", "__main__") except (AttributeError, ValueError): pass if module is not None: someenum.__module__ = module return someenum	4.65 KB	2026-01-20 12:58
📁 suite	-	2026-03-21 15:45
📄 util.py # testing/util.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php import decimal import gc import random import sys import time import types from ..util import decorator from ..util import defaultdict from ..util import inspect_getfullargspec from ..util import jython from ..util import py2k from ..util import pypy if jython: def jython_gc_collect(args): """aggressive gc.collect for tests.""" gc.collect() time.sleep(0.1) gc.collect() gc.collect() return 0 # "lazy" gc, for VM's that don't GC on refcount == 0 gc_collect = lazy_gc = jython_gc_collect elif pypy: def pypy_gc_collect(args): gc.collect() gc.collect() gc_collect = lazy_gc = pypy_gc_collect else: # assume CPython - straight gc.collect, lazy_gc() is a pass gc_collect = gc.collect def lazy_gc(): pass def picklers(): picklers = set() if py2k: try: import cPickle picklers.add(cPickle) except ImportError: pass import pickle picklers.add(pickle) # yes, this thing needs this much testing for pickle_ in picklers: for protocol in range(-2, pickle.HIGHEST_PROTOCOL): yield pickle_.loads, lambda d: pickle_.dumps(d, protocol) def round_decimal(value, prec): if isinstance(value, float): return round(value, prec) # can also use shift() here but that is 2.6 only return (value * decimal.Decimal("1" + "0" * prec)).to_integral( decimal.ROUND_FLOOR ) / pow(10, prec) class RandomSet(set): def __iter__(self): l = list(set.__iter__(self)) random.shuffle(l) return iter(l) def pop(self): index = random.randint(0, len(self) - 1) item = list(set.__iter__(self))[index] self.remove(item) return item def union(self, other): return RandomSet(set.union(self, other)) def difference(self, other): return RandomSet(set.difference(self, other)) def intersection(self, other): return RandomSet(set.intersection(self, other)) def copy(self): return RandomSet(self) def conforms_partial_ordering(tuples, sorted_elements): """True if the given sorting conforms to the given partial ordering.""" deps = defaultdict(set) for parent, child in tuples: deps[parent].add(child) for i, node in enumerate(sorted_elements): for n in sorted_elements[i:]: if node in deps[n]: return False else: return True def all_partial_orderings(tuples, elements): edges = defaultdict(set) for parent, child in tuples: edges[child].add(parent) def _all_orderings(elements): if len(elements) == 1: yield list(elements) else: for elem in elements: subset = set(elements).difference([elem]) if not subset.intersection(edges[elem]): for sub_ordering in _all_orderings(subset): yield [elem] + sub_ordering return iter(_all_orderings(elements)) def function_named(fn, name): """Return a function with a given __name__. Will assign to __name__ and return the original function if possible on the Python implementation, otherwise a new function will be constructed. This function should be phased out as much as possible in favor of @decorator. Tests that "generate" many named tests should be modernized. """ try: fn.__name__ = name except TypeError: fn = types.FunctionType( fn.__code__, fn.__globals__, name, fn.__defaults__, fn.__closure__ ) return fn def run_as_contextmanager(ctx, fn, arg, kw): """Run the given function under the given contextmanager, simulating the behavior of 'with' to support older Python versions. This is not necessary anymore as we have placed 2.6 as minimum Python version, however some tests are still using this structure. """ obj = ctx.__enter__() try: result = fn(obj, arg, *kw) ctx.__exit__(None, None, None) return result except: exc_info = sys.exc_info() raise_ = ctx.__exit__(exc_info) if not raise_: raise else: return raise_ def rowset(results): """Converts the results of sql execution into a plain set of column tuples. Useful for asserting the results of an unordered query. """ return {tuple(row) for row in results} def fail(msg): assert False, msg @decorator def provide_metadata(fn, args, kw): """Provide bound MetaData for a single test, dropping afterwards.""" from . import config from . import engines from sqlalchemy import schema metadata = schema.MetaData(config.db) self = args[0] prev_meta = getattr(self, "metadata", None) self.metadata = metadata try: return fn(args, *kw) finally: engines.drop_all_tables(metadata, config.db) self.metadata = prev_meta def flag_combinations(combinations): """A facade around @testing.combinations() oriented towards boolean keyword-based arguments. Basically generates a nice looking identifier based on the keywords and also sets up the argument names. E.g.:: @testing.flag_combinations( dict(lazy=False, passive=False), dict(lazy=True, passive=False), dict(lazy=False, passive=True), dict(lazy=False, passive=True, raiseload=True), ) would result in:: @testing.combinations( ('', False, False, False), ('lazy', True, False, False), ('lazy_passive', True, True, False), ('lazy_passive', True, True, True), id_='iaaa', argnames='lazy,passive,raiseload' ) """ from . import config keys = set() for d in combinations: keys.update(d) keys = sorted(keys) return config.combinations( [ ("_".join(k for k in keys if d.get(k, False)),) + tuple(d.get(k, False) for k in keys) for d in combinations ], id_="i" + ("a" len(keys)), argnames=",".join(keys) ) def resolve_lambda(__fn, kw): """Given a no-arg lambda and a namespace, return a new lambda that has all the values filled in. This is used so that we can have module-level fixtures that refer to instance-level variables using lambdas. """ pos_args = inspect_getfullargspec(__fn)[0] pass_pos_args = {arg: kw.pop(arg) for arg in pos_args} glb = dict(__fn.__globals__) glb.update(kw) new_fn = types.FunctionType(__fn.__code__, glb) return new_fn(pass_pos_args) def metadata_fixture(ddl="function"): """Provide MetaData for a pytest fixture.""" from . import config def decorate(fn): def run_ddl(self): from sqlalchemy import schema metadata = self.metadata = schema.MetaData() try: result = fn(self, metadata) metadata.create_all(config.db) # TODO: # somehow get a per-function dml erase fixture here yield result finally: metadata.drop_all(config.db) return config.fixture(scope=ddl)(run_ddl) return decorate def force_drop_names(names): """Force the given table names to be dropped after test complete, isolating for foreign key cycles """ from . import config from sqlalchemy import inspect @decorator def go(fn, args, *kw): try: return fn(args, *kw) finally: drop_all_tables(config.db, inspect(config.db), include_names=names) return go class adict(dict): """Dict keys available as attributes. Shadows.""" def __getattribute__(self, key): try: return self[key] except KeyError: return dict.__getattribute__(self, key) def __call__(self, keys): return tuple([self[key] for key in keys]) get_all = __call__ def drop_all_tables(engine, inspector, schema=None, include_names=None): from sqlalchemy import ( Column, Table, Integer, MetaData, ForeignKeyConstraint, ) from sqlalchemy.schema import DropTable, DropConstraint if include_names is not None: include_names = set(include_names) with engine.connect() as conn: for tname, fkcs in reversed( inspector.get_sorted_table_and_fkc_names(schema=schema) ): if tname: if include_names is not None and tname not in include_names: continue conn.execute( DropTable(Table(tname, MetaData(), schema=schema)) ) elif fkcs: if not engine.dialect.supports_alter: continue for tname, fkc in fkcs: if ( include_names is not None and tname not in include_names ): continue tb = Table( tname, MetaData(), Column("x", Integer), Column("y", Integer), schema=schema, ) conn.execute( DropConstraint( ForeignKeyConstraint([tb.c.x], [tb.c.y], name=fkc) ) ) def teardown_events(event_cls): @decorator def decorate(fn, arg, kw): try: return fn(arg, **kw) finally: event_cls._clear() return decorate	9.91 KB	2026-01-20 12:58
📄 warnings.py # testing/warnings.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php from __future__ import absolute_import import warnings from . import assertions from .. import exc as sa_exc def setup_filters(): """Set global warning behavior for the test suite.""" warnings.filterwarnings( "ignore", category=sa_exc.SAPendingDeprecationWarning ) warnings.filterwarnings("error", category=sa_exc.SADeprecationWarning) warnings.filterwarnings("error", category=sa_exc.SAWarning) warnings.filterwarnings( "ignore", category=sa_exc.SAWarning, message=r"Oracle version .* is known to have a " "maximum identifier", ) # some selected deprecations... warnings.filterwarnings("error", category=DeprecationWarning) warnings.filterwarnings( "ignore", category=DeprecationWarning, message=".StopIteration" ) warnings.filterwarnings( "ignore", category=DeprecationWarning, message=".inspect.getargspec" ) try: import pytest except ImportError: pass else: warnings.filterwarnings( "once", category=pytest.PytestDeprecationWarning ) def assert_warnings(fn, warning_msgs, regex=False): """Assert that each of the given warnings are emitted by fn. Deprecated. Please use assertions.expect_warnings(). """ with assertions._expect_warnings( sa_exc.SAWarning, warning_msgs, regex=regex ): return fn()	1.61 KB	2026-01-20 12:58

Edit File:

Terminal