Initial checkin for new xmlrpc frontend.

stefanseefeld · stefanseefeld · commit 62b5faac9a18 · 2007-03-22T02:07:44.000Z
diff --git a/roundup/scripts/roundup_xmlrpc_server.py b/roundup/scripts/roundup_xmlrpc_server.py
@@ -0,0 +1,75 @@
+#! /usr/bin/env python
+#
+# Copyright (C) 2007 Stefan Seefeld
+# All rights reserved.
+# Licensed to the public under the terms of the GNU LGPL (>= 2),
+# see the file COPYING for details.
+#
+import getopt, os, sys, socket
+from roundup.xmlrpc import RoundupServer, RoundupRequestHandler
+from roundup.instance import TrackerError
+from SimpleXMLRPCServer import SimpleXMLRPCServer
+
+def usage():
+    print """
+
+Options:
+ -i instance home  -- specify the issue tracker "home directory" to administer
+ -V                -- be verbose when importing
+ -p, --port <port> -- port to listen on
+
+"""
+
+def run():
+
+    try:
+        opts, args = getopt.getopt(sys.argv[1:],
+                                   'e:i:p:V', ['encoding=', 'port='])
+    except getopt.GetoptError, e:
+        usage()
+        return 1
+
+    verbose = False
+    tracker = ''
+    port = 8000
+    encoding = None
+
+    for opt, arg in opts:
+        if opt == '-V':
+            verbose = True
+        elif opt == '-i':
+            tracker = arg
+        elif opt in ['-p', '--port']:
+            port = int(arg)
+        elif opt in ['-e', '--encoding']:
+            encoding = encoding
+
+        if sys.version_info[0:2] < (2,5):
+            if encoding:
+                print 'encodings not supported with python < 2.5'
+                sys.exit(-1)
+            server = SimpleXMLRPCServer(('', port), RoundupRequestHandler)
+        else:
+            server = SimpleXMLRPCServer(('', port), RoundupRequestHandler,
+                                        allow_none=True, encoding=encoding)
+    if not os.path.exists(tracker):
+        print 'Instance home does not exist.'
+        sys.exit(-1)
+    try:
+        object = RoundupServer(tracker, verbose)
+    except TrackerError:
+        print 'Instance home does not exist.'
+        sys.exit(-1)
+
+    server.register_instance(object)
+
+    # Go into the main listener loop
+    print 'Roundup XMLRPC server started on %s:%d' \
+          % (socket.gethostname(), port)
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        print 'Keyboard Interrupt: exiting'
+
+if __name__ == '__main__':
+    run()
diff --git a/roundup/xmlrpc.py b/roundup/xmlrpc.py
@@ -0,0 +1,165 @@
+#
+# Copyright (C) 2007 Stefan Seefeld
+# All rights reserved.
+# Licensed to the public under the terms of the GNU LGPL (>= 2),
+# see the file COPYING for details.
+#
+
+
+import base64
+import roundup.instance
+from roundup import hyperdb
+from roundup.cgi.exceptions import *
+from roundup.admin import UsageError
+from SimpleXMLRPCServer import SimpleXMLRPCRequestHandler
+
+class RoundupRequestHandler(SimpleXMLRPCRequestHandler):
+    """A SimpleXMLRPCRequestHandler with support for basic
+    HTTP Authentication."""
+
+    def do_POST(self):
+        """Extract username and password from authorization header."""
+
+        # Try to extract username and password from HTTP Authentication.
+        self.username = None
+        self.password = None
+        authorization = self.headers.get('authorization', ' ')
+        scheme, challenge = authorization.split(' ', 1)
+
+        if scheme.lower() == 'basic':
+            decoded = base64.decodestring(challenge)
+            self.username, self.password = decoded.split(':')
+
+        SimpleXMLRPCRequestHandler.do_POST(self)
+
+    def _dispatch(self, method, params):
+        """Inject username and password into function arguments."""
+
+        # Add username and password to function arguments
+        params = [self.username, self.password] + list(params)
+        return self.server._dispatch(method, params)
+
+
+class RoundupRequest:
+    """Little helper class to handle common per-request tasks such
+    as authentication and login."""
+
+    def __init__(self, tracker, username, password):
+        """Open the database for the given tracker, using the given
+        username and password."""
+
+        self.tracker = tracker
+        self.db = self.tracker.open('admin')
+        try:
+            userid = self.db.user.lookup(username)
+        except KeyError: # No such user
+            raise Unauthorised, 'Invalid user.'
+        stored = self.db.user.get(userid, 'password')
+        if stored != password: # Wrong password
+            raise Unauthorised, 'Invalid user.'
+        self.db.setCurrentUser(username)
+        
+    def __del__(self):
+        """Close the database, after committing any changes, if needed."""
+
+        if getattr(self, 'db'):
+            try:
+                if self.db.transactions:
+                    self.db.commit()
+            finally:
+                self.db.close()
+
+
+    def get_class(self, classname):
+        """Return the class for the given classname."""
+
+        try:
+            return self.db.getclass(classname)
+        except KeyError:
+            raise UsageError, 'no such class "%s"'%classname
+
+    def props_from_args(self, cl, args):
+        """Construct a list of properties from the given arguments,
+        and return them after validation."""
+
+        props = {}
+        for arg in args:
+            if arg.find('=') == -1:
+                raise UsageError, 'argument "%s" not propname=value'%arg
+            l = arg.split('=')
+            if len(l) < 2:
+                raise UsageError, 'argument "%s" not propname=value'%arg
+            key, value = l[0], '='.join(l[1:])
+            if value:
+                try:
+                    props[key] = hyperdb.rawToHyperdb(self.db, cl, None,
+                                                      key, value)
+                except hyperdb.HyperdbValueError, message:
+                    raise UsageError, message
+            else:
+                props[key] = None
+
+        return props
+
+
+#The server object
+class RoundupServer:
+    """The RoundupServer provides the interface accessible through
+    the Python XMLRPC mapping. All methods take an additional username
+    and password argument so each request can be authenticated."""
+
+    def __init__(self, tracker, verbose = False):
+        self.tracker = roundup.instance.open(tracker)
+        self.verbose = verbose
+
+    def list(self, username, password, classname, propname = None):
+
+        r = RoundupRequest(self.tracker, username, password)
+        cl = r.get_class(classname)
+        if not propname:
+            propname = cl.labelprop()
+        return [cl.get(id, propname) for id in cl.list()]
+
+    def display(self, username, password, designator, *properties):
+
+        r = RoundupRequest(self.tracker, username, password)
+        classname, nodeid = hyperdb.splitDesignator(designator)
+        cl = r.get_class(classname)
+        props = properties and list(properties) or cl.properties.keys()
+        props.sort()
+        return dict([(property, cl.get(nodeid, property))
+                     for property in props])
+
+    def create(self, username, password, classname, *args):
+
+        r = RoundupRequest(self.tracker, username, password)
+        cl = r.get_class(classname)
+
+        # convert types
+        props = r.props_from_args(cl, args)
+
+        # check for the key property
+        key = cl.getkey()
+        if key and not props.has_key(key):
+            raise UsageError, 'you must provide the "%s" property.'%key
+
+        # do the actual create
+        try:
+            return cl.create(**props)
+        except (TypeError, IndexError, ValueError), message:
+            raise UsageError, message
+
+    def set(self, username, password, designator, *args):
+
+        r = RoundupRequest(self.tracker, username, password)
+        classname, itemid = hyperdb.splitDesignator(designator)
+        cl = r.get_class(classname)
+
+        # convert types
+        props = r.props_from_args(cl, args)
+        try:
+            cl.set(itemid, **props)
+        except (TypeError, IndexError, ValueError), message:
+            raise UsageError, message
+
+
diff --git a/test/test_xmlrpc.py b/test/test_xmlrpc.py
@@ -0,0 +1,90 @@
+#
+# Copyright (C) 2007 Stefan Seefeld
+# All rights reserved.
+# Licensed to the public under the terms of the GNU LGPL (>= 2),
+# see the file COPYING for details.
+#
+
+import unittest, os, shutil, errno, sys, difflib, cgi, re
+
+from roundup.cgi.exceptions import *
+from roundup import init, instance, password, hyperdb, date
+from roundup.xmlrpc import RoundupServer
+
+import db_test_base
+
+NEEDS_INSTANCE = 1
+
+class TestCaseBase(unittest.TestCase):
+
+    def setUp(self):
+
+        self.dirname = '_test_xmlrpc'
+        # set up and open a tracker
+        self.instance = db_test_base.setupTracker(self.dirname)
+
+        # open the database
+        self.db = self.instance.open('admin')
+        self.db.user.create(username='joe', password=password.Password('random'),
+                            address='random@home.org',
+                            realname='Joe Random', roles='User')
+
+        self.db.commit()
+        self.db.close()
+        
+        self.server = RoundupServer(self.dirname)
+
+
+    def tearDown(self):
+
+        try:
+            shutil.rmtree(self.dirname)
+        except OSError, error:
+            if error.errno not in (errno.ENOENT, errno.ESRCH): raise
+
+class AccessTestCase(TestCaseBase):
+
+    def test(self):
+
+        # Retrieve all three users.
+        results = self.server.list('joe', 'random', 'user', 'id')
+        self.assertEqual(len(results), 3)
+        # Obtain data for 'joe'.
+        userid = 'user' + results[-1]
+        results = self.server.display('joe', 'random', userid)
+        self.assertEqual(results['username'], 'joe')
+        self.assertEqual(results['realname'], 'Joe Random')
+        # Reset joe's 'realname'.
+        results = self.server.set('joe', 'random', userid, 'realname=Joe Doe')
+        results = self.server.display('joe', 'random', userid, 'realname')
+        self.assertEqual(results['realname'], 'Joe Doe')
+        # Create test
+        results = self.server.create('joe', 'random', 'issue', 'title=foo')
+        issueid = 'issue' + results
+        results = self.server.display('joe', 'random', issueid, 'title')
+        self.assertEqual(results['title'], 'foo')
+
+class AuthenticationTestCase(TestCaseBase):
+
+    def test(self):
+
+        # Unknown user (caught in XMLRPC frontend).
+        self.assertRaises(Unauthorised, self.server.list,
+                          'nobody', 'nobody', 'user', 'id')
+        # Wrong permissions (caught by roundup security module).
+        results = self.server.list('joe', 'random', 'user', 'id')
+        userid = 'user' + results[0] # admin
+        # FIXME: why doesn't the following raise an exception ?
+        # self.assertRaises(Unauthorised, self.server.set,
+        #                  'joe', 'random', userid, 'realname=someone')
+
+
+def test_suite():
+    suite = unittest.TestSuite()
+    suite.addTest(unittest.makeSuite(AccessTestCase))
+    suite.addTest(unittest.makeSuite(AuthenticationTestCase))
+    return suite
+
+if __name__ == '__main__':
+    runner = unittest.TextTestRunner()
+    unittest.main(testRunner=runner)
