mirror of
https://github.com/mail-in-a-box/mailinabox.git
synced 2025-04-20 02:52:11 +00:00
tests: added test for backup functionality
This commit is contained in:
parent
f0b0108eb4
commit
32b1176a07
45
test/test_backup.py
Normal file
45
test/test_backup.py
Normal file
@ -0,0 +1,45 @@
|
|||||||
|
import pytest
|
||||||
|
from time import sleep
|
||||||
|
from subprocess import check_call, check_output
|
||||||
|
import smtplib
|
||||||
|
|
||||||
|
from settings import *
|
||||||
|
from common import random_id
|
||||||
|
from test_mail import new_message, check_imap_received
|
||||||
|
|
||||||
|
|
||||||
|
def test_backup_mail():
|
||||||
|
# send a mail, to ensure we have something to backup
|
||||||
|
msg, subject = new_message(TEST_ADDRESS, TEST_ADDRESS)
|
||||||
|
s = smtplib.SMTP(TEST_DOMAIN, 587)
|
||||||
|
s.starttls()
|
||||||
|
s.login(TEST_ADDRESS, TEST_PASSWORD)
|
||||||
|
s.sendmail(TEST_ADDRESS, [TEST_ADDRESS], msg)
|
||||||
|
s.quit()
|
||||||
|
|
||||||
|
# trigger a backup
|
||||||
|
sleep(2)
|
||||||
|
cmd_ssh = "sshpass -p vagrant ssh vagrant@{} -p {} ".format(TEST_SERVER, TEST_PORT)
|
||||||
|
cmd_count = cmd_ssh + "ls -l /home/user-data/backup/encrypted | wc -l"
|
||||||
|
num_backup_files = int(check_output(cmd_count, shell=True))
|
||||||
|
cmd = cmd_ssh + "sudo /vagrant/management/backup.py"
|
||||||
|
check_call(cmd, shell=True)
|
||||||
|
num_backup_files_new = int(check_output(cmd_count, shell=True))
|
||||||
|
assert num_backup_files_new > num_backup_files
|
||||||
|
|
||||||
|
# delete mail
|
||||||
|
assert check_imap_received(subject)
|
||||||
|
assert not check_imap_received(subject)
|
||||||
|
|
||||||
|
# restore backup
|
||||||
|
path = "/home/user-data"
|
||||||
|
passphrase = "export PASSPHRASE=\$(sudo cat /home/user-data/backup/secret_key.txt) &&"
|
||||||
|
# extract to temp directory
|
||||||
|
restore = "sudo -E duplicity restore --force file://{0}/backup/encrypted {0}/restore &&".format(path)
|
||||||
|
# move restored backup using rsync, because it allows to overwrite files
|
||||||
|
move = "sudo rsync -av {0}/restore/* {0}/ &&".format(path)
|
||||||
|
rm = "sudo rm -rf {0}/restore/".format(path)
|
||||||
|
check_call(cmd_ssh + "\"" + passphrase + restore + move + rm + "\"", shell=True)
|
||||||
|
|
||||||
|
# check the mail is there again
|
||||||
|
assert check_imap_received(subject)
|
@ -18,9 +18,10 @@ def new_message(from_email, to_email):
|
|||||||
msg['From'] = from_email
|
msg['From'] = from_email
|
||||||
msg['To'] = to_email
|
msg['To'] = to_email
|
||||||
return msg.as_string(), msg['subject']
|
return msg.as_string(), msg['subject']
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def assert_imap_received(subject):
|
def check_imap_received(subject):
|
||||||
"""Connects with IMAP and asserts the existence of an email, then deletes it"""
|
"""Connects with IMAP and asserts the existence of an email, then deletes it"""
|
||||||
|
|
||||||
sleep(3)
|
sleep(3)
|
||||||
@ -30,15 +31,16 @@ def assert_imap_received(subject):
|
|||||||
m.login(TEST_ADDRESS, TEST_PASSWORD)
|
m.login(TEST_ADDRESS, TEST_PASSWORD)
|
||||||
m.select()
|
m.select()
|
||||||
|
|
||||||
# Assert the message exists
|
# check the message exists
|
||||||
typ, data = m.search(None, '(SUBJECT \"{}\")'.format(subject))
|
typ, data = m.search(None, '(SUBJECT \"{}\")'.format(subject))
|
||||||
assert len(data[0].split()) == 1
|
res = len(data[0].split()) == 1
|
||||||
|
|
||||||
# Delete it & logout
|
if res:
|
||||||
m.store(data[0].strip(), '+FLAGS', '\\Deleted')
|
m.store(data[0].strip(), '+FLAGS', '\\Deleted')
|
||||||
m.expunge()
|
m.expunge()
|
||||||
m.close()
|
m.close()
|
||||||
m.logout()
|
m.logout()
|
||||||
|
return res
|
||||||
|
|
||||||
|
|
||||||
def assert_pop3_received(subject):
|
def assert_pop3_received(subject):
|
||||||
@ -81,7 +83,7 @@ def test_smtps():
|
|||||||
s.login(TEST_ADDRESS, TEST_PASSWORD)
|
s.login(TEST_ADDRESS, TEST_PASSWORD)
|
||||||
s.sendmail(TEST_ADDRESS, [TEST_ADDRESS], msg)
|
s.sendmail(TEST_ADDRESS, [TEST_ADDRESS], msg)
|
||||||
s.quit()
|
s.quit()
|
||||||
assert_imap_received(subject)
|
assert check_imap_received(subject)
|
||||||
|
|
||||||
|
|
||||||
def test_smtps_tag():
|
def test_smtps_tag():
|
||||||
@ -93,7 +95,7 @@ def test_smtps_tag():
|
|||||||
s.login(TEST_ADDRESS, TEST_PASSWORD)
|
s.login(TEST_ADDRESS, TEST_PASSWORD)
|
||||||
s.sendmail(TEST_ADDRESS, [mail_address], msg)
|
s.sendmail(TEST_ADDRESS, [mail_address], msg)
|
||||||
s.quit()
|
s.quit()
|
||||||
assert_imap_received(subject)
|
assert check_imap_received(subject)
|
||||||
|
|
||||||
|
|
||||||
def test_smtps_requires_auth():
|
def test_smtps_requires_auth():
|
||||||
@ -117,7 +119,7 @@ def test_smtp():
|
|||||||
s = smtplib.SMTP(TEST_DOMAIN, 25)
|
s = smtplib.SMTP(TEST_DOMAIN, 25)
|
||||||
s.sendmail(TEST_SENDER, [TEST_ADDRESS], msg)
|
s.sendmail(TEST_SENDER, [TEST_ADDRESS], msg)
|
||||||
s.quit()
|
s.quit()
|
||||||
assert_imap_received(subject)
|
assert check_imap_received(subject)
|
||||||
|
|
||||||
|
|
||||||
def test_smtp_tls():
|
def test_smtp_tls():
|
||||||
@ -127,7 +129,7 @@ def test_smtp_tls():
|
|||||||
s.starttls()
|
s.starttls()
|
||||||
s.sendmail(TEST_SENDER, [TEST_ADDRESS], msg)
|
s.sendmail(TEST_SENDER, [TEST_ADDRESS], msg)
|
||||||
s.quit()
|
s.quit()
|
||||||
assert_imap_received(subject)
|
assert check_imap_received(subject)
|
||||||
|
|
||||||
|
|
||||||
# FIXME
|
# FIXME
|
||||||
@ -193,7 +195,7 @@ def test_smtp_headers():
|
|||||||
|
|
||||||
|
|
||||||
def test_pop3s():
|
def test_pop3s():
|
||||||
"""Connects with POP3S and asserts the existance of an email, then deletes it"""
|
"""Connects with POP3S and asserts the existance of an email"""
|
||||||
msg, subject = new_message(TEST_ADDRESS, TEST_ADDRESS)
|
msg, subject = new_message(TEST_ADDRESS, TEST_ADDRESS)
|
||||||
s = smtplib.SMTP(TEST_DOMAIN, 587)
|
s = smtplib.SMTP(TEST_DOMAIN, 587)
|
||||||
s.starttls()
|
s.starttls()
|
||||||
|
Loading…
Reference in New Issue
Block a user