Загрузить файлы в «venv/Lib/site-packages/sqlalchemy/testing/suite»

venv/Lib/site-packages/sqlalchemy/testing/suite/test_sequence.py

@@ -0,0 +1,317 @@
+# testing/suite/test_sequence.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 config
+from .. import fixtures
+from ..assertions import eq_
+from ..assertions import is_true
+from ..config import requirements
+from ..provision import normalize_sequence
+from ..schema import Column
+from ..schema import Table
+from ... import inspect
+from ... import Integer
+from ... import MetaData
+from ... import Sequence
+from ... import String
+from ... import testing
+
+
+class SequenceTest(fixtures.TablesTest):
+    __requires__ = ("sequences",)
+    __sparse_driver_backend__ = True
+
+    run_create_tables = "each"
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            "seq_pk",
+            metadata,
+            Column(
+                "id",
+                Integer,
+                normalize_sequence(config, Sequence("tab_id_seq")),
+                primary_key=True,
+            ),
+            Column("data", String(50)),
+        )
+
+        Table(
+            "seq_opt_pk",
+            metadata,
+            Column(
+                "id",
+                Integer,
+                normalize_sequence(
+                    config,
+                    Sequence("tab_id_seq", data_type=Integer, optional=True),
+                ),
+                primary_key=True,
+            ),
+            Column("data", String(50)),
+        )
+
+        Table(
+            "seq_no_returning",
+            metadata,
+            Column(
+                "id",
+                Integer,
+                normalize_sequence(config, Sequence("noret_id_seq")),
+                primary_key=True,
+            ),
+            Column("data", String(50)),
+            implicit_returning=False,
+        )
+
+        if testing.requires.schemas.enabled:
+            Table(
+                "seq_no_returning_sch",
+                metadata,
+                Column(
+                    "id",
+                    Integer,
+                    normalize_sequence(
+                        config,
+                        Sequence(
+                            "noret_sch_id_seq", schema=config.test_schema
+                        ),
+                    ),
+                    primary_key=True,
+                ),
+                Column("data", String(50)),
+                implicit_returning=False,
+                schema=config.test_schema,
+            )
+
+    def test_insert_roundtrip(self, connection):
+        connection.execute(self.tables.seq_pk.insert(), dict(data="some data"))
+        self._assert_round_trip(self.tables.seq_pk, connection)
+
+    def test_insert_lastrowid(self, connection):
+        r = connection.execute(
+            self.tables.seq_pk.insert(), dict(data="some data")
+        )
+        eq_(
+            r.inserted_primary_key, (testing.db.dialect.default_sequence_base,)
+        )
+
+    def test_nextval_direct(self, connection):
+        r = connection.scalar(self.tables.seq_pk.c.id.default)
+        eq_(r, testing.db.dialect.default_sequence_base)
+
+    @requirements.sequences_optional
+    def test_optional_seq(self, connection):
+        r = connection.execute(
+            self.tables.seq_opt_pk.insert(), dict(data="some data")
+        )
+        eq_(r.inserted_primary_key, (1,))
+
+    def _assert_round_trip(self, table, conn):
+        row = conn.execute(table.select()).first()
+        eq_(row, (testing.db.dialect.default_sequence_base, "some data"))
+
+    def test_insert_roundtrip_no_implicit_returning(self, connection):
+        connection.execute(
+            self.tables.seq_no_returning.insert(), dict(data="some data")
+        )
+        self._assert_round_trip(self.tables.seq_no_returning, connection)
+
+    @testing.combinations((True,), (False,), argnames="implicit_returning")
+    @testing.requires.schemas
+    def test_insert_roundtrip_translate(self, connection, implicit_returning):
+        seq_no_returning = Table(
+            "seq_no_returning_sch",
+            MetaData(),
+            Column(
+                "id",
+                Integer,
+                normalize_sequence(
+                    config, Sequence("noret_sch_id_seq", schema="alt_schema")
+                ),
+                primary_key=True,
+            ),
+            Column("data", String(50)),
+            implicit_returning=implicit_returning,
+            schema="alt_schema",
+        )
+
+        connection = connection.execution_options(
+            schema_translate_map={"alt_schema": config.test_schema}
+        )
+        connection.execute(seq_no_returning.insert(), dict(data="some data"))
+        self._assert_round_trip(seq_no_returning, connection)
+
+    @testing.requires.schemas
+    def test_nextval_direct_schema_translate(self, connection):
+        seq = normalize_sequence(
+            config, Sequence("noret_sch_id_seq", schema="alt_schema")
+        )
+        connection = connection.execution_options(
+            schema_translate_map={"alt_schema": config.test_schema}
+        )
+
+        r = connection.scalar(seq)
+        eq_(r, testing.db.dialect.default_sequence_base)
+
+
+class SequenceCompilerTest(testing.AssertsCompiledSQL, fixtures.TestBase):
+    __requires__ = ("sequences",)
+    __sparse_driver_backend__ = True
+
+    def test_literal_binds_inline_compile(self, connection):
+        table = Table(
+            "x",
+            MetaData(),
+            Column(
+                "y", Integer, normalize_sequence(config, Sequence("y_seq"))
+            ),
+            Column("q", Integer),
+        )
+
+        stmt = table.insert().values(q=5)
+
+        seq_nextval = connection.dialect.statement_compiler(
+            statement=None, dialect=connection.dialect
+        ).visit_sequence(normalize_sequence(config, Sequence("y_seq")))
+        self.assert_compile(
+            stmt,
+            "INSERT INTO x (y, q) VALUES (%s, 5)" % (seq_nextval,),
+            literal_binds=True,
+            dialect=connection.dialect,
+        )
+
+
+class HasSequenceTest(fixtures.TablesTest):
+    run_deletes = None
+
+    __requires__ = ("sequences",)
+    __sparse_driver_backend__ = True
+
+    @classmethod
+    def define_tables(cls, metadata):
+        normalize_sequence(config, Sequence("user_id_seq", metadata=metadata))
+        normalize_sequence(
+            config,
+            Sequence(
+                "other_seq",
+                metadata=metadata,
+                nomaxvalue=True,
+                nominvalue=True,
+            ),
+        )
+        if testing.requires.schemas.enabled:
+            normalize_sequence(
+                config,
+                Sequence(
+                    "user_id_seq", schema=config.test_schema, metadata=metadata
+                ),
+            )
+            normalize_sequence(
+                config,
+                Sequence(
+                    "schema_seq", schema=config.test_schema, metadata=metadata
+                ),
+            )
+        Table(
+            "user_id_table",
+            metadata,
+            Column("id", Integer, primary_key=True),
+        )
+
+    def test_has_sequence(self, connection):
+        eq_(inspect(connection).has_sequence("user_id_seq"), True)
+
+    def test_has_sequence_cache(self, connection, metadata):
+        insp = inspect(connection)
+        eq_(insp.has_sequence("user_id_seq"), True)
+        ss = normalize_sequence(config, Sequence("new_seq", metadata=metadata))
+        eq_(insp.has_sequence("new_seq"), False)
+        ss.create(connection)
+        try:
+            eq_(insp.has_sequence("new_seq"), False)
+            insp.clear_cache()
+            eq_(insp.has_sequence("new_seq"), True)
+        finally:
+            ss.drop(connection)
+
+    def test_has_sequence_other_object(self, connection):
+        eq_(inspect(connection).has_sequence("user_id_table"), False)
+
+    @testing.requires.schemas
+    def test_has_sequence_schema(self, connection):
+        eq_(
+            inspect(connection).has_sequence(
+                "user_id_seq", schema=config.test_schema
+            ),
+            True,
+        )
+
+    def test_has_sequence_neg(self, connection):
+        eq_(inspect(connection).has_sequence("some_sequence"), False)
+
+    @testing.requires.schemas
+    def test_has_sequence_schemas_neg(self, connection):
+        eq_(
+            inspect(connection).has_sequence(
+                "some_sequence", schema=config.test_schema
+            ),
+            False,
+        )
+
+    @testing.requires.schemas
+    def test_has_sequence_default_not_in_remote(self, connection):
+        eq_(
+            inspect(connection).has_sequence(
+                "other_sequence", schema=config.test_schema
+            ),
+            False,
+        )
+
+    @testing.requires.schemas
+    def test_has_sequence_remote_not_in_default(self, connection):
+        eq_(inspect(connection).has_sequence("schema_seq"), False)
+
+    def test_get_sequence_names(self, connection):
+        exp = {"other_seq", "user_id_seq"}
+
+        res = set(inspect(connection).get_sequence_names())
+        is_true(res.intersection(exp) == exp)
+        is_true("schema_seq" not in res)
+
+    @testing.requires.schemas
+    def test_get_sequence_names_no_sequence_schema(self, connection):
+        eq_(
+            inspect(connection).get_sequence_names(
+                schema=config.test_schema_2
+            ),
+            [],
+        )
+
+    @testing.requires.schemas
+    def test_get_sequence_names_sequences_schema(self, connection):
+        eq_(
+            sorted(
+                inspect(connection).get_sequence_names(
+                    schema=config.test_schema
+                )
+            ),
+            ["schema_seq", "user_id_seq"],
+        )
+
+
+class HasSequenceTestEmpty(fixtures.TestBase):
+    __requires__ = ("sequences",)
+    __sparse_driver_backend__ = True
+
+    def test_get_sequence_names_no_sequence(self, connection):
+        eq_(
+            inspect(connection).get_sequence_names(),
+            [],
+        )

venv/Lib/site-packages/sqlalchemy/testing/suite/test_unicode_ddl.py

@@ -0,0 +1,189 @@
+# testing/suite/test_unicode_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
+
+
+from sqlalchemy import desc
+from sqlalchemy import ForeignKey
+from sqlalchemy import Integer
+from sqlalchemy import MetaData
+from sqlalchemy import testing
+from sqlalchemy.testing import eq_
+from sqlalchemy.testing import fixtures
+from sqlalchemy.testing.schema import Column
+from sqlalchemy.testing.schema import Table
+
+
+class UnicodeSchemaTest(fixtures.TablesTest):
+    __requires__ = ("unicode_ddl",)
+    __backend__ = True
+
+    @classmethod
+    def define_tables(cls, metadata):
+        global t1, t2, t3
+
+        t1 = Table(
+            "unitable1",
+            metadata,
+            Column("méil", Integer, primary_key=True),
+            Column("\u6e2c\u8a66", Integer),
+            test_needs_fk=True,
+        )
+        t2 = Table(
+            "Unitéble2",
+            metadata,
+            Column("méil", Integer, primary_key=True, key="a"),
+            Column(
+                "\u6e2c\u8a66",
+                Integer,
+                ForeignKey("unitable1.méil"),
+                key="b",
+            ),
+            test_needs_fk=True,
+        )
+
+        # Few DBs support Unicode foreign keys
+        if testing.against("sqlite"):
+            t3 = Table(
+                "\u6e2c\u8a66",
+                metadata,
+                Column(
+                    "\u6e2c\u8a66_id",
+                    Integer,
+                    primary_key=True,
+                    autoincrement=False,
+                ),
+                Column(
+                    "unitable1_\u6e2c\u8a66",
+                    Integer,
+                    ForeignKey("unitable1.\u6e2c\u8a66"),
+                ),
+                Column("Unitéble2_b", Integer, ForeignKey("Unitéble2.b")),
+                Column(
+                    "\u6e2c\u8a66_self",
+                    Integer,
+                    ForeignKey("\u6e2c\u8a66.\u6e2c\u8a66_id"),
+                ),
+                test_needs_fk=True,
+            )
+        else:
+            t3 = Table(
+                "\u6e2c\u8a66",
+                metadata,
+                Column(
+                    "\u6e2c\u8a66_id",
+                    Integer,
+                    primary_key=True,
+                    autoincrement=False,
+                ),
+                Column("unitable1_\u6e2c\u8a66", Integer),
+                Column("Unitéble2_b", Integer),
+                Column("\u6e2c\u8a66_self", Integer),
+                test_needs_fk=True,
+            )
+
+    def test_insert(self, connection):
+        connection.execute(t1.insert(), {"méil": 1, "\u6e2c\u8a66": 5})
+        connection.execute(t2.insert(), {"a": 1, "b": 1})
+        connection.execute(
+            t3.insert(),
+            {
+                "\u6e2c\u8a66_id": 1,
+                "unitable1_\u6e2c\u8a66": 5,
+                "Unitéble2_b": 1,
+                "\u6e2c\u8a66_self": 1,
+            },
+        )
+
+        eq_(connection.execute(t1.select()).fetchall(), [(1, 5)])
+        eq_(connection.execute(t2.select()).fetchall(), [(1, 1)])
+        eq_(connection.execute(t3.select()).fetchall(), [(1, 5, 1, 1)])
+
+    def test_col_targeting(self, connection):
+        connection.execute(t1.insert(), {"méil": 1, "\u6e2c\u8a66": 5})
+        connection.execute(t2.insert(), {"a": 1, "b": 1})
+        connection.execute(
+            t3.insert(),
+            {
+                "\u6e2c\u8a66_id": 1,
+                "unitable1_\u6e2c\u8a66": 5,
+                "Unitéble2_b": 1,
+                "\u6e2c\u8a66_self": 1,
+            },
+        )
+
+        row = connection.execute(t1.select()).first()
+        eq_(row._mapping[t1.c["méil"]], 1)
+        eq_(row._mapping[t1.c["\u6e2c\u8a66"]], 5)
+
+        row = connection.execute(t2.select()).first()
+        eq_(row._mapping[t2.c["a"]], 1)
+        eq_(row._mapping[t2.c["b"]], 1)
+
+        row = connection.execute(t3.select()).first()
+        eq_(row._mapping[t3.c["\u6e2c\u8a66_id"]], 1)
+        eq_(row._mapping[t3.c["unitable1_\u6e2c\u8a66"]], 5)
+        eq_(row._mapping[t3.c["Unitéble2_b"]], 1)
+        eq_(row._mapping[t3.c["\u6e2c\u8a66_self"]], 1)
+
+    def test_reflect(self, connection):
+        connection.execute(t1.insert(), {"méil": 2, "\u6e2c\u8a66": 7})
+        connection.execute(t2.insert(), {"a": 2, "b": 2})
+        connection.execute(
+            t3.insert(),
+            {
+                "\u6e2c\u8a66_id": 2,
+                "unitable1_\u6e2c\u8a66": 7,
+                "Unitéble2_b": 2,
+                "\u6e2c\u8a66_self": 2,
+            },
+        )
+
+        meta = MetaData()
+        tt1 = Table(t1.name, meta, autoload_with=connection)
+        tt2 = Table(t2.name, meta, autoload_with=connection)
+        tt3 = Table(t3.name, meta, autoload_with=connection)
+
+        connection.execute(tt1.insert(), {"méil": 1, "\u6e2c\u8a66": 5})
+        connection.execute(tt2.insert(), {"méil": 1, "\u6e2c\u8a66": 1})
+        connection.execute(
+            tt3.insert(),
+            {
+                "\u6e2c\u8a66_id": 1,
+                "unitable1_\u6e2c\u8a66": 5,
+                "Unitéble2_b": 1,
+                "\u6e2c\u8a66_self": 1,
+            },
+        )
+
+        eq_(
+            connection.execute(tt1.select().order_by(desc("méil"))).fetchall(),
+            [(2, 7), (1, 5)],
+        )
+        eq_(
+            connection.execute(tt2.select().order_by(desc("méil"))).fetchall(),
+            [(2, 2), (1, 1)],
+        )
+        eq_(
+            connection.execute(
+                tt3.select().order_by(desc("\u6e2c\u8a66_id"))
+            ).fetchall(),
+            [(2, 7, 2, 2), (1, 5, 1, 1)],
+        )
+
+    def test_repr(self):
+        meta = MetaData()
+        t = Table("\u6e2c\u8a66", meta, Column("\u6e2c\u8a66_id", Integer))
+        eq_(
+            repr(t),
+            (
+                "Table('測試', MetaData(), "
+                "Column('測試_id', Integer(), "
+                "table=<測試>), "
+                "schema=None)"
+            ),
+        )

venv/Lib/site-packages/sqlalchemy/testing/suite/test_update_delete.py

@@ -0,0 +1,139 @@
+# testing/suite/test_update_delete.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 String
+from ... import testing
+
+
+class SimpleUpdateDeleteTest(fixtures.TablesTest):
+    run_deletes = "each"
+    __requires__ = ("sane_rowcount",)
+    __sparse_driver_backend__ = True
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            "plain_pk",
+            metadata,
+            Column("id", Integer, primary_key=True),
+            Column("data", String(50)),
+        )
+
+    @classmethod
+    def insert_data(cls, connection):
+        connection.execute(
+            cls.tables.plain_pk.insert(),
+            [
+                {"id": 1, "data": "d1"},
+                {"id": 2, "data": "d2"},
+                {"id": 3, "data": "d3"},
+            ],
+        )
+
+    def test_update(self, connection):
+        t = self.tables.plain_pk
+        r = connection.execute(
+            t.update().where(t.c.id == 2), dict(data="d2_new")
+        )
+        assert not r.is_insert
+        assert not r.returns_rows
+        assert r.rowcount == 1
+
+        eq_(
+            connection.execute(t.select().order_by(t.c.id)).fetchall(),
+            [(1, "d1"), (2, "d2_new"), (3, "d3")],
+        )
+
+    def test_delete(self, connection):
+        t = self.tables.plain_pk
+        r = connection.execute(t.delete().where(t.c.id == 2))
+        assert not r.is_insert
+        assert not r.returns_rows
+        assert r.rowcount == 1
+        eq_(
+            connection.execute(t.select().order_by(t.c.id)).fetchall(),
+            [(1, "d1"), (3, "d3")],
+        )
+
+    @testing.variation("criteria", ["rows", "norows", "emptyin"])
+    @testing.requires.update_returning
+    def test_update_returning(self, connection, criteria):
+        t = self.tables.plain_pk
+
+        stmt = t.update().returning(t.c.id, t.c.data)
+
+        if criteria.norows:
+            stmt = stmt.where(t.c.id == 10)
+        elif criteria.rows:
+            stmt = stmt.where(t.c.id == 2)
+        elif criteria.emptyin:
+            stmt = stmt.where(t.c.id.in_([]))
+        else:
+            criteria.fail()
+
+        r = connection.execute(stmt, dict(data="d2_new"))
+        assert not r.is_insert
+        assert r.returns_rows
+        eq_(r.keys(), ["id", "data"])
+
+        if criteria.rows:
+            eq_(r.all(), [(2, "d2_new")])
+        else:
+            eq_(r.all(), [])
+
+        eq_(
+            connection.execute(t.select().order_by(t.c.id)).fetchall(),
+            (
+                [(1, "d1"), (2, "d2_new"), (3, "d3")]
+                if criteria.rows
+                else [(1, "d1"), (2, "d2"), (3, "d3")]
+            ),
+        )
+
+    @testing.variation("criteria", ["rows", "norows", "emptyin"])
+    @testing.requires.delete_returning
+    def test_delete_returning(self, connection, criteria):
+        t = self.tables.plain_pk
+
+        stmt = t.delete().returning(t.c.id, t.c.data)
+
+        if criteria.norows:
+            stmt = stmt.where(t.c.id == 10)
+        elif criteria.rows:
+            stmt = stmt.where(t.c.id == 2)
+        elif criteria.emptyin:
+            stmt = stmt.where(t.c.id.in_([]))
+        else:
+            criteria.fail()
+
+        r = connection.execute(stmt)
+        assert not r.is_insert
+        assert r.returns_rows
+        eq_(r.keys(), ["id", "data"])
+
+        if criteria.rows:
+            eq_(r.all(), [(2, "d2")])
+        else:
+            eq_(r.all(), [])
+
+        eq_(
+            connection.execute(t.select().order_by(t.c.id)).fetchall(),
+            (
+                [(1, "d1"), (3, "d3")]
+                if criteria.rows
+                else [(1, "d1"), (2, "d2"), (3, "d3")]
+            ),
+        )
+
+
+__all__ = ("SimpleUpdateDeleteTest",)