(self)
| 78 | self.assertEqual(row, ([1, 2, 3],)) |
| 79 | |
| 80 | def test_sqlalchemy(self) -> None: |
| 81 | engine = sqlalchemy.engine.create_engine(MATERIALIZED_URL) |
| 82 | with engine.connect() as connection: |
| 83 | r = connection.execute( |
| 84 | sqlalchemy.text( |
| 85 | "SELECT * FROM (VALUES (1, 2), (3, 4)) AS t(col1, col2)" |
| 86 | ) |
| 87 | ) |
| 88 | results = [[c1, c2] for c1, c2 in r] |
| 89 | self.assertEqual(results, [[1, 2], [3, 4]]) |
| 90 | |
| 91 | def test_psycopg2_subscribe(self) -> None: |
| 92 | """Test SUBSCRIBE with psycopg2 via server cursors.""" |