Restore support for old-style xmlrpc url initialization #10

eulexistdb/db.py

@@ -142,7 +142,6 @@ def wrapper(*args, **kwargs):
         "time_taken", "name", "return_value", "args", "kwargs"])
 
 
-
 class ExistDB(object):
     """Connect to an eXist database, and manipulate and query it.
 
@@ -170,6 +169,10 @@ class ExistDB(object):
     # None should override a configured EXISTDB_TIMEOUT)
     DEFAULT_TIMEOUT = object()
 
+    exist_url = None
+    username = None
+    password = None
+
     def __init__(self, server_url=None, username=None, password=None,
                  resultType=None, encoding='UTF-8', verbose=False,
                  keep_alive=None, timeout=DEFAULT_TIMEOUT):
@@ -181,10 +184,17 @@ def __init__(self, server_url=None, username=None, password=None,
         # easily setting a timeout of None and have it override any
         # configured EXISTDB_TIMEOUT
 
-        # add username/password to url if set
-        self.exist_url = server_url
-        self.username = username
-        self.password = password
+        if 'xmlrpc' in server_url:
+            self._init_from_xmlrpc_url(server_url)
+        else:
+            # add username/password to url if set
+            self.exist_url = server_url
+
+        # if username/password are supplied, set them
+        if username is not None:
+            self.username = username
+        if password is not None:
+            self.password = password
 
         # if server url or timeout are not set, attempt to get from django settings
         if self.exist_url is None or timeout == ExistDB.DEFAULT_TIMEOUT:
@@ -268,6 +278,25 @@ def _serverurl_from_djangoconf(self):
         except ImportError:
             pass
 
+    def _init_from_xmlrpc_url(self, url):
+        # map old-style xmlrpc url with username/password to
+        # new-style initialization
+        parsed = urlparse.urlparse(url)
+        # add username/password if set
+        if parsed.username:
+            self.username = parsed.username
+        if parsed.password:
+            self.password = parsed.password
+
+        # construct base exist url, without xmlrpc extension
+        path = parsed.path.replace('/xmlrpc', '')
+        # parsed netloc includes username & password; reconstruct without
+        if parsed.port is not None:
+            netloc = '%s:%s' % (parsed.hostname, parsed.port)
+        else:
+            netloc = parsed.hostname
+        self.exist_url = '%s://%s%s' % (parsed.scheme, netloc, path)
+
     def restapi_path(self, path):
         # generate rest path to a collection or document
         # FIXME: getting duplicated db path, handle this better
@@ -278,7 +307,6 @@ def restapi_path(self, path):
             path = '/%s' % path
         return '%s/rest/db%s' % (self.exist_url.rstrip('/'), path)
 
-
     def getDocument(self, name):
         """Retrieve a document from the database.
 

test/test_existdb/test_db.py

@@ -137,6 +137,41 @@ def test_serverurl_from_djangoconf(self):
         settings.EXISTDB_SERVER_USER = None
         self.assertEqual(settings.EXISTDB_SERVER_URL, self.db._serverurl_from_djangoconf())
 
+    def test_init_old_format(self):
+        # support init with old-style xmlrpc url optionally including
+        # username and password
+        user = 'exister'
+        password = 'tryme'
+        url = 'example.com:8080/exist'
+        test_db = db.ExistDB('http://%s:%s@%s/xmlrpc' % (user, password, url))
+
+        self.assertEqual(user, test_db.username)
+        self.assertEqual(password, test_db.password)
+        self.assertEqual('http://%s' % url, test_db.exist_url)
+
+        # specified username/password args shoudl override xmlrpc url
+        override_user = 'existentialist'
+        override_password = 'sartre'
+        test_db = db.ExistDB('http://%s:%s@%s/xmlrpc' % (user, password, url),
+                             username=override_user, password=override_password)
+        self.assertEqual(override_user, test_db.username)
+        self.assertEqual(override_password, test_db.password)
+        self.assertEqual('http://%s' % url, test_db.exist_url)
+
+        # no username/password
+        test_db = db.ExistDB('http://%s/xmlrpc' % (url))
+        self.assertEqual(None, test_db.username)
+        self.assertEqual(None, test_db.password)
+        self.assertEqual('http://%s' % url, test_db.exist_url)
+
+        # no port, ssl
+        url = 'example.com/exist'
+        test_db = db.ExistDB('https://%s/xmlrpc' % (url))
+        self.assertEqual('https://%s' % url, test_db.exist_url)
+
+
+
+
     def test_getDocument(self):
         """Test retrieving a full document from eXist"""
         xml = self.db.getDocument(self.COLLECTION + "/hello.xml")