PBCKP-588 test partially fixed test_simple_remote.py 41/43

Author: v.shepard

testgres/node.py

@@ -519,8 +519,8 @@ def get_auth_method(t):
                 u"local\treplication\tall\t\t\t{}\n".format(auth_local),
                 u"host\treplication\tall\t127.0.0.1/32\t{}\n".format(auth_host),
                 u"host\treplication\tall\t::1/128\t\t{}\n".format(auth_host),
-                u"host\treplication\t{}\t{}/24\t\t{}\n".format(self.os_ops.username, subnet_base, auth_host),
-                u"host\tall\t{}\t{}/24\t\t{}\n".format(self.os_ops.username, subnet_base, auth_host)
+                u"host\treplication\tall\t{}/24\t\t{}\n".format(subnet_base, auth_host),
+                u"host\tall\tall\t{}/24\t\t{}\n".format(subnet_base, auth_host)
             ]  # yapf: disable
 
             # write missing lines
@@ -790,7 +790,9 @@ def restart(self, params=[]):
         ] + params  # yapf: disable
 
         try:
-            execute_utility(_params, self.utils_log_file)
+            error_code, out, error = execute_utility(_params, self.utils_log_file, verbose=True)
+            if 'could not start server' in error:
+                raise ExecUtilException
         except ExecUtilException as e:
             msg = 'Cannot restart node'
             files = self._collect_special_files()

testgres/operations/remote_ops.py

@@ -1,5 +1,6 @@
 import os
 import tempfile
+import time
 from typing import Optional
 
 import sshtunnel
@@ -46,11 +47,29 @@ def __init__(self, host="127.0.0.1", hostname='localhost', port=None, ssh_key=No
         self.remote = True
         self.ssh = self.ssh_connect()
         self.username = username or self.get_user()
+        self.tunnel = None
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.close_tunnel()
+        if getattr(self, 'ssh', None):
+            self.ssh.close()
 
     def __del__(self):
-        if self.ssh:
+        if getattr(self, 'ssh', None):
             self.ssh.close()
 
+    def close_tunnel(self):
+        if getattr(self, 'tunnel', None):
+            self.tunnel.stop(force=True)
+            start_time = time.time()
+            while self.tunnel.is_active:
+                if time.time() - start_time > sshtunnel.TUNNEL_TIMEOUT:
+                    break
+                time.sleep(0.5)
+
     def ssh_connect(self) -> Optional[SSHClient]:
         if not self.remote:
             return None
@@ -402,26 +421,27 @@ def db_connect(self, dbname, user, password=None, host="127.0.0.1", port=5432, s
         This function establishes a connection to a PostgreSQL database on the remote system using the specified
         parameters. It returns a connection object that can be used to interact with the database.
         """
-        tunnel = sshtunnel.open_tunnel(
+        self.close_tunnel()
+        self.tunnel = sshtunnel.open_tunnel(
             (host, 22),  # Remote server IP and SSH port
             ssh_username=user or self.username,
             ssh_pkey=ssh_key or self.ssh_key,
             remote_bind_address=(host, port),  # PostgreSQL server IP and PostgreSQL port
             local_bind_address=('localhost', port)  # Local machine IP and available port
         )
 
-        tunnel.start()
+        self.tunnel.start()
 
         try:
             conn = pglib.connect(
                 host=host,  # change to 'localhost' because we're connecting through a local ssh tunnel
-                port=tunnel.local_bind_port,  # use the local bind port set up by the tunnel
+                port=self.tunnel.local_bind_port,  # use the local bind port set up by the tunnel
                 dbname=dbname,
                 user=user or self.username,
                 password=password
             )
 
             return conn
         except Exception as e:
-            tunnel.stop()
+            self.tunnel.stop()
             raise e

testgres/pubsub.py

@@ -214,4 +214,4 @@ def catchup(self, username=None):
                 username=username or self.pub.username,
                 max_attempts=LOGICAL_REPL_MAX_CATCHUP_ATTEMPTS)
         except Exception as e:
-            raise_from(CatchUpException("Failed to catch up", query), e)
+            raise_from(CatchUpException("Failed to catch up"), e)

tests/test_simple_remote.py

@@ -6,7 +6,6 @@
 import subprocess
 import tempfile
 
-
 import testgres
 import time
 import six
@@ -138,6 +137,7 @@ def test_init_after_cleanup(self):
     @unittest.skipUnless(util_exists('pg_resetwal'), 'might be missing')
     @unittest.skipUnless(pg_version_ge('9.6'), 'requires 9.6+')
     def test_init_unique_system_id(self):
+        # FAIL
         # this function exists in PostgreSQL 9.6+
         query = 'select system_identifier from pg_control_system()'
 
@@ -291,7 +291,7 @@ def test_psql(self):
             node.safe_psql('copy horns from stdin (format csv)',
                            input=b"1\n2\n3\n\\.\n")
             _sum = node.safe_psql('select sum(w) from horns')
-            self.assertEqual(b'6\n', _sum)
+            self.assertEqual(_sum, b'6\n')
 
             # check psql's default args, fails
             with self.assertRaises(QueryException):
@@ -688,6 +688,7 @@ def test_poll_query_until(self):
             node.poll_query_until('select true')
 
     def test_logging(self):
+        # FAIL
         logfile = tempfile.NamedTemporaryFile('w', delete=True)
 
         log_conf = {
@@ -747,14 +748,11 @@ def test_pgbench(self):
                               options=['-q']).pgbench_run(time=2)
 
             # run TPC-B benchmark
-            proc = node.pgbench(stdout=subprocess.PIPE,
+            out = node.pgbench(stdout=subprocess.PIPE,
                                 stderr=subprocess.STDOUT,
                                 options=['-T3'])
 
-            out, _ = proc.communicate()
-            out = out.decode('utf-8')
-
-            self.assertTrue('tps' in out)
+            self.assertTrue(b'tps = ' in out)
 
     def test_pg_config(self):
         # check same instances
@@ -764,7 +762,6 @@ def test_pg_config(self):
 
         # save right before config change
         c1 = get_pg_config()
-
         # modify setting for this scope
         with scoped_config(cache_pg_config=False) as config:
 
@@ -819,12 +816,16 @@ def test_unix_sockets(self):
             node.init(unix_sockets=False, allow_streaming=True)
             node.start()
 
-            node.execute('select 1')
-            node.safe_psql('select 1')
+            res_exec = node.execute('select 1')
+            res_psql = node.safe_psql('select 1')
+            self.assertEqual(res_exec, [(1,)])
+            self.assertEqual(res_psql, b'1\n')
 
             with node.replicate().start() as r:
-                r.execute('select 1')
-                r.safe_psql('select 1')
+                res_exec = r.execute('select 1')
+                res_psql = r.safe_psql('select 1')
+                self.assertEqual(res_exec, [(1,)])
+                self.assertEqual(res_psql, b'1\n')
 
     def test_auto_name(self):
         with get_remote_node().init(allow_streaming=True).start() as m: