Merge "Remove errant assertion from unit of work"

5 files changed, 94 insertions, 19 deletions

diff --git a/doc/build/changelog/unreleased_14/5867.rst b/doc/build/changelog/unreleased_14/5867.rst
new file mode 100644
index 000000000..739377e0b
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/5867.rst
@@ -0,0 +1,9 @@
+.. change::
+    :tags: bug, regression, orm
+    :tickets: 5867
+
+    Fixed ORM unit of work regression where an errant "assert primary_key"
+    statement interferes with primary key generation sequences that don't
+    actually consider the columns in the table to use a real primary key
+    constraint, instead using :paramref:`_orm.mapper.primary_key` to establish
+    certain columns as "primary".
diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index b57963eb3..42b787580 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -1222,7 +1222,6 @@ def _emit_insert_statements(
                         )
 
                     primary_key = result.inserted_primary_key
-                    assert primary_key
                     for pk, col in zip(
                         primary_key, mapper._pks_by_table[table]
                     ):
diff --git a/test/orm/test_unitofworkv2.py b/test/orm/test_unitofworkv2.py
index 65089f773..822b47632 100644
--- a/test/orm/test_unitofworkv2.py
+++ b/test/orm/test_unitofworkv2.py
@@ -1,20 +1,25 @@
 from sqlalchemy import cast
+from sqlalchemy import DateTime
 from sqlalchemy import event
 from sqlalchemy import exc
 from sqlalchemy import FetchedValue
 from sqlalchemy import ForeignKey
 from sqlalchemy import func
+from sqlalchemy import Identity
 from sqlalchemy import inspect
 from sqlalchemy import Integer
 from sqlalchemy import JSON
 from sqlalchemy import literal
 from sqlalchemy import select
+from sqlalchemy import Sequence
 from sqlalchemy import String
 from sqlalchemy import testing
 from sqlalchemy import text
 from sqlalchemy import util
+from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import attributes
 from sqlalchemy.orm import backref
+from sqlalchemy.orm import clear_mappers
 from sqlalchemy.orm import exc as orm_exc
 from sqlalchemy.orm import mapper
 from sqlalchemy.orm import relationship
@@ -3098,3 +3103,80 @@ class EnsureCacheTest(fixtures.FutureEngineMixin, UOWTest):
 
             is_(conn._execution_options["compiled_cache"], cache)
             eq_(len(inspect(User)._compiled_cache), 3)
+
+
+class ORMOnlyPrimaryKeyTest(fixtures.TestBase):
+    @testing.requires.identity_columns
+    @testing.requires.returning
+    def test_a(self, base, run_test):
+        class A(base):
+            __tablename__ = "a"
+
+            id = Column(Integer, Identity())
+            included_col = Column(Integer)
+
+            __mapper_args__ = {"primary_key": [id], "eager_defaults": True}
+
+        run_test(A, A())
+
+    @testing.requires.sequences_as_server_defaults
+    @testing.requires.returning
+    def test_b(self, base, run_test):
+
+        seq = Sequence("x_seq")
+
+        class A(base):
+            __tablename__ = "a"
+
+            id = Column(Integer, seq, server_default=seq.next_value())
+            included_col = Column(Integer)
+
+            __mapper_args__ = {"primary_key": [id], "eager_defaults": True}
+
+        run_test(A, A())
+
+    def test_c(self, base, run_test):
+        class A(base):
+            __tablename__ = "a"
+
+            id = Column(Integer, nullable=False)
+            included_col = Column(Integer)
+
+            __mapper_args__ = {"primary_key": [id]}
+
+        a1 = A(id=1, included_col=select(1).scalar_subquery())
+        run_test(A, a1)
+
+    def test_d(self, base, run_test):
+        class A(base):
+            __tablename__ = "a"
+
+            id = Column(Integer, nullable=False)
+            updated_at = Column(DateTime, server_default=func.now())
+
+            __mapper_args__ = {"primary_key": [id], "eager_defaults": True}
+
+        a1 = A(id=1)
+        run_test(A, a1)
+
+    @testing.fixture
+    def base(self, metadata):
+        yield declarative_base(metadata=metadata)
+        clear_mappers()
+
+    @testing.fixture
+    def run_test(self, metadata, connection):
+        def go(A, a1):
+            metadata.create_all(connection)
+
+            with Session(connection) as s:
+                s.add(a1)
+
+                s.flush()
+                eq_(a1.id, 1)
+                s.commit()
+
+                aa = s.query(A).first()
+                is_(a1, aa)
+
+        return go
diff --git a/test/requirements.py b/test/requirements.py
index 3c9b39ac7..1a4c5f646 100644
--- a/test/requirements.py
+++ b/test/requirements.py
@@ -253,21 +253,6 @@ class DefaultRequirements(SuiteRequirements):
         return skip_if(["oracle"])
 
     @property
-    def identity(self):
-        """Target database must support GENERATED AS IDENTITY or a facsimile.
-
-        Includes GENERATED AS IDENTITY, AUTOINCREMENT, AUTO_INCREMENT, or other
-        column DDL feature that fills in a DB-generated identifier at
-        INSERT-time without requiring pre-execution of a SEQUENCE or other
-        artifact.
-
-        """
-        return skip_if(
-            ["firebird", "oracle", "postgresql", "sybase"],
-            "not supported by database",
-        )
-
-    @property
     def temporary_tables(self):
         """target database supports temporary tables"""
         return skip_if(["firebird", self._sqlite_file_db], "not supported (?)")
diff --git a/test/sql/test_defaults.py b/test/sql/test_defaults.py
index 7f8391708..543ae1f98 100644
--- a/test/sql/test_defaults.py
+++ b/test/sql/test_defaults.py
@@ -1082,11 +1082,11 @@ class EmptyInsertTest(fixtures.TestBase):
 
 
 class AutoIncrementTest(fixtures.TestBase):
-    __requires__ = ("identity",)
+
     __backend__ = True
 
-    @testing.provide_metadata
-    def test_autoincrement_single_col(self, connection):
+    @testing.requires.empty_inserts
+    def test_autoincrement_single_col(self, metadata, connection):
         single = Table(
             "single", self.metadata, Column("id", Integer, primary_key=True)
         )