mirror of
https://github.com/simple-login/app.git
synced 2024-11-16 05:34:45 +08:00
7ba9bcb9e2
* Initial save to disk * Store unsent messages to disk so they can be retried later * Set back not sending emails * Fixed decorator * Add general exceptions to the catchall * Have dummy server just to make sure * Added several server test cases * ADded tests for bounced and error status * Moved dir creation to config parse time * Set LOG.e Co-authored-by: Adrià Casajús <adria.casajus@proton.ch>
110 lines
3.6 KiB
Python
110 lines
3.6 KiB
Python
from email.message import Message
|
|
|
|
import pytest
|
|
from app.config import (
|
|
ALERT_COMPLAINT_FORWARD_PHASE,
|
|
ALERT_COMPLAINT_REPLY_PHASE,
|
|
ALERT_COMPLAINT_TRANSACTIONAL_PHASE,
|
|
POSTMASTER,
|
|
)
|
|
from app.db import Session
|
|
from app.email_utils import generate_verp_email
|
|
from app.handler.provider_complaint import (
|
|
handle_hotmail_complaint,
|
|
handle_yahoo_complaint,
|
|
)
|
|
from app.mail_sender import mail_sender
|
|
from app.models import (
|
|
Alias,
|
|
ProviderComplaint,
|
|
SentAlert,
|
|
EmailLog,
|
|
VerpType,
|
|
Contact,
|
|
)
|
|
from tests.utils import create_new_user, load_eml_file
|
|
|
|
origins = [
|
|
[handle_yahoo_complaint, "yahoo"],
|
|
[handle_hotmail_complaint, "hotmail"],
|
|
]
|
|
|
|
|
|
def prepare_complaint(
|
|
provider_name: str, alias: Alias, rcpt_address: str, sender_address: str
|
|
) -> Message:
|
|
contact = Contact.create(
|
|
user_id=alias.user.id,
|
|
alias_id=alias.id,
|
|
website_email="a@b.c",
|
|
reply_email="d@e.f",
|
|
commit=True,
|
|
)
|
|
elog = EmailLog.create(
|
|
user_id=alias.user.id,
|
|
mailbox_id=alias.user.default_mailbox_id,
|
|
contact_id=contact.id,
|
|
commit=True,
|
|
bounced=True,
|
|
)
|
|
return_path = generate_verp_email(VerpType.bounce_forward, elog.id)
|
|
return load_eml_file(
|
|
f"{provider_name}_complaint.eml",
|
|
{
|
|
"postmaster": POSTMASTER,
|
|
"return_path": return_path,
|
|
"rcpt": rcpt_address,
|
|
"sender": sender_address,
|
|
"rcpt_comma_list": f"{rcpt_address},other_rcpt@somwhere.net",
|
|
},
|
|
)
|
|
|
|
|
|
@mail_sender.store_emails_test_decorator
|
|
@pytest.mark.parametrize("handle_ftor,provider", origins)
|
|
def test_provider_to_user(flask_client, handle_ftor, provider):
|
|
user = create_new_user()
|
|
alias = Alias.create_new_random(user)
|
|
Session.commit()
|
|
complaint = prepare_complaint(provider, alias, user.email, "nobody@nowhere.net")
|
|
assert handle_ftor(complaint)
|
|
found = ProviderComplaint.filter_by(user_id=user.id).all()
|
|
assert len(found) == 0
|
|
alerts = SentAlert.filter_by(user_id=user.id).all()
|
|
assert len(alerts) == 1
|
|
sent_mails = mail_sender.get_stored_emails()
|
|
assert len(sent_mails) == 1
|
|
assert alerts[0].alert_type == f"{ALERT_COMPLAINT_TRANSACTIONAL_PHASE}_{provider}"
|
|
|
|
|
|
@pytest.mark.parametrize("handle_ftor,provider", origins)
|
|
def test_provider_forward_phase(flask_client, handle_ftor, provider):
|
|
user = create_new_user()
|
|
alias = Alias.create_new_random(user)
|
|
Session.commit()
|
|
complaint = prepare_complaint(provider, alias, "nobody@nowhere.net", alias.email)
|
|
assert handle_ftor(complaint)
|
|
found = ProviderComplaint.filter_by(user_id=user.id).all()
|
|
assert len(found) == 1
|
|
alerts = SentAlert.filter_by(user_id=user.id).all()
|
|
assert len(alerts) == 1
|
|
assert alerts[0].alert_type == f"{ALERT_COMPLAINT_REPLY_PHASE}_{provider}"
|
|
|
|
|
|
@mail_sender.store_emails_test_decorator
|
|
@pytest.mark.parametrize("handle_ftor,provider", origins)
|
|
def test_provider_reply_phase(flask_client, handle_ftor, provider):
|
|
mail_sender.store_emails_instead_of_sending()
|
|
mail_sender.purge_stored_emails()
|
|
user = create_new_user()
|
|
alias = Alias.create_new_random(user)
|
|
Session.commit()
|
|
complaint = prepare_complaint(provider, alias, alias.email, "no@no.no")
|
|
assert handle_ftor(complaint)
|
|
found = ProviderComplaint.filter_by(user_id=user.id).all()
|
|
assert len(found) == 0
|
|
alerts = SentAlert.filter_by(user_id=user.id).all()
|
|
assert len(alerts) == 1
|
|
sent_mails = mail_sender.get_stored_emails()
|
|
assert len(sent_mails) == 1
|
|
assert alerts[0].alert_type == f"{ALERT_COMPLAINT_FORWARD_PHASE}_{provider}"
|