from multiprocessing import Process

from sqlalchemy import create_engine, pool

import unittest

from tempfile import mkdtemp

import shutil

from iro.model.utils import WithSession

from iro.model import POOL_SIZE as DB_POOL_SIZE

import iro.model.schema as schema

from iro.main import runReactor

import iro.error as IroError

class SampleDatabase(Database):

    def createPassword(self):

        self.password="test"

        return self.password

#activates all logging we can get.

from twisted.python import log

import logging

logging.basicConfig(level=logging.DEBUG, format='%(asctime)s %(name)s(%(processName)s)-%(levelname)s: %(message)s')

observer = log.PythonLoggingObserver()

observer.start()

class XMLRPCTest(unittest.TestCase):

    """tests for the xmlrpc interface"""

    def setUp(self):

        self.s = Process(target=startReactor, args=(md.engine,))

        self.s.start()

        #the new process needs time to get stated, so this process has to sleep

        time.sleep(.2)

    def tearDown(self):

        self.__debug().stop()

        time.sleep(.2)

        self.s.join()

        self.__cleanDB()

    def __cleanDB(self):

        with WithSession(md.engine, autocommit=True) as session:

            for table in schema.__tables__:

                session.query(getattr(schema,table)).delete()

    def __debug(self):

        return xServer('http://localhost:7080/debug')

    def __rpc2(self):

        return ServerProxy('http://localhost:7080/RPC2')

    def testDebugHello(self):

        '''simple test for the connection to xmlrpc server'''

        ret=self.__debug().hello()

        self.failUnlessEqual(ret,'hello')

    def testListMethods(self):

        '''list of all offical Methods, that can be executed'''

        ret=self.__rpc2().listMethods()

    def testStatus(self):

        ''' test the status function'''

        with WithSession(md.engine, autocommit=True) as session:

            u = User(name='test',apikey='abcdef123456789')

            session.add(User(name='test',apikey='abcdef123456789'))

        self.failUnlessEqual(self.__rpc2().status('abcdef123456789'), {})

        with WithSession(md.engine, autocommit=True) as session:

            u = session.merge(u)

            j = Job(hash="a1", info='info', status="started")

            j.user=u

            session.add(j)

        status = {'a1':{"status":"started"}}

        self.failUnlessEqual(self.__rpc2().status('abcdef123456789'), status)

        self.failUnlessEqual(self.__rpc2().status('abcdef123456789',"a1"), status)

        self.failUnlessEqual(self.__rpc2().status('abcdef123456789', '', 'false'), status)

        self.failUnlessEqual(self.__rpc2().status('abcdef123456789', '', 0), status)

        #JobNotFound

        with self.assertRaises(Fault) as fault:

            self.__rpc2().status('abcdef123456789',"b")

        exc = fault.exception

        unf = IroError.JobNotFound()

        self.failUnlessEqual(exc.faultCode, unf.code)

        self.failUnlessEqual(exc.faultString, unf.msg)

        #self.failUnlessEqual(self.__rpc2().status('abcdef123456789','abcde', True), ["<User('test','abcdef123456789')>",'abcde', True])

        #self.failUnlessEqual(self.__rpc2().status('abcdef123456789', '', 'true'), ["<User('test','abcdef123456789')>", '', True])

        #self.failUnlessEqual(self.__rpc2().status('abcdef123456789', '', 1), ["<User('test','abcdef123456789')>", '', True])

    def testNoSuchUser(self):

        '''a unknown user should raise a UserNotNound Exception

        bewcause xmlrpc only has a Fault exception this Exception has to be deliverd through a xmlrpclib.Fault Exception'''

        with self.assertRaises(Fault) as fault:

            self.__rpc2().status('abcdef123456789')

        exc = fault.exception

        unf=IroError.UserNotFound()

        self.failUnlessEqual(exc.faultCode, unf.code)

        self.failUnlessEqual(exc.faultString, unf.msg)

    def testNoSuchMethod(self):

        '''a unknown mothod should raise a Exception '''

        with self.assertRaises(Fault) as fault:

            self.__rpc2().nosuchmethod()

        exc = fault.exception

        self.failUnlessEqual(exc.faultCode, 8001)

        self.failUnlessEqual(exc.faultString, "procedure nosuchmethod not found")

    def testValidationFault(self):

        '''a validate Exception should be translated to a xmlrpclib.Fault.'''

        with self.assertRaises(Fault) as fault:

            self.__rpc2().status('xxx')

        exc = fault.exception

        self.failUnlessEqual(exc.faultCode, 700)

        self.failUnlessEqual(exc.faultString, "Validation of 'apikey' failed.")

    def testRoutes(self):

        '''test the route function'''

        with WithSession(md.engine, autocommit=True) as session:

            u=User(name='test',apikey='abcdef123456789')

            o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms")

            u.rights.append(Userright(o)) 

            session.add(u)

        self.failUnlessEqual(self.__rpc2().routes('abcdef123456789','sms'),['sipgate_basic'])

        with self.assertRaises(Fault) as fault:

            self.__rpc2().routes('abcdef123456789','fax')

        exc = fault.exception

        self.failUnlessEqual(exc.faultCode, 700)

        self.failUnlessEqual(exc.faultString, "Typ is not valid.")

        with WithSession(md.engine, autocommit=True) as session:

            o=Offer(name="sipgate_plus", provider="sipgate", route="plus", typ="sms")

            u = session.query(User).filter_by(name="test").first()

            u.rights.append(Userright(o)) 

            o=Offer(name="faxde", provider="faxde", route="", typ="fax")

            session.add(o)

            session.commit()

        self.failUnlessEqual(self.__rpc2().routes('abcdef123456789','sms'),['sipgate_basic','sipgate_plus'])

        self.failUnlessEqual(self.__rpc2().routes('abcdef123456789','fax'),[])

        with WithSession(md.engine, autocommit=True) as session:

            u = session.query(User).filter_by(name="test").first()

            u.rights.append(Userright(o)) 

        self.failUnlessEqual(self.__rpc2().routes('abcdef123456789','sms'),['sipgate_basic','sipgate_plus'])

        self.failUnlessEqual(self.__rpc2().routes('abcdef123456789','fax'),['faxde'])

    def testDefaultRoutes(self):

        '''test the defaultRoute function'''

        with WithSession(md.engine, autocommit=True) as session:

            u=User(name='test',apikey='abcdef123456789')

            o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms")

            u.rights.append(Userright(o,True)) 

            o=Offer(name="sipgate_plus", provider="sipgate", route="plus", typ="sms")

            u.rights.append(Userright(o))

            session.add(u)

        self.failUnlessEqual(self.__rpc2().defaultRoute('abcdef123456789','sms'),['sipgate_basic'])

    def testTelnumbers(self):

        '''test the telefon validator'''

        self.failUnlessEqual(self.__rpc2().telnumber(["0123/456(78)","+4912346785433","00123435456-658"]),True)

        numbers=['xa','+1','1-23',';:+0','0123']

        for number in numbers:

            with self.assertRaises(Fault) as fault:

                self.__rpc2().telnumber([number])

            exc = fault.exception

            self.failUnlessEqual(exc.faultCode, 701)

            self.failUnlessEqual(exc.faultString, "No valid telnumber: '%s'"%number)

        with self.assertRaises(Fault) as fault:

            self.__rpc2().telnumber(['01234']+numbers)

        exc = fault.exception

        self.failUnlessEqual(exc.faultCode, 701)

        self.failUnlessEqual(exc.faultString, "No valid telnumber: '%s'"%numbers[0])

    def testVaildEmail(self):

        '''test vaild email adresses (got from wikipedia)'''

        validmails=["niceandsimple@example.com",

                "simplewith+symbol@example.com",

                'a.little.unusual@example.com',

                'a.little.more.unusual@dept.example.com',

                '@[10.10.10.10]',

                '@[1.1.1.1]',

                '@[200.100.100.100]',

                'user@[2001:db8:1ff::a0b:dbd0]',

                '"much.more\ unusual"@example.com',

                '"very.unusual.@.unusual.com"@example.com',

                '"very.(),:;<>[]\\".VERY.\\"very@\\\ \\"very\\".unusual"@strange.example.com',

                "!#$%&'*+-/=?^_`{}|~@example.org",

                '"()<>[]:;@,\\\"!#$%&\'*+-/=?^_`{}| ~  ? ^_`{}|~."@example.org',

                '""@example.org']

        for num in validmails:

            self.failUnlessEqual(self.__rpc2().email([num]),True)

    def testInvaildEmail(self):

        '''test invaild email adresses (got from wikipedia)'''

        invalid=["Abc.example.com",  # (an @ character must separate the local and domain parts)

                "Abc.@example.com",          # (character dot(.) is last in local part)

                "Abc..123@example.com",      # (character dot(.) is double)

                "A@b@c@example.com",         # (only one @ is allowed outside quotation marks)

                'a"b(c)d,e:f;g<h>i[j\k]l@example.com',  # (none of the special characters in this local part is allowed outside quotation marks)

                'just"not"right@example.com',           # (quoted strings must be dot separated, or the only element making up the local-part)

                'thisis."notallowed@example.com',      # (spaces, quotes, and backslashes may only exist when within quoted strings and preceded by a slash)

                'this\\ still\\"not\\allowed@example.com',   # (even if escaped (preceded by a backslash), spaces, quotes, and backslashes must still be contained by quotes)

                ]

        for number in invalid:

            with self.assertRaises(Fault) as fault:

                self.__rpc2().email([number])

            exc = fault.exception

            self.failUnlessEqual(exc.faultCode, 702)

            self.failUnlessEqual(exc.faultString, "No valid email: '%s'"%number)

def startReactor(engine):

    """starts the Rector with a special debug Clild, so that the reactor can be stopped remotly. """

    from twisted.internet import reactor

    from twisted.web import xmlrpc, resource

    from iro.view.xmlrpc import appendResource

    class XMLRPCDebug(xmlrpc.XMLRPC): 

        def xmlrpc_stop(self):

            reactor.callLater(0.1,reactor.stop)

            return ""

        def xmlrpc_hello(self):

            return "hello"

    root = resource.Resource()

    root = appendResource(root)

    root.putChild('debug', XMLRPCDebug())

    runReactor(reactor, engine, root)

class ModuleData:

    def __init__(self):

        self.tdir = mkdtemp(prefix='iro-mysql-')

        self.server = Server('%s/my.cnf'%self.tdir)

        self.db = SampleDatabase("test","test",'%s/my.cnf'%self.tdir)

        self.engine = create_engine('mysql://test:test@localhost/test?unix_socket=%s/socket'%self.tdir,

    def setUp(self):

        with open('%s/my.cnf'%self.tdir,'w') as cnf:

            cnf.write(createConfig(self.tdir))

        self.server.create()

        self.server.start()

        self.db.create()

        Base.metadata.create_all(self.engine)

    def tearDown(self):

        self.server.stop()

        shutil.rmtree(self.tdir)