Commits

Olemis Lang committed ba8a5d7

BH RPC : Introducing functional RPC testers ... UNTESTED

... but tracpc.tests.api test cases succeed

  • Participants
  • Parent commits 0e36238

Comments (0)

Files changed (1)

File trunk/tracrpc/tests/__init__.py

 Contains modifications for Apache(TM) Bloodhound compatibility
 """
 
+import abc
 import os
 import sys
 import time
 import urllib
 import urllib2
+import xmlrpclib
 
 if sys.version_info[:2] < (2, 7):
     import unittest2 as unittest
             return self.get_trac_environment()
 
 
+    class RpcTesterMixin(object):
+        """RPC extensions for functional testers.
+        """
+        __metaclass__ = abc.ABCMeta
+
+        @abc.abstractmethod
+        def server_proxy(self, rpc_url, user):
+            """Return an object of a type compatible with xmlrpclib.ServerProxy
+            """
+
+
+    def basic_auth_url(url, user, pwd=None):
+        """Generate RPC URL for a given user.
+
+        If password is not provided explicitly then assume it is equal
+        to username.
+        """
+        if pwd is None:
+            pwd = user
+        parts = urllib2.urlparse.urlsplit(url)
+        newparts = parts.__class__(parts.scheme, 
+                                   '%s:%s@%s' % (user, pwd, parts.netloc),
+                                   parts.path, parts.query, parts.fragment)
+        return newparts.geturl()
+
+
+    class XmlRpcTransportTester(RpcTesterMixin):
+        """XML-RPC over HTTP. Supports custom transport.
+
+        If transport is mnissing fall back to basic auth, default behavior
+        for Trac RPC test suite.
+        """
+
+        transport = None
+
+        def server_proxy(self, rpc_url, user=None):
+            transport = self.transport(rpc_url, user) \
+                        if self.transport is not None else None
+            if transport is None and user is not None:
+                rpc_url = basic_auth_url(rpc_url, user)
+            return xmlrpclib.ServerProxy(rpc_url, transport=transport)
+
+
+    class XmlRpcDigestTester(XmlRpcTransportTester):
+        """XML-RPC over HTTP Digest authentication.
+
+        Based upon
+        http://trac-hacks.org/wiki/XmlRpcPlugin#UsingDigestAuthenticationinpython
+        http://bytes.com/topic/python/answers/509382-solution-xml-rpc-over-proxy
+        """
+        class _DigestTransportMixin(object):
+            def __init__(self, username, pw, realm, verbose = None, 
+                         use_datetime=0):
+                self.__username = username
+                self.__pw = pw
+                self.__realm = realm
+                self.verbose = verbose
+                self._use_datetime = use_datetime
+
+            # scheme_prefix overridden in subclasses
+            scheme_prefix = None
+
+            def request(self, host, handler, request_body, verbose):
+
+                url= self.scheme_prefix + host + handler
+                if verbose or self.verbose:
+                    print "ProxyTransport URL: [%s]"%url
+
+                request = urllib2.Request(url)
+                request.add_data(request_body)
+                # Note: 'Host' and 'Content-Length' are added automatically
+                request.add_header("User-Agent", self.user_agent)
+                request.add_header("Content-Type", "text/xml") # Important
+
+                # setup digest authentication
+                authhandler = urllib2.HTTPDigestAuthHandler()
+                authhandler.add_password(self.__realm, url, self.__username,
+                                         self.__pw)
+                opener = urllib2.build_opener(authhandler)
+
+                # TODO: Add support for other features e.g. HTTP proxy ?
+                #proxy_handler=urllib2.ProxyHandler()
+                #opener=urllib2.build_opener(proxy_handler)
+                f=opener.open(request)
+                return(self.parse_response(f))            
+
+        class SafeTransport(_DigestTransportMixin, xmlrpclib.SafeTransport):
+            scheme_prefix = 'https://'
+
+        class Transport(_DigestTransportMixin, xmlrpclib.Transport):
+            scheme_prefix = 'http://'
+
+        def transport(self, rpc_url, user):
+            """Create HTTP Digest transport
+            """
+            if rpc_url.startswith('http://'):
+                cls = self.Transport
+            elif rpc_url.startswith('https://'):
+                cls = self.SafeTransport
+            else:
+                raise ValueError("Unsupport URL scheme %s" % (rpc_url,))
+            # TODO: Real parameterization
+            return cls(user, user, 'bloodhound', verbose = None, use_datetime=0)
+
+    class DefaultRpcTester(XmlRpcDigestTester,
+                           functional.BloodhoundFunctionalTester):
+        pass
+
     class RpcFunctionalTestSuite(functional.MultiproductFunctionalTestSuite):
         env_class = RpcTestEnvironment
-        tester_class = functional.BloodhoundFunctionalTester
+        tester_class = DefaultRpcTester
 
         def testenv_path(self, port=None):
             if port is None:
             """
             return self._tester.url + '/login/rpc'
 
-        def url_for(self, user):
-            """Generate RPC URL for a given user.
-            """
-            parts = urllib2.urlparse.urlsplit(self._tester.url)
-            newparts = parts.__class__(parts.scheme, 
-                                       '%s:%s@%s' % (user, user, parts.netloc),
-                                       parts.path, parts.query, parts.fragment)
-            return newparts.geturl()
-
         @property
         def url_user(self):
             """URL to force authentication upon RPC request for 'user'
             """
-            return self.url_for('user') + '/login/rpc'
+            return basic_auth_url(self._tester.url, 'user') + '/login/rpc'
 
         @property
         def url_admin(self):
             """URL to force authentication upon RPC request for 'admin'
             """
-            return self.url_for('admin') + '/login/rpc'
+            return basic_auth_url(self._tester.url, 'admin') + '/login/rpc'
 
         def failUnlessRaises(self, excClass, callableObj, *args, **kwargs):
             """Enhanced assertions to detect exceptions."""