add tests in auth_test

Author: sfalkon

tests/auth_test.py

@@ -1,14 +1,16 @@
 """
-Need to install pexpect.
-pip install pexpect
+Description:
+    The Test suite check behavior of pg_probackup utility,
+    if password is required for connection to PostgreSQL instance.
+    - https://confluence.postgrespro.ru/pages/viewpage.action?pageId=16777522
 """
 
 import os
 import unittest
 import tempfile
 import signal
 
-from .helpers.ptrack_helpers import ProbackupTest
+from .helpers.ptrack_helpers import ProbackupTest, ProbackupException
 from testgres import StartNodeException, configure_testgres
 
 module_name = 'auth_test'
@@ -89,6 +91,7 @@ def tearDown(self):
         pass
 
     def test_empty_password(self):
+        """ Test case: PGPB_AUTH03 - zero password length """
         try:
             self.assertIn("ERROR: no password supplied",
                           "".join(map(lambda x: x.decode("utf-8"),
@@ -99,6 +102,7 @@ def test_empty_password(self):
             self.fail(e.value)
 
     def test_wrong_password(self):
+        """ Test case: PGPB_AUTH04 - incorrect password """
         try:
             self.assertIn("password authentication failed",
                           "".join(map(lambda x: x.decode("utf-8"),
@@ -109,6 +113,7 @@ def test_wrong_password(self):
             self.fail(e.value)
 
     def test_right_password(self):
+        """ Test case: PGPB_AUTH01 - correct password """
         try:
             self.assertIn("completed",
                           "".join(map(lambda x: x.decode("utf-8"),
@@ -119,13 +124,61 @@ def test_right_password(self):
             self.fail(e.value)
 
     def test_ctrl_c_event(self):
+        """ Test case: PGPB_AUTH02 - send interrupt signal """
         try:
             run_pb_with_auth(self.cmd, kill=True)
         except TIMEOUT:
             self.fail("Error: CTRL+C event ignored")
 
+    def test_pgpassfile_env(self):
+        path = os.path.join(self.pb.tmp_path, module_name, 'pgpass.conf')
+        line = ":".join(['127.0.0.1', self.node.port, 'postgres', 'backup', 'password'])
+        create_pgpass(path, line)
+        os.environ["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)
+
+    def test_pgpass(self):
+        path = os.path.join(os.path.expanduser('~'), '.pgpass')
+        line = ":".join(['127.0.0.1', self.node.port, 'postgres', 'backup', 'password'])
+        create_pgpass(path, 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)
+
+    def test_pgpassword(self):
+        path = os.path.join(os.path.expanduser('~'), '.pgpass')
+        line = ":".join(['127.0.0.1', self.node.port, 'postgres', 'backup', 'wrong_password'])
+        create_pgpass(path, line)
+        os.environ["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)
+
 
 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()
@@ -134,6 +187,15 @@ def modify_pg_hba(node):
 
 
 def run_pb_with_auth(cmd, password=None, kill=False):
+    """
+    Description:
+        Runnig pg_probackup utility in interactive and send a password or the kill signal.
+    :param cmd:
+    :param password:
+    :param kill:
+    :return stdout:
+    :raises pexpect.TIMEOUT, pexpect.ExceptionPexpect:
+    """
     try:
         with spawn(" ".join(cmd), timeout=10) as probackup:
             result = probackup.expect("Password for user .*:", 5)
@@ -148,3 +210,9 @@ def run_pb_with_auth(cmd, password=None, kill=False):
         raise TIMEOUT("Timeout error.")
     except ExceptionPexpect:
         raise ExceptionPexpect("Pexpect error.")
+
+
+def create_pgpass(path, line):
+    with open(path, 'w') as passfile:
+        # host:port:db:username:password
+        passfile.write(line)