Add test coverage in core

base_external_dbsource/models/base_external_dbsource.py

@@ -140,7 +140,8 @@ def execute(self, query=None, execute_params=None, metadata=False, **kwargs):
  pass
 
  method = self._get_adapter_method('execute')
- rows, cols = method(sqlquery, sqlparams, metadata, execute_params)
+ rows, cols = method(query, execute_params, metadata)
+
  if metadata:
  return{'cols': cols, 'rows': rows}
  else:
@@ -178,13 +179,13 @@ def connection_close_postgresql(self, connection):
  def connection_open_postgresql(self):
  return psycopg2.connect(self.conn_string)
 
- def execute_postgresql(self, sqlquery, sqlparams, metadata):
- return self._execute_generic(sqlquery, sqlparams, metadata)
+ def execute_postgresql(self, query, params, metadata):
+ return self._execute_generic(query, params, metadata)
 
- def _execute_generic(self, sqlquery, sqlparams, metadata):
+ def _execute_generic(self, query, params, metadata):
  with self.connection_open() as connection:
  cur = connection.cursor()
- cur.execute(sqlquery, sqlparams)
+ cur.execute(query, params)
  cols = []
  if metadata:
  cols = [d[0] for d in cur.description]

base_external_dbsource/tests/__init__.py

@@ -1,3 +1,3 @@
 # -*- encoding: utf-8 -*-
 
-from . import test_create_dbsource
+from . import test_base_external_dbsource

base_external_dbsource/tests/test_base_external_dbsource.py

@@ -0,0 +1,110 @@
+# -*- coding: utf-8 -*-
+# Copyright 2016 LasLabs Inc.
+
+import mock
+
+from openerp.tests import common
+
+from ..exceptions import ConnectionFailedError, ConnectionSuccessError
+
+
+class TestBaseExternalDbsource(common.TransactionCase):
+
+ def setUp(self):
+ super(TestBaseExternalDbsource, self).setUp()
+ self.dbsource = self.env.ref('base_external_dbsource.demo_postgre')
+
+ def test_conn_string_full(self):
+ """ It should add password if string interpolation not detected """
+ self.dbsource.conn_string = 'conn string'
+ expect = self.dbsource.conn_string + ';PWD%s'
+ self.assertEqual(
+ self.dbsource.conn_string_full, expect,
+ )
+
+ # Interface
+
+ def test_connection_success(self):
+ """ It should raise for successful connection """
+ with self.assertRaises(ConnectionSuccessError):
+ self.dbsource.connection_test()
+
+ def test_connection_fail(self):
+ """ It should raise for failed/invalid connection """
+ # Connection without connection_string
+ self.dbsource.conn_string = ""
+ with self.assertRaises(ConnectionFailedError):
+ self.dbsource.connection_test()
+
+ def test_connection_open_calls_close(self):
+ """ It should close connection after context ends """
+ with mock.patch.object(self.dbsource, 'connection_close') as close:
+ with self.dbsource.connection_open() as connection:
+ pass
+ close.assert_called_once_with(connection)
+
+ def test_connection_close(self):
+ """ It should call adapter's close method """
+ with mock.patch.object(
+ self.dbsource, 'connection_close_postgresql'
+ ) as close:
+ expect = mock.MagicMock()
+ self.dbsource.connection_close(expect)
+ close.assert_called_once_with(expect)
+
+ def test_execute_asserts_query_arg(self):
+ """ It should raise a TypeError if query and sqlquery not in args """
+ with self.assertRaises(TypeError):
+ self.dbsource.execute()
+
+ def test_execute_calls_adapter(self):
+ """ It should call the adapter methods with proper args """
+ expect = ('query', 'execute', 'metadata')
+ with mock.patch.object(self.dbsource, 'execute_postgresql') as psql:
+ self.dbsource.execute(*expect)
+ psql.assert_called_once_with(*expect)
+
+ def test_execute_return(self):
+ """ It should return rows if not metadata """
+ with mock.patch.object(self.dbsource, 'execute_postgresql') as psql:
+ psql.return_value = 'rows', 'cols'
+ res = self.dbsource.execute(True, True, False)
+ self.assertEqual(res, 'rows')
+
+ def test_execute_return_metadata(self):
+ """ It should return rows and cols if metadata """
+ with mock.patch.object(self.dbsource, 'execute_postgresql') as psql:
+ psql.return_value = 'rows', 'cols'
+ res = self.dbsource.execute(True, True, False)
+ self.assertDictEqual(
+ res,
+ {'rows': 'rows', 'cols': 'cols'},
+ )
+
+ # Postgres
+
+ def test_execute_postgresql(self):
+ """ It should call generic executor with proper args """
+ expect = ('query', 'execute', 'metadata')
+ with mock.patch.object(self.dbsource, '_execute_generic') as execute:
+ self.dbsource.execute(*expect)
+ execute.assert_called_once_with(*expect)
+
+ # Old API Compat
+
+ def test_execute_calls_adapter_old_api(self):
+ """ It should call the adapter correctly if old kwargs provided """
+ expect = [None, None, 'metadata']
+ with mock.patch.object(self.dbsource, 'execute_postgresql') as psql:
+ self.dbsource.execute(*expect, sqlparams='params', sqlquery='query')
+ expect[0], expect[1] = 'query', 'params'
+ psql.assert_called_once_with(*expect)
+
+ def test_conn_open(self):
+ """ It should return open connection for use """
+ with mock.patch.object(self.dbsource, 'connection_open') as connection:
+ res = self.dbsource.conn_open()
+ self.assertEqual(
+ res,
+ connection().__enter__(),
+ )