insertion tests for referee

Author: kelvich

tests2/lib/bank_client.py

@@ -13,6 +13,7 @@
 import logging
 import re
 import pprint
+import uuid
 
 class MtmTxAggregate(object):
 
@@ -80,7 +81,6 @@ def __init__(self, dsns, n_accounts=100000):
         self.total = 0
         self.aggregates = [{} for e in dsns]
         keep_trying(40, 1, self.initdb, 'self.initdb')
-        self.running = True
         self.nodes_state_fields = ["id", "disabled", "disconnected", "catchUp", "slotLag",
             "avgTransDelay", "lastStatusChange", "oldestSnapshot", "SenderPid",
             "SenderStartTime ", "ReceiverPid", "ReceiverStartTime", "connStr"]
@@ -117,6 +117,7 @@ def initdb(self):
         conn.commit()
         cur.execute('drop table if exists bank_test')
         cur.execute('create table bank_test(uid int primary key, amount int)')
+        cur.execute('create table insert_test(id text primary key)')
         cur.execute('''
                 insert into bank_test
                 select *, 0 from generate_series(0, %s)''',
@@ -166,6 +167,19 @@ def no_prepared_tx(self):
         print("n_prepared = %d" % (n_prepared))
         return (n_prepared)
 
+    def insert_counts(self):
+        counts = []
+
+        for dsn in self.dsns:
+            con = psycopg2.connect(dsn)
+            cur = con.cursor()
+            cur.execute("select count(*) from insert_test;")
+            counts.append(int(cur.fetchone()[0]))
+            cur.close()
+            con.close()
+
+        return counts
+
     @asyncio.coroutine
     def status(self):
         while self.running:
@@ -267,6 +281,11 @@ def transfer_tx(self, conn, cur, agg, conn_i):
         assert(cur.rowcount == 1)
         yield from cur.execute('commit')
 
+    @asyncio.coroutine
+    def insert_tx(self, conn, cur, agg, conn_i):
+        query = "insert into insert_test values ('%s')" % (uuid.uuid4())
+        yield from cur.execute(query)
+
     @asyncio.coroutine
     def total_tx(self, conn, cur, agg, conn_i):
         yield from cur.execute("select sum(amount), count(*), count(uid), current_setting('multimaster.node_id') from bank_test")
@@ -289,16 +308,21 @@ def run(self):
         self.loop = asyncio.get_event_loop()
 
         for i, _ in enumerate(self.dsns):
-            for j in range(3):
+            for j in range(1):
                 asyncio.ensure_future(self.exec_tx(self.transfer_tx, i, 'transfer', j))
             asyncio.ensure_future(self.exec_tx(self.total_tx, i, 'sumtotal', 0))
+            for j in range(2):
+                asyncio.ensure_future(self.exec_tx(self.insert_tx, i, 'inserter', j))
 
         asyncio.ensure_future(self.status())
 
         self.loop.run_forever()
 
     def bgrun(self):
         print('Starting evloop in different process')
+
+        self.running = True
+
         self.parent_pipe, self.child_pipe = aioprocessing.AioPipe()
         self.evloop_process = multiprocessing.Process(target=self.run, args=())
         self.evloop_process.start()

tests2/lib/test_helper.py

@@ -40,15 +40,33 @@ def awaitCommit(self, node_id):
 
         while total_sleep <= TEST_MAX_RECOVERY_TIME:
             aggs = self.client.get_aggregates(clean=False, _print=False)
-            # print('=== ',aggs[node_id]['transfer']['finish'])
+            print('=== ',aggs[node_id]['transfer']['finish'])
             if ('commit' in aggs[node_id]['transfer']['finish'] and
                     aggs[node_id]['transfer']['finish']['commit'] > 10):
                 break
             time.sleep(5)
             total_sleep += 5
 
+    def awaitOnline(self, dsn):
+        total_sleep = 0
+        one = 0
 
-    def performFailure(self, failure, wait=0, node_wait_for_commit=-1):
+        while total_sleep <= TEST_MAX_RECOVERY_TIME:
+            try:
+                con = psycopg2.connect(dsn + " connect_timeout=1")
+                cur = con.cursor()
+                cur.execute("select 1")
+                one = int(cur.fetchone()[0])
+                cur.close()
+                con.close()
+                print("Online!")
+                break
+            except Exception as e:
+                print('Waiting for online:', str(e))
+                time.sleep(5)
+                total_sleep += 5
+
+    def performFailure(self, failure, wait=0, node_wait_for_commit=-1, node_wait_for_online=None, stop_load=False):
 
         time.sleep(TEST_WARMING_TIME)
 
@@ -72,12 +90,27 @@ def performFailure(self, failure, wait=0, node_wait_for_commit=-1):
 
         self.client.clean_aggregates()
 
+        if stop_load:
+            time.sleep(3)
+            self.client.get_aggregates(clean=False)
+            self.client.stop()
+
         if node_wait_for_commit >= 0:
             self.awaitCommit(node_wait_for_commit)
         else:
             time.sleep(TEST_RECOVERY_TIME)
 
+        if node_wait_for_online != None:
+            self.awaitOnline(node_wait_for_online)
+        else:
+            time.sleep(TEST_RECOVERY_TIME)
+
+        if stop_load:
+            self.client.bgrun()
+            time.sleep(3)
+
         aggs = self.client.get_aggregates()
+
         return (aggs_failure, aggs)
 
     def nodeExecute(dsn, statements):

tests2/test_referee.py

@@ -56,6 +56,22 @@ def setUp(self):
     def tearDown(self):
         print('Finish test at ',datetime.datetime.utcnow())
 
+    def test_neighbor_restart(self):
+        print('### test_neighbor_restart ###')
+
+        aggs_failure, aggs = self.performFailure(RestartNode('node2'), node_wait_for_online="dbname=regression user=postgres host=127.0.0.1 port=15433", stop_load=True)
+
+        self.assertCommits(aggs_failure[:1])
+        self.assertNoCommits(aggs_failure[1:])
+        self.assertIsolation(aggs_failure)
+
+        self.assertCommits(aggs)
+        self.assertIsolation(aggs)
+
+        counts = self.client.insert_counts()
+        self.assertEqual(counts[0], counts[1])
+
+
     def test_node_crash(self):
         print('### test_node_crash ###')
 
@@ -68,6 +84,9 @@ def test_node_crash(self):
         self.assertCommits(aggs)
         self.assertIsolation(aggs)
 
+        counts = self.client.insert_counts()
+        self.assertEqual(counts[0], counts[1])
+
 
     def test_partition_referee(self):
         print('### test_partition_referee ###')
@@ -81,6 +100,9 @@ def test_partition_referee(self):
         self.assertCommits(aggs)
         self.assertIsolation(aggs)
 
+        counts = self.client.insert_counts()
+        self.assertEqual(counts[0], counts[1])
+
     def test_double_failure_referee(self):
         print('### test_double_failure_referee ###')
 
@@ -102,6 +124,9 @@ def test_double_failure_referee(self):
         self.assertCommits(aggs)
         self.assertIsolation(aggs)
 
+        counts = self.client.insert_counts()
+        self.assertEqual(counts[0], counts[1])
+
     def test_saved_referee_decision(self):
         print('### test_saved_referee_decision ###')
         docker_api = docker.from_env()
@@ -173,6 +198,9 @@ def test_saved_referee_decision(self):
 
         self.assertEqual(decisions_count, 0)
 
+        counts = self.client.insert_counts()
+        self.assertEqual(counts[0], counts[1])
+
     def test_winner_restart(self):
         print('### test_winner_restart ###')
 
@@ -200,6 +228,9 @@ def test_winner_restart(self):
         docker_api.containers.get('node1').start()
         self.awaitCommit(0)
 
+        counts = self.client.insert_counts()
+        self.assertEqual(counts[0], counts[1])
+
 
 if __name__ == '__main__':
     unittest.main()