# -*- coding: utf-8 -*- """ The postman project - daemon.py This file is released under the BSD license, see LICENSE for more information. Francisco de Borja Lopez Rio - Soluciones Informaticas Codigo23 S.L.U. - http://codigo23.net """ 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()