(sql_context)
| 213 | |
| 214 | @pytest.mark.asyncio |
| 215 | async def test_bulk_update(sql_context): |
| 216 | db, dialect, is_psycopg = sql_context |
| 217 | obj1 = await IntFields.create(intnum=1) |
| 218 | obj2 = await IntFields.create(intnum=2) |
| 219 | obj1.intnum = obj1.intnum + 1 |
| 220 | obj2.intnum = obj2.intnum + 1 |
| 221 | sql = IntFields.bulk_update([obj1, obj2], fields=["intnum"]).sql() |
| 222 | |
| 223 | if dialect == "mysql": |
| 224 | expected = "UPDATE `intfields` SET `intnum`=CASE WHEN `id`=%s THEN %s WHEN `id`=%s THEN %s END WHERE `id` IN (%s,%s)" |
| 225 | elif dialect == "postgres": |
| 226 | if is_psycopg: |
| 227 | expected = 'UPDATE "intfields" SET "intnum"=CASE WHEN "id"=%s THEN CAST(%s AS INT) WHEN "id"=%s THEN CAST(%s AS INT) END WHERE "id" IN (%s,%s)' |
| 228 | else: |
| 229 | expected = 'UPDATE "intfields" SET "intnum"=CASE WHEN "id"=$1 THEN CAST($2 AS INT) WHEN "id"=$3 THEN CAST($4 AS INT) END WHERE "id" IN ($5,$6)' |
| 230 | else: |
| 231 | expected = 'UPDATE "intfields" SET "intnum"=CASE WHEN "id"=? THEN ? WHEN "id"=? THEN ? END WHERE "id" IN (?,?)' |
| 232 | assert sql == expected |
| 233 | |
| 234 | |
| 235 | @pytest.mark.asyncio |
nothing calls this directly
no test coverage detected
searching dependent graphs…