Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Browse files

Added tests for AdispClient.

  • Loading branch information...
commit d34d1aad78a076be0780ed372b06823d6c47dcc9 1 parent fff3a33
@FSX authored
Showing with 102 additions and 0 deletions.
  1. +3 −0  tests/README.rst
  2. +80 −0 tests/adisp_client.py
  3. +19 −0 tests/runtests.py
View
3  tests/README.rst
@@ -7,4 +7,7 @@ a new file called ``database.cfg`` and adjust the settings to your needs.
Then run one of the following tests:
- ``async_client.py``
+- ``adisp_client.py``
- ``blocking_client.py``
+
+Or run ``runtests.py`` to run all tests.
View
80 tests/adisp_client.py
@@ -0,0 +1,80 @@
+#!/usr/bin/env python
+
+import sys
+import unittest
+
+import tornado.ioloop
+import tornado.testing
+import momoko
+
+import settings
+
+
+class AdispClientTest(tornado.testing.AsyncTestCase):
+ """``AdispClient`` tests.
+ """
+ def setUp(self):
+ super(AdispClientTest, self).setUp()
+ self.db = momoko.AdispClient({
+ 'host': settings.host,
+ 'port': settings.port,
+ 'database': settings.database,
+ 'user': settings.user,
+ 'password': settings.password,
+ 'min_conn': settings.min_conn,
+ 'max_conn': settings.max_conn,
+ 'cleanup_timeout': settings.cleanup_timeout,
+ 'ioloop': self.io_loop
+ })
+
+ def tearDown(self):
+ super(AdispClientTest, self).tearDown()
+
+ @momoko.process
+ def test_single_query(self):
+ """Test executing a single SQL query.
+ """
+ cursor = yield self.db.execute('SELECT 42, 12, 40, 11;')
+ self.assertEqual(cursor.fetchall(), [(42, 12, 40, 11)])
+
+ @momoko.process
+ def test_batch_query(self):
+ """Test executing a batch query.
+ """
+ input = {
+ 'query1': ['SELECT 42, 12, %s, %s;', (23, 56)],
+ 'query2': 'SELECT 1, 2, 3, 4, 5;',
+ 'query3': 'SELECT 465767, 4567, 3454;'
+ }
+ expected = {
+ 'query1': [(42, 12, 23, 56)],
+ 'query2': [(1, 2, 3, 4, 5)],
+ 'query3': [(465767, 4567, 3454)]
+ }
+
+ cursors = yield self.db.batch(input)
+
+ for key, cursor in cursors.items():
+ self.assertEqual(cursor.fetchall(), expected[key])
+
+ @momoko.process
+ def test_chain_query(self):
+ """Test executing a chain query.
+ """
+ input = (
+ ['SELECT 42, 12, %s, 11;', (23,)],
+ 'SELECT 1, 2, 3, 4, 5;'
+ )
+ expected = (
+ [(42, 12, 23, 11)],
+ [(1, 2, 3, 4, 5)]
+ )
+
+ cursors = yield self.db.chain(input)
+
+ for index, cursor in enumerate(cursors):
+ self.assertEqual(cursor.fetchall(), expected[index])
+
+
+if __name__ == '__main__':
+ unittest.main()
View
19 tests/runtests.py
@@ -0,0 +1,19 @@
+#!/usr/bin/env python
+
+import unittest
+
+
+TEST_MODULES = [
+ 'async_client',
+ 'adisp_client',
+ 'blocking_client'
+]
+
+
+def all():
+ return unittest.defaultTestLoader.loadTestsFromNames(TEST_MODULES)
+
+
+if __name__ == '__main__':
+ import tornado.testing
+ tornado.testing.main()
Please sign in to comment.
Something went wrong with that request. Please try again.