Test SUBSCRIBE with psycopg2 via server cursors.
(self)
| 89 | self.assertEqual(results, [[1, 2], [3, 4]]) |
| 90 | |
| 91 | def test_psycopg2_subscribe(self) -> None: |
| 92 | """Test SUBSCRIBE with psycopg2 via server cursors.""" |
| 93 | conn = psycopg2.connect(MATERIALIZED_URL) |
| 94 | conn.set_session(autocommit=True) |
| 95 | with conn.cursor() as cur: |
| 96 | # Create a table with one row of data. |
| 97 | cur.execute("CREATE TABLE psycopg2_subscribe (a int, b text)") |
| 98 | cur.execute("INSERT INTO psycopg2_subscribe VALUES (1, 'a')") |
| 99 | conn.set_session(autocommit=False) |
| 100 | |
| 101 | # Start SUBSCRIBE using the binary copy protocol. |
| 102 | cur.execute("DECLARE cur CURSOR FOR SUBSCRIBE psycopg2_subscribe") |
| 103 | cur.execute("FETCH ALL cur") |
| 104 | |
| 105 | # Validate the first row, but ignore the timestamp column. |
| 106 | row = cur.fetchone() |
| 107 | if row is not None: |
| 108 | ts, diff, a, b = row |
| 109 | self.assertEqual(diff, 1) |
| 110 | self.assertEqual(a, 1) |
| 111 | self.assertEqual(b, "a") |
| 112 | else: |
| 113 | self.fail("row is None") |
| 114 | |
| 115 | self.assertEqual(cur.fetchone(), None) |
| 116 | |
| 117 | # Insert another row from another connection to simulate an |
| 118 | # update arriving. |
| 119 | with psycopg2.connect(MATERIALIZED_URL) as conn2: |
| 120 | conn2.set_session(autocommit=True) |
| 121 | with conn2.cursor() as cur2: |
| 122 | cur2.execute("INSERT INTO psycopg2_subscribe VALUES (2, 'b')") |
| 123 | |
| 124 | # Validate the new row, again ignoring the timestamp column. |
| 125 | cur.execute("FETCH ALL cur") |
| 126 | row = cur.fetchone() |
| 127 | assert row is not None |
| 128 | |
| 129 | ts, diff, a, b = row |
| 130 | self.assertEqual(diff, 1) |
| 131 | self.assertEqual(a, 2) |
| 132 | self.assertEqual(b, "b") |
| 133 | |
| 134 | self.assertEqual(cur.fetchone(), None) |
| 135 | |
| 136 | def test_psycopg3_subscribe_copy(self) -> None: |
| 137 | """Test SUBSCRIBE with psycopg3 via its new binary COPY decoding support.""" |