rpc_interface: add get_db_test rpc

get_db_test does a quick db connection, and returns ether 'Success' or


Change-Id: I9e2cd553369ff39d26a21b640d655584878db648
Reviewed-on: https://chromium-review.googlesource.com/537213
Commit-Ready: Chris Ching <chingcodes@chromium.org>
Tested-by: Chris Ching <chingcodes@chromium.org>
Reviewed-by: Prathmesh Prabhu <pprabhu@chromium.org>
diff --git a/frontend/afe/rpc_interface.py b/frontend/afe/rpc_interface.py
index 39c3c98..c186c39 100644
--- a/frontend/afe/rpc_interface.py
+++ b/frontend/afe/rpc_interface.py
@@ -37,8 +37,10 @@
 import os
 import sys
+from django.db import connection as db_connection
 from django.db import transaction
 from django.db.models import Count
+from django.db.utils import DatabaseError
 import common
 # TODO(akeshet): Replace with monarch stats once we know how to instrument rpc
@@ -1596,6 +1598,15 @@
     return datetime.datetime.now().strftime("%Y-%m-%d %H:%M")
+def ping_db():
+    """Simple connection test to db"""
+    try:
+        db_connection.cursor()
+    except DatabaseError:
+        return [False]
+    return [True]
 def get_hosts_by_attribute(attribute, value):
     Get the list of valid hosts that share the same host attribute value.
diff --git a/frontend/afe/rpc_interface_unittest.py b/frontend/afe/rpc_interface_unittest.py
index 8ff9cbd..635c7f1 100755
--- a/frontend/afe/rpc_interface_unittest.py
+++ b/frontend/afe/rpc_interface_unittest.py
@@ -73,6 +73,10 @@
+    def test_ping_db(self):
+        self.assertEquals(rpc_interface.ping_db(), [True])
     def test_get_hosts(self):
         hosts = rpc_interface.get_hosts()
         self._check_hostnames(hosts, [host.hostname for host in self.hosts])