diff --git a/venv/Lib/site-packages/sqlalchemy/testing/provision.py b/venv/Lib/site-packages/sqlalchemy/testing/provision.py new file mode 100644 index 0000000..b2a3354 --- /dev/null +++ b/venv/Lib/site-packages/sqlalchemy/testing/provision.py @@ -0,0 +1,602 @@ +# testing/provision.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 logging + +from . import config +from . import engines +from . import util +from .. import exc +from .. import inspect +from ..engine import Connection +from ..engine import Engine +from ..engine import url as sa_url +from ..schema import sort_tables_and_constraints +from ..sql import ddl +from ..sql import schema +from ..util import decorator + +log = logging.getLogger(__name__) + +FOLLOWER_IDENT = None + + +class register: + def __init__(self, decorator=None): + self.fns = {} + self.decorator = decorator + + @classmethod + def init(cls, fn): + return register().for_db("*")(fn) + + @classmethod + def init_decorator(cls, decorator): + return register(decorator).for_db("*") + + def for_db(self, *dbnames): + def decorate(fn): + if self.decorator: + fn = self.decorator(fn) + for dbname in dbnames: + self.fns[dbname] = fn + return self + + return decorate + + def call_original(self, cfg, *arg, **kw): + return self.fns["*"](cfg, *arg, **kw) + + def __call__(self, cfg, *arg, **kw): + if isinstance(cfg, str): + url = sa_url.make_url(cfg) + elif isinstance(cfg, sa_url.URL): + url = cfg + elif isinstance(cfg, (Engine, Connection)): + url = cfg.engine.url + else: + url = cfg.db.url + backend = url.get_backend_name() + if backend in self.fns: + return self.fns[backend](cfg, *arg, **kw) + else: + return self.fns["*"](cfg, *arg, **kw) + + +def create_follower_db(follower_ident): + for cfg in _configs_for_db_operation(): + log.info("CREATE database %s, URI %r", follower_ident, cfg.db.url) + create_db(cfg, cfg.db, follower_ident) + + +def setup_config(db_url, options, file_config, follower_ident): + # load the dialect, which should also have it set up its provision + # hooks + + dialect = sa_url.make_url(db_url).get_dialect() + + dialect.load_provisioning() + + if follower_ident: + db_url = follower_url_from_main(db_url, follower_ident) + db_opts = {} + update_db_opts(db_url, db_opts, options) + db_opts["scope"] = "global" + eng = engines.testing_engine(db_url, db_opts) + + post_configure_engine(db_url, eng, follower_ident) + + eng.connect().close() + + cfg = config.Config.register(eng, db_opts, options, file_config) + + # a symbolic name that tests can use if they need to disambiguate + # names across databases + if follower_ident: + config.ident = follower_ident + + if follower_ident: + configure_follower(cfg, follower_ident) + return cfg + + +def drop_follower_db(follower_ident): + for cfg in _configs_for_db_operation(): + log.info("DROP database %s, URI %r", follower_ident, cfg.db.url) + drop_db(cfg, cfg.db, follower_ident) + + +def generate_db_urls(db_urls, extra_drivers): + """Generate a set of URLs to test given configured URLs plus additional + driver names. + + Given: + + .. sourcecode:: text + + --dburi postgresql://db1 \ + --dburi postgresql://db2 \ + --dburi postgresql://db2 \ + --dbdriver=psycopg2 --dbdriver=asyncpg?async_fallback=true + + Noting that the default postgresql driver is psycopg2, the output + would be: + + .. sourcecode:: text + + postgresql+psycopg2://db1 + postgresql+asyncpg://db1 + postgresql+psycopg2://db2 + postgresql+psycopg2://db3 + + That is, for the driver in a --dburi, we want to keep that and use that + driver for each URL it's part of . For a driver that is only + in --dbdrivers, we want to use it just once for one of the URLs. + for a driver that is both coming from --dburi as well as --dbdrivers, + we want to keep it in that dburi. + + Driver specific query options can be specified by added them to the + driver name. For example, to enable the async fallback option for + asyncpg:: + + .. sourcecode:: text + + --dburi postgresql://db1 \ + --dbdriver=asyncpg?async_fallback=true + + """ + urls = set() + + backend_to_driver_we_already_have = collections.defaultdict(set) + + urls_plus_dialects = [ + (url_obj, url_obj.get_dialect()) + for url_obj in [sa_url.make_url(db_url) for db_url in db_urls] + ] + + for url_obj, dialect in urls_plus_dialects: + # use get_driver_name instead of dialect.driver to account for + # "_async" virtual drivers like oracledb and psycopg + driver_name = url_obj.get_driver_name() + backend_to_driver_we_already_have[dialect.name].add(driver_name) + + backend_to_driver_we_need = {} + + for url_obj, dialect in urls_plus_dialects: + backend = dialect.name + dialect.load_provisioning() + + if backend not in backend_to_driver_we_need: + backend_to_driver_we_need[backend] = extra_per_backend = set( + extra_drivers + ).difference(backend_to_driver_we_already_have[backend]) + else: + extra_per_backend = backend_to_driver_we_need[backend] + + for driver_url in _generate_driver_urls(url_obj, extra_per_backend): + if driver_url in urls: + continue + urls.add(driver_url) + yield driver_url + + +def _generate_driver_urls(url, extra_drivers): + main_driver = url.get_driver_name() + extra_drivers.discard(main_driver) + + url = generate_driver_url(url, main_driver, "") + yield url + + for drv in list(extra_drivers): + if "?" in drv: + driver_only, query_str = drv.split("?", 1) + + else: + driver_only = drv + query_str = None + + new_url = generate_driver_url(url, driver_only, query_str) + if new_url: + extra_drivers.remove(drv) + + yield new_url + + +@register.init +def is_preferred_driver(cfg, engine): + """Return True if the engine's URL is on the "default" driver, or + more generally the "preferred" driver to use for tests. + + Backends can override this to make a different driver the "prefeferred" + driver that's not the default. + + """ + return ( + engine.url._get_entrypoint() + is engine.url.set( + drivername=engine.url.get_backend_name() + )._get_entrypoint() + ) + + +@register.init +def generate_driver_url(url, driver, query_str): + backend = url.get_backend_name() + + new_url = url.set( + drivername="%s+%s" % (backend, driver), + ) + if query_str: + new_url = new_url.update_query_string(query_str) + + try: + new_url.get_dialect() + except exc.NoSuchModuleError: + return None + else: + return new_url + + +def _configs_for_db_operation(): + hosts = set() + + for cfg in config.Config.all_configs(): + cfg.db.dispose() + + for cfg in config.Config.all_configs(): + url = cfg.db.url + backend = url.get_backend_name() + host_conf = (backend, url.username, url.host, url.database) + + if host_conf not in hosts: + yield cfg + hosts.add(host_conf) + + for cfg in config.Config.all_configs(): + cfg.db.dispose() + + +@register.init +def drop_all_schema_objects_pre_tables(cfg, eng): + pass + + +@register.init +def drop_all_schema_objects_post_tables(cfg, eng): + pass + + +def drop_all_schema_objects(cfg, eng): + drop_all_schema_objects_pre_tables(cfg, eng) + + drop_views(cfg, eng) + + if config.requirements.materialized_views.enabled: + drop_materialized_views(cfg, eng) + + inspector = inspect(eng) + + consider_schemas = (None,) + if config.requirements.schemas.enabled_for_config(cfg): + consider_schemas += (cfg.test_schema, cfg.test_schema_2) + util.drop_all_tables(eng, inspector, consider_schemas=consider_schemas) + + drop_all_schema_objects_post_tables(cfg, eng) + + if config.requirements.sequences.enabled_for_config(cfg): + with eng.begin() as conn: + for seq in inspector.get_sequence_names(): + conn.execute(ddl.DropSequence(schema.Sequence(seq))) + if config.requirements.schemas.enabled_for_config(cfg): + for schema_name in [cfg.test_schema, cfg.test_schema_2]: + for seq in inspector.get_sequence_names( + schema=schema_name + ): + conn.execute( + ddl.DropSequence( + schema.Sequence(seq, schema=schema_name) + ) + ) + + +def drop_views(cfg, eng): + inspector = inspect(eng) + + try: + view_names = inspector.get_view_names() + except NotImplementedError: + pass + else: + with eng.begin() as conn: + for vname in view_names: + conn.execute( + ddl._DropView(schema.Table(vname, schema.MetaData())) + ) + + if config.requirements.schemas.enabled_for_config(cfg): + try: + view_names = inspector.get_view_names(schema=cfg.test_schema) + except NotImplementedError: + pass + else: + with eng.begin() as conn: + for vname in view_names: + conn.execute( + ddl._DropView( + schema.Table( + vname, + schema.MetaData(), + schema=cfg.test_schema, + ) + ) + ) + + +def drop_materialized_views(cfg, eng): + inspector = inspect(eng) + + mview_names = inspector.get_materialized_view_names() + + with eng.begin() as conn: + for vname in mview_names: + conn.exec_driver_sql(f"DROP MATERIALIZED VIEW {vname}") + + if config.requirements.schemas.enabled_for_config(cfg): + mview_names = inspector.get_materialized_view_names( + schema=cfg.test_schema + ) + with eng.begin() as conn: + for vname in mview_names: + conn.exec_driver_sql( + f"DROP MATERIALIZED VIEW {cfg.test_schema}.{vname}" + ) + + +@register.init +def create_db(cfg, eng, ident): + """Dynamically create a database for testing. + + Used when a test run will employ multiple processes, e.g., when run + via `tox` or `pytest -n4`. + """ + raise NotImplementedError( + "no DB creation routine for cfg: %s" % (eng.url,) + ) + + +@register.init +def drop_db(cfg, eng, ident): + """Drop a database that we dynamically created for testing.""" + raise NotImplementedError("no DB drop routine for cfg: %s" % (eng.url,)) + + +def _adapt_update_db_opts(fn): + insp = util.inspect_getfullargspec(fn) + if len(insp.args) == 3: + return fn + else: + return lambda db_url, db_opts, _options: fn(db_url, db_opts) + + +@register.init_decorator(_adapt_update_db_opts) +def update_db_opts(db_url, db_opts, options): + """Set database options (db_opts) for a test database that we created.""" + + +@register.init +def post_configure_engine(url, engine, follower_ident): + """Perform extra steps after configuring the main engine for testing. + + (For the internal dialects, currently only used by sqlite, oracle, mssql) + """ + + +@register.init +def post_configure_testing_engine(url, engine, options, scope): + """perform extra steps after configuring any engine within the + testing_engine() function. + + this includes the main engine as well as most ad-hoc testing engines. + + steps here should not get in the way of test cases that are looking + for events, etc. + + """ + + +@register.init +def follower_url_from_main(url, ident): + """Create a connection URL for a dynamically-created test database. + + :param url: the connection URL specified when the test run was invoked + :param ident: the pytest-xdist "worker identifier" to be used as the + database name + """ + url = sa_url.make_url(url) + return url.set(database=ident) + + +@register.init +def configure_follower(cfg, ident): + """Create dialect-specific config settings for a follower database.""" + pass + + +@register.init +def run_reap_dbs(url, ident): + """Remove databases that were created during the test process, after the + process has ended. + + This is an optional step that is invoked for certain backends that do not + reliably release locks on the database as long as a process is still in + use. For the internal dialects, this is currently only necessary for + mssql and oracle. + """ + + +def reap_dbs(idents_file): + log.info("Reaping databases...") + + urls = collections.defaultdict(set) + idents = collections.defaultdict(set) + dialects = {} + + with open(idents_file) as file_: + for line in file_: + line = line.strip() + db_name, db_url = line.split(" ") + url_obj = sa_url.make_url(db_url) + if db_name not in dialects: + dialects[db_name] = url_obj.get_dialect() + dialects[db_name].load_provisioning() + url_key = (url_obj.get_backend_name(), url_obj.host) + urls[url_key].add(db_url) + idents[url_key].add(db_name) + + for url_key in urls: + url = list(urls[url_key])[0] + ident = idents[url_key] + run_reap_dbs(url, ident) + + +@register.init +def temp_table_keyword_args(cfg, eng): + """Specify keyword arguments for creating a temporary Table. + + Dialect-specific implementations of this method will return the + kwargs that are passed to the Table method when creating a temporary + table for testing, e.g., in the define_temp_tables method of the + ComponentReflectionTest class in suite/test_reflection.py + """ + raise NotImplementedError( + "no temp table keyword args routine for cfg: %s" % (eng.url,) + ) + + +@register.init +def prepare_for_drop_tables(config, connection): + pass + + +@register.init +def stop_test_class_outside_fixtures(config, db, testcls): + pass + + +@register.init +def get_temp_table_name(cfg, eng, base_name): + """Specify table name for creating a temporary Table. + + Dialect-specific implementations of this method will return the + name to use when creating a temporary table for testing, + e.g., in the define_temp_tables method of the + ComponentReflectionTest class in suite/test_reflection.py + + Default to just the base name since that's what most dialects will + use. The mssql dialect's implementation will need a "#" prepended. + """ + return base_name + + +@register.init +def set_default_schema_on_connection(cfg, dbapi_connection, schema_name): + raise NotImplementedError( + "backend does not implement a schema name set function: %s" + % (cfg.db.url,) + ) + + +@register.init +def upsert( + cfg, + table, + returning, + *, + set_lambda=None, + sort_by_parameter_order=False, + index_elements=None, +): + """return the backends insert..on conflict / on dupe etc. construct. + + while we should add a backend-neutral upsert construct as well, such as + insert().upsert(), it's important that we continue to test the + backend-specific insert() constructs since if we do implement + insert().upsert(), that would be using a different codepath for the things + we need to test like insertmanyvalues, etc. + + """ + raise NotImplementedError( + f"backend does not include an upsert implementation: {cfg.db.url}" + ) + + +@register.init +def normalize_sequence(cfg, sequence): + """Normalize sequence parameters for dialect that don't start with 1 + by default. + + The default implementation does nothing + """ + return sequence + + +@register.init +def allow_stale_update_impl(cfg): + return contextlib.nullcontext() + + +@decorator +def allow_stale_updates(fn, *arg, **kw): + """decorator around a test function that indicates the test will + be UPDATING rows that have been read and are now stale. + + This normally doesn't require intervention except for mariadb 12 + which now raises its own error for that, and we want to turn off + that setting just within the scope of the test that needs it + to be turned off (i.e. ORM stale version tests) + + """ + with allow_stale_update_impl(config._current): + return fn(*arg, **kw) + + +@register.init +def delete_from_all_tables(connection, cfg, metadata): + """an absolutely foolproof delete from all tables routine. + + dialects should override this to add special instructions like + disable constraints etc. + + """ + savepoints = getattr(cfg.requirements, "savepoints", False) + if savepoints: + savepoints = savepoints.enabled + + inspector = inspect(connection) + + for table in reversed( + [ + t + for (t, fks) in sort_tables_and_constraints( + metadata.tables.values() + ) + if t is not None + # remember that inspector.get_table_names() is cached, + # so this emits SQL once per unique schema name + and t.name in inspector.get_table_names(schema=t.schema) + ] + ): + if savepoints: + with connection.begin_nested(): + connection.execute(table.delete()) + else: + connection.execute(table.delete()) diff --git a/venv/Lib/site-packages/sqlalchemy/testing/requirements.py b/venv/Lib/site-packages/sqlalchemy/testing/requirements.py new file mode 100644 index 0000000..26edaff --- /dev/null +++ b/venv/Lib/site-packages/sqlalchemy/testing/requirements.py @@ -0,0 +1,1945 @@ +# testing/requirements.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 + + +"""Global database feature support policy. + +Provides decorators to mark tests requiring specific feature support from the +target database. + +External dialect test suites should subclass SuiteRequirements +to provide specific inclusion/exclusions. + +""" + +from __future__ import annotations + +import os +import platform + +from . import asyncio as _test_asyncio +from . import exclusions +from .exclusions import only_on +from .. import create_engine +from .. import util +from ..pool import QueuePool + + +class Requirements: + pass + + +class SuiteRequirements(Requirements): + @property + def create_table(self): + """target platform can emit basic CreateTable DDL.""" + + return exclusions.open() + + @property + def drop_table(self): + """target platform can emit basic DropTable DDL.""" + + return exclusions.open() + + @property + def table_ddl_if_exists(self): + """target platform supports IF NOT EXISTS / IF EXISTS for tables.""" + + return exclusions.closed() + + @property + def index_ddl_if_exists(self): + """target platform supports IF NOT EXISTS / IF EXISTS for indexes.""" + + return exclusions.closed() + + @property + def uuid_data_type(self): + """Return databases that support the UUID datatype.""" + + return exclusions.closed() + + @property + def foreign_keys(self): + """Target database must support foreign keys.""" + + return exclusions.open() + + @property + def foreign_keys_reflect_as_index(self): + """Target database creates an index that's reflected for + foreign keys.""" + + return exclusions.closed() + + @property + def unique_index_reflect_as_unique_constraints(self): + """Target database reflects unique indexes as unique constrains.""" + + return exclusions.closed() + + @property + def unique_constraints_reflect_as_index(self): + """Target database reflects unique constraints as indexes.""" + + return exclusions.closed() + + @property + def table_value_constructor(self): + """Database / dialect supports a query like: + + .. sourcecode:: sql + + SELECT * FROM VALUES ( (c1, c2), (c1, c2), ...) + AS some_table(col1, col2) + + SQLAlchemy generates this with the :func:`_sql.values` function. + + """ + return exclusions.closed() + + @property + def standard_cursor_sql(self): + """Target database passes SQL-92 style statements to cursor.execute() + when a statement like select() or insert() is run. + + A very small portion of dialect-level tests will ensure that certain + conditions are present in SQL strings, and these tests use very basic + SQL that will work on any SQL-like platform in order to assert results. + + It's normally a given for any pep-249 DBAPI that a statement like + "SELECT id, name FROM table WHERE some_table.id=5" will work. + However, there are dialects that don't actually produce SQL Strings + and instead may work with symbolic objects instead, or dialects that + aren't working with SQL, so for those this requirement can be marked + as excluded. + + """ + + return exclusions.open() + + @property + def on_update_cascade(self): + """target database must support ON UPDATE..CASCADE behavior in + foreign keys.""" + + return exclusions.open() + + @property + def non_updating_cascade(self): + """target database must *not* support ON UPDATE..CASCADE behavior in + foreign keys.""" + return exclusions.closed() + + @property + def deferrable_fks(self): + return exclusions.closed() + + @property + def on_update_or_deferrable_fks(self): + # TODO: exclusions should be composable, + # somehow only_if([x, y]) isn't working here, negation/conjunctions + # getting confused. + return exclusions.only_if( + lambda: self.on_update_cascade.enabled + or self.deferrable_fks.enabled + ) + + @property + def queue_pool(self): + """target database is using QueuePool""" + + def go(config): + return isinstance(config.db.pool, QueuePool) + + return exclusions.only_if(go) + + @property + def self_referential_foreign_keys(self): + """Target database must support self-referential foreign keys.""" + + return exclusions.open() + + @property + def foreign_key_ddl(self): + """Target database must support the DDL phrases for FOREIGN KEY.""" + + return exclusions.open() + + @property + def named_constraints(self): + """target database must support names for constraints.""" + + return exclusions.open() + + @property + def implicitly_named_constraints(self): + """target database must apply names to unnamed constraints.""" + + return exclusions.open() + + @property + def unusual_column_name_characters(self): + """target database allows column names that have unusual characters + in them, such as dots, spaces, slashes, or percent signs. + + The column names are as always in such a case quoted, however the + DB still needs to support those characters in the name somehow. + + """ + return exclusions.open() + + @property + def subqueries(self): + """Target database must support subqueries.""" + + return exclusions.open() + + @property + def offset(self): + """target database can render OFFSET, or an equivalent, in a + SELECT. + """ + + return exclusions.open() + + @property + def bound_limit_offset(self): + """target database can render LIMIT and/or OFFSET using a bound + parameter + """ + + return exclusions.open() + + @property + def sql_expression_limit_offset(self): + """target database can render LIMIT and/or OFFSET with a complete + SQL expression, such as one that uses the addition operator. + parameter + """ + + return exclusions.open() + + @property + def parens_in_union_contained_select_w_limit_offset(self): + """Target database must support parenthesized SELECT in UNION + when LIMIT/OFFSET is specifically present. + + E.g. (SELECT ...) UNION (SELECT ..) + + This is known to fail on SQLite. + + """ + return exclusions.open() + + @property + def parens_in_union_contained_select_wo_limit_offset(self): + """Target database must support parenthesized SELECT in UNION + when OFFSET/LIMIT is specifically not present. + + E.g. (SELECT ... LIMIT ..) UNION (SELECT .. OFFSET ..) + + This is known to fail on SQLite. It also fails on Oracle + because without LIMIT/OFFSET, there is currently no step that + creates an additional subquery. + + """ + return exclusions.open() + + @property + def boolean_col_expressions(self): + """Target database must support boolean expressions as columns""" + + return exclusions.closed() + + @property + def nullable_booleans(self): + """Target database allows boolean columns to store NULL.""" + + return exclusions.open() + + @property + def nullsordering(self): + """Target backends that support nulls ordering.""" + + return exclusions.closed() + + @property + def standalone_binds(self): + """target database/driver supports bound parameters as column + expressions without being in the context of a typed column. + """ + return exclusions.open() + + @property + def standalone_null_binds_whereclause(self): + """target database/driver supports bound parameters with NULL in the + WHERE clause, in situations where it has to be typed. + + """ + return exclusions.open() + + @property + def intersect(self): + """Target database must support INTERSECT or equivalent.""" + return exclusions.closed() + + @property + def except_(self): + """Target database must support EXCEPT or equivalent (i.e. MINUS).""" + return exclusions.closed() + + @property + def window_functions(self): + """Target database must support window functions.""" + return exclusions.closed() + + @property + def ctes(self): + """Target database supports CTEs""" + + return exclusions.closed() + + @property + def ctes_with_update_delete(self): + """target database supports CTES that ride on top of a normal UPDATE + or DELETE statement which refers to the CTE in a correlated subquery. + + """ + + return exclusions.closed() + + @property + def ctes_with_values(self): + """target database supports CTES that ride on top of a VALUES + clause.""" + + return exclusions.closed() + + @property + def ctes_on_dml(self): + """target database supports CTES which consist of INSERT, UPDATE + or DELETE *within* the CTE, e.g. WITH x AS (UPDATE....)""" + + return exclusions.closed() + + @property + def autoincrement_insert(self): + """target platform generates new surrogate integer primary key values + when insert() is executed, excluding the pk column.""" + + return exclusions.open() + + @property + def fetch_rows_post_commit(self): + """target platform will allow cursor.fetchone() to proceed after a + COMMIT. + + Typically this refers to an INSERT statement with RETURNING which + is invoked within "autocommit". If the row can be returned + after the autocommit, then this rule can be open. + + """ + + return exclusions.open() + + @property + def group_by_complex_expression(self): + """target platform supports SQL expressions in GROUP BY + + e.g. + + SELECT x + y AS somelabel FROM table GROUP BY x + y + + """ + + return exclusions.open() + + @property + def sane_rowcount(self): + return exclusions.skip_if( + lambda config: not config.db.dialect.supports_sane_rowcount, + "driver doesn't support 'sane' rowcount", + ) + + @property + def sane_multi_rowcount(self): + return exclusions.fails_if( + lambda config: not config.db.dialect.supports_sane_multi_rowcount, + "driver %(driver)s %(doesnt_support)s 'sane' multi row count", + ) + + @property + def sane_rowcount_w_returning(self): + return exclusions.fails_if( + lambda config: not ( + config.db.dialect.supports_sane_rowcount_returning + ), + "driver doesn't support 'sane' rowcount when returning is on", + ) + + @property + def empty_inserts(self): + """target platform supports INSERT with no values, i.e. + INSERT DEFAULT VALUES or equivalent.""" + + return exclusions.only_if( + lambda config: config.db.dialect.supports_empty_insert + or config.db.dialect.supports_default_values + or config.db.dialect.supports_default_metavalue, + "empty inserts not supported", + ) + + @property + def empty_inserts_executemany(self): + """target platform supports INSERT with no values, i.e. + INSERT DEFAULT VALUES or equivalent, within executemany()""" + + return self.empty_inserts + + @property + def insert_from_select(self): + """target platform supports INSERT from a SELECT.""" + + return exclusions.open() + + @property + def delete_returning(self): + """target platform supports DELETE ... RETURNING.""" + + return exclusions.only_if( + lambda config: config.db.dialect.delete_returning, + "%(database)s %(does_support)s 'DELETE ... RETURNING'", + ) + + @property + def insert_returning(self): + """target platform supports INSERT ... RETURNING.""" + + return exclusions.only_if( + lambda config: config.db.dialect.insert_returning, + "%(database)s %(does_support)s 'INSERT ... RETURNING'", + ) + + @property + def update_returning(self): + """target platform supports UPDATE ... RETURNING.""" + + return exclusions.only_if( + lambda config: config.db.dialect.update_returning, + "%(database)s %(does_support)s 'UPDATE ... RETURNING'", + ) + + @property + def insert_executemany_returning(self): + """target platform supports RETURNING when INSERT is used with + executemany(), e.g. multiple parameter sets, indicating + as many rows come back as do parameter sets were passed. + + """ + + return exclusions.only_if( + lambda config: config.db.dialect.insert_executemany_returning, + "%(database)s %(does_support)s 'RETURNING of " + "multiple rows with INSERT executemany'", + ) + + @property + def insertmanyvalues(self): + return exclusions.only_if( + lambda config: config.db.dialect.supports_multivalues_insert + and config.db.dialect.insert_returning + and config.db.dialect.use_insertmanyvalues, + "%(database)s %(does_support)s 'insertmanyvalues functionality", + ) + + @property + def tuple_in(self): + """Target platform supports the syntax + "(x, y) IN ((x1, y1), (x2, y2), ...)" + """ + + return exclusions.closed() + + @property + def tuple_in_w_empty(self): + """Target platform tuple IN w/ empty set""" + return self.tuple_in + + @property + def duplicate_names_in_cursor_description(self): + """target platform supports a SELECT statement that has + the same name repeated more than once in the columns list.""" + + return exclusions.open() + + @property + def denormalized_names(self): + """Target database must have 'denormalized', i.e. + UPPERCASE as case insensitive names.""" + + return exclusions.skip_if( + lambda config: not config.db.dialect.requires_name_normalize, + "Backend does not require denormalized names.", + ) + + @property + def multivalues_inserts(self): + """target database must support multiple VALUES clauses in an + INSERT statement.""" + + return exclusions.skip_if( + lambda config: not config.db.dialect.supports_multivalues_insert, + "Backend does not support multirow inserts.", + ) + + @property + def implements_get_lastrowid(self): + """target dialect implements the executioncontext.get_lastrowid() + method without reliance on RETURNING. + + """ + return exclusions.open() + + @property + def arraysize(self): + """dialect includes the required pep-249 attribute + ``cursor.arraysize``""" + + return exclusions.open() + + @property + def emulated_lastrowid(self): + """target dialect retrieves cursor.lastrowid, or fetches + from a database-side function after an insert() construct executes, + within the get_lastrowid() method. + + Only dialects that "pre-execute", or need RETURNING to get last + inserted id, would return closed/fail/skip for this. + + """ + return exclusions.closed() + + @property + def emulated_lastrowid_even_with_sequences(self): + """target dialect retrieves cursor.lastrowid or an equivalent + after an insert() construct executes, even if the table has a + Sequence on it. + + """ + return exclusions.closed() + + @property + def dbapi_lastrowid(self): + """target platform includes a 'lastrowid' accessor on the DBAPI + cursor object. + + """ + return exclusions.closed() + + @property + def views(self): + """Target database must support VIEWs.""" + + return exclusions.closed() + + @property + def schemas(self): + """Target database must support external schemas, and have one + named 'test_schema'.""" + + return only_on(lambda config: config.db.dialect.supports_schemas) + + @property + def cross_schema_fk_reflection(self): + """target system must support reflection of inter-schema + foreign keys""" + return exclusions.closed() + + @property + def foreign_key_constraint_name_reflection(self): + """Target supports reflection of FOREIGN KEY constraints and + will return the name of the constraint that was used in the + "CONSTRAINT FOREIGN KEY" DDL. + + MySQL prior to version 8 and MariaDB prior to version 10.5 + don't support this. + + """ + return exclusions.closed() + + @property + def implicit_default_schema(self): + """target system has a strong concept of 'default' schema that can + be referred to implicitly. + + basically, PostgreSQL. + + """ + return exclusions.closed() + + @property + def default_schema_name_switch(self): + """target dialect implements provisioning module including + set_default_schema_on_connection""" + + return exclusions.closed() + + @property + def server_side_cursors(self): + """Target dialect must support server side cursors.""" + + return exclusions.only_if( + [lambda config: config.db.dialect.supports_server_side_cursors], + "no server side cursors support", + ) + + @property + def sequences(self): + """Target database must support SEQUENCEs.""" + + return exclusions.only_if( + [lambda config: config.db.dialect.supports_sequences], + "no sequence support", + ) + + @property + def no_sequences(self): + """the opposite of "sequences", DB does not support sequences at + all.""" + + return exclusions.NotPredicate(self.sequences) + + @property + def sequences_optional(self): + """Target database supports sequences, but also optionally + as a means of generating new PK values.""" + + return exclusions.only_if( + [ + lambda config: config.db.dialect.supports_sequences + and config.db.dialect.sequences_optional + ], + "no sequence support, or sequences not optional", + ) + + @property + def supports_lastrowid(self): + """target database / driver supports cursor.lastrowid as a means + of retrieving the last inserted primary key value. + + note that if the target DB supports sequences also, this is still + assumed to work. This is a new use case brought on by MariaDB 10.3. + + """ + return exclusions.only_if( + [lambda config: config.db.dialect.postfetch_lastrowid] + ) + + @property + def no_lastrowid_support(self): + """the opposite of supports_lastrowid""" + return exclusions.only_if( + [lambda config: not config.db.dialect.postfetch_lastrowid] + ) + + @property + def reflects_pk_names(self): + return exclusions.closed() + + @property + def table_reflection(self): + """target database has general support for table reflection""" + return exclusions.open() + + @property + def reflect_tables_no_columns(self): + """target database supports creation and reflection of tables with no + columns, or at least tables that seem to have no columns.""" + + return exclusions.closed() + + @property + def temp_table_comment_reflection(self): + """indicates if database supports comments on temp tables and + the dialect can reflect them""" + return exclusions.closed() + + @property + def comment_reflection(self): + """Indicates if the database support table comment reflection""" + return exclusions.closed() + + @property + def comment_reflection_full_unicode(self): + """Indicates if the database support table comment reflection in the + full unicode range, including emoji etc. + """ + return exclusions.closed() + + @property + def constraint_comment_reflection(self): + """indicates if the database support comments on constraints + and their reflection""" + return exclusions.closed() + + @property + def column_collation_reflection(self): + """Indicates if the database support column collation reflection. + + This requirement also uses ``get_order_by_collation`` to get + an available collation. + """ + return exclusions.closed() + + @property + def view_column_reflection(self): + """target database must support retrieval of the columns in a view, + similarly to how a table is inspected. + + This does not include the full CREATE VIEW definition. + + """ + return self.views + + @property + def view_reflection(self): + """target database must support inspection of the full CREATE VIEW + definition.""" + return self.views + + @property + def schema_reflection(self): + return self.schemas + + @property + def schema_create_delete(self): + """target database supports schema create and dropped with + 'CREATE SCHEMA' and 'DROP SCHEMA'""" + return exclusions.closed() + + @property + def primary_key_constraint_reflection(self): + return exclusions.open() + + @property + def foreign_key_constraint_reflection(self): + return exclusions.open() + + @property + def foreign_key_constraint_option_reflection_ondelete(self): + return exclusions.closed() + + @property + def fk_constraint_option_reflection_ondelete_restrict(self): + return exclusions.closed() + + @property + def fk_constraint_option_reflection_ondelete_noaction(self): + return exclusions.closed() + + @property + def foreign_key_constraint_option_reflection_onupdate(self): + return exclusions.closed() + + @property + def fk_constraint_option_reflection_onupdate_restrict(self): + return exclusions.closed() + + @property + def temp_table_reflection(self): + return exclusions.open() + + @property + def temp_table_reflect_indexes(self): + return self.temp_table_reflection + + @property + def temp_table_names(self): + """target dialect supports listing of temporary table names""" + return exclusions.closed() + + @property + def has_temp_table(self): + """target dialect supports checking a single temp table name""" + return exclusions.closed() + + @property + def temporary_tables(self): + """target database supports temporary tables""" + return exclusions.open() + + @property + def temporary_views(self): + """target database supports temporary views""" + return exclusions.closed() + + @property + def index_reflection(self): + return exclusions.open() + + @property + def index_reflects_included_columns(self): + return exclusions.closed() + + @property + def indexes_with_ascdesc(self): + """target database supports CREATE INDEX with per-column ASC/DESC.""" + return exclusions.open() + + @property + def reflect_indexes_with_ascdesc(self): + """target database supports reflecting INDEX with per-column + ASC/DESC.""" + return exclusions.open() + + @property + def reflect_indexes_with_ascdesc_as_expression(self): + """target database supports reflecting INDEX with per-column + ASC/DESC but reflects them as expressions (like oracle).""" + return exclusions.closed() + + @property + def indexes_check_column_order(self): + """target database supports CREATE INDEX with column order check.""" + return exclusions.closed() + + @property + def indexes_with_expressions(self): + """target database supports CREATE INDEX against SQL expressions.""" + return exclusions.closed() + + @property + def reflect_indexes_with_expressions(self): + """target database supports reflection of indexes with + SQL expressions.""" + return exclusions.closed() + + @property + def unique_constraint_reflection(self): + """target dialect supports reflection of unique constraints""" + return exclusions.open() + + @property + def inline_check_constraint_reflection(self): + """target dialect supports reflection of inline check constraints""" + return exclusions.closed() + + @property + def check_constraint_reflection(self): + """target dialect supports reflection of check constraints""" + return exclusions.closed() + + @property + def duplicate_key_raises_integrity_error(self): + """target dialect raises IntegrityError when reporting an INSERT + with a primary key violation. (hint: it should) + + """ + return exclusions.open() + + @property + def unbounded_varchar(self): + """Target database must support VARCHAR with no length""" + + return exclusions.open() + + @property + def nvarchar_types(self): + """target database supports NVARCHAR and NCHAR as an actual datatype""" + return exclusions.closed() + + @property + def unicode_data_no_special_types(self): + """Target database/dialect can receive / deliver / compare data with + non-ASCII characters in plain VARCHAR, TEXT columns, without the need + for special "national" datatypes like NVARCHAR or similar. + + """ + return exclusions.open() + + @property + def unicode_data(self): + """Target database/dialect must support Python unicode objects with + non-ASCII characters represented, delivered as bound parameters + as well as in result rows. + + """ + return exclusions.open() + + @property + def unicode_ddl(self): + """Target driver must support some degree of non-ascii symbol + names. + """ + return exclusions.closed() + + @property + def symbol_names_w_double_quote(self): + """Target driver can create tables with a name like 'some " table'""" + return exclusions.open() + + @property + def datetime_interval(self): + """target dialect supports rendering of a datetime.timedelta as a + literal string, e.g. via the TypeEngine.literal_processor() method. + + """ + return exclusions.closed() + + @property + def datetime_literals(self): + """target dialect supports rendering of a date, time, or datetime as a + literal string, e.g. via the TypeEngine.literal_processor() method. + + """ + + return exclusions.closed() + + @property + def datetime(self): + """target dialect supports representation of Python + datetime.datetime() objects.""" + + return exclusions.open() + + @property + def datetime_timezone(self): + """target dialect supports representation of Python + datetime.datetime() with tzinfo with DateTime(timezone=True).""" + + return exclusions.closed() + + @property + def time_timezone(self): + """target dialect supports representation of Python + datetime.time() with tzinfo with Time(timezone=True).""" + + return exclusions.closed() + + @property + def date_implicit_bound(self): + """target dialect when given a date object will bind it such + that the database server knows the object is a date, and not + a plain string. + + """ + return exclusions.open() + + @property + def time_implicit_bound(self): + """target dialect when given a time object will bind it such + that the database server knows the object is a time, and not + a plain string. + + """ + return exclusions.open() + + @property + def datetime_implicit_bound(self): + """target dialect when given a datetime object will bind it such + that the database server knows the object is a datetime, and not + a plain string. + + """ + return exclusions.open() + + @property + def datetime_microseconds(self): + """target dialect supports representation of Python + datetime.datetime() with microsecond objects.""" + + return exclusions.open() + + @property + def timestamp_microseconds(self): + """target dialect supports representation of Python + datetime.datetime() with microsecond objects but only + if TIMESTAMP is used.""" + return exclusions.closed() + + @property + def timestamp_microseconds_implicit_bound(self): + """target dialect when given a datetime object which also includes + a microseconds portion when using the TIMESTAMP data type + will bind it such that the database server knows + the object is a datetime with microseconds, and not a plain string. + + """ + return self.timestamp_microseconds + + @property + def datetime_historic(self): + """target dialect supports representation of Python + datetime.datetime() objects with historic (pre 1970) values.""" + + return exclusions.closed() + + @property + def date(self): + """target dialect supports representation of Python + datetime.date() objects.""" + + return exclusions.open() + + @property + def date_coerces_from_datetime(self): + """target dialect accepts a datetime object as the target + of a date column.""" + + return exclusions.open() + + @property + def date_historic(self): + """target dialect supports representation of Python + datetime.datetime() objects with historic (pre 1970) values.""" + + return exclusions.closed() + + @property + def time(self): + """target dialect supports representation of Python + datetime.time() objects.""" + + return exclusions.open() + + @property + def time_microseconds(self): + """target dialect supports representation of Python + datetime.time() with microsecond objects.""" + + return exclusions.open() + + @property + def binary_comparisons(self): + """target database/driver can allow BLOB/BINARY fields to be compared + against a bound parameter value. + """ + + return exclusions.open() + + @property + def binary_literals(self): + """target backend supports simple binary literals, e.g. an + expression like: + + .. sourcecode:: sql + + SELECT CAST('foo' AS BINARY) + + Where ``BINARY`` is the type emitted from :class:`.LargeBinary`, + e.g. it could be ``BLOB`` or similar. + + Basically fails on Oracle. + + """ + + return exclusions.open() + + @property + def autocommit(self): + """target dialect supports 'AUTOCOMMIT' as an isolation_level""" + return exclusions.closed() + + @property + def skip_autocommit_rollback(self): + """target dialect supports the detect_autocommit_setting() method and + uses the default implementation of do_rollback()""" + + return exclusions.closed() + + @property + def isolation_level(self): + """target dialect supports general isolation level settings. + + Note that this requirement, when enabled, also requires that + the get_isolation_levels() method be implemented. + + """ + return exclusions.closed() + + def get_isolation_levels(self, config): + """Return a structure of supported isolation levels for the current + testing dialect. + + The structure indicates to the testing suite what the expected + "default" isolation should be, as well as the other values that + are accepted. The dictionary has two keys, "default" and "supported". + The "supported" key refers to a list of all supported levels and + it should include AUTOCOMMIT if the dialect supports it. + + If the :meth:`.DefaultRequirements.isolation_level` requirement is + not open, then this method has no return value. + + E.g.:: + + >>> testing.requirements.get_isolation_levels() + { + "default": "READ_COMMITTED", + "supported": [ + "SERIALIZABLE", "READ UNCOMMITTED", + "READ COMMITTED", "REPEATABLE READ", + "AUTOCOMMIT" + ] + } + """ + with config.db.connect() as conn: + try: + supported = conn.dialect.get_isolation_level_values( + conn.connection.dbapi_connection + ) + except NotImplementedError: + return None + else: + return { + "default": conn.dialect.default_isolation_level, + "supported": supported, + } + + @property + def get_isolation_level_values(self): + """target dialect supports the + :meth:`_engine.Dialect.get_isolation_level_values` + method added in SQLAlchemy 2.0. + + """ + + def go(config): + with config.db.connect() as conn: + try: + conn.dialect.get_isolation_level_values( + conn.connection.dbapi_connection + ) + except NotImplementedError: + return False + else: + return True + + return exclusions.only_if(go) + + @property + def dialect_level_isolation_level_param(self): + """test that the dialect allows the 'isolation_level' argument + to be handled by DefaultDialect""" + + def go(config): + try: + e = create_engine( + config.db.url, isolation_level="READ COMMITTED" + ) + except: + return False + else: + return ( + e.dialect._on_connect_isolation_level == "READ COMMITTED" + ) + + return exclusions.only_if(go) + + @property + def array_type(self): + """Target platform implements a native ARRAY type""" + return exclusions.closed() + + @property + def json_type(self): + """target platform implements a native JSON type.""" + + return exclusions.closed() + + @property + def json_array_indexes(self): + """target platform supports numeric array indexes + within a JSON structure""" + + return self.json_type + + @property + def json_index_supplementary_unicode_element(self): + return exclusions.open() + + @property + def legacy_unconditional_json_extract(self): + """Backend has a JSON_EXTRACT or similar function that returns a + valid JSON string in all cases. + + Used to test a legacy feature and is not needed. + + """ + return exclusions.closed() + + @property + def precision_numerics_general(self): + """target backend has general support for moderately high-precision + numerics.""" + return exclusions.open() + + @property + def precision_numerics_enotation_small(self): + """target backend supports Decimal() objects using E notation + to represent very small values.""" + return exclusions.closed() + + @property + def precision_numerics_enotation_large(self): + """target backend supports Decimal() objects using E notation + to represent very large values.""" + return exclusions.open() + + @property + def precision_numerics_many_significant_digits(self): + """target backend supports values with many digits on both sides, + such as 319438950232418390.273596, 87673.594069654243 + + """ + return exclusions.closed() + + @property + def cast_precision_numerics_many_significant_digits(self): + """same as precision_numerics_many_significant_digits but within the + context of a CAST statement (hello MySQL) + + """ + return self.precision_numerics_many_significant_digits + + @property + def server_defaults(self): + """Target backend supports server side defaults for columns""" + + return exclusions.closed() + + @property + def expression_server_defaults(self): + """Target backend supports server side defaults with SQL expressions + for columns""" + + return exclusions.closed() + + @property + def implicit_decimal_binds(self): + """target backend will return a selected Decimal as a Decimal, not + a string. + + e.g.:: + + expr = decimal.Decimal("15.7563") + + value = e.scalar(select(literal(expr))) + + assert value == expr + + See :ticket:`4036` + + """ + + return exclusions.open() + + @property + def numeric_received_as_decimal_untyped(self): + """target backend will return result columns that are explicitly + against NUMERIC or similar precision-numeric datatypes (not including + FLOAT or INT types) as Python Decimal objects, and not as floats + or ints, including when no SQLAlchemy-side typing information is + associated with the statement (e.g. such as a raw SQL string). + + This should be enabled if either the DBAPI itself returns Decimal + objects, or if the dialect has set up DBAPI-specific return type + handlers such that Decimal objects come back automatically. + + """ + return exclusions.open() + + @property + def nested_aggregates(self): + """target database can select an aggregate from a subquery that's + also using an aggregate + + """ + return exclusions.open() + + @property + def recursive_fk_cascade(self): + """target database must support ON DELETE CASCADE on a self-referential + foreign key + + """ + return exclusions.open() + + @property + def precision_numerics_retains_significant_digits(self): + """A precision numeric type will return empty significant digits, + i.e. a value such as 10.000 will come back in Decimal form with + the .000 maintained.""" + + return exclusions.closed() + + @property + def infinity_floats(self): + """The Float type can persist and load float('inf'), float('-inf').""" + + return exclusions.closed() + + @property + def float_or_double_precision_behaves_generically(self): + return exclusions.closed() + + @property + def precision_generic_float_type(self): + """target backend will return native floating point numbers with at + least seven decimal places when using the generic Float type. + + """ + return exclusions.open() + + @property + def literal_float_coercion(self): + """target backend will return the exact float value 15.7563 + with only four significant digits from this statement: + + SELECT :param + + where :param is the Python float 15.7563 + + i.e. it does not return 15.75629997253418 + + """ + return exclusions.open() + + @property + def floats_to_four_decimals(self): + """target backend can return a floating-point number with four + significant digits (such as 15.7563) accurately + (i.e. without FP inaccuracies, such as 15.75629997253418). + + """ + return exclusions.open() + + @property + def fetch_null_from_numeric(self): + """target backend doesn't crash when you try to select a NUMERIC + value that has a value of NULL. + + Added to support Pyodbc bug #351. + """ + + return exclusions.open() + + @property + def float_is_numeric(self): + """target backend uses Numeric for Float/Dual""" + + return exclusions.open() + + @property + def text_type(self): + """Target database must support an unbounded Text() " + "type such as TEXT or CLOB""" + + return exclusions.open() + + @property + def empty_strings_varchar(self): + """target database can persist/return an empty string with a + varchar. + + """ + return exclusions.open() + + @property + def empty_strings_text(self): + """target database can persist/return an empty string with an + unbounded text.""" + + return exclusions.open() + + @property + def expressions_against_unbounded_text(self): + """target database supports use of an unbounded textual field in a + WHERE clause.""" + + return exclusions.open() + + @property + def selectone(self): + """target driver must support the literal statement 'select 1'""" + return exclusions.open() + + @property + def savepoints(self): + """Target database must support savepoints.""" + + return exclusions.closed() + + @property + def two_phase_transactions(self): + """Target database must support two-phase transactions.""" + + return exclusions.closed() + + @property + def update_from(self): + """Target must support UPDATE..FROM syntax""" + return exclusions.closed() + + @property + def delete_from(self): + """Target must support DELETE FROM..FROM or DELETE..USING syntax""" + return exclusions.closed() + + @property + def update_where_target_in_subquery(self): + """Target must support UPDATE (or DELETE) where the same table is + present in a subquery in the WHERE clause. + + This is an ANSI-standard syntax that apparently MySQL can't handle, + such as: + + .. sourcecode:: sql + + UPDATE documents SET flag=1 WHERE documents.title IN + (SELECT max(documents.title) AS title + FROM documents GROUP BY documents.user_id + ) + + """ + return exclusions.open() + + @property + def mod_operator_as_percent_sign(self): + """target database must use a plain percent '%' as the 'modulus' + operator.""" + return exclusions.closed() + + @property + def percent_schema_names(self): + """target backend supports weird identifiers with percent signs + in them, e.g. 'some % column'. + + this is a very weird use case but often has problems because of + DBAPIs that use python formatting. It's not a critical use + case either. + + """ + return exclusions.closed() + + @property + def order_by_col_from_union(self): + """target database supports ordering by a column from a SELECT + inside of a UNION + + E.g.: + + .. sourcecode:: sql + + (SELECT id, ...) UNION (SELECT id, ...) ORDER BY id + + """ + return exclusions.open() + + @property + def order_by_label_with_expression(self): + """target backend supports ORDER BY a column label within an + expression. + + Basically this: + + .. sourcecode:: sql + + select data as foo from test order by foo || 'bar' + + Lots of databases including PostgreSQL don't support this, + so this is off by default. + + """ + return exclusions.closed() + + @property + def order_by_collation(self): + def check(config): + try: + self.get_order_by_collation(config) + return False + except NotImplementedError: + return True + + return exclusions.skip_if(check) + + def get_order_by_collation(self, config): + raise NotImplementedError() + + @property + def unicode_connections(self): + """Target driver must support non-ASCII characters being passed at + all. + """ + return exclusions.open() + + @property + def graceful_disconnects(self): + """Target driver must raise a DBAPI-level exception, such as + InterfaceError, when the underlying connection has been closed + and the execute() method is called. + """ + return exclusions.open() + + @property + def independent_connections(self): + """ + Target must support simultaneous, independent database connections. + """ + return exclusions.open() + + @property + def independent_readonly_connections(self): + """ + Target must support simultaneous, independent database connections + that will be used in a readonly fashion. + + """ + return exclusions.open() + + @property + def skip_mysql_on_windows(self): + """Catchall for a large variety of MySQL on Windows failures""" + return exclusions.open() + + @property + def ad_hoc_engines(self): + """Test environment must allow ad-hoc engine/connection creation. + + No longer used in any tests; is a no-op + + """ + return exclusions.open() + + @property + def no_windows(self): + return exclusions.skip_if(self._running_on_windows()) + + def _running_on_windows(self): + return exclusions.LambdaPredicate( + lambda: platform.system() == "Windows", + description="running on Windows", + ) + + @property + def timing_intensive(self): + from . import config + + return config.add_to_marker.timing_intensive + + @property + def posix(self): + return exclusions.skip_if(lambda: os.name != "posix") + + @property + def memory_intensive(self): + from . import config + + return config.add_to_marker.memory_intensive + + @property + def threading_with_mock(self): + """Mark tests that use threading and mock at the same time - stability + issues have been observed with coverage + + """ + return exclusions.skip_if( + lambda config: config.options.has_coverage, + "Stability issues with coverage", + ) + + @property + def sqlalchemy2_stubs(self): + def check(config): + try: + __import__("sqlalchemy-stubs.ext.mypy") + except ImportError: + return False + else: + return True + + return exclusions.only_if(check) + + @property + def no_sqlalchemy2_stubs(self): + def check(config): + try: + __import__("sqlalchemy-stubs.ext.mypy") + except ImportError: + return False + else: + return True + + return exclusions.skip_if(check) + + @property + def up_to_date_typealias_type(self): + # this checks a particular quirk found in typing_extensions <=4.12.0 + # using older python versions like 3.10 or 3.9, we use TypeAliasType + # from typing_extensions which does not provide for sufficient + # introspection prior to 4.13.0 + def check(config): + import typing + import typing_extensions + + TypeAliasType = getattr( + typing, "TypeAliasType", typing_extensions.TypeAliasType + ) + TV = typing.TypeVar("TV") + TA_generic = TypeAliasType( # type: ignore + "TA_generic", typing.List[TV], type_params=(TV,) + ) + return hasattr(TA_generic[int], "__value__") + + return exclusions.only_if(check) + + @property + def python38(self): + return exclusions.only_if( + lambda: util.py38, "Python 3.8 or above required" + ) + + @property + def python39(self): + return exclusions.only_if( + lambda: util.py39, "Python 3.9 or above required" + ) + + @property + def python310(self): + return exclusions.only_if( + lambda: util.py310, "Python 3.10 or above required" + ) + + @property + def python311(self): + return exclusions.only_if( + lambda: util.py311, "Python 3.11 or above required" + ) + + @property + def python312(self): + return exclusions.only_if( + lambda: util.py312, "Python 3.12 or above required" + ) + + @property + def python314(self): + return exclusions.only_if( + lambda: util.py314, "Python 3.14 or above required" + ) + + @property + def fail_python314b1(self): + return exclusions.fails_if( + lambda: util.compat.py314b1, "Fails as of python 3.14.0b1" + ) + + @property + def not_python314(self): + """This requirement is interim to assist with backporting of + issue #12405. + + SQLAlchemy 2.0 still includes the ``await_fallback()`` method that + makes use of ``asyncio.get_event_loop_policy()``. This is removed + in SQLAlchemy 2.1. + + """ + return exclusions.skip_if( + lambda: util.py314, "Python 3.14 or above not supported" + ) + + @property + def pep649(self): + """pep649 deferred evaluation of annotations without future mode""" + return self.python314 + + @property + def cpython(self): + return exclusions.only_if( + lambda: util.cpython, "cPython interpreter needed" + ) + + @property + def gil_enabled(self): + return exclusions.only_if( + lambda: not util.freethreading, "GIL-enabled build needed" + ) + + @property + def is64bit(self): + return exclusions.only_if(lambda: util.is64bit, "64bit required") + + @property + def patch_library(self): + def check_lib(): + try: + __import__("patch") + except ImportError: + return False + else: + return True + + return exclusions.only_if(check_lib, "patch library needed") + + @property + def predictable_gc(self): + """target platform must remove all cycles unconditionally when + gc.collect() is called, as well as clean out unreferenced subclasses. + + """ + return self.cpython + self.gil_enabled + + @property + def no_coverage(self): + """Test should be skipped if coverage is enabled. + + This is to block tests that exercise libraries that seem to be + sensitive to coverage, such as PostgreSQL notice logging. + + """ + return exclusions.skip_if( + lambda config: config.options.has_coverage, + "Issues observed when coverage is enabled", + ) + + def _has_mysql_on_windows(self, config): + return False + + def _has_mysql_fully_case_sensitive(self, config): + return False + + @property + def sqlite(self): + return exclusions.skip_if(lambda: not self._has_sqlite()) + + @property + def cextensions(self): + return exclusions.skip_if( + lambda: not util.has_compiled_ext(), + "Cython extensions not installed", + ) + + def _has_sqlite(self): + from sqlalchemy import create_engine + + try: + create_engine("sqlite://") + return True + except ImportError: + return False + + @property + def async_dialect(self): + """dialect makes use of await_() to invoke operations on the DBAPI.""" + + return exclusions.closed() + + @property + def asyncio(self): + return self.greenlet + + @property + def no_greenlet(self): + def go(config): + try: + import greenlet # noqa: F401 + except ImportError: + return True + else: + return False + + return exclusions.only_if(go) + + @property + def greenlet(self): + def go(config): + if not _test_asyncio.ENABLE_ASYNCIO: + return False + + try: + import greenlet # noqa: F401 + except ImportError: + return False + else: + return True + + return exclusions.only_if(go) + + @property + def computed_columns(self): + "Supports computed columns" + return exclusions.closed() + + @property + def computed_columns_stored(self): + "Supports computed columns with `persisted=True`" + return exclusions.closed() + + @property + def computed_columns_virtual(self): + "Supports computed columns with `persisted=False`" + return exclusions.closed() + + @property + def computed_columns_default_persisted(self): + """If the default persistence is virtual or stored when `persisted` + is omitted""" + return exclusions.closed() + + @property + def computed_columns_reflect_persisted(self): + """If persistence information is returned by the reflection of + computed columns""" + return exclusions.closed() + + @property + def supports_distinct_on(self): + """If a backend supports the DISTINCT ON in a select""" + return exclusions.closed() + + @property + def supports_is_distinct_from(self): + """Supports some form of "x IS [NOT] DISTINCT FROM y" construct. + Different dialects will implement their own flavour, e.g., + sqlite will emit "x IS NOT y" instead of "x IS DISTINCT FROM y". + + .. seealso:: + + :meth:`.ColumnOperators.is_distinct_from` + + """ + return exclusions.skip_if( + lambda config: not config.db.dialect.supports_is_distinct_from, + "driver doesn't support an IS DISTINCT FROM construct", + ) + + @property + def identity_columns(self): + """If a backend supports GENERATED { ALWAYS | BY DEFAULT } + AS IDENTITY""" + return exclusions.closed() + + @property + def identity_columns_standard(self): + """If a backend supports GENERATED { ALWAYS | BY DEFAULT } + AS IDENTITY with a standard syntax. + This is mainly to exclude MSSql. + """ + return exclusions.closed() + + @property + def regexp_match(self): + """backend supports the regexp_match operator.""" + return exclusions.closed() + + @property + def regexp_replace(self): + """backend supports the regexp_replace operator.""" + return exclusions.closed() + + @property + def fetch_first(self): + """backend supports the fetch first clause.""" + return exclusions.closed() + + @property + def fetch_percent(self): + """backend supports the fetch first clause with percent.""" + return exclusions.closed() + + @property + def fetch_ties(self): + """backend supports the fetch first clause with ties.""" + return exclusions.closed() + + @property + def fetch_no_order_by(self): + """backend supports the fetch first without order by""" + return exclusions.closed() + + @property + def fetch_offset_with_options(self): + """backend supports the offset when using fetch first with percent + or ties. basically this is "not mssql" + """ + return exclusions.closed() + + @property + def fetch_expression(self): + """backend supports fetch / offset with expression in them, like + + SELECT * FROM some_table + OFFSET 1 + 1 ROWS FETCH FIRST 1 + 1 ROWS ONLY + """ + return exclusions.closed() + + @property + def autoincrement_without_sequence(self): + """If autoincrement=True on a column does not require an explicit + sequence. This should be false only for oracle. + """ + return exclusions.open() + + @property + def generic_classes(self): + "If X[Y] can be implemented with ``__class_getitem__``. py3.7+" + return exclusions.open() + + @property + def json_deserializer_binary(self): + "indicates if the json_deserializer function is called with bytes" + return exclusions.closed() + + @property + def reflect_table_options(self): + """Target database must support reflecting table_options.""" + return exclusions.closed() + + @property + def materialized_views(self): + """Target database must support MATERIALIZED VIEWs.""" + return exclusions.closed() + + @property + def materialized_views_reflect_pk(self): + """Target database reflect MATERIALIZED VIEWs pks.""" + return exclusions.closed() + + @property + def supports_bitwise_or(self): + """Target database supports bitwise or""" + return exclusions.closed() + + @property + def supports_bitwise_and(self): + """Target database supports bitwise and""" + return exclusions.closed() + + @property + def supports_bitwise_not(self): + """Target database supports bitwise not""" + return exclusions.closed() + + @property + def supports_bitwise_xor(self): + """Target database supports bitwise xor""" + return exclusions.closed() + + @property + def supports_bitwise_shift(self): + """Target database supports bitwise left or right shift""" + return exclusions.closed() + + @property + def like_escapes(self): + """Target backend supports custom ESCAPE characters + with LIKE comparisons""" + return exclusions.open() diff --git a/venv/Lib/site-packages/sqlalchemy/testing/schema.py b/venv/Lib/site-packages/sqlalchemy/testing/schema.py new file mode 100644 index 0000000..5d2600e --- /dev/null +++ b/venv/Lib/site-packages/sqlalchemy/testing/schema.py @@ -0,0 +1,198 @@ +# testing/schema.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 sys + +from . import config +from . import exclusions +from .. import event +from .. import schema +from .. import types as sqltypes +from ..orm import mapped_column as _orm_mapped_column +from ..util import OrderedDict + +__all__ = ["Table", "Column"] + +table_options = {} + + +def Table(*args, **kw) -> schema.Table: + """A schema.Table wrapper/hook for dialect-specific tweaks.""" + + # pop out local options; these are not used at the moment + _ = {k: kw.pop(k) for k in list(kw) if k.startswith("test_")} + + kw.update(table_options) + + return schema.Table(*args, **kw) + + +def mapped_column(*args, **kw): + """An orm.mapped_column wrapper/hook for dialect-specific tweaks.""" + + return _schema_column(_orm_mapped_column, args, kw) + + +def Column(*args, **kw): + """A schema.Column wrapper/hook for dialect-specific tweaks.""" + + return _schema_column(schema.Column, args, kw) + + +def _schema_column(factory, args, kw): + test_opts = {k: kw.pop(k) for k in list(kw) if k.startswith("test_")} + + if not config.requirements.foreign_key_ddl.enabled_for_config(config): + args = [arg for arg in args if not isinstance(arg, schema.ForeignKey)] + + construct = factory(*args, **kw) + + if factory is schema.Column: + col = construct + else: + col = construct.column + + if test_opts.get("test_needs_autoincrement", False) and kw.get( + "primary_key", False + ): + if col.default is None and col.server_default is None: + col.autoincrement = True + + # allow any test suite to pick up on this + col.info["test_needs_autoincrement"] = True + + # hardcoded rule for oracle; this should + # be moved out + if exclusions.against(config._current, "oracle"): + + def add_seq(c, tbl): + c._init_items( + schema.Sequence( + _truncate_name( + config.db.dialect, tbl.name + "_" + c.name + "_seq" + ), + optional=True, + ) + ) + + event.listen(col, "after_parent_attach", add_seq, propagate=True) + return construct + + +class eq_type_affinity: + """Helper to compare types inside of datastructures based on affinity. + + E.g.:: + + eq_( + inspect(connection).get_columns("foo"), + [ + { + "name": "id", + "type": testing.eq_type_affinity(sqltypes.INTEGER), + "nullable": False, + "default": None, + "autoincrement": False, + }, + { + "name": "data", + "type": testing.eq_type_affinity(sqltypes.NullType), + "nullable": True, + "default": None, + "autoincrement": False, + }, + ], + ) + + """ + + def __init__(self, target): + self.target = sqltypes.to_instance(target) + + def __eq__(self, other): + return self.target._type_affinity is other._type_affinity + + def __ne__(self, other): + return self.target._type_affinity is not other._type_affinity + + +class eq_compile_type: + """similar to eq_type_affinity but uses compile""" + + def __init__(self, target): + self.target = target + + def __eq__(self, other): + return self.target == other.compile() + + def __ne__(self, other): + return self.target != other.compile() + + +class eq_clause_element: + """Helper to compare SQL structures based on compare()""" + + def __init__(self, target): + self.target = target + + def __eq__(self, other): + return self.target.compare(other) + + def __ne__(self, other): + return not self.target.compare(other) + + +def _truncate_name(dialect, name): + if len(name) > dialect.max_identifier_length: + return ( + name[0 : max(dialect.max_identifier_length - 6, 0)] + + "_" + + hex(hash(name) % 64)[2:] + ) + else: + return name + + +def pep435_enum(name): + # Implements PEP 435 in the minimal fashion needed by SQLAlchemy + __members__ = OrderedDict() + + def __init__(self, name, value, alias=None): + self.name = name + self.value = value + self.__members__[name] = self + value_to_member[value] = self + setattr(self.__class__, name, self) + if alias: + self.__members__[alias] = self + setattr(self.__class__, alias, self) + + value_to_member = {} + + @classmethod + def get(cls, value): + return value_to_member[value] + + someenum = type( + name, + (object,), + {"__members__": __members__, "__init__": __init__, "get": get}, + ) + + # getframe() trick for pickling I don't understand courtesy + # Python namedtuple() + try: + module = sys._getframe(1).f_globals.get("__name__", "__main__") + except (AttributeError, ValueError): + pass + if module is not None: + someenum.__module__ = module + + return someenum diff --git a/venv/Lib/site-packages/sqlalchemy/testing/util.py b/venv/Lib/site-packages/sqlalchemy/testing/util.py new file mode 100644 index 0000000..8656c54 --- /dev/null +++ b/venv/Lib/site-packages/sqlalchemy/testing/util.py @@ -0,0 +1,534 @@ +# testing/util.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 deque +from collections import namedtuple +import contextlib +import decimal +import gc +from itertools import chain +import pickle +import random +import sys +from sys import getsizeof +import time +import types +from typing import Any + +from . import config +from . import mock +from .. import inspect +from ..engine import Connection +from ..schema import Column +from ..schema import DropConstraint +from ..schema import DropTable +from ..schema import ForeignKeyConstraint +from ..schema import MetaData +from ..schema import Table +from ..sql import schema +from ..sql.sqltypes import Integer +from ..util import decorator +from ..util import defaultdict +from ..util import has_refcount_gc +from ..util import inspect_getfullargspec + +if not has_refcount_gc: + + def non_refcount_gc_collect(*args): + gc.collect() + gc.collect() + + gc_collect = lazy_gc = non_refcount_gc_collect +else: + # assume CPython - straight gc.collect, lazy_gc() is a pass + gc_collect = gc.collect + + def lazy_gc(): + pass + + +def picklers(): + nt = namedtuple("picklers", ["loads", "dumps"]) + + for protocol in range(-2, pickle.HIGHEST_PROTOCOL + 1): + yield nt(pickle.loads, lambda d: pickle.dumps(d, protocol)) + + +def random_choices(population, k=1): + return random.choices(population, k=k) + + +def round_decimal(value, prec): + if isinstance(value, float): + return round(value, prec) + + # can also use shift() here but that is 2.6 only + return (value * decimal.Decimal("1" + "0" * prec)).to_integral( + decimal.ROUND_FLOOR + ) / pow(10, prec) + + +class RandomSet(set): + def __iter__(self): + l = list(set.__iter__(self)) + random.shuffle(l) + return iter(l) + + def pop(self): + index = random.randint(0, len(self) - 1) + item = list(set.__iter__(self))[index] + self.remove(item) + return item + + def union(self, other): + return RandomSet(set.union(self, other)) + + def difference(self, other): + return RandomSet(set.difference(self, other)) + + def intersection(self, other): + return RandomSet(set.intersection(self, other)) + + def copy(self): + return RandomSet(self) + + +def conforms_partial_ordering(tuples, sorted_elements): + """True if the given sorting conforms to the given partial ordering.""" + + deps = defaultdict(set) + for parent, child in tuples: + deps[parent].add(child) + for i, node in enumerate(sorted_elements): + for n in sorted_elements[i:]: + if node in deps[n]: + return False + else: + return True + + +def all_partial_orderings(tuples, elements): + edges = defaultdict(set) + for parent, child in tuples: + edges[child].add(parent) + + def _all_orderings(elements): + if len(elements) == 1: + yield list(elements) + else: + for elem in elements: + subset = set(elements).difference([elem]) + if not subset.intersection(edges[elem]): + for sub_ordering in _all_orderings(subset): + yield [elem] + sub_ordering + + return iter(_all_orderings(elements)) + + +def function_named(fn, name): + """Return a function with a given __name__. + + Will assign to __name__ and return the original function if possible on + the Python implementation, otherwise a new function will be constructed. + + This function should be phased out as much as possible + in favor of @decorator. Tests that "generate" many named tests + should be modernized. + + """ + try: + fn.__name__ = name + except TypeError: + fn = types.FunctionType( + fn.__code__, fn.__globals__, name, fn.__defaults__, fn.__closure__ + ) + return fn + + +def run_as_contextmanager(ctx, fn, *arg, **kw): + """Run the given function under the given contextmanager, + simulating the behavior of 'with' to support older + Python versions. + + This is not necessary anymore as we have placed 2.6 + as minimum Python version, however some tests are still using + this structure. + + """ + + obj = ctx.__enter__() + try: + result = fn(obj, *arg, **kw) + ctx.__exit__(None, None, None) + return result + except: + exc_info = sys.exc_info() + raise_ = ctx.__exit__(*exc_info) + if not raise_: + raise + else: + return raise_ + + +def rowset(results): + """Converts the results of sql execution into a plain set of column tuples. + + Useful for asserting the results of an unordered query. + """ + + return {tuple(row) for row in results} + + +def fail(msg): + assert False, msg + + +@decorator +def provide_metadata(fn, *args, **kw): + """Provide bound MetaData for a single test, dropping afterwards. + + Legacy; use the "metadata" pytest fixture. + + """ + + from . import fixtures + + metadata = schema.MetaData() + self = args[0] + prev_meta = getattr(self, "metadata", None) + self.metadata = metadata + try: + return fn(*args, **kw) + finally: + # close out some things that get in the way of dropping tables. + # when using the "metadata" fixture, there is a set ordering + # of things that makes sure things are cleaned up in order, however + # the simple "decorator" nature of this legacy function means + # we have to hardcode some of that cleanup ahead of time. + + # close ORM sessions + fixtures.close_all_sessions() + + # integrate with the "connection" fixture as there are many + # tests where it is used along with provide_metadata + cfc = fixtures.base._connection_fixture_connection + if cfc: + # TODO: this warning can be used to find all the places + # this is used with connection fixture + # warn("mixing legacy provide metadata with connection fixture") + drop_all_tables_from_metadata(metadata, cfc) + # as the provide_metadata fixture is often used with "testing.db", + # when we do the drop we have to commit the transaction so that + # the DB is actually updated as the CREATE would have been + # committed + cfc.get_transaction().commit() + else: + drop_all_tables_from_metadata(metadata, config.db) + self.metadata = prev_meta + + +def flag_combinations(*combinations): + """A facade around @testing.combinations() oriented towards boolean + keyword-based arguments. + + Basically generates a nice looking identifier based on the keywords + and also sets up the argument names. + + E.g.:: + + @testing.flag_combinations( + dict(lazy=False, passive=False), + dict(lazy=True, passive=False), + dict(lazy=False, passive=True), + dict(lazy=False, passive=True, raiseload=True), + ) + def test_fn(lazy, passive, raiseload): ... + + would result in:: + + @testing.combinations( + ("", False, False, False), + ("lazy", True, False, False), + ("lazy_passive", True, True, False), + ("lazy_passive", True, True, True), + id_="iaaa", + argnames="lazy,passive,raiseload", + ) + def test_fn(lazy, passive, raiseload): ... + + """ + + keys = set() + + for d in combinations: + keys.update(d) + + keys = sorted(keys) + + return config.combinations( + *[ + ("_".join(k for k in keys if d.get(k, False)),) + + tuple(d.get(k, False) for k in keys) + for d in combinations + ], + id_="i" + ("a" * len(keys)), + argnames=",".join(keys), + ) + + +def lambda_combinations(lambda_arg_sets, **kw): + args = inspect_getfullargspec(lambda_arg_sets) + + arg_sets = lambda_arg_sets(*[mock.Mock() for arg in args[0]]) + + def create_fixture(pos): + def fixture(**kw): + return lambda_arg_sets(**kw)[pos] + + fixture.__name__ = "fixture_%3.3d" % pos + return fixture + + return config.combinations( + *[(create_fixture(i),) for i in range(len(arg_sets))], **kw + ) + + +def resolve_lambda(__fn, **kw): + """Given a no-arg lambda and a namespace, return a new lambda that + has all the values filled in. + + This is used so that we can have module-level fixtures that + refer to instance-level variables using lambdas. + + """ + + pos_args = inspect_getfullargspec(__fn)[0] + pass_pos_args = {arg: kw.pop(arg) for arg in pos_args} + glb = dict(__fn.__globals__) + glb.update(kw) + new_fn = types.FunctionType(__fn.__code__, glb) + return new_fn(**pass_pos_args) + + +def metadata_fixture(ddl="function"): + """Provide MetaData for a pytest fixture.""" + + def decorate(fn): + def run_ddl(self): + metadata = self.metadata = schema.MetaData() + try: + result = fn(self, metadata) + metadata.create_all(config.db) + # TODO: + # somehow get a per-function dml erase fixture here + yield result + finally: + metadata.drop_all(config.db) + + return config.fixture(scope=ddl)(run_ddl) + + return decorate + + +def force_drop_names(*names): + """Force the given table names to be dropped after test complete, + isolating for foreign key cycles + + """ + + @decorator + def go(fn, *args, **kw): + try: + return fn(*args, **kw) + finally: + drop_all_tables(config.db, inspect(config.db), include_names=names) + + return go + + +class adict(dict): + """Dict keys available as attributes. Shadows.""" + + def __getattribute__(self, key): + try: + return self[key] + except KeyError: + return dict.__getattribute__(self, key) + + def __call__(self, *keys): + return tuple([self[key] for key in keys]) + + get_all = __call__ + + +def drop_all_tables_from_metadata(metadata, engine_or_connection): + from . import engines + + def go(connection): + engines.testing_reaper.prepare_for_drop_tables(connection) + + if not connection.dialect.supports_alter: + from . import assertions + + with assertions.expect_warnings( + "Can't sort tables", assert_=False + ): + metadata.drop_all(connection) + else: + metadata.drop_all(connection) + + if not isinstance(engine_or_connection, Connection): + with engine_or_connection.begin() as connection: + go(connection) + else: + go(engine_or_connection) + + +def drop_all_tables( + engine, + inspector, + schema=None, + consider_schemas=(None,), + include_names=None, +): + if include_names is not None: + include_names = set(include_names) + + if schema is not None: + assert consider_schemas == ( + None, + ), "consider_schemas and schema are mutually exclusive" + consider_schemas = (schema,) + + with engine.begin() as conn: + for table_key, fkcs in reversed( + inspector.sort_tables_on_foreign_key_dependency( + consider_schemas=consider_schemas + ) + ): + if table_key: + if ( + include_names is not None + and table_key[1] not in include_names + ): + continue + conn.execute( + DropTable( + Table(table_key[1], MetaData(), schema=table_key[0]) + ) + ) + elif fkcs: + if not engine.dialect.supports_alter: + continue + for t_key, fkc in fkcs: + if ( + include_names is not None + and t_key[1] not in include_names + ): + continue + tb = Table( + t_key[1], + MetaData(), + Column("x", Integer), + Column("y", Integer), + schema=t_key[0], + ) + conn.execute( + DropConstraint( + ForeignKeyConstraint([tb.c.x], [tb.c.y], name=fkc) + ) + ) + + +def teardown_events(event_cls): + @decorator + def decorate(fn, *arg, **kw): + try: + return fn(*arg, **kw) + finally: + event_cls._clear() + + return decorate + + +def total_size(o): + """Returns the approximate memory footprint an object and all of its + contents. + + source: https://code.activestate.com/recipes/577504/ + + + """ + + def dict_handler(d): + return chain.from_iterable(d.items()) + + all_handlers = { + tuple: iter, + list: iter, + deque: iter, + dict: dict_handler, + set: iter, + frozenset: iter, + } + seen = set() # track which object id's have already been seen + default_size = getsizeof(0) # estimate sizeof object without __sizeof__ + + def sizeof(o): + if id(o) in seen: # do not double count the same object + return 0 + seen.add(id(o)) + s = getsizeof(o, default_size) + + for typ, handler in all_handlers.items(): + if isinstance(o, typ): + s += sum(map(sizeof, handler(o))) + break + return s + + return sizeof(o) + + +def count_cache_key_tuples(tup): + """given a cache key tuple, counts how many instances of actual + tuples are found. + + used to alert large jumps in cache key complexity. + + """ + stack = [tup] + + sentinel = object() + num_elements = 0 + + while stack: + elem = stack.pop(0) + if elem is sentinel: + num_elements += 1 + elif isinstance(elem, tuple): + if elem: + stack = list(elem) + [sentinel] + stack + return num_elements + + +@contextlib.contextmanager +def skip_if_timeout(seconds: float, cleanup: Any = None): + + now = time.time() + yield + sec = time.time() - now + if sec > seconds: + try: + cleanup() + finally: + config.skip_test( + f"test took too long ({sec:.4f} seconds > {seconds})" + ) diff --git a/venv/Lib/site-packages/sqlalchemy/testing/warnings.py b/venv/Lib/site-packages/sqlalchemy/testing/warnings.py new file mode 100644 index 0000000..a96ec36 --- /dev/null +++ b/venv/Lib/site-packages/sqlalchemy/testing/warnings.py @@ -0,0 +1,52 @@ +# testing/warnings.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 warnings + +from . import assertions +from .. import exc +from .. import exc as sa_exc +from ..exc import SATestSuiteWarning +from ..util.langhelpers import _warnings_warn + + +def warn_test_suite(message): + _warnings_warn(message, category=SATestSuiteWarning) + + +def setup_filters(): + """hook for setting up warnings filters. + + SQLAlchemy-specific classes must only be here and not in pytest config, + as we need to delay importing SQLAlchemy until conftest.py has been + processed. + + NOTE: filters on subclasses of DeprecationWarning or + PendingDeprecationWarning have no effect if added here, since pytest + will add at each test the following filters + ``always::PendingDeprecationWarning`` and ``always::DeprecationWarning`` + that will take precedence over any added here. + + """ + warnings.filterwarnings("error", category=exc.SAWarning) + warnings.filterwarnings("always", category=exc.SATestSuiteWarning) + + +def assert_warnings(fn, warning_msgs, regex=False): + """Assert that each of the given warnings are emitted by fn. + + Deprecated. Please use assertions.expect_warnings(). + + """ + + with assertions._expect_warnings( + sa_exc.SAWarning, warning_msgs, regex=regex + ): + return fn()