source: mailjam/postman/tests/daemon.py@ 10:d5329a2a05b7

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

import os, multiprocessing, xmlrpclib, time
from SimpleXMLRPCServer import SimpleXMLRPCServer
from unittest import TestCase

from postman.daemon import Postman, PostmanXMLRPC, PostmanDaemon
from postman.models import Member, MailingList
from postman.storage import JsonStorage as Storage


class TestPostman(TestCase):
    """
    postman.daemon.Postman tests.

    Remember to call the .clear() method of postman after each test, so
    the temp storage files are deleted
    """
    def setUp(self):
        self.configfile = os.path.join(os.path.dirname(__file__), 'postman.conf')
        self.mailing_list = MailingList('test_list', 'test_list@example.com',
                                        members={}, configfile=self.configfile)
        self.member =  Member('test@example.com')

    def test___init__(self):
        postman = Postman(configfile=self.configfile)
        self.assertIsInstance(postman, Postman)
        self.assertEqual(postman.mailings, {})
        self.assertEqual(postman.mailings_addresses, [])
        self.assertIsInstance(postman.dbs, dict)
        self.assertTrue('mailings' in postman.dbs.keys())
        self.assertTrue('members' in postman.dbs.keys())
        self.assertIsInstance(postman.dbs['mailings'], Storage)
        self.assertIsInstance(postman.dbs['members'], Storage)

    def test_save(self):
        postman = Postman(configfile=self.configfile)
        self.assertFalse(postman.save())
        postman.add_mailing_list(self.mailing_list.info())
        self.assertTrue(postman.save())
        # FIXME: We have to test here that the generated json file
        # contains the data it should contain

        # Clear the files created by the tests
        postman.clear()

    def test_load(self):
        postman = Postman(configfile=self.configfile)
        self.assertFalse(postman.mailings)
        self.assertFalse(postman.mailings_addresses)
        self.assertFalse(postman.load())
        postman.add_mailing_list(self.mailing_list.info())
        self.assertTrue(postman.load())

        # Check that another postman instance is able to load the saved data
        postman_load = Postman(configfile=self.configfile)
        self.assertFalse(postman_load.mailings)
        self.assertFalse(postman_load.mailings_addresses)
        postman_load.load()
        self.assertTrue(postman_load.mailings)
        self.assertIsInstance(postman_load.mailings, dict)
        self.assertTrue(postman_load.mailings_addresses)
        self.assertIsInstance(postman_load.mailings_addresses, list)        

        # Clear the files created by the tests
        postman.clear()

    def test_clear(self):
        postman = Postman(configfile=self.configfile)
        self.assertFalse(postman.clear())
        postman.add_mailing_list(self.mailing_list.info())
        self.assertTrue(postman.clear())

    def test_add_mailing_list(self):
        postman = Postman(configfile=self.configfile)
        with self.assertRaises(TypeError):
            # test improper info values
            postman.add_mailing_list(['a list', 'is an', 'invalid parameter'])
            postman.add_mailing_list(self.mailing_list)
        with self.assertRaises(ValueError):
            #test incomplete/missing info values
            postman.add_mailing_list()
            postman.add_mailing_list({'name': 'missing info'})
            postman.add_mailing_list({'address': 'missing info'})
            postman.add_mailing_list({'name': 'missing info',
                                      'address': 'missing info'})
        # test mailing lists can be added
        self.assertTrue(postman.add_mailing_list(self.mailing_list.info()))
        self.assertTrue(postman.mailings)
        self.assertIsInstance(postman.mailings, dict)
        self.assertTrue(postman.mailings_addresses)
        self.assertIsInstance(postman.mailings_addresses, list)
        with self.assertRaises(IndexError):
            # test what happens when the mailing has been already added
            postman.add_mailing_list(self.mailing_list.info())

        # Clear the files created by the tests
        postman.clear()

    def test_add_mailing_member(self):
        postman = Postman(configfile=self.configfile)
        postman.add_mailing_list(self.mailing_list.info())
        with self.assertRaises(ValueError):
            # test what happens if we call the method without proper
            # parameters
            postman.add_mailing_member()
            postman.add_mailing_member(None, None)
            postman.add_mailing_member(None, 'test_list@example.net')
            postman.add_mailing_member('test@example.net', None)
            postman.add_mailing_member('test@example', 'test_list@example.net')
        with self.assertRaises(IndexError):
            # test if we try to add a member to a non-existing mailing list
            postman.add_mailing_member('test@example.net',
                                       'test_list_b@example.net')
        # Test adding a member
        self.assertTrue(postman.add_mailing_member('test@example.net',
                                                   self.mailing_list.address))
        # Test trying to re-add that user
        self.assertFalse(postman.add_mailing_member('test@example.net',
                                                    self.mailing_list.address))

        # Clear the files created by the tests
        postman.clear()


class TestPostmanDaemon(TestCase):
    """
    postman.daemon.PostmanDaemon tests.

    Remember to call the .clear() method of postman after each test, so
    the temp storage files are deleted
    """
    def setUp(self):
        self.configfile = os.path.join(os.path.dirname(__file__),
                                       'postman.conf')
        self.mailing_list = MailingList('test_xmlrpc',
                                        'test_xmlrpc@example.com', members={},
                                        configfile=self.configfile)
        self.member =  Member('test@example.com')

    def test___init__(self):
        daemon = PostmanDaemon(self.configfile)
        self.assertIsInstance(daemon, PostmanDaemon)
        self.assertFalse(daemon.ready_to_serve)
        # FIXME: More tests should be added here once the configuration
        # file feature is added        

    def test_create_server(self):
        daemon = PostmanDaemon(self.configfile)
        daemon.port = 9001
        self.assertTrue(daemon.create_server())
        self.assertIsInstance(daemon.server, SimpleXMLRPCServer)
        self.assertFalse(daemon.ready_to_serve)
        
    def test_add_methods(self):
        daemon = PostmanDaemon(self.configfile)
        daemon.port = 9002
        self.assertTrue(daemon.add_methods())
        self.assertTrue(daemon.ready_to_serve)

        daemon = PostmanDaemon(self.configfile)
        daemon.port = 9003        
        daemon.create_server()
        self.assertTrue(daemon.add_methods())
        self.assertTrue(daemon.ready_to_serve)
        
    def test_run(self):
        daemon = PostmanDaemon(self.configfile)
        daemon.port = 9004
        # start the daemon in another process, so we can start communicating
        # with itjobs = []
        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(2)
        
        # FIXME: Hardcoded url here, should be picked from a config file        
        client = xmlrpclib.ServerProxy('http://localhost:9004')

        # Check that we can perform an XMLRPC call and that the list of
        # available public methods contains the list of methods we have
        # defined in our base XMLRPC class
        set_class_methods = set(PostmanXMLRPC()._listMethods())
        set_xmlrpc_methods = set(client.system.listMethods())
        self.assertTrue(set_class_methods.issubset(set_xmlrpc_methods))
        
        # Stop the server
        p.terminate()