Adding DB.from_url method and tests

buildbot/db.py

@@ -20,6 +20,7 @@
 #
 # Contributor(s):
 #   Brian Warner <warner@lothar.com>
+#   Chris AtLee <catlee@mozilla.com>
 #
 # Alternatively, the contents of this file may be used under the terms of
 # either the GNU General Public License Version 2 or later (the "GPL"), or
@@ -35,7 +36,7 @@
 #
 # ***** END LICENSE BLOCK *****
 
-import sys, time, collections, base64, textwrap
+import sys, time, collections, base64, textwrap, os, cgi, re
 
 try:
     import simplejson
@@ -281,6 +282,61 @@ def __init__(self, dbapiName, *connargs, **connkw):
         self.connargs = connargs
         self.connkw = connkw
 
+    @classmethod
+    def from_url(cls, url, basedir=None):
+        """Parses a URL of the format
+        driver://[username:password@]host:port/database[?args]
+
+        and returns a DB object representing this URL
+        """
+        match = re.match(r"""
+        ^(?P<driver>\w+)://
+        (
+            ((?P<user>\w+)(:(?P<passwd>\S+))?@)?
+            ((?P<host>[-A-Za-z0-9.]+)(:(?P<port>\d+))?)?/
+            (?P<database>\S+?)(\?(?P<args>.*))?
+        )?$""", url, re.X)
+        if not match:
+            raise ValueError("Malformed url")
+
+        d = match.groupdict()
+        driver = d['driver']
+        user = d['user']
+        passwd = d['passwd']
+        host = d['host']
+        port = d['port']
+        if port is not None:
+            port = int(port)
+        database = d['database']
+        args = {}
+        if d['args']:
+            for key, value in cgi.parse_qsl(d['args']):
+                args[key] = value
+
+        if driver == "sqlite":
+            # user, passwd, host, and port must all be None
+            if not user == passwd == host == port == None:
+                raise ValueError("user, passwd, host, port must all be None")
+            if not database:
+                database = ":memory:"
+            else:
+                database = database % dict(basedir=basedir)
+                database = os.path.join(basedir, database)
+            return cls("sqlite3", database, **args)
+        elif driver == "mysql":
+            args['host'] = host
+            args['db'] = database
+            if user:
+                args['user'] = user
+            if passwd:
+                args['passwd'] = passwd
+            if port:
+                args['port'] = port
+
+            return cls("MySQLdb", **args)
+        else:
+            raise ValueError("Unsupported dbapi %s" % driver)
+
 def get_sqlite_dbapi_name():
     # see which dbapi we can use, and import it as 'buildbot.db.sqlite3'
     sqlite_dbapi_name = None

buildbot/test/unit/test_config.py

@@ -400,6 +400,88 @@ def startService(self):
 BuildmasterConfig = c
 """
 
+class TestDBUrl(unittest.TestCase):
+    def testSQLiteRelative(self):
+        basedir = "/foo/bar"
+        d = db.DB.from_url("sqlite:///state.sqlite", basedir=basedir)
+        self.failUnlessEqual(d.dbapiName, "pysqlite2.dbapi2")
+        self.failUnlessEqual(d.connargs, (os.path.join(basedir, "state.sqlite"),))
+
+    def testSQLiteBasedir(self):
+        basedir = "/foo/bar"
+        d = db.DB.from_url("sqlite:///%(basedir)s/baz/state.sqlite", basedir=basedir)
+        self.failUnlessEqual(d.dbapiName, "pysqlite2.dbapi2")
+        self.failUnlessEqual(d.connargs, (os.path.join(basedir, "baz/state.sqlite"),))
+
+    def testSQLiteAbsolute(self):
+        basedir = "/foo/bar"
+        d = db.DB.from_url("sqlite:////tmp/state.sqlite", basedir=basedir)
+        self.failUnlessEqual(d.dbapiName, "pysqlite2.dbapi2")
+        self.failUnlessEqual(d.connargs, ("/tmp/state.sqlite",))
+
+    def testSQLiteMemory(self):
+        basedir = "/foo/bar"
+        d = db.DB.from_url("sqlite://", basedir=basedir)
+        self.failUnlessEqual(d.dbapiName, "pysqlite2.dbapi2")
+        self.failUnlessEqual(d.connargs, (":memory:",))
+
+    def testSQLiteArgs(self):
+        basedir = "/foo/bar"
+        d = db.DB.from_url("sqlite:///state.sqlite?foo=bar", basedir=basedir)
+        self.failUnlessEqual(d.dbapiName, "pysqlite2.dbapi2")
+        self.failUnlessEqual(d.connargs, (os.path.join(basedir, "state.sqlite"),))
+        self.failUnlessEqual(d.connkw, dict(foo="bar"))
+
+    def testBadUrls(self):
+        self.failUnlessRaises(ValueError, db.DB.from_url, "state.sqlite")
+        self.failUnlessRaises(ValueError, db.DB.from_url, "sqlite/state.sqlite")
+        self.failUnlessRaises(ValueError, db.DB.from_url, "sqlite:/state.sqlite")
+        self.failUnlessRaises(ValueError, db.DB.from_url, "sqlite:state.sqlite")
+        self.failUnlessRaises(ValueError, db.DB.from_url, "mysql://foo")
+        self.failUnlessRaises(ValueError, db.DB.from_url, "unknowndb://foo/bar")
+        self.failUnlessRaises(ValueError, db.DB.from_url, "mysql://somehost.com:badport/db")
+
+    def testMysql(self):
+        basedir = "/foo/bar"
+        d = db.DB.from_url("mysql://somehost.com/database_name", basedir=basedir)
+        self.failUnlessEqual(d.dbapiName, "MySQLdb")
+        self.failUnlessEqual(d.connkw, dict(host="somehost.com", db="database_name"))
+
+    def testMysqlPort(self):
+        basedir = "/foo/bar"
+        d = db.DB.from_url("mysql://somehost.com:9000/database_name", basedir=basedir)
+        self.failUnlessEqual(d.dbapiName, "MySQLdb")
+        self.failUnlessEqual(d.connkw, dict(host="somehost.com",
+            db="database_name", port=9000))
+
+    def testMysqlLocal(self):
+        basedir = "/foo/bar"
+        d = db.DB.from_url("mysql:///database_name", basedir=basedir)
+        self.failUnlessEqual(d.dbapiName, "MySQLdb")
+        self.failUnlessEqual(d.connkw, dict(host=None, db="database_name"))
+
+    def testMysqlAuth(self):
+        basedir = "/foo/bar"
+        d = db.DB.from_url("mysql://user:password@somehost.com/database_name",
+                basedir=basedir)
+        self.failUnlessEqual(d.dbapiName, "MySQLdb")
+        self.failUnlessEqual(d.connkw, dict(host="somehost.com", db="database_name",
+            user="user", passwd="password"))
+
+    def testMysqlAuthNoPass(self):
+        basedir = "/foo/bar"
+        d = db.DB.from_url("mysql://user@somehost.com/database_name", basedir=basedir)
+        self.failUnlessEqual(d.dbapiName, "MySQLdb")
+        self.failUnlessEqual(d.connkw, dict(host="somehost.com", db="database_name",
+            user="user"))
+
+    def testMysqlArgs(self):
+        basedir = "/foo/bar"
+        d = db.DB.from_url("mysql://somehost.com/database_name?foo=bar", basedir=basedir)
+        self.failUnlessEqual(d.dbapiName, "MySQLdb")
+        self.failUnlessEqual(d.connkw, dict(host="somehost.com", db="database_name",
+            foo="bar"))
+
 class SetupBase:
     def setUp(self):
         # this class generates several deprecation warnings, which the user