forked from stove/dataset
201 lines
6.8 KiB
Python
201 lines
6.8 KiB
Python
from yaml import load, SafeLoader
|
|
from os import environ
|
|
import pytest
|
|
import datetime
|
|
|
|
from imaplib2 import IMAP4_SSL
|
|
from smtplib import SMTP, SMTPNotSupportedError, SMTPAuthenticationError
|
|
|
|
|
|
|
|
conf_file = f'{environ["MACHINE_TEST_DIR"]}/imap.yml'
|
|
with open(conf_file) as yaml:
|
|
data = load(yaml, Loader=SafeLoader)
|
|
parameters = (
|
|
(1, 5, 'user', data['username'], data['username'], data['username'], [data['password']]),
|
|
(2, 5, 'user', data['username'], data['username'], 'alias_' + data['username'], [data['password']]),
|
|
(1, 3, 'family', data['username_family'], data['username_family'], data['username_family'], [data['password_family'], data['password_family'] + "2"]),
|
|
(3, 5, 'user', data['username'], data['ext_username'], data['username'], [data['password']]),
|
|
(4, 5, 'user', data['username'], data['ext_username'], 'alias_' + data['username'], [data['password']]),
|
|
(2, 3, 'family', data['username_family'], data['ext_username'], data['username_family'], [data['password_family'], data['password_family'] + "2"]),
|
|
)
|
|
|
|
|
|
def get_msg(username, dest, msg='MESSAGE', with_date=True):
|
|
date = datetime.datetime.now()
|
|
ret = f'From: {username}\r\nTo: {dest}\r\n\r\nSubject: TEST\r\n{msg}\r\n'
|
|
if with_date:
|
|
date_str = date.strftime('%a, %d %b %Y %H:%M:%S +0200 (CEST)')
|
|
ret = f'Date: {date_str}\r\n{ret}'
|
|
return ret
|
|
|
|
|
|
@pytest.mark.parametrize('idx, maxi, typ, login_username, username, dest, passwords', parameters)
|
|
def test_imap_wrong_password(idx, maxi, typ, login_username, username, dest, passwords):
|
|
if username == data['ext_username']:
|
|
return
|
|
imap = IMAP4_SSL(data['address'])
|
|
try:
|
|
imap.LOGIN(username, 'b')
|
|
except:
|
|
pass
|
|
else:
|
|
raise Exception('wrong login !')
|
|
|
|
|
|
@pytest.mark.parametrize('idx, maxi, typ, login_username, username, dest, passwords', parameters)
|
|
def test_imap_migration(idx, maxi, typ, login_username, username, dest, passwords):
|
|
if dest.startswith('alias_'):
|
|
return
|
|
if username == data['ext_username']:
|
|
return
|
|
msg = get_msg(username, dest, 'MIGRATION', False)
|
|
if 'FIRST_RUN' in environ:
|
|
smtp = SMTP(data['address'], '587')
|
|
smtp.starttls()
|
|
error = None
|
|
for password in passwords:
|
|
try:
|
|
smtp.login(username, password)
|
|
break
|
|
except SMTPAuthenticationError as err:
|
|
error = err
|
|
else:
|
|
raise error from error
|
|
smtp.sendmail(username, dest, msg)
|
|
smtp.quit()
|
|
imap = IMAP4_SSL(data['address'])
|
|
error = None
|
|
for password in passwords:
|
|
try:
|
|
imap.LOGIN(username, password)
|
|
break
|
|
except Exception as err:
|
|
error = err
|
|
else:
|
|
raise error from error
|
|
imap.SELECT(readonly=True)
|
|
typ, req = imap.SEARCH(None, 'ALL')
|
|
assert typ == 'OK'
|
|
assert len(req) == 1
|
|
assert req[0] == b'1'
|
|
field = imap.FETCH('1', '(RFC822)')
|
|
assert field[0] == 'OK'
|
|
assert field[1][-2][-1].decode().endswith(msg)
|
|
imap.CLOSE()
|
|
imap.LOGOUT()
|
|
|
|
|
|
@pytest.mark.parametrize('idx, maxi, typ, login_username, username, dest, passwords', parameters)
|
|
def test_smtp_no_tls(idx, maxi, typ, login_username, username, dest, passwords):
|
|
if username == data['ext_username']:
|
|
return
|
|
smtp = SMTP(data['address'], '587')
|
|
with pytest.raises(SMTPNotSupportedError):
|
|
smtp.login(username, passwords[0])
|
|
|
|
|
|
@pytest.mark.parametrize('idx, maxi, typ, login_username, username, dest, passwords', parameters)
|
|
def test_smtp_wrong_passwd(idx, maxi, typ, login_username, username, dest, passwords):
|
|
if username == data['ext_username']:
|
|
return
|
|
smtp = SMTP(data['address'], '587')
|
|
smtp.starttls()
|
|
with pytest.raises(SMTPAuthenticationError):
|
|
smtp.login(username, 'a')
|
|
smtp.quit()
|
|
|
|
|
|
@pytest.mark.parametrize('idx, maxi, typ, login_username, username, dest, passwords', parameters)
|
|
def test_smtp_login(idx, maxi, typ, login_username, username, dest, passwords):
|
|
if username == data['ext_username']:
|
|
return
|
|
smtp = SMTP(data['address'], '587')
|
|
smtp.starttls()
|
|
error = None
|
|
for password in passwords:
|
|
try:
|
|
smtp.login(username, password)
|
|
break
|
|
except SMTPAuthenticationError as err:
|
|
error = err
|
|
else:
|
|
raise error from error
|
|
smtp.quit()
|
|
|
|
|
|
@pytest.mark.parametrize('idx, maxi, typ, login_username, username, dest, passwords', parameters)
|
|
def test_smtp_sendmail(idx, maxi, typ, login_username, username, dest, passwords):
|
|
if username == data['ext_username']:
|
|
smtp = SMTP(data['smtp'], '25')
|
|
else:
|
|
smtp = SMTP(data['address'], '587')
|
|
smtp.starttls()
|
|
error = None
|
|
for password in passwords:
|
|
try:
|
|
smtp.login(username, password)
|
|
break
|
|
except SMTPAuthenticationError as err:
|
|
error = err
|
|
else:
|
|
raise error from error
|
|
smtp.sendmail(username, dest, get_msg(username, dest))
|
|
smtp.quit()
|
|
|
|
|
|
@pytest.mark.parametrize('idx, maxi, typ, login_username, username, dest, passwords', parameters)
|
|
def test_imap_read_mail(idx, maxi, typ, login_username, username, dest, passwords):
|
|
imap = IMAP4_SSL(data['address'])
|
|
error = None
|
|
for password in passwords:
|
|
try:
|
|
imap.LOGIN(login_username, password)
|
|
break
|
|
except Exception as err:
|
|
error = err
|
|
else:
|
|
raise error from error
|
|
imap.SELECT(readonly=True)
|
|
typ, req = imap.SEARCH(None, 'ALL')
|
|
assert typ == 'OK'
|
|
assert len(req) == 1
|
|
msg = get_msg(username, dest, with_date=False)
|
|
msg_no = req[0].split()
|
|
assert len(msg_no) == maxi
|
|
num = msg_no[idx]
|
|
field = imap.FETCH(num, '(RFC822)')
|
|
assert field[0] == 'OK'
|
|
fdata = field[1][-2][-1].decode().split('\r\n')
|
|
if fdata[-2].startswith('--'):
|
|
fdata = fdata[:-2]
|
|
fdata = '\r\n'.join(fdata)
|
|
assert 'Undelivered' not in fdata
|
|
assert fdata.endswith(msg)
|
|
imap.CLOSE()
|
|
imap.LOGOUT()
|
|
|
|
|
|
@pytest.mark.parametrize('idx, maxi, typ, login_username, username, dest, passwords', parameters)
|
|
def test_imap_delete_mail(idx, maxi, typ, login_username, username, dest, passwords):
|
|
if username == data['ext_username']:
|
|
return
|
|
imap = IMAP4_SSL(data['address'])
|
|
error = None
|
|
for password in passwords:
|
|
try:
|
|
imap.LOGIN(login_username, password)
|
|
break
|
|
except Exception as err:
|
|
error = err
|
|
else:
|
|
raise error from error
|
|
imap.SELECT()
|
|
typ, req = imap.SEARCH(None, 'ALL')
|
|
msg_no = req[0].split()
|
|
for num in msg_no[1:]:
|
|
ret = imap.store(num, '+FLAGS', '\\Deleted')
|
|
assert ret[0] == 'OK', f'error when deleting mail: {ret}'
|
|
imap.expunge()
|
|
imap.CLOSE()
|
|
imap.LOGOUT()
|