test_email.py

from fastapi import BackgroundTasks, status
import pytest
from sqlalchemy.orm import Session
# from starlette.status import HTTP_302_FOUND

from app.database.models import User, UserEvent
# from app.internal.security.dependancies import current_user
from app.internal.email import (mail, send, send_email_file,
                                send_email_invitation,
                                send_email_to_event_participants,
                                verify_email_pattern)
from app.internal.utils import create_model, delete_instance, save


def test_email_send(client, user, event, smtpd):
    mail.config.SUPPRESS_SEND = 1
    mail.config.MAIL_SERVER = smtpd.hostname
    mail.config.MAIL_PORT = smtpd.port
    mail.config.USE_CREDENTIALS = False
    mail.config.MAIL_TLS = False
    with mail.record_messages() as outbox:
        response = client.post(
            "/email/send", data={
                "event_used": event.id, "user_to_send": user.id,
                "title": "Testing",
                "background_tasks": BackgroundTasks})
        assert len(outbox) == 1
        assert response.ok


def test_failed_email_send(client, user, event, smtpd):
    mail.config.SUPPRESS_SEND = 1
    mail.config.MAIL_SERVER = smtpd.hostname
    mail.config.MAIL_PORT = smtpd.port
    with mail.record_messages() as outbox:
        response = client.post(
            "/email/send", data={
                "event_used": event.id + 1, "user_to_send": user.id,
                "title": "Testing",
                "background_tasks": BackgroundTasks})
        assert len(outbox) == 0
        assert not response.ok


@pytest.fixture
def configured_smtpd(smtpd):
    """
    Overrides the SMTP related configuration to use a mock SMTP server
    :param smtpd: the smtpdfix fixture that represents an SMTP server
    :return: smtpd
    """

    mail.config.SUPPRESS_SEND = 1
    mail.config.MAIL_SERVER = smtpd.hostname
    mail.config.MAIL_PORT = smtpd.port
    mail.config.USE_CREDENTIALS = False
    mail.config.MAIL_TLS = False
    yield smtpd


def test_send_mail_no_body(client, configured_smtpd):
    with mail.record_messages() as outbox:
        response = client.post("/email/invitation/")
        assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY

        assert response.json() == {'detail': [{
            'loc': ['body'],
            'msg': 'field required',
            'type': 'value_error.missing'}]}
        assert not outbox


def test_send_mail_invalid_email(client, configured_smtpd):
    with mail.record_messages() as outbox:
        response = client.post("/email/invitation/", json={
            "sender_name": "string",
            "recipient_name": "string",
            "recipient_mail": "test#mail.com"
        })

        assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
        assert response.json() == {
            "detail": "Please enter valid email address"}
        assert not outbox


def assert_validation_error_missing_body_fields(validation_msg,
                                                missing_fields):
    """
    helper function for asserting with open api validation errors
    look at https://fastapi.tiangolo.com/tutorial/path-params/#data-validation
    :param validation_msg: the response message after json
    :param missing_fields: a list of fields that are asserted missing
    """
    assert isinstance(validation_msg, dict)
    assert 1 == len(validation_msg)
    assert "detail" in validation_msg
    details = validation_msg["detail"]
    assert isinstance(details, list)
    assert len(missing_fields) == len(details)
    for detail in details:
        assert 3 == len(detail)
        assert "type" in detail
        assert "value_error.missing" == detail["type"]
        assert "msg" in detail
        assert "field required" == detail["msg"]
        assert "loc" in detail
        loc = detail["loc"]
        assert isinstance(loc, list)
        assert 2 == len(loc)
        assert "body" == loc[0]
        assert loc[1] in missing_fields


@pytest.mark.parametrize("body, missing_fields", [
    (
            {"sender_name": "string", "recipient_name": "string"},
            ["recipient_mail"],
    ),

    (
            {"sender_name": "string", "recipient_mail": "test@mail.com"},
            ["recipient_name"],
    ),
    (
            {"recipient_name": "string", "recipient_mail": "test@mail.com"},
            ["sender_name"],
    ),
    (
            {"sender_name": "string"},
            ["recipient_name", "recipient_mail"],
    ),
    (
            {"recipient_name": "string"},
            ["sender_name", "recipient_mail"],
    ),
    (
            {"recipient_mail": "test@mail.com"},
            ["sender_name", "recipient_name"],
    ),
])
def test_send_mail_partial_body(body, missing_fields,
                                client, configured_smtpd):
    with mail.record_messages() as outbox:
        response = client.post("/email/invitation/", json=body)
        assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
        assert_validation_error_missing_body_fields(response.json(),
                                                    missing_fields)
        assert not outbox


def test_send_mail_valid_email(client, configured_smtpd):
    with mail.record_messages() as outbox:
        response = client.post("/email/invitation/", json={
            "sender_name": "string",
            "recipient_name": "string",
            "recipient_mail": "test@mail.com"
        }
                               )
        assert response.ok
        assert outbox


@pytest.mark.parametrize("sender_name,recipient_name,recipient_mail", [
    ("", "other_person", "other@mail.com"),
    ("us_person", "", "other@mail.com"),
])
def test_send_mail_bad_invitation(client,
                                  configured_smtpd,
                                  sender_name,
                                  recipient_name,
                                  recipient_mail):
    with mail.record_messages() as outbox:
        response = client.post("/email/invitation/", json={
            "sender_name": sender_name,
            "recipient_name": recipient_name,
            "recipient_mail": recipient_mail
        }
                               )
        assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
        assert response.json() == {
            "detail": "Couldn't send the email!"}
        assert not outbox


@pytest.mark.parametrize("sender_name,recipient_name,recipient_mail", [
    ("", "other_person", "other@mail.com"),
    ("us_person", "", "other@mail.com"),
    ("us_person", "other_person", "other#mail.com"),
])
def test_send_mail_bad_invitation_internal(client,
                                           configured_smtpd,
                                           sender_name,
                                           recipient_name,
                                           recipient_mail):
    background_task = BackgroundTasks()
    assert not send_email_invitation(sender_name,
                                     recipient_name,
                                     recipient_mail,
                                     background_task)


@pytest.mark.parametrize("recipient_mail,file_path", [
    ("other@mail.com", "non_existing_file"),
    ("other#mail.com", __file__),
])
def test_send_mail_bad_file_internal(client,
                                     configured_smtpd,
                                     recipient_mail,
                                     file_path):
    background_task = BackgroundTasks()
    assert not send_email_file(file_path, recipient_mail, background_task)


def test_send_mail_good_file_internal(client, configured_smtpd):
    background_task = BackgroundTasks()
    assert send_email_file(__file__, "good@mail.com", background_task)


@pytest.fixture
def bad_user(session: Session) -> User:
    test_user = create_model(
        session, User,
        username='test_username',
        password='test_password',
        email='test.email#gmail.com',
        language_id=1,
    )
    yield test_user
    delete_instance(session, test_user)


def test_send(session, bad_user, event):
    background_task = BackgroundTasks()
    assert not send(session=session,
                    event_used=1,
                    user_to_send=1,
                    title="Test",
                    background_tasks=background_task)


@pytest.mark.parametrize("email", ["test#mail.com",
                                   "test_mail.com",
                                   "test@mail-com"])
def test_verify_email_pattern(email):
    assert not verify_email_pattern(email)


def test_sending_mailing_list_with_no_user(session, no_event_user,
                                           event_owning_user,
                                           user1,
                                           event_example):
    """this test assures a wrong user won't be able to use the mailing list"""
    association = UserEvent(
        user_id=no_event_user.id,
        event_id=event_example.id
    )
    save(session, association)

    association2 = UserEvent(
        user_id=user1.id,
        event_id=event_example.id
    )
    save(session, association2)

    num_emails_send = send_email_to_event_participants(
        session, event_example.id, 'this mail example', 'booboo')
    assert num_emails_send == 0


def test_sending_mailing_list_from_event_owner(session, no_event_user,
                                               event_owning_user,
                                               user1,
                                               event_example, client,
                                               security_test_client):
    """this test assures mailing list is sent successfuly from
    the event owner. assiciations were created already at the test above."""
    # logged_user_data = {'username': event_owning_user.username,
    #                     'password': event_owning_user.password}
    pass
    # res = security_test_client.post(
    #     security_test_client.app.url_path_for('login'),
    #     data=logged_user_data)
    # TODO: log in event_owning_user to assure successful mailing list send

    # if res.status_code == HTTP_302_FOUND:
    #     num_emails_send = send_email_to_event_participants(
    #         session, event_example.id, 'this mail example', 'booboo')
    #     assert num_emails_send == 2