raftable test for node out

Author: kelvich

contrib/raftable/tests/lib/bank_client.py

@@ -25,9 +25,8 @@ def start(self):
             client.start()
 
     def stop(self):
-        print('collection stop called', self._clients)
+        print('Terminating clients')
         for client in self._clients:
-            print('stop coll')
             client.stop()
 
 
@@ -52,26 +51,33 @@ def check(self):
             value = random.randrange(1, 1000000)
 
             event_id = self.history.register_start('setkey')
+            #print(self.node_id, 'value <- ')
             cur.execute("select raftable('rush', '%d', 100)" % (value))
-            print(self.node_id, 'value <- ', value)
+            #print(self.node_id, 'value <- ', value)
             self.history.register_finish(event_id, 'commit')
 
             event_id = self.history.register_start('readkey')
+            #print(self.node_id, 'value -> ')
             cur.execute("select raftable('rush')")
             value = cur.fetchone()[0]
-            print(self.node_id, 'value -> ', value)
+            #print(self.node_id, 'value -> ', value)
             self.history.register_finish(event_id, 'commit')
 
         cur.close()
         conn.close()
 
+        # we should clear to queue before exit as non-empty queue
+        # will block join()
+        self._history.close()
+
+
     def start(self):
         self.check_process = Process(target=self.check, args=())
         self.check_process.start()
 
     def stop(self):
-        print('Stopping!');
         self.run.value = False
-        self.check_process.join()
+        self.check_process.terminate()
+
 
 

contrib/raftable/tests/lib/event_history.py

@@ -94,4 +94,14 @@ def aggregate(self):
     def aggregate_by(self, period):
         return
 
+    def close(self):
+        print('closing queue')
+        self.queue.close()
+        print('clearing queue')
+        self.load_queue()
+        print('joining queue')
+        self.queue.cancel_join_thread()
+
+        
+
 

contrib/raftable/tests/test_recovery.py

@@ -15,44 +15,34 @@ def setUp(self):
         self.clients.start()
 
     def tearDown(self):
-        print('tearDown')
         self.clients.stop()
         subprocess.check_call(['blockade','join'])
 
     def test_0_normal_operation(self):
         print('### normalOpsTest ###')
-        print('Waiting 10s to check operability')
-        time.sleep(10)
-
-
-#     def test_1_node_disconnect(self):
-#         print('### disconnectTest ###')
-# 
-#         subprocess.check_call(['blockade','partition','node3'])
-#         print('Node3 disconnected')
-# 
-#         print('Waiting 12s to discover failure')
-#         time.sleep(12)
-#         for client in self.clients:
-#             agg = client.history.aggregate()
-#             print(agg)
-# 
-#         print('Waiting 3s to check operability')
-#         time.sleep(3)
-#         for client in self.clients:
-#             agg = client.history.aggregate()
-#             print(agg)
-# 
-#         subprocess.check_call(['blockade','join'])
-#         print('Node3 connected back')
-# 
-#         print('Waiting 12s for catch-up')
-#         time.sleep(12)
-# 
-#         for client in self.clients:
-#             agg = client.history.aggregate()
-#             print(agg)
+        print('Waiting 5s to check operability')
+        time.sleep(5)
 
+        for client in self.clients:
+            agg = client.history.aggregate()
+            print(agg)
+            # naively check that we did at least some set ops
+            self.assertTrue(agg['setkey']['commit'] > 10)
+
+    def test_1_node_disconnect(self):
+        print('### disconnectTest ###')
+
+        subprocess.check_call(['blockade','partition','node3'])
+        print('Node3 disconnected')
+
+        print('Waiting 5s to discover failure')
+        time.sleep(5)
+
+        for client in self.clients:
+            agg = client.history.aggregate()
+            print(agg)
+            # check we didn't stuck in set op
+            self.assertTrue(agg['setkey']['running_latency'] < 3)
 
 if __name__ == '__main__':
     unittest.main()