Загрузить файлы в «venv/Lib/site-packages/sqlalchemy/testing»
This commit is contained in:
96
venv/Lib/site-packages/sqlalchemy/testing/__init__.py
Normal file
96
venv/Lib/site-packages/sqlalchemy/testing/__init__.py
Normal file
@@ -0,0 +1,96 @@
|
||||
# testing/__init__.py
|
||||
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
|
||||
# <see AUTHORS file>
|
||||
#
|
||||
# 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
|
||||
994
venv/Lib/site-packages/sqlalchemy/testing/assertions.py
Normal file
994
venv/Lib/site-packages/sqlalchemy/testing/assertions.py
Normal file
@@ -0,0 +1,994 @@
|
||||
# testing/assertions.py
|
||||
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
|
||||
# <see AUTHORS file>
|
||||
#
|
||||
# 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"],
|
||||
)
|
||||
520
venv/Lib/site-packages/sqlalchemy/testing/assertsql.py
Normal file
520
venv/Lib/site-packages/sqlalchemy/testing/assertsql.py
Normal file
@@ -0,0 +1,520 @@
|
||||
# testing/assertsql.py
|
||||
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
|
||||
# <see AUTHORS file>
|
||||
#
|
||||
# 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()
|
||||
135
venv/Lib/site-packages/sqlalchemy/testing/asyncio.py
Normal file
135
venv/Lib/site-packages/sqlalchemy/testing/asyncio.py
Normal file
@@ -0,0 +1,135 @@
|
||||
# testing/asyncio.py
|
||||
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
|
||||
# <see AUTHORS file>
|
||||
#
|
||||
# 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
|
||||
434
venv/Lib/site-packages/sqlalchemy/testing/config.py
Normal file
434
venv/Lib/site-packages/sqlalchemy/testing/config.py
Normal file
@@ -0,0 +1,434 @@
|
||||
# testing/config.py
|
||||
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
|
||||
# <see AUTHORS file>
|
||||
#
|
||||
# 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)
|
||||
Reference in New Issue
Block a user