# -*- test-case-name: twisted.web.test.test_xmlrpc -*-
#
# Copyright (c) 2001-2004 Twisted Matrix Laboratories.
# See LICENSE for details.

"""Test XML-RPC support."""

try:
    import xmlrpclib
except ImportError:
    xmlrpclib = None
    class XMLRPC: pass
else:
    from twisted.web import xmlrpc
    from twisted.web.xmlrpc import XMLRPC, addIntrospection

from twisted.trial import unittest
from twisted.web import server, static
from twisted.internet import reactor, defer
from twisted.python import log

import time

class TestRuntimeError(RuntimeError):
    pass

class TestValueError(ValueError):
    pass



class Test(XMLRPC):

    FAILURE = 666
    NOT_FOUND = 23
    SESSION_EXPIRED = 42

    # the doc string is part of the test
    def xmlrpc_add(self, a, b):
        """This function add two numbers."""
        return a + b

    xmlrpc_add.signature = [['int', 'int', 'int'],
                            ['double', 'double', 'double']]

    # the doc string is part of the test
    def xmlrpc_pair(self, string, num):
        """This function puts the two arguments in an array."""
        return [string, num]

    xmlrpc_pair.signature = [['array', 'string', 'int']]

    # the doc string is part of the test
    def xmlrpc_defer(self, x):
        """Help for defer."""
        return defer.succeed(x)

    def xmlrpc_deferFail(self):
        return defer.fail(TestValueError())

    # don't add a doc string, it's part of the test
    def xmlrpc_fail(self):
        raise TestRuntimeError

    def xmlrpc_fault(self):
        return xmlrpc.Fault(12, "hello")

    def xmlrpc_deferFault(self):
        return defer.fail(xmlrpc.Fault(17, "hi"))

    def xmlrpc_complex(self):
        return {"a": ["b", "c", 12, []], "D": "foo"}

    def xmlrpc_dict(self, map, key):
        return map[key]

    def _getFunction(self, functionPath):
        try:
            return XMLRPC._getFunction(self, functionPath)
        except xmlrpc.NoSuchFunction:
            if functionPath.startswith("SESSION"):
                raise xmlrpc.Fault(self.SESSION_EXPIRED, "Session non-existant/expired.")
            else:
                raise

    xmlrpc_dict.help = 'Help for dict.'

class TestAuthHeader(Test):
    """
    This is used to get the header info so that we can test
    authentication.
    """
    def __init__(self):
        Test.__init__(self)
        self.request = None

    def render(self, request):
        self.request = request
        return Test.render(self, request)

    def xmlrpc_authinfo(self):
        return self.request.getUser(), self.request.getPassword()

class XMLRPCTestCase(unittest.TestCase):

    def setUp(self):
        self.p = reactor.listenTCP(0, server.Site(Test()),
                                   interface="127.0.0.1")
        self.port = self.p.getHost().port

    def tearDown(self):
        return self.p.stopListening()

    def proxy(self):
        return xmlrpc.Proxy("http://127.0.0.1:%d/" % self.port)

    def testResults(self):
        inputOutput = [
            ("add", (2, 3), 5),
            ("defer", ("a",), "a"),
            ("dict", ({"a": 1}, "a"), 1),
            ("pair", ("a", 1), ["a", 1]),
            ("complex", (), {"a": ["b", "c", 12, []], "D": "foo"})]

        dl = []
        for meth, args, outp in inputOutput:
            d = self.proxy().callRemote(meth, *args)
            d.addCallback(self.assertEquals, outp)
            dl.append(d)
        return defer.DeferredList(dl, fireOnOneErrback=True)

    def testErrors(self):
        dl = []
        for code, methodName in [(666, "fail"), (666, "deferFail"),
                                 (12, "fault"), (23, "noSuchMethod"),
                                 (17, "deferFault"), (42, "SESSION_TEST")]:
            d = self.proxy().callRemote(methodName)
            d = self.assertFailure(d, xmlrpc.Fault)
            d.addCallback(lambda exc, code=code: self.assertEquals(exc.faultCode, code))
            dl.append(d)
        d = defer.DeferredList(dl, fireOnOneErrback=True)
        d.addCallback(lambda ign: log.flushErrors(TestRuntimeError, TestValueError))
        return d


class XMLRPCTestCase2(XMLRPCTestCase):
    """Test with proxy that doesn't add a slash."""

    def proxy(self):
        return xmlrpc.Proxy("http://127.0.0.1:%d" % self.port)



class XMLRPCAllowNoneTestCase(unittest.TestCase):
    """
    Test with allowNone set to True.

    These are not meant to be exhaustive serialization tests, since
    L{xmlrpclib} does all of the actual serialization work.  They are just
    meant to exercise a few codepaths to make sure we are calling into
    xmlrpclib correctly.
    """

    def setUp(self):
        self.p = reactor.listenTCP(
            0, server.Site(Test(allowNone=True)), interface="127.0.0.1")
        self.port = self.p.getHost().port


    def tearDown(self):
        return self.p.stopListening()


    def proxy(self):
        return xmlrpc.Proxy("http://127.0.0.1:%d" % (self.port,), allowNone=True)


    def test_deferredNone(self):
        """
        Test that passing a C{None} as an argument to a remote method and
        returning a L{Deferred} which fires with C{None} properly passes
        </nil> over the network if allowNone is set to True.
        """
        d = self.proxy().callRemote('defer', None)
        d.addCallback(self.assertEquals, None)
        return d


    def test_dictWithNoneValue(self):
        """
        Test that return a C{dict} with C{None} as a value works properly.
        """
        d = self.proxy().callRemote('defer', {'a': None})
        d.addCallback(self.assertEquals, {'a': None})
        return d



class XMLRPCTestAuthenticated(XMLRPCTestCase):
    """
    Test with authenticated proxy. We run this with the same inout/ouput as
    above.
    """
    user = "username"
    password = "asecret"

    def setUp(self):
        self.p = reactor.listenTCP(0, server.Site(TestAuthHeader()),
                                   interface="127.0.0.1")
        self.port = self.p.getHost().port


    def testAuthInfoInURL(self):
        p = xmlrpc.Proxy("http://%s:%s@127.0.0.1:%d/" % (self.user, self.password, self.port))
        return p.callRemote("authinfo").addCallback(self.assertEquals, [self.user, self.password])


    def testExplicitAuthInfo(self):
        p = xmlrpc.Proxy("http://127.0.0.1:%d/" % (self.port,), self.user, self.password)
        return p.callRemote("authinfo").addCallback(self.assertEquals, [self.user, self.password])


    def testExplicitAuthInfoOverride(self):
        p = xmlrpc.Proxy("http://wrong:info@127.0.0.1:%d/" % (self.port,), self.user, self.password)
        return p.callRemote("authinfo").addCallback(self.assertEquals, [self.user, self.password])



class XMLRPCTestIntrospection(XMLRPCTestCase):

    def setUp(self):
        xmlrpc = Test()
        addIntrospection(xmlrpc)
        self.p = reactor.listenTCP(0, server.Site(xmlrpc),interface="127.0.0.1")
        self.port = self.p.getHost().port

    def testListMethods(self):

        def cbMethods(meths):
            meths.sort()
            self.failUnlessEqual(
                meths,
                ['add', 'complex', 'defer', 'deferFail',
                 'deferFault', 'dict', 'fail', 'fault',
                 'pair', 'system.listMethods',
                 'system.methodHelp',
                 'system.methodSignature'])

        d = self.proxy().callRemote("system.listMethods")
        d.addCallback(cbMethods)
        return d

    def testMethodHelp(self):
        inputOutputs = [
            ("defer", "Help for defer."),
            ("fail", ""),
            ("dict", "Help for dict.")]

        dl = []
        for meth, expected in inputOutputs:
            d = self.proxy().callRemote("system.methodHelp", meth)
            d.addCallback(self.assertEquals, expected)
            dl.append(d)
        return defer.DeferredList(dl, fireOnOneErrback=True)

    def testMethodSignature(self):
        inputOutputs = [
            ("defer", ""),
            ("add", [['int', 'int', 'int'],
                     ['double', 'double', 'double']]),
            ("pair", [['array', 'string', 'int']])]

        dl = []
        for meth, expected in inputOutputs:
            d = self.proxy().callRemote("system.methodSignature", meth)
            d.addCallback(self.assertEquals, expected)
            dl.append(d)
        return defer.DeferredList(dl, fireOnOneErrback=True)


class XMLRPCClientErrorHandling(unittest.TestCase):
    def setUp(self):
        self.resource = static.File(__file__)
        self.resource.isLeaf = True
        self.port = reactor.listenTCP(0, server.Site(self.resource), interface='127.0.0.1')

    def tearDown(self):
        return self.port.stopListening()

    def testErroneousResponse(self):
        proxy = xmlrpc.Proxy("http://127.0.0.1:%d/" % (self.port.getHost().port,))
        return self.assertFailure(proxy.callRemote("someMethod"), Exception)
