(
self,
connection,
metadata,
)
| 1603 | |
| 1604 | @testing.requires.aggregate_order_by |
| 1605 | def test_aggregate_order_by( |
| 1606 | self, |
| 1607 | connection, |
| 1608 | metadata, |
| 1609 | ): |
| 1610 | |
| 1611 | values_t = Table( |
| 1612 | "values", |
| 1613 | metadata, |
| 1614 | Column("value", String(2)), |
| 1615 | Column("ordering", String(2)), |
| 1616 | ) |
| 1617 | metadata.create_all(connection) |
| 1618 | connection.execute( |
| 1619 | values_t.insert(), |
| 1620 | [ |
| 1621 | {"value": "a", "ordering": "1"}, |
| 1622 | {"value": "b", "ordering": "3"}, |
| 1623 | {"value": "c", "ordering": "2"}, |
| 1624 | ], |
| 1625 | ) |
| 1626 | |
| 1627 | if testing.against("postgresql", "mssql"): |
| 1628 | fn = lambda expr: func.string_agg( # noqa: E731 |
| 1629 | expr, literal_column("''") |
| 1630 | ) |
| 1631 | expected = "bca" |
| 1632 | elif testing.against(["mysql", "mariadb", "sqlite"]): |
| 1633 | fn = func.group_concat |
| 1634 | expected = "b,c,a" |
| 1635 | elif testing.against("oracle"): |
| 1636 | fn = func.listagg |
| 1637 | expected = "bca" |
| 1638 | else: |
| 1639 | assert False |
| 1640 | |
| 1641 | stmt = select( |
| 1642 | fn(values_t.c.value).aggregate_order_by(values_t.c.ordering.desc()) |
| 1643 | ) |
| 1644 | eq_(connection.scalar(stmt), expected) |
| 1645 | |
| 1646 | @testing.fails_on_everything_except("postgresql") |
| 1647 | def test_as_from(self, connection): |
nothing calls this directly
no test coverage detected