reimplement support for pg_dump, fix tests, provide a hack for pg8000

Author: funbringer

testgres/testgres.py

@@ -228,7 +228,12 @@ def execute(self, query, *args):
         self.cursor.execute(query, args)
 
         try:
-            return self.cursor.fetchall()
+            res = self.cursor.fetchall()
+
+            if isinstance(res, tuple):
+                res = [tuple(t) for t in res]
+
+            return res
         except Exception:
             return None
 
@@ -752,37 +757,41 @@ def safe_psql(self, dbname, query, username=None):
             raise QueryException(six.text_type(err))
         return out
 
-    def dump(self, dbname, filename):
+    def dump(self, dbname, filename=None):
         """
         Dump database using pg_dump.
 
         Args:
             dbname: database name to connect to (str).
             filename: output file (str).
+
+        Returns:
+            Path to file containing dump.
         """
 
-        path = os.path.join(self.base_dir, filename)
+        f, filename = filename or tempfile.mkstemp()
+        os.close(f)
+
         _params = [
             "-p{}".format(self.port),
-            "-f", path, dbname
+            "-f{}".format(filename),
+            dbname
         ]
 
         _execute_utility("pg_dump", _params, self.utils_logname)
 
-    def restore(self, dbname, filename, node=None):
+        return filename
+
+    def restore(self, dbname, filename):
         """
         Restore database from pg_dump's file.
 
         Args:
             dbname: database name to connect to (str).
-            filename: output file (str).
+            filename: database dump taken by pg_dump (str).
         """
 
-        if not node:
-            node = self
-
-        path = os.path.join(node.base_dir, filename)
-        self.psql(dbname, filename=path)
+        self.psql(dbname, filename=filename)
 
     def poll_query_until(self, dbname, query):
         """

testgres/tests/test_simple.py

@@ -97,11 +97,11 @@ def test_simple_queries(self):
             self.assertEqual(res, b'1\n')
 
             res = node.execute('postgres', 'select 1')
-            self.assertEqual(res, ([1], ))
+            self.assertListEqual(res, [(1, )])
 
             with node.connect('postgres') as con:
                 res = con.execute('select 1')
-                self.assertEqual(res, ([1], ))
+                self.assertListEqual(res, [(1, )])
 
     def test_transactions(self):
         with get_new_node('test') as node:
@@ -116,12 +116,12 @@ def test_transactions(self):
                 con.begin()
                 con.execute('insert into test values (2)')
                 res = con.execute('select * from test order by val asc')
-                self.assertEqual(res, ([1], [2]))
+                self.assertListEqual(res, [(1, ), (2, )])
                 con.rollback()
 
                 con.begin()
                 res = con.execute('select * from test')
-                self.assertEqual(res, ([1], ))
+                self.assertListEqual(res, [(1, )])
                 con.rollback()
 
                 con.begin()
@@ -165,7 +165,7 @@ def test_backup_simple(self):
                     slave.start()
                     res = slave.execute('postgres',
                                         'select * from test order by i asc')
-                    self.assertEqual(res, ([1], [2], [3], [4]))
+                    self.assertListEqual(res, [(1, ), (2, ), (3, ), (4, )])
 
     def test_backup_multiple(self):
         with get_new_node('node') as node:
@@ -210,53 +210,52 @@ def test_backup_and_replication(self):
 
             backup = node.backup()
 
-            replica = backup.spawn_replica('replica')
-            replica.start()
-            res = replica.execute('postgres', 'select * from abc')
-            self.assertEqual(len(res), 1)
-            self.assertEqual(res[0][0], 1)
-            self.assertEqual(res[0][1], 2)
-
-            cur_ver = LooseVersion(get_pg_config()['VERSION_NUM'])
-            min_ver = LooseVersion('10')
-
-            # Prepare the query which would check whether record reached replica
-            # (It is slightly different for Postgres 9.6 and Postgres 10+)
-            if cur_ver >= min_ver:
-                wait_lsn = 'SELECT pg_current_wal_lsn() <= replay_lsn '           \
-                    'FROM pg_stat_replication WHERE application_name = \'%s\''    \
-                    % replica.name
-            else:
-                wait_lsn = 'SELECT pg_current_xlog_location() <= replay_location '\
-                    'FROM pg_stat_replication WHERE application_name = \'%s\''    \
-                    % replica.name
-
-            # Insert into master node
-            node.psql('postgres', 'insert into abc values (3, 4)')
-            # Wait until data syncronizes
-            node.poll_query_until('postgres', wait_lsn)
-            # Check that this record was exported to replica
-            res = replica.execute('postgres', 'select * from abc')
-            self.assertEqual(len(res), 2)
-            self.assertEqual(res[1][0], 3)
-            self.assertEqual(res[1][1], 4)
-
-            # check manual cleanup
-            replica.cleanup()
-            replica.free_port()
+            with backup.spawn_replica('replica') as replica:
+                replica.start()
+                res = replica.execute('postgres', 'select * from abc')
+                self.assertListEqual(res, [(1, 2)])
+
+                cur_ver = LooseVersion(get_pg_config()['VERSION_NUM'])
+                min_ver = LooseVersion('10')
+
+                # Prepare the query which would check whether record reached replica
+                # (It is slightly different for Postgres 9.6 and Postgres 10+)
+                if cur_ver >= min_ver:
+                    wait_lsn = 'SELECT pg_current_wal_lsn() <= replay_lsn '           \
+                        'FROM pg_stat_replication WHERE application_name = \'%s\''    \
+                        % replica.name
+                else:
+                    wait_lsn = 'SELECT pg_current_xlog_location() <= replay_location '\
+                        'FROM pg_stat_replication WHERE application_name = \'%s\''    \
+                        % replica.name
+
+                # Insert into master node
+                node.psql('postgres', 'insert into abc values (3, 4)')
+                # Wait until data syncronizes
+                node.poll_query_until('postgres', wait_lsn)
+                # Check that this record was exported to replica
+                res = replica.execute('postgres', 'select * from abc')
+                self.assertListEqual(res, [(1, 2), (3, 4)])
 
     def test_dump(self):
-        with get_new_node('test') as node:
-            node.init().start()
-            node.safe_psql(
-                'postgres', 'create table abc as '
-                'select g as a, g as b from generate_series(1, 10) as g')
-            node.psql('postgres', 'create database test')
-            node.dump('postgres', 'test.sql')
-            node.restore('test', 'test.sql')
-            self.assertEqual(
-                node.psql('postgres', 'select * from abc'),
-                node.psql('test', 'select * from abc'), )
+        with get_new_node('node1') as node1:
+            node1.init().start()
+
+            with node1.connect('postgres') as con:
+                con.begin()
+                con.execute('create table test (val int)')
+                con.execute('insert into test values (1), (2)')
+                con.commit()
+
+            # take a new dump
+            dump = node1.dump('postgres')
+
+            with get_new_node('node2') as node2:
+                node2.init().start().restore('postgres', dump)
+
+                res = node2.execute('postgres',
+                                    'select * from test order by val asc')
+                self.assertListEqual(res, [(1, ), (2, )])
 
     def test_users(self):
         with get_new_node('master') as node: