Fix auth_test.py

Author: funny-falcon

tests/__init__.py

@@ -27,7 +27,7 @@ def load_tests(loader, tests, pattern):
         if os.environ['PG_PROBACKUP_LONG'] == 'ON':
             suite.addTests(loader.loadTestsFromModule(time_consuming))
 
-#    suite.addTests(loader.loadTestsFromModule(auth_test))
+    suite.addTests(loader.loadTestsFromModule(auth_test))
     suite.addTests(loader.loadTestsFromModule(archive))
     suite.addTests(loader.loadTestsFromModule(backup))
     suite.addTests(loader.loadTestsFromModule(catchup))

tests/auth_test.py

@@ -126,24 +126,16 @@ def test_backup_via_unprivileged_user(self):
             node.safe_psql(
                 "postgres",
                 "GRANT EXECUTE ON FUNCTION pg_stop_backup(boolean) TO backup")
-        elif self.get_vestion(node) < self.version_to_num('15.0'):
+        elif self.get_version(node) < self.version_to_num('15.0'):
             node.safe_psql(
                 "postgres",
-                "GRANT EXECUTE ON FUNCTION "
-                "pg_stop_backup(boolean, boolean) TO backup")
-            # Do this for ptrack backups
-            node.safe_psql(
-                "postgres",
-                "GRANT EXECUTE ON FUNCTION pg_stop_backup() TO backup")
+                "GRANT EXECUTE ON FUNCTION pg_stop_backup() TO backup; "
+                "GRANT EXECUTE ON FUNCTION pg_stop_backup(boolean, boolean) TO backup;")
         else:
             node.safe_psql(
                 "postgres",
-                "GRANT EXECUTE ON FUNCTION "
-                "pg_backup_stop(boolean) TO backup")
-            # Do this for ptrack backups
-            node.safe_psql(
-                "postgres",
-                "GRANT EXECUTE ON FUNCTION pg_backup_stop() TO backup")
+                "GRANT EXECUTE ON FUNCTION pg_backup_stop() TO backup; "
+                "GRANT EXECUTE ON FUNCTION pg_backup_stop(boolean) TO backup;")
 
         self.backup_node(
                 backup_dir, 'node', node, options=['-U', 'backup'])
@@ -193,7 +185,10 @@ def setUpClass(cls):
             set_replication=True,
             initdb_params=['--data-checksums', '--auth-host=md5']
         )
-        modify_pg_hba(cls.node)
+
+        cls.username = cls.pb.get_username()
+
+        cls.modify_pg_hba(cls.node)
 
         cls.pb.init_pb(cls.backup_dir)
         cls.pb.add_instance(cls.backup_dir, cls.node.name, cls.node)
@@ -203,7 +198,7 @@ def setUpClass(cls):
         except StartNodeException:
             raise unittest.skip("Node hasn't started")
 
-        if cls.pb.get_version(cls.node) < 150000:
+        if cls.pb.get_version(cls.node) < 100000:
             cls.node.safe_psql(
                 "postgres",
                 "CREATE ROLE backup WITH LOGIN PASSWORD 'password'; "
@@ -218,6 +213,21 @@ def setUpClass(cls):
                 "GRANT EXECUTE ON FUNCTION txid_current() TO backup; "
                 "GRANT EXECUTE ON FUNCTION txid_current_snapshot() TO backup; "
                 "GRANT EXECUTE ON FUNCTION txid_snapshot_xmax(txid_snapshot) TO backup;")
+        elif cls.pb.get_version(cls.node) < 150000:
+            cls.node.safe_psql(
+                "postgres",
+                "CREATE ROLE backup WITH LOGIN PASSWORD 'password'; "
+                "GRANT USAGE ON SCHEMA pg_catalog TO backup; "
+                "GRANT EXECUTE ON FUNCTION current_setting(text) TO backup; "
+                "GRANT EXECUTE ON FUNCTION pg_is_in_recovery() TO backup; "
+                "GRANT EXECUTE ON FUNCTION pg_start_backup(text, boolean, boolean) TO backup; "
+                "GRANT EXECUTE ON FUNCTION pg_stop_backup() TO backup; "
+                "GRANT EXECUTE ON FUNCTION pg_stop_backup(boolean, boolean) TO backup; "
+                "GRANT EXECUTE ON FUNCTION pg_create_restore_point(text) TO backup; "
+                "GRANT EXECUTE ON FUNCTION pg_switch_wal() TO backup; "
+                "GRANT EXECUTE ON FUNCTION txid_current() TO backup; "
+                "GRANT EXECUTE ON FUNCTION txid_current_snapshot() TO backup; "
+                "GRANT EXECUTE ON FUNCTION txid_snapshot_xmax(txid_snapshot) TO backup;")
         else:
             cls.node.safe_psql(
                 "postgres",
@@ -229,7 +239,7 @@ def setUpClass(cls):
                 "GRANT EXECUTE ON FUNCTION pg_backup_stop() TO backup; "
                 "GRANT EXECUTE ON FUNCTION pg_backup_stop(boolean) TO backup; "
                 "GRANT EXECUTE ON FUNCTION pg_create_restore_point(text) TO backup; "
-                "GRANT EXECUTE ON FUNCTION pg_switch_xlog() TO backup; "
+                "GRANT EXECUTE ON FUNCTION pg_switch_wal() TO backup; "
                 "GRANT EXECUTE ON FUNCTION txid_current() TO backup; "
                 "GRANT EXECUTE ON FUNCTION txid_current_snapshot() TO backup; "
                 "GRANT EXECUTE ON FUNCTION txid_snapshot_xmax(txid_snapshot) TO backup;")
@@ -244,12 +254,13 @@ def tearDownClass(cls):
 
     @unittest.skipIf(skip_test, "Module pexpect isn't installed. You need to install it.")
     def setUp(self):
-        self.cmd = ['backup',
+        self.pb_cmd = ['backup',
                     '-B', self.backup_dir,
                     '--instance', self.node.name,
                     '-h', '127.0.0.1',
                     '-p', str(self.node.port),
                     '-U', 'backup',
+                    '-d', 'postgres',
                     '-b', 'FULL'
                     ]
 
@@ -269,136 +280,106 @@ def test_empty_password(self):
         """ Test case: PGPB_AUTH03 - zero password length """
         try:
             self.assertIn("ERROR: no password supplied",
-                          str(run_pb_with_auth([self.pb.probackup_path] + self.cmd, '\0\r\n'))
-                          )
+                          self.run_pb_with_auth('\0\r\n'))
         except (TIMEOUT, ExceptionPexpect) as e:
             self.fail(e.value)
 
     def test_wrong_password(self):
         """ Test case: PGPB_AUTH04 - incorrect password """
-        try:
-            self.assertIn("password authentication failed",
-                          str(run_pb_with_auth([self.pb.probackup_path] + self.cmd, 'wrong_password\r\n'))
-                          )
-        except (TIMEOUT, ExceptionPexpect) as e:
-            self.fail(e.value)
+        self.assertIn("password authentication failed",
+                      self.run_pb_with_auth('wrong_password\r\n'))
 
     def test_right_password(self):
         """ Test case: PGPB_AUTH01 - correct password """
-        try:
-            self.assertIn("completed",
-                          str(run_pb_with_auth([self.pb.probackup_path] + self.cmd, 'password\r\n'))
-                          )
-        except (TIMEOUT, ExceptionPexpect) as e:
-            self.fail(e.value)
+        self.assertIn("completed",
+                      self.run_pb_with_auth('password\r\n'))
 
     def test_right_password_and_wrong_pgpass(self):
         """ Test case: PGPB_AUTH05 - correct password and incorrect .pgpass (-W)"""
         line = ":".join(['127.0.0.1', str(self.node.port), 'postgres', 'backup', 'wrong_password'])
-        create_pgpass(self.pgpass_file, line)
-        try:
-            self.assertIn("completed",
-                          str(run_pb_with_auth([self.pb.probackup_path] + self.cmd + ['-W'], 'password\r\n'))
-                          )
-        except (TIMEOUT, ExceptionPexpect) as e:
-            self.fail(e.value)
+        self.create_pgpass(self.pgpass_file, line)
+        self.assertIn("completed",
+                      self.run_pb_with_auth('password\r\n', add_args=["-W"]))
 
     def test_ctrl_c_event(self):
         """ Test case: PGPB_AUTH02 - send interrupt signal """
         try:
-            run_pb_with_auth([self.pb.probackup_path] + self.cmd, kill=True)
+            self.run_pb_with_auth(kill=True)
         except TIMEOUT:
             self.fail("Error: CTRL+C event ignored")
 
     def test_pgpassfile_env(self):
         """ Test case: PGPB_AUTH06 - set environment var PGPASSFILE """
         path = os.path.join(self.pb.tmp_path, module_name, 'pgpass.conf')
         line = ":".join(['127.0.0.1', str(self.node.port), 'postgres', 'backup', 'password'])
-        create_pgpass(path, line)
+        self.create_pgpass(path, line)
         self.pb.test_env["PGPASSFILE"] = path
-        try:
-            self.assertEqual(
-                "OK",
-                self.pb.show_pb(self.backup_dir, self.node.name, self.pb.run_pb(self.cmd + ['-w']))["status"],
-                "ERROR: Full backup status is not valid."
-            )
-        except ProbackupException as e:
-            self.fail(e)
+        self.assertEqual(
+            "OK",
+            self.pb.show_pb(self.backup_dir, self.node.name, self.pb.run_pb(self.pb_cmd + ['-w']))["status"],
+            "ERROR: Full backup status is not valid."
+        )
 
     def test_pgpass(self):
         """ Test case: PGPB_AUTH07 - Create file .pgpass in home dir. """
         line = ":".join(['127.0.0.1', str(self.node.port), 'postgres', 'backup', 'password'])
-        create_pgpass(self.pgpass_file, line)
-        try:
-            self.assertEqual(
-                "OK",
-                self.pb.show_pb(self.backup_dir, self.node.name, self.pb.run_pb(self.cmd + ['-w']))["status"],
-                "ERROR: Full backup status is not valid."
-            )
-        except ProbackupException as e:
-            self.fail(e)
+        self.create_pgpass(self.pgpass_file, line)
+        self.assertEqual(
+            "OK",
+            self.pb.show_pb(self.backup_dir, self.node.name, self.pb.run_pb(self.pb_cmd + ['-w']))["status"],
+            "ERROR: Full backup status is not valid."
+        )
 
     def test_pgpassword(self):
         """ Test case: PGPB_AUTH08 - set environment var PGPASSWORD """
         self.pb.test_env["PGPASSWORD"] = "password"
-        try:
-            self.assertEqual(
-                "OK",
-                self.pb.show_pb(self.backup_dir, self.node.name, self.pb.run_pb(self.cmd + ['-w']))["status"],
-                "ERROR: Full backup status is not valid."
-            )
-        except ProbackupException as e:
-            self.fail(e)
+        self.assertEqual(
+            "OK",
+            self.pb.show_pb(self.backup_dir, self.node.name, self.pb.run_pb(self.pb_cmd + ['-w']))["status"],
+            "ERROR: Full backup status is not valid."
+        )
 
     def test_pgpassword_and_wrong_pgpass(self):
         """ Test case: PGPB_AUTH09 - Check priority between PGPASSWORD and .pgpass file"""
         line = ":".join(['127.0.0.1', str(self.node.port), 'postgres', 'backup', 'wrong_password'])
-        create_pgpass(self.pgpass_file, line)
+        self.create_pgpass(self.pgpass_file, line)
         self.pb.test_env["PGPASSWORD"] = "password"
-        try:
-            self.assertEqual(
-                "OK",
-                self.pb.show_pb(self.backup_dir, self.node.name, self.pb.run_pb(self.cmd + ['-w']))["status"],
-                "ERROR: Full backup status is not valid."
-            )
-        except ProbackupException as e:
-            self.fail(e)
-
+        self.assertEqual(
+            "OK",
+            self.pb.show_pb(self.backup_dir, self.node.name, self.pb.run_pb(self.pb_cmd + ['-w']))["status"],
+            "ERROR: Full backup status is not valid."
+        )
 
-def run_pb_with_auth(cmd, password=None, kill=False):
-    try:
-        with spawn(" ".join(cmd), encoding='utf-8', timeout=10) as probackup:
+    def run_pb_with_auth(self, password=None, add_args = [], kill=False):
+        with spawn(self.pb.probackup_path, self.pb_cmd + add_args, encoding='utf-8', timeout=10) as probackup:
             result = probackup.expect(u"Password for user .*:", 5)
             if kill:
                 probackup.kill(signal.SIGINT)
             elif result == 0:
                 probackup.sendline(password)
                 probackup.expect(EOF)
-                return probackup.before
+                return str(probackup.before)
             else:
                 raise ExceptionPexpect("Other pexpect errors.")
-    except TIMEOUT:
-        raise TIMEOUT("Timeout error.")
-    except ExceptionPexpect:
-        raise ExceptionPexpect("Pexpect error.")
-
-
-def modify_pg_hba(node):
-    """
-    Description:
-        Add trust authentication for user postgres. Need for add new role and set grant.
-    :param node:
-    :return None:
-    """
-    hba_conf = os.path.join(node.data_dir, "pg_hba.conf")
-    with open(hba_conf, 'r+') as fio:
-        data = fio.read()
-        fio.seek(0)
-        fio.write('host\tall\tpostgres\t127.0.0.1/0\ttrust\n' + data)
-
-
-def create_pgpass(path, line):
-    with open(path, 'w') as passfile:
-        # host:port:db:username:password
-        passfile.write(line)
-    os.chmod(path, 0o600)
+
+
+    @classmethod
+    def modify_pg_hba(cls, node):
+        """
+        Description:
+            Add trust authentication for user postgres. Need for add new role and set grant.
+        :param node:
+        :return None:
+        """
+        hba_conf = os.path.join(node.data_dir, "pg_hba.conf")
+        with open(hba_conf, 'r+') as fio:
+            data = fio.read()
+            fio.seek(0)
+            fio.write('host\tall\t%s\t127.0.0.1/0\ttrust\n%s' % (cls.username, data))
+
+
+    def create_pgpass(self, path, line):
+        with open(path, 'w') as passfile:
+            # host:port:db:username:password
+            passfile.write(line)
+        os.chmod(path, 0o600)