tests: remove psycopg2

Author: gsmolk

tests/Readme.md

@@ -32,7 +32,6 @@ Run suit of basic simple tests:
 
 Usage:
  pip install testgres
- pip install psycopg2
  export PG_CONFIG=/path/to/pg_config
  python -m unittest [-v] tests[.specific_module][.class.test]
 ```

tests/delta.py

@@ -5,6 +5,7 @@
 from testgres import QueryException
 import subprocess
 import time
+from threading import Thread
 
 
 module_name = 'delta'
@@ -522,48 +523,40 @@ def test_delta_vacuum_full(self):
             " as id from generate_series(0,1000000) i"
             )
 
-        # create async connection
-        conn = self.get_async_connect(port=node.port)
+        pg_connect = node.connect("postgres", autocommit=True)
 
-        self.wait(conn)
-
-        acurs = conn.cursor()
-        acurs.execute("select pg_backend_pid()")
-
-        self.wait(conn)
-        pid = acurs.fetchall()[0][0]
-        print(pid)
-
-        gdb = self.gdb_attach(pid)
+        gdb = self.gdb_attach(pg_connect.pid)
         gdb.set_breakpoint('reform_and_rewrite_tuple')
 
         gdb.continue_execution_until_running()
 
-        acurs.execute("VACUUM FULL t_heap")
+        process = Thread(
+            target=pg_connect.execute, args=["VACUUM FULL t_heap"])
+        process.start()
 
-        if gdb.stopped_in_breakpoint():
-            gdb.continue_execution_until_break(20)
+        while not gdb.stopped_in_breakpoint:
+            sleep(1)
 
-        self.backup_node(
-            backup_dir, 'node', node,
-            backup_type='delta', options=['--stream']
-        )
+        gdb.continue_execution_until_break(20)
 
         self.backup_node(
             backup_dir, 'node', node,
-            backup_type='delta', options=['--stream']
-        )
+            backup_type='delta', options=['--stream'])
+
         if self.paranoia:
             pgdata = self.pgdata_content(node.data_dir)
 
+        gdb.remove_all_breakpoints()
+        gdb._execute('detach')
+        process.join()
+
         old_tablespace = self.get_tblspace_path(node, 'somedata')
         new_tablespace = self.get_tblspace_path(node_restored, 'somedata_new')
 
         self.restore_node(
             backup_dir, 'node', node_restored,
             options=["-j", "4", "-T", "{0}={1}".format(
-                old_tablespace, new_tablespace)]
-        )
+                old_tablespace, new_tablespace)])
 
         # Physical comparison
         if self.paranoia:
@@ -1125,7 +1118,6 @@ def test_delta_corruption_heal_via_ptrack_1(self):
             backup_type="delta",
             options=["-j", "4", "--stream", '--log-level-file=verbose'])
 
-
         # open log file and check
         with open(os.path.join(backup_dir, 'log', 'pg_probackup.log')) as f:
             log_content = f.read()
@@ -1254,9 +1246,9 @@ def test_delta_nullified_heap_page_backup(self):
             backup_dir, 'node', node)
 
         # Nullify some block in PostgreSQL
-        file = os.path.join(node.data_dir, file_path).replace("\\","/")
+        file = os.path.join(node.data_dir, file_path).replace("\\", "/")
         if os.name == 'nt':
-            file = file.replace("\\","/")
+            file = file.replace("\\", "/")
 
         with open(file, 'r+b', 0) as f:
             f.seek(8192)
@@ -1281,7 +1273,7 @@ def test_delta_nullified_heap_page_backup(self):
                 content)
             self.assertNotIn(
                 "Skipping blknum: 1 in file: {0}".format(file),
-                 content)
+                content)
 
         # Restore DELTA backup
         node_restored = self.make_simple_node(

tests/helpers/ptrack_helpers.py

@@ -9,7 +9,6 @@
 import re
 import getpass
 import select
-import psycopg2
 from time import sleep
 import re
 import json
@@ -1408,31 +1407,6 @@ def compare_pgdata(self, original_pgdata, restored_pgdata):
                 fail = True
         self.assertFalse(fail, error_message)
 
-    def get_async_connect(self, database=None, host=None, port=5432):
-        if not database:
-            database = 'postgres'
-        if not host:
-            host = '127.0.0.1'
-
-        return psycopg2.connect(
-            database='postgres',
-            host='127.0.0.1',
-            port=port,
-            async_=True
-        )
-
-    def wait(self, connection):
-        while True:
-            state = connection.poll()
-            if state == psycopg2.extensions.POLL_OK:
-                break
-            elif state == psycopg2.extensions.POLL_WRITE:
-                select.select([], [connection.fileno()], [])
-            elif state == psycopg2.extensions.POLL_READ:
-                select.select([connection.fileno()], [], [])
-            else:
-                raise psycopg2.OperationalError('poll() returned %s' % state)
-
     def gdb_attach(self, pid):
         return GDBobj([str(pid)], self.verbose, attach=True)
 
@@ -1540,7 +1514,7 @@ def remove_all_breakpoints(self):
                 return
 
         raise GdbException(
-            'Failed to set breakpoint.\n Output:\n {0}'.format(result)
+            'Failed to remove breakpoints.\n Output:\n {0}'.format(result)
         )
 
     def run_until_break(self):
@@ -1632,6 +1606,18 @@ def _execute(self, cmd, running=True):
         self.proc.stdin.flush()
         self.proc.stdin.write(cmd + '\n')
         self.proc.stdin.flush()
+        sleep(1)
+
+        # look for command we just send
+        while True:
+            line = self.proc.stdout.readline()
+            if self.verbose:
+                print(repr(line))
+
+            if cmd not in line:
+                continue
+            else:
+                break
 
         while True:
             line = self.proc.stdout.readline()

tests/ptrack.py

@@ -7,6 +7,7 @@
 import shutil
 import sys
 import time
+from threading import Thread
 
 
 module_name = 'ptrack'
@@ -213,27 +214,21 @@ def test_ptrack_vacuum_full(self):
             " as id from generate_series(0,1000000) i"
             )
 
-        # create async connection
-        conn = self.get_async_connect(port=node.port)
+        pg_connect = node.connect("postgres", autocommit=True)
 
-        self.wait(conn)
-
-        acurs = conn.cursor()
-        acurs.execute("select pg_backend_pid()")
-
-        self.wait(conn)
-        pid = acurs.fetchall()[0][0]
-        print(pid)
-
-        gdb = self.gdb_attach(pid)
+        gdb = self.gdb_attach(pg_connect.pid)
         gdb.set_breakpoint('reform_and_rewrite_tuple')
 
         gdb.continue_execution_until_running()
 
-        acurs.execute("VACUUM FULL t_heap")
+        process = Thread(
+            target=pg_connect.execute, args=["VACUUM FULL t_heap"])
+        process.start()
 
-        if gdb.stopped_in_breakpoint():
-            gdb.continue_execution_until_break(20)
+        while not gdb.stopped_in_breakpoint:
+            sleep(1)
+
+        gdb.continue_execution_until_break(20)
 
         self.backup_node(
             backup_dir, 'node', node, backup_type='ptrack')
@@ -244,6 +239,10 @@ def test_ptrack_vacuum_full(self):
         if self.paranoia:
             pgdata = self.pgdata_content(node.data_dir)
 
+        gdb.remove_all_breakpoints()
+        gdb._execute('detach')
+        process.join()
+
         old_tablespace = self.get_tblspace_path(node, 'somedata')
         new_tablespace = self.get_tblspace_path(node_restored, 'somedata_new')