Method test_derived_dml_bulk

test/ext/test_hybrid.py:1929–1973

(self, crud, derived_dml, decl_base, connection)

    @testing.variation("crud", ["insert", "update"])
    def test_derived_dml_bulk(self, crud, derived_dml, decl_base, connection):
        A = derived_dml
        decl_base.metadata.create_all(connection)

        with self.sql_execution_asserter(connection) as asserter:
            with Session(connection) as session:
                session.execute(
                    insert(A),
                    [
                        {"rate": 1.5, "adjusted_amount": 25},
                        {"rate": 2.5, "adjusted_amount": 25},
                    ],
                )

                if crud.update:
                    session.execute(
                        update(A),
                        [
                            {"id": 1, "rate": 1.8, "adjusted_amount": 30},
                            {"id": 2, "rate": 2.8, "adjusted_amount": 40},
                        ],
                    )
        asserter.assert_(
            CompiledSQL(
                "INSERT INTO a (amount, rate) VALUES (:amount, :rate)",
                [
                    {"amount": 25 / 1.5, "rate": 1.5},
                    {"amount": 25 / 2.5, "rate": 2.5},
                ],
            ),
            Conditional(
                crud.update,
                [
                    CompiledSQL(
                        "UPDATE a SET amount=:amount, rate=:rate "
                        "WHERE a.id = :a_id",
                        [
                            {"amount": 30 / 1.8, "rate": 1.8, "a_id": 1},
                            {"amount": 40 / 2.8, "rate": 2.8, "a_id": 2},
                        ],
                    )
                ],
                [],
            ),
        )


class SpecialObjectTest(fixtures.TestBase, AssertsCompiledSQL):