Method test_upsert_downgrades

test/sql/test_insert_exec.py:2775–2887

(
        self,
        metadata,
        connection,
        pk_type: testing.Variation,
        sort_by_parameter_order,
        randomize_returning,
        sentinel_type,
        warn_for_downgrades,
    )

    )
    @testing.requires.provisioned_upsert
    def test_upsert_downgrades(
        self,
        metadata,
        connection,
        pk_type: testing.Variation,
        sort_by_parameter_order,
        randomize_returning,
        sentinel_type,
        warn_for_downgrades,
    ):
        if pk_type.serverside:
            pk_col = Column(
                "id",
                Integer(),
                primary_key=True,
                insert_sentinel=bool(sentinel_type.use_pk_explicit),
            )
        elif pk_type.clientside:
            pk_col = Column(
                "id",
                Uuid(),
                default=uuid.uuid4,
                primary_key=True,
                insert_sentinel=bool(sentinel_type.use_pk_explicit),
            )
        else:
            pk_type.fail()

        if sentinel_type.separate_uuid:
            extra_col = Column(
                "sent_col",
                Uuid(),
                default=uuid.uuid4,
                insert_sentinel=True,
                nullable=False,
            )
        elif sentinel_type.separate_sentinel:
            extra_col = insert_sentinel("sent_col")
        else:
            extra_col = Column("sent_col", Integer)

        t1 = Table(
            "upsert_table",
            metadata,
            pk_col,
            Column("data", String(50)),
            extra_col,
            Column(
                "has_server_default",
                String(30),
                server_default="some_server_default",
            ),
        )
        metadata.create_all(connection)

        result = connection.execute(
            insert(t1).returning(
                t1.c.id, t1.c.data, sort_by_parameter_order=True