瀏覽代碼

HUE-1715 [dbquery] test using sqlite

Abraham Elmahrek 12 年之前
父節點
當前提交
b818b96
共有 1 個文件被更改,包括 62 次插入3 次删除
  1. 62 3
      apps/rdbms/src/rdbms/tests.py

+ 62 - 3
apps/rdbms/src/rdbms/tests.py

@@ -15,12 +15,17 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-# import re
+import json
+import os
+import uuid
 
-from nose.tools import assert_true
+from django.core.urlresolvers import reverse
+from nose.tools import assert_true, assert_equal
 
 from desktop.lib.django_test_util import make_logged_in_client
-from beeswax.server import dbms
+
+from rdbms import conf as rdbms_conf
+from rdbms.server import dbms
 
 
 class MockRdbms:
@@ -46,3 +51,57 @@ class TestMockedRdbms:
   def test_basic_flow(self):
     response = self.client.get("/rdbms/")
     assert_true('DB Query' in response.content, response.content)
+
+
+class TestSQLiteRdbmsBase(object):
+  @classmethod
+  def setup_class(cls):
+    cls.database = '/tmp/%s.db' % uuid.uuid4()
+    cls.prefillDatabase()
+
+  @classmethod
+  def teardown_class(cls):
+    os.remove(cls.database)
+
+  def setUp(self):
+    self.client = make_logged_in_client()
+    self.finish = rdbms_conf.RDBMS.set_for_testing({
+      'sqlitee': {
+        'name': self.database,
+        'engine': 'sqlite'
+      }
+    })
+
+  def tearDown(self):
+    self.finish()
+
+  @classmethod
+  def prefillDatabase(cls):
+    import sqlite3
+    connection = sqlite3.connect(cls.database)
+    connection.execute("CREATE TABLE test1 (date text, trans text, symbol text, qty real, price real)")
+    connection.execute("INSERT INTO test1 VALUES ('2006-01-05','BUY','RHAT',100,35.14)")
+    connection.commit()
+    connection.close()
+
+
+class TestAPI(TestSQLiteRdbmsBase):
+  def test_get_servers(self):
+    response = self.client.get(reverse('rdbms:api_servers'))
+    response_dict = json.loads(response.content)
+    assert_true('sqlitee' in response_dict['servers'], response_dict)
+
+  def test_get_databases(self):
+    response = self.client.get(reverse('rdbms:api_databases', args=['sqlitee']))
+    response_dict = json.loads(response.content)
+    assert_true(self.database in response_dict['databases'], response_dict)
+
+  def test_execute_query(self):
+    data = {
+      'server': 'sqlitee',
+      'database': self.database,
+      'query': 'SELECT * FROM test1'
+    }
+    response = self.client.post(reverse('rdbms:api_execute_query'), data, follow=True)
+    response_dict = json.loads(response.content)
+    assert_equal(1, len(response_dict['results']['rows']), response_dict)