Method test_a_in_b

test/sql/test_defaults.py:904–951

(self, a, b, connection)

        argnames="a, b",
    )
    def test_a_in_b(self, a, b, connection):
        q = self.tables.q
        p = self.tables.p

        conn = connection
        if a == "delete":
            conn.execute(q.insert().values(y=10, z=1))
            cte = q.delete().where(q.c.z == 1).returning(q.c.z).cte("c")
            expected = None
        elif a == "insert":
            cte = q.insert().values(z=1, y=10).returning(q.c.z).cte("c")
            expected = (2, 10)
        elif a == "update":
            conn.execute(q.insert().values(x=5, y=10, z=1))
            cte = (
                q.update()
                .where(q.c.z == 1)
                .values(x=7)
                .returning(q.c.z)
                .cte("c")
            )
            expected = (7, 5)
        elif a == "select":
            conn.execute(q.insert().values(x=5, y=10, z=1))
            cte = sa.select(q.c.z).cte("c")
            expected = (5, 10)

        if b == "select":
            conn.execute(p.insert().values(s=1))
            stmt = select(p.c.s, cte.c.z).where(p.c.s == cte.c.z)
        elif b == "insert":
            sel = select(1, cte.c.z)
            stmt = (
                p.insert().from_select(["s", "t"], sel).returning(p.c.s, p.c.t)
            )
        elif b == "delete":
            stmt = p.insert().values(s=1, t=cte.c.z).returning(p.c.s, cte.c.z)
        elif b == "update":
            conn.execute(p.insert().values(s=1))
            stmt = (
                p.update()
                .values(t=5)
                .where(p.c.s == cte.c.z)
                .returning(p.c.u, cte.c.z)
            )
        eq_(list(conn.execute(stmt)), [(1, 1)])

        eq_(conn.execute(select(q.c.x, q.c.y)).first(), expected)


class PKDefaultTest(fixtures.TestBase):