(
self, metadata, connection, stmt_type, callable_type
)
| 1580 | @testing.variation("stmt_type", ["lambda_stmt", "lambda_crit"]) |
| 1581 | @testing.variation("callable_type", ["none", "closure", "parameter"]) |
| 1582 | def test_9029_integration( |
| 1583 | self, metadata, connection, stmt_type, callable_type |
| 1584 | ): |
| 1585 | t = Table( |
| 1586 | "t", |
| 1587 | metadata, |
| 1588 | Column("id", Integer, primary_key=True), |
| 1589 | Column("data", JSON), |
| 1590 | ) |
| 1591 | |
| 1592 | t.create(connection) |
| 1593 | |
| 1594 | connection.execute( |
| 1595 | t.insert(), |
| 1596 | { |
| 1597 | "id": 12, |
| 1598 | "data": {"key": "value", "key2": {"subkey": [1, 2, 3]}}, |
| 1599 | }, |
| 1600 | ) |
| 1601 | |
| 1602 | d = {"key": "value", "key2": {"subkey": [1, 2, 3]}} |
| 1603 | |
| 1604 | if callable_type.none: |
| 1605 | if stmt_type.lambda_stmt: |
| 1606 | stmt = lambda_stmt(lambda: select(t).filter(t.c.data == d)) |
| 1607 | elif stmt_type.lambda_crit: |
| 1608 | stmt = select(t).filter(lambda: t.c.data == d) |
| 1609 | else: |
| 1610 | stmt_type.fail() |
| 1611 | |
| 1612 | to_run = stmt |
| 1613 | |
| 1614 | elif callable_type.closure: |
| 1615 | |
| 1616 | def go(): |
| 1617 | if stmt_type.lambda_stmt: |
| 1618 | stmt = lambda_stmt(lambda: select(t).filter(t.c.data == d)) |
| 1619 | elif stmt_type.lambda_crit: |
| 1620 | stmt = select(t).filter(lambda: t.c.data == d) |
| 1621 | else: |
| 1622 | stmt_type.fail() |
| 1623 | return stmt |
| 1624 | |
| 1625 | to_run = go() |
| 1626 | |
| 1627 | elif callable_type.parameter: |
| 1628 | |
| 1629 | def go(data): |
| 1630 | if stmt_type.lambda_stmt: |
| 1631 | stmt = lambda_stmt( |
| 1632 | lambda: select(t).filter(t.c.data == data) |
| 1633 | ) |
| 1634 | elif stmt_type.lambda_crit: |
| 1635 | stmt = select(t).filter(lambda: t.c.data == data) |
| 1636 | else: |
| 1637 | stmt_type.fail() |
| 1638 | |
| 1639 | return stmt |
nothing calls this directly
no test coverage detected