diff options
Diffstat (limited to 'test/dialect')
| -rw-r--r-- | test/dialect/mssql/test_compiler.py | 327 | ||||
| -rw-r--r-- | test/dialect/mssql/test_deprecations.py | 244 | ||||
| -rw-r--r-- | test/dialect/mssql/test_engine.py | 13 | ||||
| -rw-r--r-- | test/dialect/mssql/test_query.py | 206 | ||||
| -rw-r--r-- | test/dialect/mssql/test_reflection.py | 5 | ||||
| -rw-r--r-- | test/dialect/mysql/test_on_duplicate.py | 34 | ||||
| -rw-r--r-- | test/dialect/oracle/test_compiler.py | 39 | ||||
| -rw-r--r-- | test/dialect/postgresql/test_compiler.py | 15 |
8 files changed, 588 insertions, 295 deletions
diff --git a/test/dialect/mssql/test_compiler.py b/test/dialect/mssql/test_compiler.py index 83a610888..67120e8fe 100644 --- a/test/dialect/mssql/test_compiler.py +++ b/test/dialect/mssql/test_compiler.py @@ -2,8 +2,10 @@ from sqlalchemy import Column from sqlalchemy import Computed from sqlalchemy import delete +from sqlalchemy import exc from sqlalchemy import extract from sqlalchemy import func +from sqlalchemy import Identity from sqlalchemy import Index from sqlalchemy import insert from sqlalchemy import Integer @@ -27,6 +29,7 @@ from sqlalchemy.dialects.mssql.base import try_cast from sqlalchemy.sql import column from sqlalchemy.sql import quoted_name from sqlalchemy.sql import table +from sqlalchemy.testing import assert_raises_message from sqlalchemy.testing import AssertsCompiledSQL from sqlalchemy.testing import eq_ from sqlalchemy.testing import fixtures @@ -1116,96 +1119,6 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): checkparams={"x_1": 5, "param_1": 0}, ) - def test_primary_key_no_identity(self): - metadata = MetaData() - tbl = Table( - "test", - metadata, - Column("id", Integer, autoincrement=False, primary_key=True), - ) - self.assert_compile( - schema.CreateTable(tbl), - "CREATE TABLE test (id INTEGER NOT NULL, " "PRIMARY KEY (id))", - ) - - def test_primary_key_defaults_to_identity(self): - metadata = MetaData() - tbl = Table("test", metadata, Column("id", Integer, primary_key=True)) - self.assert_compile( - schema.CreateTable(tbl), - "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,1), " - "PRIMARY KEY (id))", - ) - - def test_identity_no_primary_key(self): - metadata = MetaData() - tbl = Table( - "test", metadata, Column("id", Integer, autoincrement=True) - ) - self.assert_compile( - schema.CreateTable(tbl), - "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,1)" ")", - ) - - def test_identity_separate_from_primary_key(self): - metadata = MetaData() - tbl = Table( - "test", - metadata, - Column("id", Integer, autoincrement=False, primary_key=True), - Column("x", Integer, autoincrement=True), - ) - self.assert_compile( - schema.CreateTable(tbl), - "CREATE TABLE test (id INTEGER NOT NULL, " - "x INTEGER NOT NULL IDENTITY(1,1), " - "PRIMARY KEY (id))", - ) - - def test_identity_illegal_two_autoincrements(self): - metadata = MetaData() - tbl = Table( - "test", - metadata, - Column("id", Integer, autoincrement=True), - Column("id2", Integer, autoincrement=True), - ) - # this will be rejected by the database, just asserting this is what - # the two autoincrements will do right now - self.assert_compile( - schema.CreateTable(tbl), - "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,1), " - "id2 INTEGER NOT NULL IDENTITY(1,1))", - ) - - def test_identity_start_0(self): - metadata = MetaData() - tbl = Table( - "test", - metadata, - Column("id", Integer, mssql_identity_start=0, primary_key=True), - ) - self.assert_compile( - schema.CreateTable(tbl), - "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(0,1), " - "PRIMARY KEY (id))", - ) - - def test_identity_increment_5(self): - metadata = MetaData() - tbl = Table( - "test", - metadata, - Column( - "id", Integer, mssql_identity_increment=5, primary_key=True - ), - ) - self.assert_compile( - schema.CreateTable(tbl), - "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,5), " - "PRIMARY KEY (id))", - ) - def test_table_pkc_clustering(self): metadata = MetaData() tbl = Table( @@ -1388,6 +1301,240 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): ) +class CompileIdentityTest(fixtures.TestBase, AssertsCompiledSQL): + __dialect__ = mssql.dialect() + + def assert_compile_with_warning(self, *args, **kwargs): + with testing.expect_deprecated( + "The dialect options 'mssql_identity_start' and " + "'mssql_identity_increment' are deprecated. " + "Use the 'Identity' object instead." + ): + return self.assert_compile(*args, **kwargs) + + def test_primary_key_no_identity(self): + metadata = MetaData() + tbl = Table( + "test", + metadata, + Column("id", Integer, autoincrement=False, primary_key=True), + ) + self.assert_compile( + schema.CreateTable(tbl), + "CREATE TABLE test (id INTEGER NOT NULL, PRIMARY KEY (id))", + ) + + def test_primary_key_defaults_to_identity(self): + metadata = MetaData() + tbl = Table("test", metadata, Column("id", Integer, primary_key=True)) + self.assert_compile( + schema.CreateTable(tbl), + "CREATE TABLE test (id INTEGER NOT NULL IDENTITY, " + "PRIMARY KEY (id))", + ) + + def test_primary_key_with_identity_object(self): + metadata = MetaData() + tbl = Table( + "test", + metadata, + Column( + "id", + Integer, + Identity(start=3, increment=42), + primary_key=True, + ), + ) + self.assert_compile( + schema.CreateTable(tbl), + "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(3,42), " + "PRIMARY KEY (id))", + ) + + def test_identity_no_primary_key(self): + metadata = MetaData() + tbl = Table( + "test", metadata, Column("id", Integer, autoincrement=True) + ) + self.assert_compile( + schema.CreateTable(tbl), + "CREATE TABLE test (id INTEGER NOT NULL IDENTITY)", + ) + + def test_identity_object_no_primary_key(self): + metadata = MetaData() + tbl = Table( + "test", metadata, Column("id", Integer, Identity(increment=42)), + ) + self.assert_compile( + schema.CreateTable(tbl), + "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,42))", + ) + + def test_identity_object_1_1(self): + metadata = MetaData() + tbl = Table( + "test", + metadata, + Column("id", Integer, Identity(start=1, increment=1)), + ) + self.assert_compile( + schema.CreateTable(tbl), + "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,1))", + ) + + def test_identity_object_no_primary_key_non_nullable(self): + metadata = MetaData() + tbl = Table( + "test", + metadata, + Column("id", Integer, Identity(start=3), nullable=False,), + ) + self.assert_compile( + schema.CreateTable(tbl), + "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(3,1)" ")", + ) + + def test_identity_separate_from_primary_key(self): + metadata = MetaData() + tbl = Table( + "test", + metadata, + Column("id", Integer, autoincrement=False, primary_key=True), + Column("x", Integer, autoincrement=True), + ) + self.assert_compile( + schema.CreateTable(tbl), + "CREATE TABLE test (id INTEGER NOT NULL, " + "x INTEGER NOT NULL IDENTITY, " + "PRIMARY KEY (id))", + ) + + def test_identity_object_separate_from_primary_key(self): + metadata = MetaData() + tbl = Table( + "test", + metadata, + Column("id", Integer, autoincrement=False, primary_key=True), + Column("x", Integer, Identity(start=3, increment=42),), + ) + self.assert_compile( + schema.CreateTable(tbl), + "CREATE TABLE test (id INTEGER NOT NULL, " + "x INTEGER NOT NULL IDENTITY(3,42), " + "PRIMARY KEY (id))", + ) + + def test_identity_illegal_two_autoincrements(self): + metadata = MetaData() + tbl = Table( + "test", + metadata, + Column("id", Integer, autoincrement=True), + Column("id2", Integer, autoincrement=True), + ) + # this will be rejected by the database, just asserting this is what + # the two autoincrements will do right now + self.assert_compile( + schema.CreateTable(tbl), + "CREATE TABLE test (id INTEGER NOT NULL IDENTITY, " + "id2 INTEGER NOT NULL IDENTITY)", + ) + + def test_identity_object_illegal_two_autoincrements(self): + metadata = MetaData() + tbl = Table( + "test", + metadata, + Column( + "id", + Integer, + Identity(start=3, increment=42), + autoincrement=True, + ), + Column("id2", Integer, Identity(start=7, increment=2),), + ) + # this will be rejected by the database, just asserting this is what + # the two autoincrements will do right now + self.assert_compile( + schema.CreateTable(tbl), + "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(3,42), " + "id2 INTEGER NOT NULL IDENTITY(7,2))", + ) + + def test_identity_start_0(self): + metadata = MetaData() + tbl = Table( + "test", + metadata, + Column("id", Integer, mssql_identity_start=0, primary_key=True), + ) + self.assert_compile_with_warning( + schema.CreateTable(tbl), + "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(0,1), " + "PRIMARY KEY (id))", + ) + + def test_identity_increment_5(self): + metadata = MetaData() + tbl = Table( + "test", + metadata, + Column( + "id", Integer, mssql_identity_increment=5, primary_key=True + ), + ) + self.assert_compile_with_warning( + schema.CreateTable(tbl), + "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,5), " + "PRIMARY KEY (id))", + ) + + @testing.combinations( + schema.CreateTable( + Table( + "test", + MetaData(), + Column( + "id", + Integer, + Identity(start=2, increment=2), + mssql_identity_start=0, + ), + ) + ), + schema.CreateTable( + Table( + "test1", + MetaData(), + Column( + "id2", + Integer, + Identity(start=3, increment=3), + mssql_identity_increment=5, + ), + ) + ), + ) + def test_identity_options_ignored_with_identity_object(self, create_table): + assert_raises_message( + exc.CompileError, + "Cannot specify options 'mssql_identity_start' and/or " + "'mssql_identity_increment' while also using the " + "'Identity' construct.", + create_table.compile, + dialect=self.__dialect__, + ) + + def test_identity_object_no_options(self): + metadata = MetaData() + tbl = Table("test", metadata, Column("id", Integer, Identity()),) + self.assert_compile( + schema.CreateTable(tbl), + "CREATE TABLE test (id INTEGER NOT NULL IDENTITY)", + ) + + class SchemaTest(fixtures.TestBase): def setup(self): t = Table( diff --git a/test/dialect/mssql/test_deprecations.py b/test/dialect/mssql/test_deprecations.py new file mode 100644 index 000000000..c0df266b1 --- /dev/null +++ b/test/dialect/mssql/test_deprecations.py @@ -0,0 +1,244 @@ +# -*- encoding: utf-8 +from sqlalchemy import Column +from sqlalchemy import engine_from_config +from sqlalchemy import Integer +from sqlalchemy import MetaData +from sqlalchemy import select +from sqlalchemy import String +from sqlalchemy import Table +from sqlalchemy import testing +from sqlalchemy.dialects.mssql import base as mssql +from sqlalchemy.sql import column +from sqlalchemy.sql import table +from sqlalchemy.testing import assertions +from sqlalchemy.testing import AssertsCompiledSQL +from sqlalchemy.testing import engines +from sqlalchemy.testing import eq_ +from sqlalchemy.testing import fixtures +from sqlalchemy.testing import is_ +from sqlalchemy.testing.mock import Mock + + +def _legacy_schema_aliasing_warning(): + return assertions.expect_deprecated("The legacy_schema_aliasing parameter") + + +class LegacySchemaAliasingTest(fixtures.TestBase, AssertsCompiledSQL): + """Legacy behavior tried to prevent schema-qualified tables + from being rendered as dotted names, and were instead aliased. + + This behavior no longer seems to be required. + + """ + + def setup(self): + metadata = MetaData() + self.t1 = table( + "t1", + column("a", Integer), + column("b", String), + column("c", String), + ) + self.t2 = Table( + "t2", + metadata, + Column("a", Integer), + Column("b", Integer), + Column("c", Integer), + schema="schema", + ) + + def _assert_sql(self, element, legacy_sql, modern_sql=None): + dialect = self._legacy_dialect() + + self.assert_compile(element, legacy_sql, dialect=dialect) + + dialect = mssql.dialect() + self.assert_compile(element, modern_sql or "foob", dialect=dialect) + + def _legacy_dialect(self): + with _legacy_schema_aliasing_warning(): + return mssql.dialect(legacy_schema_aliasing=True) + + @testing.combinations( + ( + { + "sqlalchemy.url": "mssql://foodsn", + "sqlalchemy.legacy_schema_aliasing": "true", + }, + True, + ), + ( + { + "sqlalchemy.url": "mssql://foodsn", + "sqlalchemy.legacy_schema_aliasing": "false", + }, + False, + ), + ) + def test_legacy_schema_flag(self, cfg, expected): + with testing.expect_deprecated("The legacy_schema_aliasing parameter"): + e = engine_from_config( + cfg, module=Mock(version="MS SQL Server 11.0.92") + ) + is_(e.dialect.legacy_schema_aliasing, expected) + + def test_result_map(self): + s = self.t2.select() + c = s.compile(dialect=self._legacy_dialect()) + assert self.t2.c.a in set(c._create_result_map()["a"][1]) + + def test_result_map_use_labels(self): + s = self.t2.select(use_labels=True) + c = s.compile(dialect=self._legacy_dialect()) + assert self.t2.c.a in set(c._create_result_map()["schema_t2_a"][1]) + + def test_straight_select(self): + self._assert_sql( + self.t2.select(), + "SELECT t2_1.a, t2_1.b, t2_1.c FROM [schema].t2 AS t2_1", + "SELECT [schema].t2.a, [schema].t2.b, " + "[schema].t2.c FROM [schema].t2", + ) + + def test_straight_select_use_labels(self): + self._assert_sql( + self.t2.select(use_labels=True), + "SELECT t2_1.a AS schema_t2_a, t2_1.b AS schema_t2_b, " + "t2_1.c AS schema_t2_c FROM [schema].t2 AS t2_1", + "SELECT [schema].t2.a AS schema_t2_a, " + "[schema].t2.b AS schema_t2_b, " + "[schema].t2.c AS schema_t2_c FROM [schema].t2", + ) + + def test_join_to_schema(self): + t1, t2 = self.t1, self.t2 + self._assert_sql( + t1.join(t2, t1.c.a == t2.c.a).select(), + "SELECT t1.a, t1.b, t1.c, t2_1.a, t2_1.b, t2_1.c FROM t1 " + "JOIN [schema].t2 AS t2_1 ON t2_1.a = t1.a", + "SELECT t1.a, t1.b, t1.c, [schema].t2.a, [schema].t2.b, " + "[schema].t2.c FROM t1 JOIN [schema].t2 ON [schema].t2.a = t1.a", + ) + + def test_union_schema_to_non(self): + t1, t2 = self.t1, self.t2 + s = ( + select(t2.c.a, t2.c.b) + .apply_labels() + .union(select(t1.c.a, t1.c.b).apply_labels()) + .alias() + .select() + ) + self._assert_sql( + s, + "SELECT anon_1.schema_t2_a, anon_1.schema_t2_b FROM " + "(SELECT t2_1.a AS schema_t2_a, t2_1.b AS schema_t2_b " + "FROM [schema].t2 AS t2_1 UNION SELECT t1.a AS t1_a, " + "t1.b AS t1_b FROM t1) AS anon_1", + "SELECT anon_1.schema_t2_a, anon_1.schema_t2_b FROM " + "(SELECT [schema].t2.a AS schema_t2_a, [schema].t2.b AS " + "schema_t2_b FROM [schema].t2 UNION SELECT t1.a AS t1_a, " + "t1.b AS t1_b FROM t1) AS anon_1", + ) + + def test_column_subquery_to_alias(self): + a1 = self.t2.alias("a1") + s = select([self.t2, select(a1.c.a).scalar_subquery()]) + self._assert_sql( + s, + "SELECT t2_1.a, t2_1.b, t2_1.c, " + "(SELECT a1.a FROM [schema].t2 AS a1) " + "AS anon_1 FROM [schema].t2 AS t2_1", + "SELECT [schema].t2.a, [schema].t2.b, [schema].t2.c, " + "(SELECT a1.a FROM [schema].t2 AS a1) AS anon_1 FROM [schema].t2", + ) + + +class LegacySchemaAliasingBackendTest( + testing.AssertsExecutionResults, fixtures.TestBase +): + __only_on__ = "mssql" + + @testing.provide_metadata + def test_insertid_schema(self): + meta = self.metadata + + with _legacy_schema_aliasing_warning(): + eng = engines.testing_engine( + options=dict(legacy_schema_aliasing=False) + ) + + tbl = Table( + "test", + meta, + Column("id", Integer, primary_key=True), + schema=testing.config.test_schema, + ) + + with eng.connect() as conn: + tbl.create(conn) + conn.execute(tbl.insert(), {"id": 1}) + eq_(conn.scalar(tbl.select()), 1) + + @testing.provide_metadata + def test_insertid_schema_legacy(self): + meta = self.metadata + + tbl = Table( + "test", + meta, + Column("id", Integer, primary_key=True), + schema=testing.config.test_schema, + ) + + with _legacy_schema_aliasing_warning(): + eng = engines.testing_engine( + options=dict(legacy_schema_aliasing=True) + ) + + with eng.connect() as conn: + + tbl.create(conn) + conn.execute(tbl.insert(), {"id": 1}) + eq_(conn.scalar(tbl.select()), 1) + + @testing.provide_metadata + def test_delete_schema(self, connection): + meta = self.metadata + + is_(connection.dialect.legacy_schema_aliasing, False) + + tbl = Table( + "test", + meta, + Column("id", Integer, primary_key=True), + schema=testing.config.test_schema, + ) + tbl.create(connection) + connection.execute(tbl.insert(), {"id": 1}) + eq_(connection.scalar(tbl.select()), 1) + connection.execute(tbl.delete(tbl.c.id == 1)) + eq_(connection.scalar(tbl.select()), None) + + @testing.provide_metadata + def test_delete_schema_legacy(self): + meta = self.metadata + with _legacy_schema_aliasing_warning(): + eng = engines.testing_engine( + options=dict(legacy_schema_aliasing=True) + ) + + tbl = Table( + "test", + meta, + Column("id", Integer, primary_key=True), + schema=testing.config.test_schema, + ) + + with eng.connect() as conn: + tbl.create(conn) + conn.execute(tbl.insert(), {"id": 1}) + eq_(conn.scalar(tbl.select()), 1) + conn.execute(tbl.delete(tbl.c.id == 1)) + eq_(conn.scalar(tbl.select()), None) diff --git a/test/dialect/mssql/test_engine.py b/test/dialect/mssql/test_engine.py index 220941489..27e92d5b5 100644 --- a/test/dialect/mssql/test_engine.py +++ b/test/dialect/mssql/test_engine.py @@ -1,7 +1,6 @@ # -*- encoding: utf-8 from sqlalchemy import Column -from sqlalchemy import engine_from_config from sqlalchemy import event from sqlalchemy import exc from sqlalchemy import Integer @@ -339,18 +338,6 @@ class ParseConnectTest(fixtures.TestBase): ) -class EngineFromConfigTest(fixtures.TestBase): - def test_legacy_schema_flag(self): - cfg = { - "sqlalchemy.url": "mssql://foodsn", - "sqlalchemy.legacy_schema_aliasing": "false", - } - e = engine_from_config( - cfg, module=Mock(version="MS SQL Server 11.0.92") - ) - eq_(e.dialect.legacy_schema_aliasing, False) - - class FastExecutemanyTest(fixtures.TestBase): __only_on__ = "mssql" __backend__ = True diff --git a/test/dialect/mssql/test_query.py b/test/dialect/mssql/test_query.py index 1e73b8b49..e37b388e8 100644 --- a/test/dialect/mssql/test_query.py +++ b/test/dialect/mssql/test_query.py @@ -6,6 +6,7 @@ from sqlalchemy import desc from sqlalchemy import event from sqlalchemy import ForeignKey from sqlalchemy import func +from sqlalchemy import Identity from sqlalchemy import Integer from sqlalchemy import literal from sqlalchemy import MetaData @@ -17,8 +18,6 @@ from sqlalchemy import Table from sqlalchemy import testing from sqlalchemy import util from sqlalchemy.dialects.mssql import base as mssql -from sqlalchemy.sql import column -from sqlalchemy.sql import table from sqlalchemy.testing import AssertsCompiledSQL from sqlalchemy.testing import engines from sqlalchemy.testing import eq_ @@ -27,120 +26,11 @@ from sqlalchemy.testing.assertsql import CursorSQL from sqlalchemy.testing.assertsql import DialectSQL from sqlalchemy.util import ue - metadata = None cattable = None matchtable = None -class LegacySchemaAliasingTest(fixtures.TestBase, AssertsCompiledSQL): - """Legacy behavior tried to prevent schema-qualified tables - from being rendered as dotted names, and were instead aliased. - - This behavior no longer seems to be required. - - """ - - def setup(self): - metadata = MetaData() - self.t1 = table( - "t1", - column("a", Integer), - column("b", String), - column("c", String), - ) - self.t2 = Table( - "t2", - metadata, - Column("a", Integer), - Column("b", Integer), - Column("c", Integer), - schema="schema", - ) - - def _assert_sql(self, element, legacy_sql, modern_sql=None): - dialect = mssql.dialect(legacy_schema_aliasing=True) - - self.assert_compile(element, legacy_sql, dialect=dialect) - - dialect = mssql.dialect() - self.assert_compile(element, modern_sql or "foob", dialect=dialect) - - def _legacy_dialect(self): - return mssql.dialect(legacy_schema_aliasing=True) - - def test_result_map(self): - s = self.t2.select() - c = s.compile(dialect=self._legacy_dialect()) - assert self.t2.c.a in set(c._create_result_map()["a"][1]) - - def test_result_map_use_labels(self): - s = self.t2.select(use_labels=True) - c = s.compile(dialect=self._legacy_dialect()) - assert self.t2.c.a in set(c._create_result_map()["schema_t2_a"][1]) - - def test_straight_select(self): - self._assert_sql( - self.t2.select(), - "SELECT t2_1.a, t2_1.b, t2_1.c FROM [schema].t2 AS t2_1", - "SELECT [schema].t2.a, [schema].t2.b, " - "[schema].t2.c FROM [schema].t2", - ) - - def test_straight_select_use_labels(self): - self._assert_sql( - self.t2.select(use_labels=True), - "SELECT t2_1.a AS schema_t2_a, t2_1.b AS schema_t2_b, " - "t2_1.c AS schema_t2_c FROM [schema].t2 AS t2_1", - "SELECT [schema].t2.a AS schema_t2_a, " - "[schema].t2.b AS schema_t2_b, " - "[schema].t2.c AS schema_t2_c FROM [schema].t2", - ) - - def test_join_to_schema(self): - t1, t2 = self.t1, self.t2 - self._assert_sql( - t1.join(t2, t1.c.a == t2.c.a).select(), - "SELECT t1.a, t1.b, t1.c, t2_1.a, t2_1.b, t2_1.c FROM t1 " - "JOIN [schema].t2 AS t2_1 ON t2_1.a = t1.a", - "SELECT t1.a, t1.b, t1.c, [schema].t2.a, [schema].t2.b, " - "[schema].t2.c FROM t1 JOIN [schema].t2 ON [schema].t2.a = t1.a", - ) - - def test_union_schema_to_non(self): - t1, t2 = self.t1, self.t2 - s = ( - select(t2.c.a, t2.c.b) - .apply_labels() - .union(select(t1.c.a, t1.c.b).apply_labels()) - .alias() - .select() - ) - self._assert_sql( - s, - "SELECT anon_1.schema_t2_a, anon_1.schema_t2_b FROM " - "(SELECT t2_1.a AS schema_t2_a, t2_1.b AS schema_t2_b " - "FROM [schema].t2 AS t2_1 UNION SELECT t1.a AS t1_a, " - "t1.b AS t1_b FROM t1) AS anon_1", - "SELECT anon_1.schema_t2_a, anon_1.schema_t2_b FROM " - "(SELECT [schema].t2.a AS schema_t2_a, [schema].t2.b AS " - "schema_t2_b FROM [schema].t2 UNION SELECT t1.a AS t1_a, " - "t1.b AS t1_b FROM t1) AS anon_1", - ) - - def test_column_subquery_to_alias(self): - a1 = self.t2.alias("a1") - s = select([self.t2, select(a1.c.a).scalar_subquery()]) - self._assert_sql( - s, - "SELECT t2_1.a, t2_1.b, t2_1.c, " - "(SELECT a1.a FROM [schema].t2 AS a1) " - "AS anon_1 FROM [schema].t2 AS t2_1", - "SELECT [schema].t2.a, [schema].t2.b, [schema].t2.c, " - "(SELECT a1.a FROM [schema].t2 AS a1) AS anon_1 FROM [schema].t2", - ) - - class IdentityInsertTest(fixtures.TestBase, AssertsCompiledSQL): __only_on__ = "mssql" __dialect__ = mssql.MSDialect() @@ -309,7 +199,7 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase): t1 = Table( "t1", meta, - Column("id", Integer, mssql_identity_start=100, primary_key=True), + Column("id", Integer, Identity(start=100), primary_key=True), Column("descr", String(200)), # the following flag will prevent the # MSSQLCompiler.returning_clause from getting called, @@ -321,7 +211,7 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase): t2 = Table( "t2", meta, - Column("id", Integer, mssql_identity_start=200, primary_key=True), + Column("id", Integer, Identity(start=200), primary_key=True), Column("descr", String(200)), ) @@ -432,44 +322,18 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase): ) @testing.provide_metadata - def test_insertid_schema(self): - meta = self.metadata - eng = engines.testing_engine( - options=dict(legacy_schema_aliasing=False) - ) - meta.bind = eng - conn = eng.connect() - conn.exec_driver_sql("create schema paj") - - @event.listens_for(meta, "after_drop") - def cleanup(target, connection, **kw): - connection.exec_driver_sql("drop schema paj") - - tbl = Table( - "test", meta, Column("id", Integer, primary_key=True), schema="paj" - ) - tbl.create(conn) - conn.execute(tbl.insert(), {"id": 1}) - eq_(conn.scalar(tbl.select()), 1) - - @testing.provide_metadata - def test_insertid_schema_legacy(self): + def test_insertid_schema(self, connection): meta = self.metadata - eng = engines.testing_engine(options=dict(legacy_schema_aliasing=True)) - meta.bind = eng - conn = eng.connect() - conn.exec_driver_sql("create schema paj") - - @event.listens_for(meta, "after_drop") - def cleanup(target, connection, **kw): - connection.exec_driver_sql("drop schema paj") tbl = Table( - "test", meta, Column("id", Integer, primary_key=True), schema="paj" + "test", + meta, + Column("id", Integer, primary_key=True), + schema=testing.config.test_schema, ) - tbl.create() - tbl.insert().execute({"id": 1}) - eq_(tbl.select().scalar(), 1) + tbl.create(connection) + connection.execute(tbl.insert(), {"id": 1}) + eq_(connection.scalar(tbl.select()), 1) @testing.provide_metadata def test_returning_no_autoinc(self, connection): @@ -489,48 +353,20 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase): eq_(result.fetchall(), [(1, "somestring")]) @testing.provide_metadata - def test_delete_schema(self): + def test_delete_schema(self, connection): meta = self.metadata - eng = engines.testing_engine( - options=dict(legacy_schema_aliasing=False) - ) - meta.bind = eng - conn = eng.connect() - conn.exec_driver_sql("create schema paj") - - @event.listens_for(meta, "after_drop") - def cleanup(target, connection, **kw): - connection.exec_driver_sql("drop schema paj") tbl = Table( - "test", meta, Column("id", Integer, primary_key=True), schema="paj" - ) - tbl.create(conn) - conn.execute(tbl.insert(), {"id": 1}) - eq_(conn.scalar(tbl.select()), 1) - conn.execute(tbl.delete(tbl.c.id == 1)) - eq_(conn.scalar(tbl.select()), None) - - @testing.provide_metadata - def test_delete_schema_legacy(self): - meta = self.metadata - eng = engines.testing_engine(options=dict(legacy_schema_aliasing=True)) - meta.bind = eng - conn = eng.connect() - conn.exec_driver_sql("create schema paj") - - @event.listens_for(meta, "after_drop") - def cleanup(target, connection, **kw): - connection.exec_driver_sql("drop schema paj") - - tbl = Table( - "test", meta, Column("id", Integer, primary_key=True), schema="paj" + "test", + meta, + Column("id", Integer, primary_key=True), + schema=testing.config.test_schema, ) - tbl.create(conn) - conn.execute(tbl.insert(), {"id": 1}) - eq_(conn.scalar(tbl.select()), 1) - conn.execute(tbl.delete(tbl.c.id == 1)) - eq_(conn.scalar(tbl.select()), None) + tbl.create(connection) + connection.execute(tbl.insert(), {"id": 1}) + eq_(connection.scalar(tbl.select()), 1) + connection.execute(tbl.delete(tbl.c.id == 1)) + eq_(connection.scalar(tbl.select()), None) @testing.provide_metadata def test_insertid_reserved(self, connection): diff --git a/test/dialect/mssql/test_reflection.py b/test/dialect/mssql/test_reflection.py index 176d3d2ec..6e4038eb4 100644 --- a/test/dialect/mssql/test_reflection.py +++ b/test/dialect/mssql/test_reflection.py @@ -133,7 +133,10 @@ class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL): primary_key=True, ), ) - table.create() + with testing.expect_deprecated( + "The dialect options 'mssql_identity_start' and" + ): + table.create() meta2 = MetaData(testing.db) table2 = Table("identity_test", meta2, autoload=True) diff --git a/test/dialect/mysql/test_on_duplicate.py b/test/dialect/mysql/test_on_duplicate.py index 95aabc776..ed88121a5 100644 --- a/test/dialect/mysql/test_on_duplicate.py +++ b/test/dialect/mysql/test_on_duplicate.py @@ -47,7 +47,7 @@ class OnDuplicateTest(fixtures.TablesTest): {"id": 2, "bar": "baz"}, ) - def test_on_duplicate_key_update(self): + def test_on_duplicate_key_update_multirow(self): foos = self.tables.foos with testing.db.connect() as conn: conn.execute(insert(foos, dict(id=1, bar="b", baz="bz"))) @@ -55,14 +55,34 @@ class OnDuplicateTest(fixtures.TablesTest): [dict(id=1, bar="ab"), dict(id=2, bar="b")] ) stmt = stmt.on_duplicate_key_update(bar=stmt.inserted.bar) + result = conn.execute(stmt) - eq_(result.inserted_primary_key, (2,)) + + # multirow, so its ambiguous. this is a behavioral change + # in 1.4 + eq_(result.inserted_primary_key, (None,)) eq_( conn.execute(foos.select().where(foos.c.id == 1)).fetchall(), [(1, "ab", "bz", False)], ) - def test_on_duplicate_key_update_null(self): + def test_on_duplicate_key_update_singlerow(self): + foos = self.tables.foos + with testing.db.connect() as conn: + conn.execute(insert(foos, dict(id=1, bar="b", baz="bz"))) + stmt = insert(foos).values(dict(id=2, bar="b")) + stmt = stmt.on_duplicate_key_update(bar=stmt.inserted.bar) + + result = conn.execute(stmt) + + # only one row in the INSERT so we do inserted_primary_key + eq_(result.inserted_primary_key, (2,)) + eq_( + conn.execute(foos.select().where(foos.c.id == 1)).fetchall(), + [(1, "b", "bz", False)], + ) + + def test_on_duplicate_key_update_null_multirow(self): foos = self.tables.foos with testing.db.connect() as conn: conn.execute(insert(foos, dict(id=1, bar="b", baz="bz"))) @@ -71,13 +91,15 @@ class OnDuplicateTest(fixtures.TablesTest): ) stmt = stmt.on_duplicate_key_update(updated_once=None) result = conn.execute(stmt) - eq_(result.inserted_primary_key, (2,)) + + # ambiguous + eq_(result.inserted_primary_key, (None,)) eq_( conn.execute(foos.select().where(foos.c.id == 1)).fetchall(), [(1, "b", "bz", None)], ) - def test_on_duplicate_key_update_expression(self): + def test_on_duplicate_key_update_expression_multirow(self): foos = self.tables.foos with testing.db.connect() as conn: conn.execute(insert(foos, dict(id=1, bar="b", baz="bz"))) @@ -88,7 +110,7 @@ class OnDuplicateTest(fixtures.TablesTest): bar=func.concat(stmt.inserted.bar, "_foo") ) result = conn.execute(stmt) - eq_(result.inserted_primary_key, (2,)) + eq_(result.inserted_primary_key, (None,)) eq_( conn.execute(foos.select().where(foos.c.id == 1)).fetchall(), [(1, "ab_foo", "bz", False)], diff --git a/test/dialect/oracle/test_compiler.py b/test/dialect/oracle/test_compiler.py index 21a3b04ba..97a204630 100644 --- a/test/dialect/oracle/test_compiler.py +++ b/test/dialect/oracle/test_compiler.py @@ -6,6 +6,7 @@ from sqlalchemy import exc from sqlalchemy import except_ from sqlalchemy import ForeignKey from sqlalchemy import func +from sqlalchemy import Identity from sqlalchemy import Index from sqlalchemy import Integer from sqlalchemy import literal @@ -1249,6 +1250,44 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): dialect=oracle.dialect(), ) + def test_column_identity(self): + # all other tests are in test_identity_column.py + m = MetaData() + t = Table( + "t", + m, + Column("y", Integer, Identity(always=True, start=4, increment=7)), + ) + self.assert_compile( + schema.CreateTable(t), + "CREATE TABLE t (y INTEGER GENERATED ALWAYS AS IDENTITY " + "(INCREMENT BY 7 START WITH 4))", + ) + + def test_column_identity_no_generated(self): + m = MetaData() + t = Table("t", m, Column("y", Integer, Identity(always=None))) + self.assert_compile( + schema.CreateTable(t), + "CREATE TABLE t (y INTEGER GENERATED AS IDENTITY)", + ) + + @testing.combinations( + (True, True, "ALWAYS ON NULL"), # this would error when executed + (False, None, "BY DEFAULT"), + (False, False, "BY DEFAULT"), + (False, True, "BY DEFAULT ON NULL"), + ) + def test_column_identity_on_null(self, always, on_null, text): + m = MetaData() + t = Table( + "t", m, Column("y", Integer, Identity(always, on_null=on_null)) + ) + self.assert_compile( + schema.CreateTable(t), + "CREATE TABLE t (y INTEGER GENERATED %s AS IDENTITY)" % text, + ) + class SequenceTest(fixtures.TestBase, AssertsCompiledSQL): def test_basic(self): diff --git a/test/dialect/postgresql/test_compiler.py b/test/dialect/postgresql/test_compiler.py index 708dbe147..6196b52f2 100644 --- a/test/dialect/postgresql/test_compiler.py +++ b/test/dialect/postgresql/test_compiler.py @@ -9,6 +9,7 @@ from sqlalchemy import delete from sqlalchemy import Enum from sqlalchemy import exc from sqlalchemy import func +from sqlalchemy import Identity from sqlalchemy import Index from sqlalchemy import Integer from sqlalchemy import MetaData @@ -1705,6 +1706,20 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): dialect=postgresql.dialect(), ) + def test_column_identity(self): + # all other tests are in test_identity_column.py + m = MetaData() + t = Table( + "t", + m, + Column("y", Integer, Identity(always=True, start=4, increment=7)), + ) + self.assert_compile( + schema.CreateTable(t), + "CREATE TABLE t (y INTEGER GENERATED ALWAYS AS IDENTITY " + "(INCREMENT BY 7 START WITH 4))", + ) + class InsertOnConflictTest(fixtures.TestBase, AssertsCompiledSQL): __dialect__ = postgresql.dialect() |