From f8781ae910dc000f510135661aa76c54784afc66 Mon Sep 17 00:00:00 2001 From: Dave Lasley Date: Sat, 10 Dec 2016 09:56:04 -0800 Subject: [PATCH] Add test coverage in core --- .../models/base_external_dbsource.py | 11 +- base_external_dbsource/tests/__init__.py | 2 +- .../tests/test_base_external_dbsource.py | 110 ++++++++++++++++++ .../tests/test_create_dbsource.py | 58 --------- 4 files changed, 117 insertions(+), 64 deletions(-) create mode 100644 base_external_dbsource/tests/test_base_external_dbsource.py delete mode 100644 base_external_dbsource/tests/test_create_dbsource.py diff --git a/base_external_dbsource/models/base_external_dbsource.py b/base_external_dbsource/models/base_external_dbsource.py index 3b273a8dc9c..f50682dab8e 100644 --- a/base_external_dbsource/models/base_external_dbsource.py +++ b/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] diff --git a/base_external_dbsource/tests/__init__.py b/base_external_dbsource/tests/__init__.py index 4c52d80b3a9..768d7990a6f 100644 --- a/base_external_dbsource/tests/__init__.py +++ b/base_external_dbsource/tests/__init__.py @@ -1,3 +1,3 @@ # -*- encoding: utf-8 -*- -from . import test_create_dbsource +from . import test_base_external_dbsource diff --git a/base_external_dbsource/tests/test_base_external_dbsource.py b/base_external_dbsource/tests/test_base_external_dbsource.py new file mode 100644 index 00000000000..08c592f1de8 --- /dev/null +++ b/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__(), + ) diff --git a/base_external_dbsource/tests/test_create_dbsource.py b/base_external_dbsource/tests/test_create_dbsource.py deleted file mode 100644 index f255fb62c7d..00000000000 --- a/base_external_dbsource/tests/test_create_dbsource.py +++ /dev/null @@ -1,58 +0,0 @@ -# -*- coding: utf-8 -*- - -from openerp.exceptions import Warning as UserError -from openerp.tests import common -import logging - - -class TestCreateDbsource(common.TransactionCase): - """Test class for base_external_dbsource.""" - - def test_create_dbsource(self): - """source creation should succeed.""" - dbsource = self.env.ref('base_external_dbsource.demo_postgre') - try: - dbsource.connection_test() - except UserError as e: - logging.warning("Log = " + str(e)) - self.assertTrue(u'Everything seems properly set up!' in str(e)) - - def test_create_dbsource_failed(self): - """source creation without connection string should failed.""" - dbsource = self.env.ref('base_external_dbsource.demo_postgre') - - # Connection without connection_string - dbsource.conn_string = "" - try: - dbsource.connection_test() - except UserError as e: - logging.warning("Log = " + str(e)) - self.assertTrue(u'Here is what we got instead:' in str(e)) - - def test_create_dbsource_without_connector_failed(self): - """source creation with other connector should failed.""" - dbsource = self.env.ref('base_external_dbsource.demo_postgre') - - # Connection to mysql - try: - dbsource.connector = "mysql" - dbsource.connection_test() - except ValueError as e: - logging.warning("Log = " + str(e)) - self.assertTrue(u'Wrong value for' in str(e)) - - # Connection to mysql - try: - dbsource.connector = "pyodbc" - dbsource.connection_test() - except ValueError as e: - logging.warning("Log = " + str(e)) - self.assertTrue(u'Wrong value for' in str(e)) - - # Connection to oracle - try: - dbsource.connector = "cx_Oracle" - dbsource.connection_test() - except ValueError as e: - logging.warning("Log = " + str(e)) - self.assertTrue(u'Wrong value for' in str(e))