app-MAIL-temp/tests/handler/test_provider_complaints.py

import email
from email.message import Message
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText

import pytest

from app.config import (
    ALERT_COMPLAINT_FORWARD_PHASE,
    ALERT_COMPLAINT_REPLY_PHASE,
    ALERT_COMPLAINT_TRANSACTIONAL_PHASE,
)
from app.db import Session
from app.email import headers
from app.handler.provider_complaint import (
    handle_hotmail_complaint,
    handle_yahoo_complaint,
)
from app.models import Alias, ProviderComplaint, SentAlert
from tests.utils import create_new_user

origins = [
    [handle_yahoo_complaint, "yahoo", 6],
    [handle_hotmail_complaint, "hotmail", 3],
]


def prepare_complaint(message: Message, part_num: int) -> Message:
    complaint = MIMEMultipart("related")
    # When walking, part 0 is the full message so we -1, and we want to be part N so -1 again
    for i in range(part_num - 2):
        document = MIMEText("text", "plain")
        document.set_payload(f"Part {i}")
        complaint.attach(document)
    complaint.attach(message)

    return email.message_from_bytes(complaint.as_bytes())


@pytest.mark.parametrize("handle_ftor,provider,part_num", origins)
def test_provider_to_user(flask_client, handle_ftor, provider, part_num):
    user = create_new_user()
    original_message = Message()
    original_message[headers.TO] = user.email
    original_message[headers.FROM] = "nobody@nowhere.net"
    original_message.set_payload("Contents")

    complaint = prepare_complaint(original_message, part_num)
    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
    assert alerts[0].alert_type == f"{ALERT_COMPLAINT_TRANSACTIONAL_PHASE}_{provider}"


@pytest.mark.parametrize("handle_ftor,provider,part_num", origins)
def test_provider_forward_phase(flask_client, handle_ftor, provider, part_num):
    user = create_new_user()
    alias = Alias.create_new_random(user)
    Session.commit()
    original_message = Message()
    original_message[headers.TO] = "nobody@nowhere.net"
    original_message[headers.FROM] = alias.email
    original_message.set_payload("Contents")

    complaint = prepare_complaint(original_message, part_num)
    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}"


@pytest.mark.parametrize("handle_ftor,provider,part_num", origins)
def test_provider_reply_phase(flask_client, handle_ftor, provider, part_num):
    user = create_new_user()
    alias = Alias.create_new_random(user)
    Session.commit()
    original_message = Message()
    original_message[headers.TO] = alias.email
    original_message[headers.FROM] = "no@no.no"
    original_message.set_payload("Contents")

    complaint = prepare_complaint(original_message, part_num)
    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
    assert alerts[0].alert_type == f"{ALERT_COMPLAINT_FORWARD_PHASE}_{provider}"