Загрузить файлы в «venv/Lib/site-packages/sqlalchemy/testing/suite»
This commit is contained in:
19
venv/Lib/site-packages/sqlalchemy/testing/suite/__init__.py
Normal file
19
venv/Lib/site-packages/sqlalchemy/testing/suite/__init__.py
Normal file
@@ -0,0 +1,19 @@
|
||||
# testing/suite/__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
|
||||
from .test_cte import * # noqa
|
||||
from .test_ddl import * # noqa
|
||||
from .test_deprecations import * # noqa
|
||||
from .test_dialect import * # noqa
|
||||
from .test_insert import * # noqa
|
||||
from .test_reflection import * # noqa
|
||||
from .test_results import * # noqa
|
||||
from .test_rowcount import * # noqa
|
||||
from .test_select import * # noqa
|
||||
from .test_sequence import * # noqa
|
||||
from .test_types import * # noqa
|
||||
from .test_unicode_ddl import * # noqa
|
||||
from .test_update_delete import * # noqa
|
||||
237
venv/Lib/site-packages/sqlalchemy/testing/suite/test_cte.py
Normal file
237
venv/Lib/site-packages/sqlalchemy/testing/suite/test_cte.py
Normal file
@@ -0,0 +1,237 @@
|
||||
# testing/suite/test_cte.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 .. import fixtures
|
||||
from ..assertions import eq_
|
||||
from ..schema import Column
|
||||
from ..schema import Table
|
||||
from ... import column
|
||||
from ... import ForeignKey
|
||||
from ... import Integer
|
||||
from ... import select
|
||||
from ... import String
|
||||
from ... import testing
|
||||
from ... import values
|
||||
|
||||
|
||||
class CTETest(fixtures.TablesTest):
|
||||
__sparse_driver_backend__ = True
|
||||
__requires__ = ("ctes",)
|
||||
|
||||
run_inserts = "each"
|
||||
run_deletes = "each"
|
||||
|
||||
@classmethod
|
||||
def define_tables(cls, metadata):
|
||||
Table(
|
||||
"some_table",
|
||||
metadata,
|
||||
Column("id", Integer, primary_key=True),
|
||||
Column("data", String(50)),
|
||||
Column("parent_id", ForeignKey("some_table.id")),
|
||||
)
|
||||
|
||||
Table(
|
||||
"some_other_table",
|
||||
metadata,
|
||||
Column("id", Integer, primary_key=True),
|
||||
Column("data", String(50)),
|
||||
Column("parent_id", Integer),
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def insert_data(cls, connection):
|
||||
connection.execute(
|
||||
cls.tables.some_table.insert(),
|
||||
[
|
||||
{"id": 1, "data": "d1", "parent_id": None},
|
||||
{"id": 2, "data": "d2", "parent_id": 1},
|
||||
{"id": 3, "data": "d3", "parent_id": 1},
|
||||
{"id": 4, "data": "d4", "parent_id": 3},
|
||||
{"id": 5, "data": "d5", "parent_id": 3},
|
||||
],
|
||||
)
|
||||
|
||||
def test_select_nonrecursive_round_trip(self, connection):
|
||||
some_table = self.tables.some_table
|
||||
|
||||
cte = (
|
||||
select(some_table)
|
||||
.where(some_table.c.data.in_(["d2", "d3", "d4"]))
|
||||
.cte("some_cte")
|
||||
)
|
||||
result = connection.execute(
|
||||
select(cte.c.data).where(cte.c.data.in_(["d4", "d5"]))
|
||||
)
|
||||
eq_(result.fetchall(), [("d4",)])
|
||||
|
||||
def test_select_recursive_round_trip(self, connection):
|
||||
some_table = self.tables.some_table
|
||||
|
||||
cte = (
|
||||
select(some_table)
|
||||
.where(some_table.c.data.in_(["d2", "d3", "d4"]))
|
||||
.cte("some_cte", recursive=True)
|
||||
)
|
||||
|
||||
cte_alias = cte.alias("c1")
|
||||
st1 = some_table.alias()
|
||||
# note that SQL Server requires this to be UNION ALL,
|
||||
# can't be UNION
|
||||
cte = cte.union_all(
|
||||
select(st1).where(st1.c.id == cte_alias.c.parent_id)
|
||||
)
|
||||
result = connection.execute(
|
||||
select(cte.c.data)
|
||||
.where(cte.c.data != "d2")
|
||||
.order_by(cte.c.data.desc())
|
||||
)
|
||||
eq_(
|
||||
result.fetchall(),
|
||||
[("d4",), ("d3",), ("d3",), ("d1",), ("d1",), ("d1",)],
|
||||
)
|
||||
|
||||
def test_insert_from_select_round_trip(self, connection):
|
||||
some_table = self.tables.some_table
|
||||
some_other_table = self.tables.some_other_table
|
||||
|
||||
cte = (
|
||||
select(some_table)
|
||||
.where(some_table.c.data.in_(["d2", "d3", "d4"]))
|
||||
.cte("some_cte")
|
||||
)
|
||||
connection.execute(
|
||||
some_other_table.insert().from_select(
|
||||
["id", "data", "parent_id"], select(cte)
|
||||
)
|
||||
)
|
||||
eq_(
|
||||
connection.execute(
|
||||
select(some_other_table).order_by(some_other_table.c.id)
|
||||
).fetchall(),
|
||||
[(2, "d2", 1), (3, "d3", 1), (4, "d4", 3)],
|
||||
)
|
||||
|
||||
@testing.requires.ctes_with_update_delete
|
||||
@testing.requires.update_from
|
||||
def test_update_from_round_trip(self, connection):
|
||||
some_table = self.tables.some_table
|
||||
some_other_table = self.tables.some_other_table
|
||||
|
||||
connection.execute(
|
||||
some_other_table.insert().from_select(
|
||||
["id", "data", "parent_id"], select(some_table)
|
||||
)
|
||||
)
|
||||
|
||||
cte = (
|
||||
select(some_table)
|
||||
.where(some_table.c.data.in_(["d2", "d3", "d4"]))
|
||||
.cte("some_cte")
|
||||
)
|
||||
connection.execute(
|
||||
some_other_table.update()
|
||||
.values(parent_id=5)
|
||||
.where(some_other_table.c.data == cte.c.data)
|
||||
)
|
||||
eq_(
|
||||
connection.execute(
|
||||
select(some_other_table).order_by(some_other_table.c.id)
|
||||
).fetchall(),
|
||||
[
|
||||
(1, "d1", None),
|
||||
(2, "d2", 5),
|
||||
(3, "d3", 5),
|
||||
(4, "d4", 5),
|
||||
(5, "d5", 3),
|
||||
],
|
||||
)
|
||||
|
||||
@testing.requires.ctes_with_update_delete
|
||||
@testing.requires.delete_from
|
||||
def test_delete_from_round_trip(self, connection):
|
||||
some_table = self.tables.some_table
|
||||
some_other_table = self.tables.some_other_table
|
||||
|
||||
connection.execute(
|
||||
some_other_table.insert().from_select(
|
||||
["id", "data", "parent_id"], select(some_table)
|
||||
)
|
||||
)
|
||||
|
||||
cte = (
|
||||
select(some_table)
|
||||
.where(some_table.c.data.in_(["d2", "d3", "d4"]))
|
||||
.cte("some_cte")
|
||||
)
|
||||
connection.execute(
|
||||
some_other_table.delete().where(
|
||||
some_other_table.c.data == cte.c.data
|
||||
)
|
||||
)
|
||||
eq_(
|
||||
connection.execute(
|
||||
select(some_other_table).order_by(some_other_table.c.id)
|
||||
).fetchall(),
|
||||
[(1, "d1", None), (5, "d5", 3)],
|
||||
)
|
||||
|
||||
@testing.requires.ctes_with_update_delete
|
||||
def test_delete_scalar_subq_round_trip(self, connection):
|
||||
some_table = self.tables.some_table
|
||||
some_other_table = self.tables.some_other_table
|
||||
|
||||
connection.execute(
|
||||
some_other_table.insert().from_select(
|
||||
["id", "data", "parent_id"], select(some_table)
|
||||
)
|
||||
)
|
||||
|
||||
cte = (
|
||||
select(some_table)
|
||||
.where(some_table.c.data.in_(["d2", "d3", "d4"]))
|
||||
.cte("some_cte")
|
||||
)
|
||||
connection.execute(
|
||||
some_other_table.delete().where(
|
||||
some_other_table.c.data
|
||||
== select(cte.c.data)
|
||||
.where(cte.c.id == some_other_table.c.id)
|
||||
.scalar_subquery()
|
||||
)
|
||||
)
|
||||
eq_(
|
||||
connection.execute(
|
||||
select(some_other_table).order_by(some_other_table.c.id)
|
||||
).fetchall(),
|
||||
[(1, "d1", None), (5, "d5", 3)],
|
||||
)
|
||||
|
||||
@testing.variation("values_named", [True, False])
|
||||
@testing.variation("cte_named", [True, False])
|
||||
@testing.variation("literal_binds", [True, False])
|
||||
@testing.requires.ctes_with_values
|
||||
def test_values_named_via_cte(
|
||||
self, connection, values_named, cte_named, literal_binds
|
||||
):
|
||||
|
||||
cte1 = (
|
||||
values(
|
||||
column("col1", String),
|
||||
column("col2", Integer),
|
||||
literal_binds=bool(literal_binds),
|
||||
name="some name" if values_named else None,
|
||||
)
|
||||
.data([("a", 2), ("b", 3)])
|
||||
.cte("cte1" if cte_named else None)
|
||||
)
|
||||
|
||||
stmt = select(cte1)
|
||||
|
||||
rows = connection.execute(stmt).all()
|
||||
eq_(rows, [("a", 2), ("b", 3)])
|
||||
389
venv/Lib/site-packages/sqlalchemy/testing/suite/test_ddl.py
Normal file
389
venv/Lib/site-packages/sqlalchemy/testing/suite/test_ddl.py
Normal file
@@ -0,0 +1,389 @@
|
||||
# testing/suite/test_ddl.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
|
||||
|
||||
import random
|
||||
|
||||
from . import testing
|
||||
from .. import config
|
||||
from .. import fixtures
|
||||
from .. import util
|
||||
from ..assertions import eq_
|
||||
from ..assertions import is_false
|
||||
from ..assertions import is_true
|
||||
from ..config import requirements
|
||||
from ..schema import Table
|
||||
from ... import CheckConstraint
|
||||
from ... import Column
|
||||
from ... import ForeignKeyConstraint
|
||||
from ... import Index
|
||||
from ... import inspect
|
||||
from ... import Integer
|
||||
from ... import schema
|
||||
from ... import String
|
||||
from ... import UniqueConstraint
|
||||
|
||||
|
||||
class TableDDLTest(fixtures.TestBase):
|
||||
__sparse_driver_backend__ = True
|
||||
|
||||
def _simple_fixture(self, schema=None):
|
||||
return Table(
|
||||
"test_table",
|
||||
self.metadata,
|
||||
Column("id", Integer, primary_key=True, autoincrement=False),
|
||||
Column("data", String(50)),
|
||||
schema=schema,
|
||||
)
|
||||
|
||||
def _underscore_fixture(self):
|
||||
return Table(
|
||||
"_test_table",
|
||||
self.metadata,
|
||||
Column("id", Integer, primary_key=True, autoincrement=False),
|
||||
Column("_data", String(50)),
|
||||
)
|
||||
|
||||
def _table_index_fixture(self, schema=None):
|
||||
table = self._simple_fixture(schema=schema)
|
||||
idx = Index("test_index", table.c.data)
|
||||
return table, idx
|
||||
|
||||
def _simple_roundtrip(self, table):
|
||||
with config.db.begin() as conn:
|
||||
conn.execute(table.insert().values((1, "some data")))
|
||||
result = conn.execute(table.select())
|
||||
eq_(result.first(), (1, "some data"))
|
||||
|
||||
@requirements.create_table
|
||||
@util.provide_metadata
|
||||
def test_create_table(self):
|
||||
table = self._simple_fixture()
|
||||
table.create(config.db, checkfirst=False)
|
||||
self._simple_roundtrip(table)
|
||||
|
||||
@requirements.create_table
|
||||
@requirements.schemas
|
||||
@util.provide_metadata
|
||||
def test_create_table_schema(self):
|
||||
table = self._simple_fixture(schema=config.test_schema)
|
||||
table.create(config.db, checkfirst=False)
|
||||
self._simple_roundtrip(table)
|
||||
|
||||
@requirements.drop_table
|
||||
@util.provide_metadata
|
||||
def test_drop_table(self):
|
||||
table = self._simple_fixture()
|
||||
table.create(config.db, checkfirst=False)
|
||||
table.drop(config.db, checkfirst=False)
|
||||
|
||||
@requirements.create_table
|
||||
@util.provide_metadata
|
||||
def test_underscore_names(self):
|
||||
table = self._underscore_fixture()
|
||||
table.create(config.db, checkfirst=False)
|
||||
self._simple_roundtrip(table)
|
||||
|
||||
@requirements.comment_reflection
|
||||
@util.provide_metadata
|
||||
def test_add_table_comment(self, connection):
|
||||
table = self._simple_fixture()
|
||||
table.create(connection, checkfirst=False)
|
||||
table.comment = "a comment"
|
||||
connection.execute(schema.SetTableComment(table))
|
||||
eq_(
|
||||
inspect(connection).get_table_comment("test_table"),
|
||||
{"text": "a comment"},
|
||||
)
|
||||
|
||||
@requirements.comment_reflection
|
||||
@util.provide_metadata
|
||||
def test_drop_table_comment(self, connection):
|
||||
table = self._simple_fixture()
|
||||
table.create(connection, checkfirst=False)
|
||||
table.comment = "a comment"
|
||||
connection.execute(schema.SetTableComment(table))
|
||||
connection.execute(schema.DropTableComment(table))
|
||||
eq_(
|
||||
inspect(connection).get_table_comment("test_table"), {"text": None}
|
||||
)
|
||||
|
||||
@requirements.table_ddl_if_exists
|
||||
@util.provide_metadata
|
||||
def test_create_table_if_not_exists(self, connection):
|
||||
table = self._simple_fixture()
|
||||
|
||||
connection.execute(schema.CreateTable(table, if_not_exists=True))
|
||||
|
||||
is_true(inspect(connection).has_table("test_table"))
|
||||
connection.execute(schema.CreateTable(table, if_not_exists=True))
|
||||
|
||||
@requirements.index_ddl_if_exists
|
||||
@util.provide_metadata
|
||||
def test_create_index_if_not_exists(self, connection):
|
||||
table, idx = self._table_index_fixture()
|
||||
|
||||
connection.execute(schema.CreateTable(table, if_not_exists=True))
|
||||
is_true(inspect(connection).has_table("test_table"))
|
||||
is_false(
|
||||
"test_index"
|
||||
in [
|
||||
ix["name"]
|
||||
for ix in inspect(connection).get_indexes("test_table")
|
||||
]
|
||||
)
|
||||
|
||||
connection.execute(schema.CreateIndex(idx, if_not_exists=True))
|
||||
|
||||
is_true(
|
||||
"test_index"
|
||||
in [
|
||||
ix["name"]
|
||||
for ix in inspect(connection).get_indexes("test_table")
|
||||
]
|
||||
)
|
||||
|
||||
connection.execute(schema.CreateIndex(idx, if_not_exists=True))
|
||||
|
||||
@requirements.table_ddl_if_exists
|
||||
@util.provide_metadata
|
||||
def test_drop_table_if_exists(self, connection):
|
||||
table = self._simple_fixture()
|
||||
|
||||
table.create(connection)
|
||||
|
||||
is_true(inspect(connection).has_table("test_table"))
|
||||
|
||||
connection.execute(schema.DropTable(table, if_exists=True))
|
||||
|
||||
is_false(inspect(connection).has_table("test_table"))
|
||||
|
||||
connection.execute(schema.DropTable(table, if_exists=True))
|
||||
|
||||
@requirements.index_ddl_if_exists
|
||||
@util.provide_metadata
|
||||
def test_drop_index_if_exists(self, connection):
|
||||
table, idx = self._table_index_fixture()
|
||||
|
||||
table.create(connection)
|
||||
|
||||
is_true(
|
||||
"test_index"
|
||||
in [
|
||||
ix["name"]
|
||||
for ix in inspect(connection).get_indexes("test_table")
|
||||
]
|
||||
)
|
||||
|
||||
connection.execute(schema.DropIndex(idx, if_exists=True))
|
||||
|
||||
is_false(
|
||||
"test_index"
|
||||
in [
|
||||
ix["name"]
|
||||
for ix in inspect(connection).get_indexes("test_table")
|
||||
]
|
||||
)
|
||||
|
||||
connection.execute(schema.DropIndex(idx, if_exists=True))
|
||||
|
||||
|
||||
class FutureTableDDLTest(fixtures.FutureEngineMixin, TableDDLTest):
|
||||
pass
|
||||
|
||||
|
||||
class LongNameBlowoutTest(fixtures.TestBase):
|
||||
"""test the creation of a variety of DDL structures and ensure
|
||||
label length limits pass on backends
|
||||
|
||||
"""
|
||||
|
||||
__sparse_driver_backend__ = True
|
||||
|
||||
def fk(self, metadata, connection):
|
||||
convention = {
|
||||
"fk": "foreign_key_%(table_name)s_"
|
||||
"%(column_0_N_name)s_"
|
||||
"%(referred_table_name)s_"
|
||||
+ (
|
||||
"_".join(
|
||||
"".join(random.choice("abcdef") for j in range(20))
|
||||
for i in range(10)
|
||||
)
|
||||
),
|
||||
}
|
||||
metadata.naming_convention = convention
|
||||
|
||||
Table(
|
||||
"a_things_with_stuff",
|
||||
metadata,
|
||||
Column("id_long_column_name", Integer, primary_key=True),
|
||||
test_needs_fk=True,
|
||||
)
|
||||
|
||||
cons = ForeignKeyConstraint(
|
||||
["aid"], ["a_things_with_stuff.id_long_column_name"]
|
||||
)
|
||||
Table(
|
||||
"b_related_things_of_value",
|
||||
metadata,
|
||||
Column(
|
||||
"aid",
|
||||
),
|
||||
cons,
|
||||
test_needs_fk=True,
|
||||
)
|
||||
actual_name = cons.name
|
||||
|
||||
metadata.create_all(connection)
|
||||
|
||||
if testing.requires.foreign_key_constraint_name_reflection.enabled:
|
||||
insp = inspect(connection)
|
||||
fks = insp.get_foreign_keys("b_related_things_of_value")
|
||||
reflected_name = fks[0]["name"]
|
||||
|
||||
return actual_name, reflected_name
|
||||
else:
|
||||
return actual_name, None
|
||||
|
||||
def pk(self, metadata, connection):
|
||||
convention = {
|
||||
"pk": "primary_key_%(table_name)s_"
|
||||
"%(column_0_N_name)s"
|
||||
+ (
|
||||
"_".join(
|
||||
"".join(random.choice("abcdef") for j in range(30))
|
||||
for i in range(10)
|
||||
)
|
||||
),
|
||||
}
|
||||
metadata.naming_convention = convention
|
||||
|
||||
a = Table(
|
||||
"a_things_with_stuff",
|
||||
metadata,
|
||||
Column("id_long_column_name", Integer, primary_key=True),
|
||||
Column("id_another_long_name", Integer, primary_key=True),
|
||||
)
|
||||
cons = a.primary_key
|
||||
actual_name = cons.name
|
||||
|
||||
metadata.create_all(connection)
|
||||
insp = inspect(connection)
|
||||
pk = insp.get_pk_constraint("a_things_with_stuff")
|
||||
reflected_name = pk["name"]
|
||||
return actual_name, reflected_name
|
||||
|
||||
def ix(self, metadata, connection):
|
||||
convention = {
|
||||
"ix": "index_%(table_name)s_"
|
||||
"%(column_0_N_name)s"
|
||||
+ (
|
||||
"_".join(
|
||||
"".join(random.choice("abcdef") for j in range(30))
|
||||
for i in range(10)
|
||||
)
|
||||
),
|
||||
}
|
||||
metadata.naming_convention = convention
|
||||
|
||||
a = Table(
|
||||
"a_things_with_stuff",
|
||||
metadata,
|
||||
Column("id_long_column_name", Integer, primary_key=True),
|
||||
Column("id_another_long_name", Integer),
|
||||
)
|
||||
cons = Index(None, a.c.id_long_column_name, a.c.id_another_long_name)
|
||||
actual_name = cons.name
|
||||
|
||||
metadata.create_all(connection)
|
||||
insp = inspect(connection)
|
||||
ix = insp.get_indexes("a_things_with_stuff")
|
||||
reflected_name = ix[0]["name"]
|
||||
return actual_name, reflected_name
|
||||
|
||||
def uq(self, metadata, connection):
|
||||
convention = {
|
||||
"uq": "unique_constraint_%(table_name)s_"
|
||||
"%(column_0_N_name)s"
|
||||
+ (
|
||||
"_".join(
|
||||
"".join(random.choice("abcdef") for j in range(30))
|
||||
for i in range(10)
|
||||
)
|
||||
),
|
||||
}
|
||||
metadata.naming_convention = convention
|
||||
|
||||
cons = UniqueConstraint("id_long_column_name", "id_another_long_name")
|
||||
Table(
|
||||
"a_things_with_stuff",
|
||||
metadata,
|
||||
Column("id_long_column_name", Integer, primary_key=True),
|
||||
Column("id_another_long_name", Integer),
|
||||
cons,
|
||||
)
|
||||
actual_name = cons.name
|
||||
|
||||
metadata.create_all(connection)
|
||||
insp = inspect(connection)
|
||||
uq = insp.get_unique_constraints("a_things_with_stuff")
|
||||
reflected_name = uq[0]["name"]
|
||||
return actual_name, reflected_name
|
||||
|
||||
def ck(self, metadata, connection):
|
||||
convention = {
|
||||
"ck": "check_constraint_%(table_name)s"
|
||||
+ (
|
||||
"_".join(
|
||||
"".join(random.choice("abcdef") for j in range(30))
|
||||
for i in range(10)
|
||||
)
|
||||
),
|
||||
}
|
||||
metadata.naming_convention = convention
|
||||
|
||||
cons = CheckConstraint("some_long_column_name > 5")
|
||||
Table(
|
||||
"a_things_with_stuff",
|
||||
metadata,
|
||||
Column("id_long_column_name", Integer, primary_key=True),
|
||||
Column("some_long_column_name", Integer),
|
||||
cons,
|
||||
)
|
||||
actual_name = cons.name
|
||||
|
||||
metadata.create_all(connection)
|
||||
insp = inspect(connection)
|
||||
ck = insp.get_check_constraints("a_things_with_stuff")
|
||||
reflected_name = ck[0]["name"]
|
||||
return actual_name, reflected_name
|
||||
|
||||
@testing.combinations(
|
||||
("fk",),
|
||||
("pk",),
|
||||
("ix",),
|
||||
("ck", testing.requires.check_constraint_reflection.as_skips()),
|
||||
("uq", testing.requires.unique_constraint_reflection.as_skips()),
|
||||
argnames="type_",
|
||||
)
|
||||
def test_long_convention_name(self, type_, metadata, connection):
|
||||
actual_name, reflected_name = getattr(self, type_)(
|
||||
metadata, connection
|
||||
)
|
||||
|
||||
assert len(actual_name) > 255
|
||||
|
||||
if reflected_name is not None:
|
||||
overlap = actual_name[0 : len(reflected_name)]
|
||||
if len(overlap) < len(actual_name):
|
||||
eq_(overlap[0:-5], reflected_name[0 : len(overlap) - 5])
|
||||
else:
|
||||
eq_(overlap, reflected_name)
|
||||
|
||||
|
||||
__all__ = ("TableDDLTest", "FutureTableDDLTest", "LongNameBlowoutTest")
|
||||
@@ -0,0 +1,153 @@
|
||||
# testing/suite/test_deprecations.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 .. import fixtures
|
||||
from ..assertions import eq_
|
||||
from ..schema import Column
|
||||
from ..schema import Table
|
||||
from ... import Integer
|
||||
from ... import select
|
||||
from ... import testing
|
||||
from ... import union
|
||||
|
||||
|
||||
class DeprecatedCompoundSelectTest(fixtures.TablesTest):
|
||||
__backend__ = True
|
||||
|
||||
@classmethod
|
||||
def define_tables(cls, metadata):
|
||||
Table(
|
||||
"some_table",
|
||||
metadata,
|
||||
Column("id", Integer, primary_key=True),
|
||||
Column("x", Integer),
|
||||
Column("y", Integer),
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def insert_data(cls, connection):
|
||||
connection.execute(
|
||||
cls.tables.some_table.insert(),
|
||||
[
|
||||
{"id": 1, "x": 1, "y": 2},
|
||||
{"id": 2, "x": 2, "y": 3},
|
||||
{"id": 3, "x": 3, "y": 4},
|
||||
{"id": 4, "x": 4, "y": 5},
|
||||
],
|
||||
)
|
||||
|
||||
def _assert_result(self, conn, select, result, params=()):
|
||||
eq_(conn.execute(select, params).fetchall(), result)
|
||||
|
||||
def test_plain_union(self, connection):
|
||||
table = self.tables.some_table
|
||||
s1 = select(table).where(table.c.id == 2)
|
||||
s2 = select(table).where(table.c.id == 3)
|
||||
|
||||
u1 = union(s1, s2)
|
||||
with testing.expect_deprecated(
|
||||
"The SelectBase.c and SelectBase.columns "
|
||||
"attributes are deprecated"
|
||||
):
|
||||
self._assert_result(
|
||||
connection, u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]
|
||||
)
|
||||
|
||||
# note we've had to remove one use case entirely, which is this
|
||||
# one. the Select gets its FROMS from the WHERE clause and the
|
||||
# columns clause, but not the ORDER BY, which means the old ".c" system
|
||||
# allowed you to "order_by(s.c.foo)" to get an unnamed column in the
|
||||
# ORDER BY without adding the SELECT into the FROM and breaking the
|
||||
# query. Users will have to adjust for this use case if they were doing
|
||||
# it before.
|
||||
def _dont_test_select_from_plain_union(self, connection):
|
||||
table = self.tables.some_table
|
||||
s1 = select(table).where(table.c.id == 2)
|
||||
s2 = select(table).where(table.c.id == 3)
|
||||
|
||||
u1 = union(s1, s2).alias().select()
|
||||
with testing.expect_deprecated(
|
||||
"The SelectBase.c and SelectBase.columns "
|
||||
"attributes are deprecated"
|
||||
):
|
||||
self._assert_result(
|
||||
connection, u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]
|
||||
)
|
||||
|
||||
@testing.requires.order_by_col_from_union
|
||||
@testing.requires.parens_in_union_contained_select_w_limit_offset
|
||||
def test_limit_offset_selectable_in_unions(self, connection):
|
||||
table = self.tables.some_table
|
||||
s1 = select(table).where(table.c.id == 2).limit(1).order_by(table.c.id)
|
||||
s2 = select(table).where(table.c.id == 3).limit(1).order_by(table.c.id)
|
||||
|
||||
u1 = union(s1, s2).limit(2)
|
||||
with testing.expect_deprecated(
|
||||
"The SelectBase.c and SelectBase.columns "
|
||||
"attributes are deprecated"
|
||||
):
|
||||
self._assert_result(
|
||||
connection, u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]
|
||||
)
|
||||
|
||||
@testing.requires.parens_in_union_contained_select_wo_limit_offset
|
||||
def test_order_by_selectable_in_unions(self, connection):
|
||||
table = self.tables.some_table
|
||||
s1 = select(table).where(table.c.id == 2).order_by(table.c.id)
|
||||
s2 = select(table).where(table.c.id == 3).order_by(table.c.id)
|
||||
|
||||
u1 = union(s1, s2).limit(2)
|
||||
with testing.expect_deprecated(
|
||||
"The SelectBase.c and SelectBase.columns "
|
||||
"attributes are deprecated"
|
||||
):
|
||||
self._assert_result(
|
||||
connection, u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]
|
||||
)
|
||||
|
||||
def test_distinct_selectable_in_unions(self, connection):
|
||||
table = self.tables.some_table
|
||||
s1 = select(table).where(table.c.id == 2).distinct()
|
||||
s2 = select(table).where(table.c.id == 3).distinct()
|
||||
|
||||
u1 = union(s1, s2).limit(2)
|
||||
with testing.expect_deprecated(
|
||||
"The SelectBase.c and SelectBase.columns "
|
||||
"attributes are deprecated"
|
||||
):
|
||||
self._assert_result(
|
||||
connection, u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]
|
||||
)
|
||||
|
||||
def test_limit_offset_aliased_selectable_in_unions(self, connection):
|
||||
table = self.tables.some_table
|
||||
s1 = (
|
||||
select(table)
|
||||
.where(table.c.id == 2)
|
||||
.limit(1)
|
||||
.order_by(table.c.id)
|
||||
.alias()
|
||||
.select()
|
||||
)
|
||||
s2 = (
|
||||
select(table)
|
||||
.where(table.c.id == 3)
|
||||
.limit(1)
|
||||
.order_by(table.c.id)
|
||||
.alias()
|
||||
.select()
|
||||
)
|
||||
|
||||
u1 = union(s1, s2).limit(2)
|
||||
with testing.expect_deprecated(
|
||||
"The SelectBase.c and SelectBase.columns "
|
||||
"attributes are deprecated"
|
||||
):
|
||||
self._assert_result(
|
||||
connection, u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]
|
||||
)
|
||||
776
venv/Lib/site-packages/sqlalchemy/testing/suite/test_dialect.py
Normal file
776
venv/Lib/site-packages/sqlalchemy/testing/suite/test_dialect.py
Normal file
@@ -0,0 +1,776 @@
|
||||
# testing/suite/test_dialect.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
|
||||
|
||||
|
||||
import importlib
|
||||
|
||||
from . import testing
|
||||
from .. import assert_raises
|
||||
from .. import config
|
||||
from .. import engines
|
||||
from .. import eq_
|
||||
from .. import fixtures
|
||||
from .. import is_not_none
|
||||
from .. import is_true
|
||||
from .. import mock
|
||||
from .. import ne_
|
||||
from .. import provide_metadata
|
||||
from ..assertions import expect_raises
|
||||
from ..assertions import expect_raises_message
|
||||
from ..config import requirements
|
||||
from ..provision import set_default_schema_on_connection
|
||||
from ..schema import Column
|
||||
from ..schema import Table
|
||||
from ... import bindparam
|
||||
from ... import dialects
|
||||
from ... import event
|
||||
from ... import exc
|
||||
from ... import Integer
|
||||
from ... import literal_column
|
||||
from ... import select
|
||||
from ... import String
|
||||
from ...sql.compiler import Compiled
|
||||
from ...util import inspect_getfullargspec
|
||||
|
||||
|
||||
class PingTest(fixtures.TestBase):
|
||||
__backend__ = True
|
||||
|
||||
def test_do_ping(self):
|
||||
with testing.db.connect() as conn:
|
||||
is_true(
|
||||
testing.db.dialect.do_ping(conn.connection.dbapi_connection)
|
||||
)
|
||||
|
||||
|
||||
class ArgSignatureTest(fixtures.TestBase):
|
||||
"""test that all visit_XYZ() in :class:`_sql.Compiler` subclasses have
|
||||
``**kw``, for #8988.
|
||||
|
||||
This test uses runtime code inspection. Does not need to be a
|
||||
``__backend__`` test as it only needs to run once provided all target
|
||||
dialects have been imported.
|
||||
|
||||
For third party dialects, the suite would be run with that third
|
||||
party as a "--dburi", which means its compiler classes will have been
|
||||
imported by the time this test runs.
|
||||
|
||||
"""
|
||||
|
||||
def _all_subclasses(): # type: ignore # noqa
|
||||
for d in dialects.__all__:
|
||||
if not d.startswith("_"):
|
||||
importlib.import_module("sqlalchemy.dialects.%s" % d)
|
||||
|
||||
stack = [Compiled]
|
||||
|
||||
while stack:
|
||||
cls = stack.pop(0)
|
||||
stack.extend(cls.__subclasses__())
|
||||
yield cls
|
||||
|
||||
@testing.fixture(params=list(_all_subclasses()))
|
||||
def all_subclasses(self, request):
|
||||
yield request.param
|
||||
|
||||
def test_all_visit_methods_accept_kw(self, all_subclasses):
|
||||
cls = all_subclasses
|
||||
|
||||
for k in cls.__dict__:
|
||||
if k.startswith("visit_"):
|
||||
meth = getattr(cls, k)
|
||||
|
||||
insp = inspect_getfullargspec(meth)
|
||||
is_not_none(
|
||||
insp.varkw,
|
||||
f"Compiler visit method {cls.__name__}.{k}() does "
|
||||
"not accommodate for **kw in its argument signature",
|
||||
)
|
||||
|
||||
|
||||
class ExceptionTest(fixtures.TablesTest):
|
||||
"""Test basic exception wrapping.
|
||||
|
||||
DBAPIs vary a lot in exception behavior so to actually anticipate
|
||||
specific exceptions from real round trips, we need to be conservative.
|
||||
|
||||
"""
|
||||
|
||||
run_deletes = "each"
|
||||
|
||||
__backend__ = True
|
||||
|
||||
@classmethod
|
||||
def define_tables(cls, metadata):
|
||||
Table(
|
||||
"manual_pk",
|
||||
metadata,
|
||||
Column("id", Integer, primary_key=True, autoincrement=False),
|
||||
Column("data", String(50)),
|
||||
)
|
||||
|
||||
@requirements.duplicate_key_raises_integrity_error
|
||||
def test_integrity_error(self):
|
||||
with config.db.connect() as conn:
|
||||
trans = conn.begin()
|
||||
conn.execute(
|
||||
self.tables.manual_pk.insert(), {"id": 1, "data": "d1"}
|
||||
)
|
||||
|
||||
assert_raises(
|
||||
exc.IntegrityError,
|
||||
conn.execute,
|
||||
self.tables.manual_pk.insert(),
|
||||
{"id": 1, "data": "d1"},
|
||||
)
|
||||
|
||||
trans.rollback()
|
||||
|
||||
def test_exception_with_non_ascii(self):
|
||||
with config.db.connect() as conn:
|
||||
try:
|
||||
# try to create an error message that likely has non-ascii
|
||||
# characters in the DBAPI's message string. unfortunately
|
||||
# there's no way to make this happen with some drivers like
|
||||
# mysqlclient, pymysql. this at least does produce a non-
|
||||
# ascii error message for cx_oracle, psycopg2
|
||||
conn.execute(select(literal_column("méil")))
|
||||
assert False
|
||||
except exc.DBAPIError as err:
|
||||
err_str = str(err)
|
||||
|
||||
assert str(err.orig) in str(err)
|
||||
|
||||
assert isinstance(err_str, str)
|
||||
|
||||
|
||||
class IsolationLevelTest(fixtures.TestBase):
|
||||
__backend__ = True
|
||||
|
||||
__requires__ = ("isolation_level",)
|
||||
|
||||
def _get_non_default_isolation_level(self):
|
||||
levels = requirements.get_isolation_levels(config)
|
||||
|
||||
default = levels["default"]
|
||||
supported = levels["supported"]
|
||||
|
||||
s = set(supported).difference(["AUTOCOMMIT", default])
|
||||
if s:
|
||||
return s.pop()
|
||||
else:
|
||||
config.skip_test("no non-default isolation level available")
|
||||
|
||||
def test_default_isolation_level(self):
|
||||
eq_(
|
||||
config.db.dialect.default_isolation_level,
|
||||
requirements.get_isolation_levels(config)["default"],
|
||||
)
|
||||
|
||||
def test_non_default_isolation_level(self):
|
||||
non_default = self._get_non_default_isolation_level()
|
||||
|
||||
with config.db.connect() as conn:
|
||||
existing = conn.get_isolation_level()
|
||||
|
||||
ne_(existing, non_default)
|
||||
|
||||
conn.execution_options(isolation_level=non_default)
|
||||
|
||||
eq_(conn.get_isolation_level(), non_default)
|
||||
|
||||
conn.dialect.reset_isolation_level(
|
||||
conn.connection.dbapi_connection
|
||||
)
|
||||
|
||||
eq_(conn.get_isolation_level(), existing)
|
||||
|
||||
def test_all_levels(self):
|
||||
levels = requirements.get_isolation_levels(config)
|
||||
|
||||
all_levels = levels["supported"]
|
||||
|
||||
for level in set(all_levels).difference(["AUTOCOMMIT"]):
|
||||
with config.db.connect() as conn:
|
||||
conn.execution_options(isolation_level=level)
|
||||
|
||||
eq_(conn.get_isolation_level(), level)
|
||||
|
||||
trans = conn.begin()
|
||||
trans.rollback()
|
||||
|
||||
eq_(conn.get_isolation_level(), level)
|
||||
|
||||
with config.db.connect() as conn:
|
||||
eq_(
|
||||
conn.get_isolation_level(),
|
||||
levels["default"],
|
||||
)
|
||||
|
||||
@testing.requires.get_isolation_level_values
|
||||
def test_invalid_level_execution_option(self, connection_no_trans):
|
||||
"""test for the new get_isolation_level_values() method"""
|
||||
|
||||
connection = connection_no_trans
|
||||
with expect_raises_message(
|
||||
exc.ArgumentError,
|
||||
"Invalid value '%s' for isolation_level. "
|
||||
"Valid isolation levels for '%s' are %s"
|
||||
% (
|
||||
"FOO",
|
||||
connection.dialect.name,
|
||||
", ".join(
|
||||
requirements.get_isolation_levels(config)["supported"]
|
||||
),
|
||||
),
|
||||
):
|
||||
connection.execution_options(isolation_level="FOO")
|
||||
|
||||
@testing.requires.get_isolation_level_values
|
||||
@testing.requires.dialect_level_isolation_level_param
|
||||
def test_invalid_level_engine_param(self, testing_engine):
|
||||
"""test for the new get_isolation_level_values() method
|
||||
and support for the dialect-level 'isolation_level' parameter.
|
||||
|
||||
"""
|
||||
|
||||
eng = testing_engine(options=dict(isolation_level="FOO"))
|
||||
with expect_raises_message(
|
||||
exc.ArgumentError,
|
||||
"Invalid value '%s' for isolation_level. "
|
||||
"Valid isolation levels for '%s' are %s"
|
||||
% (
|
||||
"FOO",
|
||||
eng.dialect.name,
|
||||
", ".join(
|
||||
requirements.get_isolation_levels(config)["supported"]
|
||||
),
|
||||
),
|
||||
):
|
||||
eng.connect()
|
||||
|
||||
@testing.requires.independent_readonly_connections
|
||||
def test_dialect_user_setting_is_restored(self, testing_engine):
|
||||
levels = requirements.get_isolation_levels(config)
|
||||
default = levels["default"]
|
||||
supported = (
|
||||
sorted(
|
||||
set(levels["supported"]).difference([default, "AUTOCOMMIT"])
|
||||
)
|
||||
)[0]
|
||||
|
||||
e = testing_engine(options={"isolation_level": supported})
|
||||
|
||||
with e.connect() as conn:
|
||||
eq_(conn.get_isolation_level(), supported)
|
||||
|
||||
with e.connect() as conn:
|
||||
conn.execution_options(isolation_level=default)
|
||||
eq_(conn.get_isolation_level(), default)
|
||||
|
||||
with e.connect() as conn:
|
||||
eq_(conn.get_isolation_level(), supported)
|
||||
|
||||
|
||||
class AutocommitIsolationTest(fixtures.TablesTest):
|
||||
run_deletes = "each"
|
||||
|
||||
__requires__ = ("autocommit",)
|
||||
|
||||
__backend__ = True
|
||||
|
||||
@classmethod
|
||||
def define_tables(cls, metadata):
|
||||
Table(
|
||||
"some_table",
|
||||
metadata,
|
||||
Column("id", Integer, primary_key=True, autoincrement=False),
|
||||
Column("data", String(50)),
|
||||
test_needs_acid=True,
|
||||
)
|
||||
|
||||
def _test_conn_autocommits(self, conn, autocommit, ensure_table=False):
|
||||
if ensure_table:
|
||||
self.tables.some_table.create(conn, checkfirst=True)
|
||||
conn.commit()
|
||||
|
||||
trans = conn.begin()
|
||||
conn.execute(
|
||||
self.tables.some_table.insert(), {"id": 1, "data": "some data"}
|
||||
)
|
||||
trans.rollback()
|
||||
|
||||
eq_(
|
||||
conn.scalar(select(self.tables.some_table.c.id)),
|
||||
1 if autocommit else None,
|
||||
)
|
||||
conn.rollback()
|
||||
|
||||
with conn.begin():
|
||||
conn.execute(self.tables.some_table.delete())
|
||||
|
||||
def test_autocommit_on(self, connection_no_trans):
|
||||
conn = connection_no_trans
|
||||
c2 = conn.execution_options(isolation_level="AUTOCOMMIT")
|
||||
self._test_conn_autocommits(c2, True)
|
||||
|
||||
c2.dialect.reset_isolation_level(c2.connection.dbapi_connection)
|
||||
|
||||
self._test_conn_autocommits(conn, False)
|
||||
|
||||
def test_autocommit_off(self, connection_no_trans):
|
||||
conn = connection_no_trans
|
||||
self._test_conn_autocommits(conn, False)
|
||||
|
||||
def test_turn_autocommit_off_via_default_iso_level(
|
||||
self, connection_no_trans
|
||||
):
|
||||
conn = connection_no_trans
|
||||
conn = conn.execution_options(isolation_level="AUTOCOMMIT")
|
||||
self._test_conn_autocommits(conn, True)
|
||||
|
||||
conn.execution_options(
|
||||
isolation_level=requirements.get_isolation_levels(config)[
|
||||
"default"
|
||||
]
|
||||
)
|
||||
self._test_conn_autocommits(conn, False)
|
||||
|
||||
@testing.requires.skip_autocommit_rollback
|
||||
@testing.variation("autocommit_setting", ["false", "engine", "option"])
|
||||
@testing.variation("block_rollback", [True, False])
|
||||
def test_autocommit_block(
|
||||
self, testing_engine, autocommit_setting, block_rollback
|
||||
):
|
||||
kw = {}
|
||||
if bool(block_rollback):
|
||||
kw["skip_autocommit_rollback"] = True
|
||||
if autocommit_setting.engine:
|
||||
kw["isolation_level"] = "AUTOCOMMIT"
|
||||
|
||||
engine = testing_engine(options=kw)
|
||||
|
||||
conn = engine.connect()
|
||||
if autocommit_setting.option:
|
||||
conn.execution_options(isolation_level="AUTOCOMMIT")
|
||||
self._test_conn_autocommits(
|
||||
conn,
|
||||
autocommit_setting.engine or autocommit_setting.option,
|
||||
ensure_table=True,
|
||||
)
|
||||
with mock.patch.object(
|
||||
conn.connection, "rollback", wraps=conn.connection.rollback
|
||||
) as check_rollback:
|
||||
conn.close()
|
||||
if autocommit_setting.false or not block_rollback:
|
||||
eq_(check_rollback.mock_calls, [mock.call()])
|
||||
else:
|
||||
eq_(check_rollback.mock_calls, [])
|
||||
|
||||
@testing.requires.independent_readonly_connections
|
||||
@testing.variation("use_dialect_setting", [True, False])
|
||||
def test_dialect_autocommit_is_restored(
|
||||
self, testing_engine, use_dialect_setting
|
||||
):
|
||||
"""test #10147"""
|
||||
|
||||
if use_dialect_setting:
|
||||
e = testing_engine(options={"isolation_level": "AUTOCOMMIT"})
|
||||
else:
|
||||
e = testing_engine().execution_options(
|
||||
isolation_level="AUTOCOMMIT"
|
||||
)
|
||||
|
||||
levels = requirements.get_isolation_levels(config)
|
||||
|
||||
default = levels["default"]
|
||||
|
||||
with e.connect() as conn:
|
||||
self._test_conn_autocommits(conn, True)
|
||||
|
||||
with e.connect() as conn:
|
||||
conn.execution_options(isolation_level=default)
|
||||
self._test_conn_autocommits(conn, False)
|
||||
|
||||
with e.connect() as conn:
|
||||
self._test_conn_autocommits(conn, True)
|
||||
|
||||
|
||||
class EscapingTest(fixtures.TestBase):
|
||||
@provide_metadata
|
||||
def test_percent_sign_round_trip(self):
|
||||
"""test that the DBAPI accommodates for escaped / nonescaped
|
||||
percent signs in a way that matches the compiler
|
||||
|
||||
"""
|
||||
m = self.metadata
|
||||
t = Table("t", m, Column("data", String(50)))
|
||||
t.create(config.db)
|
||||
with config.db.begin() as conn:
|
||||
conn.execute(t.insert(), dict(data="some % value"))
|
||||
conn.execute(t.insert(), dict(data="some %% other value"))
|
||||
|
||||
eq_(
|
||||
conn.scalar(
|
||||
select(t.c.data).where(
|
||||
t.c.data == literal_column("'some % value'")
|
||||
)
|
||||
),
|
||||
"some % value",
|
||||
)
|
||||
|
||||
eq_(
|
||||
conn.scalar(
|
||||
select(t.c.data).where(
|
||||
t.c.data == literal_column("'some %% other value'")
|
||||
)
|
||||
),
|
||||
"some %% other value",
|
||||
)
|
||||
|
||||
|
||||
class WeCanSetDefaultSchemaWEventsTest(fixtures.TestBase):
|
||||
__backend__ = True
|
||||
|
||||
__requires__ = ("default_schema_name_switch",)
|
||||
|
||||
def test_control_case(self):
|
||||
default_schema_name = config.db.dialect.default_schema_name
|
||||
|
||||
eng = engines.testing_engine()
|
||||
with eng.connect():
|
||||
pass
|
||||
|
||||
eq_(eng.dialect.default_schema_name, default_schema_name)
|
||||
|
||||
def test_wont_work_wo_insert(self):
|
||||
default_schema_name = config.db.dialect.default_schema_name
|
||||
|
||||
eng = engines.testing_engine()
|
||||
|
||||
@event.listens_for(eng, "connect")
|
||||
def on_connect(dbapi_connection, connection_record):
|
||||
set_default_schema_on_connection(
|
||||
config, dbapi_connection, config.test_schema
|
||||
)
|
||||
|
||||
with eng.connect() as conn:
|
||||
what_it_should_be = eng.dialect._get_default_schema_name(conn)
|
||||
eq_(what_it_should_be, config.test_schema)
|
||||
|
||||
eq_(eng.dialect.default_schema_name, default_schema_name)
|
||||
|
||||
def test_schema_change_on_connect(self):
|
||||
eng = engines.testing_engine()
|
||||
|
||||
@event.listens_for(eng, "connect", insert=True)
|
||||
def on_connect(dbapi_connection, connection_record):
|
||||
set_default_schema_on_connection(
|
||||
config, dbapi_connection, config.test_schema
|
||||
)
|
||||
|
||||
with eng.connect() as conn:
|
||||
what_it_should_be = eng.dialect._get_default_schema_name(conn)
|
||||
eq_(what_it_should_be, config.test_schema)
|
||||
|
||||
eq_(eng.dialect.default_schema_name, config.test_schema)
|
||||
|
||||
def test_schema_change_works_w_transactions(self):
|
||||
eng = engines.testing_engine()
|
||||
|
||||
@event.listens_for(eng, "connect", insert=True)
|
||||
def on_connect(dbapi_connection, *arg):
|
||||
set_default_schema_on_connection(
|
||||
config, dbapi_connection, config.test_schema
|
||||
)
|
||||
|
||||
with eng.connect() as conn:
|
||||
trans = conn.begin()
|
||||
what_it_should_be = eng.dialect._get_default_schema_name(conn)
|
||||
eq_(what_it_should_be, config.test_schema)
|
||||
trans.rollback()
|
||||
|
||||
what_it_should_be = eng.dialect._get_default_schema_name(conn)
|
||||
eq_(what_it_should_be, config.test_schema)
|
||||
|
||||
eq_(eng.dialect.default_schema_name, config.test_schema)
|
||||
|
||||
|
||||
class FutureWeCanSetDefaultSchemaWEventsTest(
|
||||
fixtures.FutureEngineMixin, WeCanSetDefaultSchemaWEventsTest
|
||||
):
|
||||
pass
|
||||
|
||||
|
||||
class DifficultParametersTest(fixtures.TestBase):
|
||||
__backend__ = True
|
||||
|
||||
tough_parameters = testing.combinations(
|
||||
("boring",),
|
||||
("per cent",),
|
||||
("per % cent",),
|
||||
("%percent",),
|
||||
("par(ens)",),
|
||||
("percent%(ens)yah",),
|
||||
("col:ons",),
|
||||
("_starts_with_underscore",),
|
||||
("dot.s",),
|
||||
("more :: %colons%",),
|
||||
("_name",),
|
||||
("___name",),
|
||||
("[BracketsAndCase]",),
|
||||
("42numbers",),
|
||||
("percent%signs",),
|
||||
("has spaces",),
|
||||
("/slashes/",),
|
||||
("more/slashes",),
|
||||
("q?marks",),
|
||||
("1param",),
|
||||
("1col:on",),
|
||||
argnames="paramname",
|
||||
)
|
||||
|
||||
@tough_parameters
|
||||
@config.requirements.unusual_column_name_characters
|
||||
def test_round_trip_same_named_column(
|
||||
self, paramname, connection, metadata
|
||||
):
|
||||
name = paramname
|
||||
|
||||
t = Table(
|
||||
"t",
|
||||
metadata,
|
||||
Column("id", Integer, primary_key=True),
|
||||
Column(name, String(50), nullable=False),
|
||||
)
|
||||
|
||||
# table is created
|
||||
t.create(connection)
|
||||
|
||||
# automatic param generated by insert
|
||||
connection.execute(t.insert().values({"id": 1, name: "some name"}))
|
||||
|
||||
# automatic param generated by criteria, plus selecting the column
|
||||
stmt = select(t.c[name]).where(t.c[name] == "some name")
|
||||
|
||||
eq_(connection.scalar(stmt), "some name")
|
||||
|
||||
# use the name in a param explicitly
|
||||
stmt = select(t.c[name]).where(t.c[name] == bindparam(name))
|
||||
|
||||
row = connection.execute(stmt, {name: "some name"}).first()
|
||||
|
||||
# name works as the key from cursor.description
|
||||
eq_(row._mapping[name], "some name")
|
||||
|
||||
# use expanding IN
|
||||
stmt = select(t.c[name]).where(
|
||||
t.c[name].in_(["some name", "some other_name"])
|
||||
)
|
||||
|
||||
connection.execute(stmt).first()
|
||||
|
||||
@testing.fixture
|
||||
def multirow_fixture(self, metadata, connection):
|
||||
mytable = Table(
|
||||
"mytable",
|
||||
metadata,
|
||||
Column("myid", Integer),
|
||||
Column("name", String(50)),
|
||||
Column("desc", String(50)),
|
||||
)
|
||||
|
||||
mytable.create(connection)
|
||||
|
||||
connection.execute(
|
||||
mytable.insert(),
|
||||
[
|
||||
{"myid": 1, "name": "a", "desc": "a_desc"},
|
||||
{"myid": 2, "name": "b", "desc": "b_desc"},
|
||||
{"myid": 3, "name": "c", "desc": "c_desc"},
|
||||
{"myid": 4, "name": "d", "desc": "d_desc"},
|
||||
],
|
||||
)
|
||||
yield mytable
|
||||
|
||||
@tough_parameters
|
||||
def test_standalone_bindparam_escape(
|
||||
self, paramname, connection, multirow_fixture
|
||||
):
|
||||
tbl1 = multirow_fixture
|
||||
stmt = select(tbl1.c.myid).where(
|
||||
tbl1.c.name == bindparam(paramname, value="x")
|
||||
)
|
||||
res = connection.scalar(stmt, {paramname: "c"})
|
||||
eq_(res, 3)
|
||||
|
||||
@tough_parameters
|
||||
def test_standalone_bindparam_escape_expanding(
|
||||
self, paramname, connection, multirow_fixture
|
||||
):
|
||||
tbl1 = multirow_fixture
|
||||
stmt = (
|
||||
select(tbl1.c.myid)
|
||||
.where(tbl1.c.name.in_(bindparam(paramname, value=["a", "b"])))
|
||||
.order_by(tbl1.c.myid)
|
||||
)
|
||||
|
||||
res = connection.scalars(stmt, {paramname: ["d", "a"]}).all()
|
||||
eq_(res, [1, 4])
|
||||
|
||||
|
||||
class ReturningGuardsTest(fixtures.TablesTest):
|
||||
"""test that the various 'returning' flags are set appropriately"""
|
||||
|
||||
__backend__ = True
|
||||
|
||||
@classmethod
|
||||
def define_tables(cls, metadata):
|
||||
Table(
|
||||
"t",
|
||||
metadata,
|
||||
Column("id", Integer, primary_key=True, autoincrement=False),
|
||||
Column("data", String(50)),
|
||||
)
|
||||
|
||||
@testing.fixture
|
||||
def run_stmt(self, connection):
|
||||
t = self.tables.t
|
||||
|
||||
def go(stmt, executemany, id_param_name, expect_success):
|
||||
stmt = stmt.returning(t.c.id)
|
||||
|
||||
if executemany:
|
||||
if not expect_success:
|
||||
# for RETURNING executemany(), we raise our own
|
||||
# error as this is independent of general RETURNING
|
||||
# support
|
||||
with expect_raises_message(
|
||||
exc.StatementError,
|
||||
rf"Dialect {connection.dialect.name}\+"
|
||||
f"{connection.dialect.driver} with "
|
||||
f"current server capabilities does not support "
|
||||
f".*RETURNING when executemany is used",
|
||||
):
|
||||
connection.execute(
|
||||
stmt,
|
||||
[
|
||||
{id_param_name: 1, "data": "d1"},
|
||||
{id_param_name: 2, "data": "d2"},
|
||||
{id_param_name: 3, "data": "d3"},
|
||||
],
|
||||
)
|
||||
else:
|
||||
result = connection.execute(
|
||||
stmt,
|
||||
[
|
||||
{id_param_name: 1, "data": "d1"},
|
||||
{id_param_name: 2, "data": "d2"},
|
||||
{id_param_name: 3, "data": "d3"},
|
||||
],
|
||||
)
|
||||
eq_(result.all(), [(1,), (2,), (3,)])
|
||||
else:
|
||||
if not expect_success:
|
||||
# for RETURNING execute(), we pass all the way to the DB
|
||||
# and let it fail
|
||||
with expect_raises(exc.DBAPIError):
|
||||
connection.execute(
|
||||
stmt, {id_param_name: 1, "data": "d1"}
|
||||
)
|
||||
else:
|
||||
result = connection.execute(
|
||||
stmt, {id_param_name: 1, "data": "d1"}
|
||||
)
|
||||
eq_(result.all(), [(1,)])
|
||||
|
||||
return go
|
||||
|
||||
def test_insert_single(self, connection, run_stmt):
|
||||
t = self.tables.t
|
||||
|
||||
stmt = t.insert()
|
||||
|
||||
run_stmt(stmt, False, "id", connection.dialect.insert_returning)
|
||||
|
||||
def test_insert_many(self, connection, run_stmt):
|
||||
t = self.tables.t
|
||||
|
||||
stmt = t.insert()
|
||||
|
||||
run_stmt(
|
||||
stmt, True, "id", connection.dialect.insert_executemany_returning
|
||||
)
|
||||
|
||||
def test_update_single(self, connection, run_stmt):
|
||||
t = self.tables.t
|
||||
|
||||
connection.execute(
|
||||
t.insert(),
|
||||
[
|
||||
{"id": 1, "data": "d1"},
|
||||
{"id": 2, "data": "d2"},
|
||||
{"id": 3, "data": "d3"},
|
||||
],
|
||||
)
|
||||
|
||||
stmt = t.update().where(t.c.id == bindparam("b_id"))
|
||||
|
||||
run_stmt(stmt, False, "b_id", connection.dialect.update_returning)
|
||||
|
||||
def test_update_many(self, connection, run_stmt):
|
||||
t = self.tables.t
|
||||
|
||||
connection.execute(
|
||||
t.insert(),
|
||||
[
|
||||
{"id": 1, "data": "d1"},
|
||||
{"id": 2, "data": "d2"},
|
||||
{"id": 3, "data": "d3"},
|
||||
],
|
||||
)
|
||||
|
||||
stmt = t.update().where(t.c.id == bindparam("b_id"))
|
||||
|
||||
run_stmt(
|
||||
stmt, True, "b_id", connection.dialect.update_executemany_returning
|
||||
)
|
||||
|
||||
def test_delete_single(self, connection, run_stmt):
|
||||
t = self.tables.t
|
||||
|
||||
connection.execute(
|
||||
t.insert(),
|
||||
[
|
||||
{"id": 1, "data": "d1"},
|
||||
{"id": 2, "data": "d2"},
|
||||
{"id": 3, "data": "d3"},
|
||||
],
|
||||
)
|
||||
|
||||
stmt = t.delete().where(t.c.id == bindparam("b_id"))
|
||||
|
||||
run_stmt(stmt, False, "b_id", connection.dialect.delete_returning)
|
||||
|
||||
def test_delete_many(self, connection, run_stmt):
|
||||
t = self.tables.t
|
||||
|
||||
connection.execute(
|
||||
t.insert(),
|
||||
[
|
||||
{"id": 1, "data": "d1"},
|
||||
{"id": 2, "data": "d2"},
|
||||
{"id": 3, "data": "d3"},
|
||||
],
|
||||
)
|
||||
|
||||
stmt = t.delete().where(t.c.id == bindparam("b_id"))
|
||||
|
||||
run_stmt(
|
||||
stmt, True, "b_id", connection.dialect.delete_executemany_returning
|
||||
)
|
||||
Reference in New Issue
Block a user