Improve metadata bwc test for Logical Replication

Author: jeeminso

tests/bwc/test_rolling_upgrade.py

@@ -1,3 +1,4 @@
+import time
 import unittest
 from crate.client import connect
 from crate.client.exceptions import ProgrammingError
@@ -88,6 +89,10 @@ def _test_rolling_upgrade(self, path, nodes):
         }
         cluster = self._new_cluster(path.from_version, nodes, settings=settings)
         cluster.start()
+        replica_cluster = None
+        if int(path.from_version.split('.')[0]) >= 5 and int(path.from_version.split('.')[1]) >= 10:
+            replica_cluster = self._new_cluster(path.from_version, 1, settings=settings, explicit_discovery=False)
+            replica_cluster.start()
         with connect(cluster.node().http_url, error_trace=True) as conn:
             c = conn.cursor()
             c.execute("create user arthur with (password = 'secret')")
@@ -152,6 +157,19 @@ def _test_rolling_upgrade(self, path, nodes):
             # Add the shards of the new partition primaries
             expected_active_shards += shards
 
+            # Set up tables for logical replications
+            if int(path.from_version.split('.')[0]) >= 5 and int(path.from_version.split('.')[1]) >= 10:
+                c.execute("create table doc.x (a int) clustered into 1 shards with (number_of_replicas=0)")
+                expected_active_shards += 1
+                c.execute("create publication p for table doc.x")
+                with connect(replica_cluster.node().http_url, error_trace=True) as replica_conn:
+                    rc = replica_conn.cursor()
+                    rc.execute("create table doc.rx (a int) clustered into 1 shards with (number_of_replicas=0)")
+                    rc.execute("create publication rp for table doc.rx")
+                    rc.execute(f"create subscription rs connection 'crate://localhost:{cluster.node().addresses.transport.port}?user=crate&sslmode=sniff' publication p")
+                c.execute(f"create subscription s connection 'crate://localhost:{replica_cluster.node().addresses.transport.port}?user=crate&sslmode=sniff' publication rp")
+                expected_active_shards += 1
+
         for idx, node in enumerate(cluster):
             # Enforce an old version node be a handler to make sure that an upgraded node can serve 'select *' from an old version node.
             # Otherwise upgraded node simply requests N-1 columns from old version with N columns and it always works.
@@ -282,6 +300,28 @@ def _test_rolling_upgrade(self, path, nodes):
                         c.execute("select version['created'] from information_schema.table_partitions where table_name = 't3' and values['a'] = ?", [idx])
                         self.assertEqual(c.fetchall(), [[partition_version]])
 
+                # Ensure logical replications works
+                if int(path.from_version.split('.')[0]) >= 5 and int(path.from_version.split('.')[1]) >= 10:
+                    with connect(replica_cluster.node().http_url, error_trace=True) as replica_conn:
+                        rc = replica_conn.cursor()
+                        #wait_for_active_shards(c)
+                        #wait_for_active_shards(rc)
+                        time.sleep(10)
+                        # Ensure publishing to remote cluster works
+                        rc.execute("select count(*) from doc.x")
+                        count = rc.fetchall()[0][0]
+                        c.execute("insert into doc.x values (1)")
+                        time.sleep(10)
+                        rc.execute("select count(*) from doc.x")
+                        self.assertEqual(rc.fetchall()[0][0], count + 1)
+                        # Ensure subscription from remote cluster works
+                        c.execute("select count(*) from doc.rx")
+                        count = c.fetchall()[0][0]
+                        rc.execute("insert into doc.rx values (1)")
+                        time.sleep(10)
+                        c.execute("select count(*) from doc.rx")
+                        self.assertEqual(c.fetchall()[0][0], count + 1)
+
         # Finally validate that all shards (primaries and replicas) of all partitions are started
         # and writes into the partitioned table while upgrading were successful
         with connect(cluster.node().http_url, error_trace=True) as conn: