Add actual asserts to mmts recovery test

Author: kelvich

contrib/mmts/Dockerfile

@@ -31,7 +31,7 @@ USER postgres
 ENV CFLAGS -O0
 WORKDIR /pg
 
-ENV REBUILD 2
+ENV REBUILD 5
 
 RUN cd /pg && \
 	git clone https://github.com/postgrespro/postgres_cluster.git --depth 1 && \

contrib/mmts/tests2/docker-entrypoint.sh

@@ -68,13 +68,15 @@ if [ "$1" = 'postgres' ]; then
 			listen_addresses='*' 
 			max_prepared_transactions = 100
 			synchronous_commit = off
+			fsync = off
 			wal_level = logical
 			max_worker_processes = 30
 			max_replication_slots = 10
 			max_wal_senders = 10
 			shared_preload_libraries = 'raftable,multimaster'
-            		default_transaction_isolation = 'repeatable read'
+			default_transaction_isolation = 'repeatable read'
 			log_checkpoints = on
+			checkpoint_timeout = 30
 			log_autovacuum_min_duration = 0
 
 			raftable.id = $NODE_ID

contrib/mmts/tests2/lib/bank_client.py

@@ -33,11 +33,14 @@ def stop(self):
         for client in self._clients:
             client.stop()
 
-    def print_agg(self):
+    def aggregate(self, echo=True):
         aggs = []
         for client in self._clients:
             aggs.append(client.history.aggregate())
 
+        if not echo:
+            return aggs
+
         columns = ['running', 'running_latency', 'max_latency', 'finish']
 
         print("\t\t", end="")
@@ -61,10 +64,7 @@ def print_agg(self):
 
         print("")
 
-    def set_acc_to_tx(self, max_acc):
-        for client in self._clients:
-            client.set_acc_to_tx(max_acc)
-
+        return aggs
 
 class BankClient(object):
 
@@ -141,7 +141,7 @@ def exec_tx(self, name, tx_block):
                 tx_block(conn, cur)    
                 self.history.register_finish(event_id, 'Commit')
             except psycopg2.Error as e:
-                print("=== node%d: %s" % (self.node_id, e.pgerror))
+                # print("=== node%d: %s" % (self.node_id, e.pgerror))
                 self.history.register_finish(event_id, e.pgerror)
                 #time.sleep(0.2)
 

contrib/mmts/tests2/lib/event_history.py

@@ -85,7 +85,7 @@ def aggregate(self):
         self.events = []
 
         for value in self.running_events.itervalues():
-            print(value)
+            # print(value)
 
             if value['name'] not in agg:
                 agg[value['name']] = copy.deepcopy(self.agg_template)

contrib/mmts/tests2/test_recovery.py

@@ -6,105 +6,56 @@
 class RecoveryTest(unittest.TestCase):
     @classmethod
     def setUpClass(self):
-        #subprocess.check_call(['blockade','up'])
         self.clients = ClientCollection([
             "dbname=postgres host=127.0.0.1 user=postgres",
             "dbname=postgres host=127.0.0.1 user=postgres port=5433",
             "dbname=postgres host=127.0.0.1 user=postgres port=5434"
         ])
+        self.clients.start()
+        time.sleep(5)
 
     @classmethod
     def tearDownClass(self):
         print('tearDown')
-        #subprocess.check_call(['blockade','join'])
-
-        # in case of error
         self.clients.stop()
-        #self.clients[0].cleanup()
 
-    def test_0_0_0(self):
-        print('### normalOpsTest ###')
+    # def test_normal_operations(self):
+    #     print('### normalOpsTest ###')
+
+    #     for i in range(3):
+    #         time.sleep(3)
+    #         aggs = self.clients.aggregate()
+    #         for agg in aggs:
+    #             # there were some commits
+    #             self.assertTrue( agg['transfer'] > 0 )
+
+    def test_node_prtition(self):
+        print('### nodePartitionTest ###')
 
-        time.sleep(10)
         subprocess.check_call(['blockade','partition','node3'])
         print('### blockade node3 ###')
 
-        self.clients.set_acc_to_tx(10000)
-        self.clients.start()
+        # clear tx history
+        self.clients.aggregate(echo=False)
 
-        for i in range(10):
+        for i in range(3):
             time.sleep(3)
-            self.clients.print_agg()
-            print("\n")
+            aggs = self.clients.aggregate()
+            self.assertTrue( aggs[0]['transfer']['finish']['Commit'] > 0 )
+            self.assertTrue( aggs[1]['transfer']['finish']['Commit'] > 0 )
+            self.assertTrue( 'Commit' not in aggs[2]['transfer'] )
 
         subprocess.check_call(['blockade','join'])
         print('### deblockade node3 ###')
 
+        # clear tx history
+        self.clients.aggregate(echo=False)
+
         for i in range(1000):
             time.sleep(3)
-            t = datetime.datetime.now()
-            self.clients.print_agg()
-            print("took %f seconds" % ( (datetime.datetime.now()-t).total_seconds(),))
-            print("\n")
-
-        self.clients.stop()
-
-
+            aggs = self.clients.aggregate()
+            print(i, aggs)
 
-#    def test_0_normal_operation(self):
-#        print('### normalOpsTest ###')
-#
-#        self.clients.set_acc_to_tx(10000)
-#        self.clients.start()
-#
-#        for i in range(5):
-#            time.sleep(3)
-#            for client in self.clients:
-#                agg = client.history.aggregate()
-#                print(agg)
-#                self.assertTrue(agg['transfer']['finish']['Commit'] > 0)
-#            print("\n")
-#
-#        self.clients.stop()
-#
-#    def test_1_distributed_deadlock(self):
-#        print('### DDD test ###')
-#
-#        self.clients.set_acc_to_tx(10)
-#        self.clients.start()
-#
-#        for i in range(5):
-#            time.sleep(3)
-#            for client in self.clients:
-#                agg = client.history.aggregate()
-#                print(agg)
-#                self.assertTrue(agg['transfer']['finish']['Commit'] > 0)
-#            print("\n")
-#
-#        self.clients.stop()
-#
-#    def test_2_node_disconnect(self):
-#        print('### disconnectTest ###')
-#
-#        self.clients.set_acc_to_tx(10000)
-#        self.clients.start()
-#
-#        subprocess.check_call(['blockade','partition','node3'])
-#        print('Node3 disconnected')
-#
-#        # give cluster some time to discover problem
-#        time.sleep(3)
-#
-#        for i in range(5):
-#            time.sleep(3)
-#            for client in self.clients:
-#                agg = client.history.aggregate()
-#                print(agg)
-#                self.assertTrue(agg['transfer']['finish']['Commit'] > 0)
-#            print("\n")
-#
-#        subprocess.check_call(['blockade','join'])
-#        self.clients.stop()
 
 if __name__ == '__main__':
     unittest.main()