Method test_delete_cascade

test/orm/test_dynamic.py:1466–1523

(
        self, delete_cascade_configured, user_address_fixture
    )

    @testing.combinations(True, False, argnames="delete_cascade_configured")
    def test_delete_cascade(
        self, delete_cascade_configured, user_address_fixture
    ):
        addresses = self.tables.addresses
        User, Address = user_address_fixture(
            addresses_args={
                "order_by": addresses.c.id,
                "backref": "user",
                "cascade": (
                    "save-update"
                    if not delete_cascade_configured
                    else "all, delete"
                ),
            }
        )

        sess = fixture_session(
            autoflush=True,
        )
        u = User(name="ed")
        u.addresses.add_all(
            [Address(email_address=letter) for letter in "abcdef"]
        )
        sess.add(u)
        sess.commit()

        from sqlalchemy import case

        # the byzantine syntax here is so the query works on MSSQL
        isnull_stmt = select(
            case((addresses.c.user_id == None, True), else_=False),
            func.count("*"),
        ).group_by(
            case((addresses.c.user_id == None, True), else_=False),
            addresses.c.user_id,
        )

        eq_(
            {isnull: count for isnull, count in sess.execute(isnull_stmt)},
            {False: 6},
        )

        sess.delete(u)

        sess.commit()

        if not delete_cascade_configured:
            eq_(
                {isnull: count for isnull, count in sess.execute(isnull_stmt)},
                {True: 6},
            )
        else:
            eq_(
                sess.connection()
                .execute(select(func.count("*")).select_from(addresses))
                .scalar(),
                0,
            )