From 66d671367545133f52f2f0f34308648a5763243a Mon Sep 17 00:00:00 2001 From: Polina Date: Thu, 2 Jul 2026 20:39:07 +0000 Subject: [PATCH] =?UTF-8?q?=D0=97=D0=B0=D0=B3=D1=80=D1=83=D0=B7=D0=B8?= =?UTF-8?q?=D1=82=D1=8C=20=D1=84=D0=B0=D0=B9=D0=BB=D1=8B=20=D0=B2=20=C2=AB?= =?UTF-8?q?venv/Lib/site-packages/sqlalchemy/testing=C2=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../sqlalchemy/testing/__init__.py | 96 ++ .../sqlalchemy/testing/assertions.py | 994 ++++++++++++++++++ .../sqlalchemy/testing/assertsql.py | 520 +++++++++ .../sqlalchemy/testing/asyncio.py | 135 +++ .../sqlalchemy/testing/config.py | 434 ++++++++ 5 files changed, 2179 insertions(+) create mode 100644 venv/Lib/site-packages/sqlalchemy/testing/__init__.py create mode 100644 venv/Lib/site-packages/sqlalchemy/testing/assertions.py create mode 100644 venv/Lib/site-packages/sqlalchemy/testing/assertsql.py create mode 100644 venv/Lib/site-packages/sqlalchemy/testing/asyncio.py create mode 100644 venv/Lib/site-packages/sqlalchemy/testing/config.py diff --git a/venv/Lib/site-packages/sqlalchemy/testing/__init__.py b/venv/Lib/site-packages/sqlalchemy/testing/__init__.py new file mode 100644 index 0000000..72ef549 --- /dev/null +++ b/venv/Lib/site-packages/sqlalchemy/testing/__init__.py @@ -0,0 +1,96 @@ +# testing/__init__.py +# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors +# +# +# This module is part of SQLAlchemy and is released under +# the MIT License: https://www.opensource.org/licenses/mit-license.php +# mypy: ignore-errors + + +from unittest import mock + +from . import config +from .assertions import assert_raises +from .assertions import assert_raises_context_ok +from .assertions import assert_raises_message +from .assertions import assert_raises_message_context_ok +from .assertions import assert_warns +from .assertions import assert_warns_message +from .assertions import AssertsCompiledSQL +from .assertions import AssertsExecutionResults +from .assertions import ComparesIndexes +from .assertions import ComparesTables +from .assertions import emits_warning +from .assertions import emits_warning_on +from .assertions import eq_ +from .assertions import eq_ignore_whitespace +from .assertions import eq_regex +from .assertions import expect_deprecated +from .assertions import expect_deprecated_20 +from .assertions import expect_raises +from .assertions import expect_raises_message +from .assertions import expect_warnings +from .assertions import in_ +from .assertions import int_within_variance +from .assertions import is_ +from .assertions import is_false +from .assertions import is_instance_of +from .assertions import is_none +from .assertions import is_not +from .assertions import is_not_ +from .assertions import is_not_none +from .assertions import is_true +from .assertions import le_ +from .assertions import ne_ +from .assertions import not_in +from .assertions import not_in_ +from .assertions import startswith_ +from .assertions import uses_deprecated +from .config import add_to_marker +from .config import async_test +from .config import combinations +from .config import combinations_list +from .config import db +from .config import fixture +from .config import requirements as requires +from .config import skip_test +from .config import Variation +from .config import variation +from .config import variation_fixture +from .exclusions import _is_excluded +from .exclusions import _server_version +from .exclusions import against as _against +from .exclusions import db_spec +from .exclusions import exclude +from .exclusions import fails +from .exclusions import fails_if +from .exclusions import fails_on +from .exclusions import fails_on_everything_except +from .exclusions import future +from .exclusions import only_if +from .exclusions import only_on +from .exclusions import skip +from .exclusions import skip_if +from .schema import eq_clause_element +from .schema import eq_type_affinity +from .util import adict +from .util import fail +from .util import flag_combinations +from .util import force_drop_names +from .util import lambda_combinations +from .util import metadata_fixture +from .util import provide_metadata +from .util import resolve_lambda +from .util import rowset +from .util import run_as_contextmanager +from .util import skip_if_timeout +from .util import teardown_events +from .warnings import assert_warnings +from .warnings import warn_test_suite + + +def against(*queries): + return _against(config._current, *queries) + + +crashes = skip diff --git a/venv/Lib/site-packages/sqlalchemy/testing/assertions.py b/venv/Lib/site-packages/sqlalchemy/testing/assertions.py new file mode 100644 index 0000000..d72fd97 --- /dev/null +++ b/venv/Lib/site-packages/sqlalchemy/testing/assertions.py @@ -0,0 +1,994 @@ +# testing/assertions.py +# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors +# +# +# This module is part of SQLAlchemy and is released under +# the MIT License: https://www.opensource.org/licenses/mit-license.php +# mypy: ignore-errors + + +from __future__ import annotations + +from collections import defaultdict +import contextlib +from copy import copy +from itertools import filterfalse +import re +import sys +import warnings + +from . import assertsql +from . import config +from . import engines +from . import mock +from .exclusions import db_spec +from .util import fail +from .. import exc as sa_exc +from .. import schema +from .. import sql +from .. import types as sqltypes +from .. import util +from ..engine import default +from ..engine import url +from ..sql.selectable import LABEL_STYLE_TABLENAME_PLUS_COL +from ..util import decorator + + +def expect_warnings(*messages, **kw): + """Context manager which expects one or more warnings. + + With no arguments, squelches all SAWarning 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. + + """ # noqa + return _expect_warnings_sqla_only(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, str) 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_sqla_only( + sa_exc.SADeprecationWarning, messages, **kw + ) + + +def expect_deprecated_20(*messages, **kw): + return _expect_warnings_sqla_only( + sa_exc.Base20DeprecationWarning, 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 + + +_FILTERS = None +_SEEN = None +_EXC_CLS = None + + +def _expect_warnings_sqla_only( + exc_cls, + messages, + regex=True, + search_msg=False, + assert_=True, +): + """SQLAlchemy internal use only _expect_warnings(). + + Alembic is using _expect_warnings() directly, and should be updated + to use this new interface. + + """ + return _expect_warnings( + exc_cls, + messages, + regex=regex, + search_msg=search_msg, + assert_=assert_, + raise_on_any_unexpected=True, + ) + + +@contextlib.contextmanager +def _expect_warnings( + exc_cls, + messages, + regex=True, + search_msg=False, + assert_=True, + raise_on_any_unexpected=False, + squelch_other_warnings=False, +): + global _FILTERS, _SEEN, _EXC_CLS + + if regex or search_msg: + filters = [re.compile(msg, re.I | re.S) for msg in messages] + else: + filters = list(messages) + + if _FILTERS is not None: + # nested call; update _FILTERS and _SEEN, return. outer + # block will assert our messages + assert _SEEN is not None + assert _EXC_CLS is not None + _FILTERS.extend(filters) + _SEEN.update(filters) + _EXC_CLS += (exc_cls,) + yield + else: + seen = _SEEN = set(filters) + _FILTERS = filters + _EXC_CLS = (exc_cls,) + + if raise_on_any_unexpected: + + def real_warn(msg, *arg, **kw): + if isinstance(msg, sa_exc.SATestSuiteWarning): + warnings.warn(msg, *arg, **kw) + else: + raise AssertionError("Got unexpected warning: %r" % msg) + + else: + real_warn = warnings.warn + + def our_warn(msg, *arg, **kw): + if isinstance(msg, _EXC_CLS): + exception = type(msg) + msg = str(msg) + elif arg: + exception = arg[0] + else: + exception = None + + if not exception or not issubclass(exception, _EXC_CLS): + if not squelch_other_warnings: + return real_warn(msg, *arg, **kw) + else: + return + + if not filters and not raise_on_any_unexpected: + return + + for filter_ in filters: + if ( + (search_msg and filter_.search(msg)) + or (regex and filter_.match(msg)) + or (not regex and filter_ == msg) + ): + seen.discard(filter_) + break + else: + if not squelch_other_warnings: + real_warn(msg, *arg, **kw) + + with mock.patch("warnings.warn", our_warn): + try: + yield + finally: + _SEEN = _FILTERS = _EXC_CLS = None + + if assert_: + 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() + + +def _assert_no_stray_pool_connections(): + engines.testing_reaper.assert_all_closed() + + +def int_within_variance(expected, received, variance): + deviance = int(expected * variance) + assert ( + abs(received - expected) < deviance + ), "Given int value %s is not within %d%% of expected value %s" % ( + received, + variance * 100, + expected, + ) + + +def eq_regex(a, b, msg=None, flags=0): + assert re.match(b, a, flags), 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_none(a, msg=None): + is_(a, None, msg=msg) + + +def is_not_none(a, msg=None): + is_not(a, None, msg=msg) + + +def is_true(a, msg=None): + is_(bool(a), True, msg=msg) + + +def is_false(a, msg=None): + is_(bool(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) + a = re.sub(r"\t", "", a) + b = re.sub(r"^\s+?|\n", "", b) + b = re.sub(r" {2,}", " ", b) + b = re.sub(r"\t", "", 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 ( + 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): + return _assert_raises(except_cls, callable_, args, kw, check_context=True) + + +def assert_raises_context_ok(except_cls, callable_, *args, **kw): + return _assert_raises(except_cls, callable_, args, kw) + + +def assert_raises_message(except_cls, msg, callable_, *args, **kwargs): + return _assert_raises( + except_cls, callable_, args, kwargs, msg=msg, check_context=True + ) + + +def assert_warns(except_cls, callable_, *args, **kwargs): + """legacy adapter function for functions that were previously using + assert_raises with SAWarning or similar. + + has some workarounds to accommodate the fact that the callable completes + with this approach rather than stopping at the exception raise. + + + """ + with _expect_warnings_sqla_only(except_cls, [".*"]): + return callable_(*args, **kwargs) + + +def assert_warns_message(except_cls, msg, callable_, *args, **kwargs): + """legacy adapter function for functions that were previously using + assert_raises with SAWarning or similar. + + has some workarounds to accommodate the fact that the callable completes + with this approach rather than stopping at the exception raise. + + Also uses regex.search() to match the given message to the error string + rather than regex.match(). + + """ + with _expect_warnings_sqla_only( + except_cls, + [msg], + search_msg=True, + regex=False, + ): + return callable_(*args, **kwargs) + + +def assert_raises_message_context_ok( + except_cls, msg, callable_, *args, **kwargs +): + return _assert_raises(except_cls, callable_, args, kwargs, msg=msg) + + +def _assert_raises( + except_cls, callable_, args, kwargs, msg=None, check_context=False +): + with _expect_raises(except_cls, msg, check_context) as ec: + callable_(*args, **kwargs) + return ec.error + + +class _ErrorContainer: + error = None + + +@contextlib.contextmanager +def _expect_raises(except_cls, msg=None, check_context=False): + if ( + isinstance(except_cls, type) + and issubclass(except_cls, Warning) + or isinstance(except_cls, Warning) + ): + raise TypeError( + "Use expect_warnings for warnings, not " + "expect_raises / assert_raises" + ) + ec = _ErrorContainer() + if check_context: + are_we_already_in_a_traceback = sys.exc_info()[0] + try: + yield ec + success = False + except except_cls as err: + ec.error = err + success = True + if msg is not None: + # I'm often pdbing here, and "err" above isn't + # in scope, so assign the string explicitly + error_as_string = str(err) + assert re.search(msg, error_as_string, re.UNICODE), "%r !~ %s" % ( + msg, + error_as_string, + ) + if check_context and not are_we_already_in_a_traceback: + _assert_proper_exception_context(err) + print(str(err).encode("utf-8")) + + # it's generally a good idea to not carry traceback objects outside + # of the except: block, but in this case especially we seem to have + # hit some bug in either python 3.10.0b2 or greenlet or both which + # this seems to fix: + # https://github.com/python-greenlet/greenlet/issues/242 + del ec + + # assert outside the block so it works for AssertionError too ! + assert success, "Callable did not raise an exception" + + +def expect_raises(except_cls, check_context=True): + return _expect_raises(except_cls, check_context=check_context) + + +def expect_raises_message(except_cls, msg, check_context=True): + return _expect_raises(except_cls, msg=msg, check_context=check_context) + + +class AssertsCompiledSQL: + def assert_compile( + self, + clause, + result, + params=None, + checkparams=None, + for_executemany=False, + check_literal_execute=None, + check_post_param=None, + dialect=None, + checkpositional=None, + check_prefetch=None, + use_default_dialect=False, + allow_dialect_select=False, + supports_default_values=True, + supports_native_boolean=False, + supports_default_metavalue=True, + literal_binds=False, + render_postcompile=False, + schema_translate_map=None, + render_schema_translate=False, + default_schema_name=None, + from_linting=False, + check_param_order=True, + use_literal_execute_for_simple_int=False, + ): + if use_default_dialect: + dialect = default.DefaultDialect() + dialect.supports_default_values = supports_default_values + dialect.supports_default_metavalue = supports_default_metavalue + dialect.supports_native_boolean = supports_native_boolean + 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" or dialect == "default_qmark": + if dialect == "default": + dialect = default.DefaultDialect() + else: + dialect = default.DefaultDialect("qmark") + dialect.supports_default_values = supports_default_values + dialect.supports_default_metavalue = supports_default_metavalue + elif dialect == "default_enhanced": + dialect = default.StrCompileDialect() + elif isinstance(dialect, str): + dialect = url.URL.create(dialect).get_dialect()() + + if default_schema_name: + dialect.default_schema_name = default_schema_name + + 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 + + if render_postcompile: + compile_kwargs["render_postcompile"] = True + + if use_literal_execute_for_simple_int: + compile_kwargs["use_literal_execute_for_simple_int"] = True + + if for_executemany: + kw["for_executemany"] = True + + if render_schema_translate: + kw["render_schema_translate"] = True + + if from_linting or getattr(self, "assert_from_linting", False): + kw["linting"] = sql.FROM_LINTING + + from sqlalchemy import orm + + if isinstance(clause, orm.Query): + stmt = clause._statement_20() + stmt._label_style = LABEL_STYLE_TABLENAME_PLUS_COL + clause = stmt + + if compile_kwargs: + kw["compile_kwargs"] = compile_kwargs + + class DontAccess: + def __getattribute__(self, key): + raise NotImplementedError( + "compiler accessed .statement; use " + "compiler.current_executable" + ) + + class CheckCompilerAccess: + def __init__(self, test_statement): + self.test_statement = test_statement + self._annotations = {} + self.supports_execution = getattr( + test_statement, "supports_execution", False + ) + + if self.supports_execution: + self._execution_options = test_statement._execution_options + + if hasattr(test_statement, "_returning"): + self._returning = test_statement._returning + if hasattr(test_statement, "_inline"): + self._inline = test_statement._inline + if hasattr(test_statement, "_return_defaults"): + self._return_defaults = test_statement._return_defaults + + @property + def _variant_mapping(self): + return self.test_statement._variant_mapping + + 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) + + if isinstance(clause, sqltypes.TypeEngine): + cache_key_no_warnings = clause._static_cache_key + if cache_key_no_warnings: + hash(cache_key_no_warnings) + else: + cache_key_no_warnings = clause._generate_cache_key() + if cache_key_no_warnings: + hash(cache_key_no_warnings[0]) + + param_str = repr(getattr(c, "params", {})) + param_str = param_str.encode("utf-8").decode("ascii", "ignore") + print(("\nSQL String:\n" + str(c) + param_str).encode("utf-8")) + + cc = re.sub(r"[\n\t]", "", str(c)) + + eq_(cc, result, "%r != %r on dialect %r" % (cc, result, dialect)) + + if checkparams is not None: + if render_postcompile: + expanded_state = c.construct_expanded_state( + params, escape_names=False + ) + eq_(expanded_state.parameters, checkparams) + else: + eq_(c.construct_params(params), checkparams) + if checkpositional is not None: + if render_postcompile: + expanded_state = c.construct_expanded_state( + params, escape_names=False + ) + eq_( + tuple( + [ + expanded_state.parameters[x] + for x in expanded_state.positiontup + ] + ), + checkpositional, + ) + else: + p = c.construct_params(params, escape_names=False) + eq_(tuple([p[x] for x in c.positiontup]), checkpositional) + if check_prefetch is not None: + eq_(c.prefetch, check_prefetch) + if check_literal_execute is not None: + eq_( + { + c.bind_names[b]: b.effective_value + for b in c.literal_execute_params + }, + check_literal_execute, + ) + if check_post_param is not None: + eq_( + { + c.bind_names[b]: b.effective_value + for b in c.post_compile_params + }, + check_post_param, + ) + if check_param_order and getattr(c, "params", None): + + def get_dialect(paramstyle, positional): + cp = copy(dialect) + cp.paramstyle = paramstyle + cp.positional = positional + return cp + + pyformat_dialect = get_dialect("pyformat", False) + pyformat_c = clause.compile(dialect=pyformat_dialect, **kw) + stmt = re.sub(r"[\n\t]", "", str(pyformat_c)) + + qmark_dialect = get_dialect("qmark", True) + qmark_c = clause.compile(dialect=qmark_dialect, **kw) + values = list(qmark_c.positiontup) + escaped = qmark_c.escaped_bind_names + + for post_param in ( + qmark_c.post_compile_params | qmark_c.literal_execute_params + ): + name = qmark_c.bind_names[post_param] + if name in values: + values = [v for v in values if v != name] + positions = [] + pos_by_value = defaultdict(list) + for v in values: + try: + if v in pos_by_value: + start = pos_by_value[v][-1] + else: + start = 0 + esc = escaped.get(v, v) + pos = stmt.index("%%(%s)s" % (esc,), start) + 2 + positions.append(pos) + pos_by_value[v].append(pos) + except ValueError: + msg = "Expected to find bindparam %r in %r" % (v, stmt) + assert False, msg + + ordered = all( + positions[i - 1] < positions[i] + for i in range(1, len(positions)) + ) + + expected = [v for _, v in sorted(zip(positions, values))] + + msg = ( + "Order of parameters %s does not match the order " + "in the statement %s. Statement %r" % (values, expected, stmt) + ) + + is_true(ordered, msg) + + +class ComparesTables: + def assert_tables_equal( + self, + table, + reflected_table, + strict_types=False, + strict_constraints=True, + ): + 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] + + if strict_constraints: + 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) + + if strict_constraints: + 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 + ) + + if strict_constraints: + 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: + 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 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): + return self.assert_sql_execution( + db, callable_, 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)) + + @contextlib.contextmanager + def assert_statement_count_multi_db(self, dbs, 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: + yield + finally: + for asserter, (ctx, db, count) in zip(asserters, recs): + ctx.__exit__(None, None, None) + asserter.assert_(assertsql.CountStatements(count)) + + +class ComparesIndexes: + def compare_table_index_with_expected( + self, table: schema.Table, expected: list, dialect_name: str + ): + eq_(len(table.indexes), len(expected)) + idx_dict = {idx.name: idx for idx in table.indexes} + for exp in expected: + idx = idx_dict[exp["name"]] + eq_(idx.unique, exp["unique"]) + cols = [c for c in exp["column_names"] if c is not None] + eq_(len(idx.columns), len(cols)) + for c in cols: + is_true(c in idx.columns) + exprs = exp.get("expressions") + if exprs: + eq_(len(idx.expressions), len(exprs)) + for idx_exp, expr, col in zip( + idx.expressions, exprs, exp["column_names"] + ): + if col is None: + eq_(idx_exp.text, expr) + if ( + exp.get("dialect_options") + and f"{dialect_name}_include" in exp["dialect_options"] + ): + eq_( + idx.dialect_options[dialect_name]["include"], + exp["dialect_options"][f"{dialect_name}_include"], + ) diff --git a/venv/Lib/site-packages/sqlalchemy/testing/assertsql.py b/venv/Lib/site-packages/sqlalchemy/testing/assertsql.py new file mode 100644 index 0000000..0570e25 --- /dev/null +++ b/venv/Lib/site-packages/sqlalchemy/testing/assertsql.py @@ -0,0 +1,520 @@ +# testing/assertsql.py +# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors +# +# +# This module is part of SQLAlchemy and is released under +# the MIT License: https://www.opensource.org/licenses/mit-license.php +# mypy: ignore-errors + + +from __future__ import annotations + +import collections +import contextlib +import itertools +import re + +from .. import event +from ..engine import url +from ..engine.default import DefaultDialect +from ..schema import BaseDDLElement + + +class AssertRule: + 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): + def __init__(self, statement, params=None, consume_statement=True): + self.statement = statement + self.params = params + self.consume_statement = consume_statement + + 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.consume_statement = True + 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", enable_returning=True + ): + self.statement = statement + self.params = params + self.dialect = dialect + self.enable_returning = enable_returning + + 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": + dialect = DefaultDialect() + # this is currently what tests are expecting + # dialect.supports_default_values = True + dialect.supports_default_metavalue = True + + if self.enable_returning: + dialect.insert_returning = dialect.update_returning = ( + dialect.delete_returning + ) = True + dialect.use_insertmanyvalues = True + dialect.supports_multivalues_insert = True + dialect.update_returning_multifrom = True + dialect.delete_returning_multifrom = True + # dialect.favor_returning_over_lastrowid = True + # dialect.insert_null_pk_still_autoincrements = True + + # this is calculated but we need it to be True for this + # to look like all the current RETURNING dialects + assert dialect.insert_executemany_returning + + return dialect + else: + return url.URL.create(self.dialect).get_dialect()() + + 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) + + # received_statement runs a full compile(). we should not need to + # consider extracted_parameters; if we do this indicates some state + # is being sent from a previous cached query, which some misbehaviors + # in the ORM can cause, see #6881 + cache_key = None # execute_observed.context.compiled.cache_key + extracted_parameters = ( + None # execute_observed.context.extracted_parameters + ) + + if "schema_translate_map" in context.execution_options: + map_ = context.execution_options["schema_translate_map"] + else: + map_ = None + + if isinstance(execute_observed.clauseelement, BaseDDLElement): + compiled = execute_observed.clauseelement.compile( + dialect=compare_dialect, + schema_translate_map=map_, + ) + else: + compiled = execute_observed.clauseelement.compile( + cache_key=cache_key, + dialect=compare_dialect, + column_keys=context.compiled.column_keys, + for_executemany=context.compiled.for_executemany, + schema_translate_map=map_, + ) + _received_statement = re.sub(r"[\n\t]", "", str(compiled)) + parameters = execute_observed.parameters + + if not parameters: + _received_parameters = [ + compiled.construct_params( + extracted_parameters=extracted_parameters + ) + ] + else: + _received_parameters = [ + compiled.construct_params( + m, extracted_parameters=extracted_parameters + ) + 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( + execute_observed, params + ) % { + "received_statement": _received_statement, + "received_parameters": _received_parameters, + } + + def _all_params(self, context): + if self.params: + if 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, execute_observed, expected_params): + return ( + "Testing for compiled statement\n%r partial params %s, " + "received\n%%(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", enable_returning=False + ): + SQLMatchRule.__init__(self) + self.regex = re.compile(regex) + self.orig_regex = regex + self.params = params + self.dialect = dialect + self.enable_returning = enable_returning + + def _failure_message(self, execute_observed, 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()._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.context.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 _dialect_adjusted_statement(self, dialect): + paramstyle = dialect.paramstyle + stmt = re.sub(r"[\n\t]", "", self.statement) + + # temporarily escape out PG double colons + stmt = stmt.replace("::", "!!") + + 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.startswith("numeric"): + counter = itertools.count(1) + + num_identifier = "$" if paramstyle == "numeric_dollar" else ":" + + def repl(m): + return f"{num_identifier}{next(counter)}" + + stmt = re.sub(r":([\w_]+)", repl, stmt) + + # put them back + stmt = stmt.replace("!!", "::") + + return stmt + + def _compare_sql(self, execute_observed, received_statement): + stmt = self._dialect_adjusted_statement( + execute_observed.context.dialect + ) + return received_statement == stmt + + def _failure_message(self, execute_observed, expected_params): + return ( + "Testing for compiled statement\n%r partial params %s, " + "received\n%%(received_statement)r with params " + "%%(received_parameters)r" + % ( + self._dialect_adjusted_statement( + execute_observed.context.dialect + ).replace("%", "%%"), + repr(expected_params).replace("%", "%%"), + ) + ) + + +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): + if not self.rules: + self.is_consumed = True + self.consume_statement = False + + 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().no_more_statements() + + +class Conditional(EachOf): + def __init__(self, condition, rules, else_rules): + if condition: + super().__init__(*rules) + else: + super().__init__(*else_rules) + + +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: + def __init__(self, context, clauseelement, multiparams, params): + self.context = context + self.clauseelement = clauseelement + + if multiparams: + self.parameters = multiparams + elif params: + self.parameters = [params] + else: + self.parameters = [] + self.statements = [] + + def __repr__(self): + return str(self.statements) + + +class SQLCursorExecuteObserved( + collections.namedtuple( + "SQLCursorExecuteObserved", + ["statement", "parameters", "context", "executemany"], + ) +): + pass + + +class SQLAsserter: + 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:\n%s" % observed + 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, execution_options + ): + conn._WORKAROUND_ISSUE_13018 = True + # 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() diff --git a/venv/Lib/site-packages/sqlalchemy/testing/asyncio.py b/venv/Lib/site-packages/sqlalchemy/testing/asyncio.py new file mode 100644 index 0000000..7effed4 --- /dev/null +++ b/venv/Lib/site-packages/sqlalchemy/testing/asyncio.py @@ -0,0 +1,135 @@ +# testing/asyncio.py +# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors +# +# +# This module is part of SQLAlchemy and is released under +# the MIT License: https://www.opensource.org/licenses/mit-license.php +# mypy: ignore-errors + + +# functions and wrappers to run tests, fixtures, provisioning and +# setup/teardown in an asyncio event loop, conditionally based on the +# current DB driver being used for a test. + +# note that SQLAlchemy's asyncio integration also supports a method +# of running individual asyncio functions inside of separate event loops +# using "async_fallback" mode; however running whole functions in the event +# loop is a more accurate test for how SQLAlchemy's asyncio features +# would run in the real world. + + +from __future__ import annotations + +from functools import wraps +import inspect + +from . import config +from ..util.concurrency import _AsyncUtil + +# may be set to False if the +# --disable-asyncio flag is passed to the test runner. +ENABLE_ASYNCIO = True +_async_util = _AsyncUtil() # it has lazy init so just always create one + + +def _shutdown(): + """called when the test finishes""" + _async_util.close() + + +def _run_coroutine_function(fn, *args, **kwargs): + return _async_util.run(fn, *args, **kwargs) + + +def _assume_async(fn, *args, **kwargs): + """Run a function in an asyncio loop unconditionally. + + This function is used for provisioning features like + testing a database connection for server info. + + Note that for blocking IO database drivers, this means they block the + event loop. + + """ + + if not ENABLE_ASYNCIO: + return fn(*args, **kwargs) + + return _async_util.run_in_greenlet(fn, *args, **kwargs) + + +def _maybe_async_provisioning(fn, *args, **kwargs): + """Run a function in an asyncio loop if any current drivers might need it. + + This function is used for provisioning features that take + place outside of a specific database driver being selected, so if the + current driver that happens to be used for the provisioning operation + is an async driver, it will run in asyncio and not fail. + + Note that for blocking IO database drivers, this means they block the + event loop. + + """ + if not ENABLE_ASYNCIO: + return fn(*args, **kwargs) + + if config.any_async: + return _async_util.run_in_greenlet(fn, *args, **kwargs) + else: + return fn(*args, **kwargs) + + +def _maybe_async(fn, *args, **kwargs): + """Run a function in an asyncio loop if the current selected driver is + async. + + This function is used for test setup/teardown and tests themselves + where the current DB driver is known. + + + """ + if not ENABLE_ASYNCIO: + return fn(*args, **kwargs) + + is_async = config._current.is_async + + if is_async: + return _async_util.run_in_greenlet(fn, *args, **kwargs) + else: + return fn(*args, **kwargs) + + +def _maybe_async_wrapper(fn): + """Apply the _maybe_async function to an existing function and return + as a wrapped callable, supporting generator functions as well. + + This is currently used for pytest fixtures that support generator use. + + """ + + if inspect.isgeneratorfunction(fn): + _stop = object() + + def call_next(gen): + try: + return next(gen) + # can't raise StopIteration in an awaitable. + except StopIteration: + return _stop + + @wraps(fn) + def wrap_fixture(*args, **kwargs): + gen = fn(*args, **kwargs) + while True: + value = _maybe_async(call_next, gen) + if value is _stop: + break + yield value + + else: + + @wraps(fn) + def wrap_fixture(*args, **kwargs): + return _maybe_async(fn, *args, **kwargs) + + return wrap_fixture diff --git a/venv/Lib/site-packages/sqlalchemy/testing/config.py b/venv/Lib/site-packages/sqlalchemy/testing/config.py new file mode 100644 index 0000000..a915f83 --- /dev/null +++ b/venv/Lib/site-packages/sqlalchemy/testing/config.py @@ -0,0 +1,434 @@ +# testing/config.py +# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors +# +# +# This module is part of SQLAlchemy and is released under +# the MIT License: https://www.opensource.org/licenses/mit-license.php +# mypy: ignore-errors + + +from __future__ import annotations + +from argparse import Namespace +import collections +import inspect +import typing +from typing import Any +from typing import Callable +from typing import Iterable +from typing import NoReturn +from typing import Optional +from typing import Tuple +from typing import TypeVar +from typing import Union + +from . import mock +from . import requirements as _requirements +from .util import fail +from .. import util + +# default requirements; this is replaced by plugin_base when pytest +# is run +requirements = _requirements.SuiteRequirements() + +db = None +db_url = None +db_opts = None +file_config = None +test_schema = None +test_schema_2 = None +any_async = False +_current = None +ident = "main" +options: Namespace = None # type: ignore + +if typing.TYPE_CHECKING: + from .plugin.plugin_base import FixtureFunctions + + _fixture_functions: FixtureFunctions +else: + + class _NullFixtureFunctions: + def _null_decorator(self): + def go(fn): + return fn + + return go + + def skip_test_exception(self, *arg, **kw): + return Exception() + + @property + def add_to_marker(self): + return mock.Mock() + + def mark_base_test_class(self): + return self._null_decorator() + + def combinations(self, *arg_sets, **kw): + return self._null_decorator() + + def param_ident(self, *parameters): + return self._null_decorator() + + def fixture(self, *arg, **kw): + return self._null_decorator() + + def get_current_test_name(self): + return None + + def async_test(self, fn): + return fn + + # default fixture functions; these are replaced by plugin_base when + # pytest runs + _fixture_functions = _NullFixtureFunctions() + + +_FN = TypeVar("_FN", bound=Callable[..., Any]) + + +def combinations( + *comb: Union[Any, Tuple[Any, ...]], + argnames: Optional[str] = None, + id_: Optional[str] = None, + **kw: str, +) -> Callable[[_FN], _FN]: + 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: + + .. sourcecode:: text + + 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, id_=id_, argnames=argnames, **kw + ) + + +def combinations_list(arg_iterable: Iterable[Tuple[Any, ...]], **kw): + "As combination, but takes a single iterable" + return combinations(*arg_iterable, **kw) + + +class Variation: + __slots__ = ("_name", "_argname") + + def __init__(self, case, argname, case_names): + self._name = case + self._argname = argname + for casename in case_names: + setattr(self, casename, casename == case) + + if typing.TYPE_CHECKING: + + def __getattr__(self, key: str) -> bool: ... + + @property + def name(self): + return self._name + + def __bool__(self): + return self._name == self._argname + + def __nonzero__(self): + return not self.__bool__() + + def __str__(self): + return f"{self._argname}={self._name!r}" + + def __repr__(self): + return str(self) + + def fail(self) -> NoReturn: + fail(f"Unknown {self}") + + @classmethod + def idfn(cls, variation): + return variation.name + + @classmethod + def generate_cases(cls, argname, cases): + case_names = [ + argname if c is True else "not_" + argname if c is False else c + for c in cases + ] + + typ = type( + argname, + (Variation,), + { + "__slots__": tuple(case_names), + }, + ) + + return [typ(casename, argname, case_names) for casename in case_names] + + +def variation(argname_or_fn, cases=None): + """a helper around testing.combinations that provides a single namespace + that can be used as a switch. + + e.g.:: + + @testing.variation("querytyp", ["select", "subquery", "legacy_query"]) + @testing.variation("lazy", ["select", "raise", "raise_on_sql"]) + def test_thing(self, querytyp, lazy, decl_base): + class Thing(decl_base): + __tablename__ = "thing" + + # use name directly + rel = relationship("Rel", lazy=lazy.name) + + # use as a switch + if querytyp.select: + stmt = select(Thing) + elif querytyp.subquery: + stmt = select(Thing).subquery() + elif querytyp.legacy_query: + stmt = Session.query(Thing) + else: + querytyp.fail() + + The variable provided is a slots object of boolean variables, as well + as the name of the case itself under the attribute ".name" + + """ + + if inspect.isfunction(argname_or_fn): + argname = argname_or_fn.__name__ + cases = argname_or_fn(None) + + @variation_fixture(argname, cases) + def go(self, request): + yield request.param + + return go + else: + argname = argname_or_fn + cases_plus_limitations = [ + ( + entry + if (isinstance(entry, tuple) and len(entry) == 2) + else (entry, None) + ) + for entry in cases + ] + + variations = Variation.generate_cases( + argname, [c for c, l in cases_plus_limitations] + ) + return combinations( + *[ + ( + (variation._name, variation, limitation) + if limitation is not None + else (variation._name, variation) + ) + for variation, (case, limitation) in zip( + variations, cases_plus_limitations + ) + ], + id_="ia", + argnames=argname, + ) + + +def variation_fixture(argname, cases, scope="function"): + return fixture( + params=Variation.generate_cases(argname, cases), + ids=Variation.idfn, + scope=scope, + ) + + +def fixture(*arg: Any, **kw: Any) -> Any: + return _fixture_functions.fixture(*arg, **kw) + + +def get_current_test_name() -> str: + return _fixture_functions.get_current_test_name() + + +def mark_base_test_class() -> Any: + return _fixture_functions.mark_base_test_class() + + +class _AddToMarker: + def __getattr__(self, attr: str) -> Any: + return getattr(_fixture_functions.add_to_marker, attr) + + +add_to_marker = _AddToMarker() + + +class Config: + 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" + + self.is_async = db.dialect.is_async and not util.asbool( + db.url.query.get("async_fallback", False) + ) + + from . import provision + + self.is_default_dialect = provision.is_preferred_driver(self, db) + + _stack = collections.deque() + _configs = set() + + def __repr__(self): + return ( + f"sqlalchemy.testing.config.Config" + f"({self.db.name}+{self.db.driver}, " + f"{self.db.dialect.server_version_info})" + ) + + def _set_name(self, db): + suffix = "_async" if db.dialect.is_async else "" + if db.dialect.server_version_info: + svi = ".".join(str(tok) for tok in db.dialect.server_version_info) + self.name = "%s+%s%s_[%s]" % (db.name, db.driver, suffix, svi) + else: + self.name = "%s+%s%s" % (db.name, db.driver, suffix) + + @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". + """ + global any_async + + cfg = Config(db, db_opts, options, file_config) + + # if any backends include an async driver, then ensure + # all setup/teardown and tests are wrapped in the maybe_async() + # decorator that will set up a greenlet context for async drivers. + any_async = any_async or cfg.is_async + + 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 pop(cls, namespace): + if cls._stack: + # a failed test w/ -x option can call reset() ahead of time + _current = cls._stack[-1] + del cls._stack[-1] + cls.set_as_current(_current, 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) + + +def async_test(fn): + return _fixture_functions.async_test(fn)