db/pg_probackup
db/pg_probackup
1
0
Fork 0
mirror of https://github.com/postgrespro/pg_probackup.git synced 2025-03-28 22:48:39 +02:00
Code Issues Releases Activity

fix auth_test

Browse Source
This commit is contained in:
sfalkon 2017-12-05 17:38:00 +03:00
parent 065fb0e539
commit d8953f580a
1 changed files with 21 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
tests/auth_test.py
View File

@ -11,7 +11,7 @@ import tempfile
import signal
from .helpers.ptrack_helpers import ProbackupTest, ProbackupException
from testgres import StartNodeException, configure_testgres
from testgres import StartNodeException
module_name = 'auth_test'
skip_test = False
@ -77,7 +77,7 @@ class AuthTest(unittest.TestCase):
@unittest.skipIf(skip_test, "Module pexpect isn't installed. You need to install it.")
def setUp(self):
self.cmd = [self.pb.probackup_path, 'backup',
self.cmd = ['backup',
'-B', self.backup_dir,
'--instance', self.node.name,
'-h', '127.0.0.1',
@ -94,7 +94,7 @@ class AuthTest(unittest.TestCase):
try:
self.assertIn("ERROR: no password supplied",
"".join(map(lambda x: x.decode("utf-8"),
run_pb_with_auth(self.cmd, '\0\r\n'))
self.run_pb_with_auth(self.cmd, '\0\r\n'))
)
)
except (TIMEOUT, ExceptionPexpect) as e:
@ -105,7 +105,7 @@ class AuthTest(unittest.TestCase):
try:
self.assertIn("password authentication failed",
"".join(map(lambda x: x.decode("utf-8"),
run_pb_with_auth(self.cmd, 'wrong_password\r\n'))
self.run_pb_with_auth(self.cmd, 'wrong_password\r\n'))
)
)
except (TIMEOUT, ExceptionPexpect) as e:
@ -116,7 +116,7 @@ class AuthTest(unittest.TestCase):
try:
self.assertIn("completed",
"".join(map(lambda x: x.decode("utf-8"),
run_pb_with_auth(self.cmd, 'password\r\n'))
self.run_pb_with_auth(self.cmd, 'password\r\n'))
)
)
except (TIMEOUT, ExceptionPexpect) as e:
@ -125,7 +125,7 @@ class AuthTest(unittest.TestCase):
def test_ctrl_c_event(self):
""" Test case: PGPB_AUTH02 - send interrupt signal """
try:
run_pb_with_auth(self.cmd, kill=True)
self.run_pb_with_auth(self.cmd, kill=True)
except TIMEOUT:
self.fail("Error: CTRL+C event ignored")
@ -170,6 +170,21 @@ class AuthTest(unittest.TestCase):
except ProbackupException as e:
self.fail(e)
def run_pb_with_auth(self, password=None, kill=False):
try:
with spawn(" ".join(self.pb.probackup_path + self.cmd), timeout=10) as probackup:
result = probackup.expect("Password for user .*:", 5)
if kill:
probackup.kill(signal.SIGINT)
elif result == 0:
probackup.sendline(password)
return probackup.readlines()
else:
raise TIMEOUT("")
except TIMEOUT:
raise TIMEOUT("Timeout error.")
except ExceptionPexpect:
raise ExceptionPexpect("Pexpect error.")
def modify_pg_hba(node):
"""
@ -184,33 +199,6 @@ def modify_pg_hba(node):
fio.seek(0)
fio.write('host\tall\tpostgres\t127.0.0.1/0\ttrust\n' + data)
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)
if kill:
probackup.kill(signal.SIGINT)
elif result == 0:
probackup.sendline(password)
return probackup.readlines()
else:
raise TIMEOUT("")
except TIMEOUT:
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