source: mailjam/mailjam/tests/mta.py@ 16:cd4170142d87

# -*- coding: utf-8 -*-

import os, sys, multiprocessing, time
from unittest import TestCase
from mailjam.mta import MTAClient
from mailjam.daemon import MailjamDaemon
from mailjam.models import Member, MailingList


class TestMTAClient(TestCase):
    """
    FIXME: These are dummy tests, they cover almost nothing from the
    real mailjam mta client (yet)
    """
    def setUp(self):
        self.mta_configfile = os.path.join(os.path.dirname(__file__),
                                           'mailjam-mta.conf')
        self.configfile = os.path.join(os.path.dirname(__file__),
                                       'mailjam.conf')
        self.mailing_list = MailingList('test_list', 'test_list@example.com',
                                        members={}, configfile=self.configfile)
        self.member =  Member('test@example.com')
        self.raw_email_file = os.path.join(os.path.dirname(__file__),
                                           'sample_raw_email.txt')
        tmp_to_read = open(self.raw_email_file, 'r')
        self.raw_email = tmp_to_read.read()
        tmp_to_read.close()

    def test___init__(self):
        # in order to start mta client instances, we need to have a mailjam
        # daemon running on the background
        # This should be added to the setUp() method, but then it would be
        # more difficult to terminate the process after all the methods
        # were called
        daemon = MailjamDaemon(self.configfile)
        daemon.port = 9100
        p = multiprocessing.Process(target=daemon.run)
        p.start()
        # Add a delay here to allow the server to be fully started before
        # starting to send requests from the client
        time.sleep(1)

        with self.assertRaises(ValueError):
            mta = MTAClient(self.mailing_list, self.mta_configfile)
            mta = MTAClient(self.member, self.mta_configfile)
            mta = MTAClient(None, self.mta_configfile)
        mta = MTAClient(self.mailing_list.address, self.mta_configfile)
        self.assertTrue(isinstance(mta, MTAClient))
        self.assertEqual(mta.address, self.mailing_list.address)
        self.assertEqual(mta.suscriptors, self.mailing_list.members_addresses)
        self.assertEqual(mta.reply_to, self.mailing_list.address)

        p.terminate()

    def test_get_raw_email(self):
        # in order to start mta client instances, we need to have a mailjam
        # daemon running on the background
        # This should be added to the setUp() method, but then it would be
        # more difficult to terminate the process after all the methods
        # were called
        daemon = MailjamDaemon(self.configfile)
        daemon.port = 9100
        p = multiprocessing.Process(target=daemon.run)
        p.start()
        # Add a delay here to allow the server to be fully started before
        # starting to send requests from the client
        time.sleep(1)

        mta = MTAClient(self.mailing_list.address, self.mta_configfile)

        sys_stdin = sys.stdin
        sys.stdin = open(self.raw_email_file, 'r')
        self.assertEqual(mta.get_raw_email(),
                         self.raw_email)
        sys.stdin.close()
        with self.assertRaises(IOError):
            mta.get_raw_email()

        tmp_file= open(self.raw_email_file('r'))
        raw_data = tmp_file.read()
        tmp_file.close()
        self.assertEqual(mta.get_raw_email(raw_data),
                         self.raw_email)

        p.terminate()

    #def save_raw_email(self):
    #    sys_stdin = sys.stdin
    #    sys.stdin = open(self.raw_email_file, 'r')
    #    self.mta.save_raw_email.filename = '/tmp/mailjam-test-mta-save-raw-email'