mirror of
https://github.com/OCA/server-backend.git
synced 2025-02-18 09:52:42 +02:00
0a5ccebdfc
Before, postgres configuration was hardcoded. This worked OK for OCA's CI infrastructure, but failed in any other out there that had different defaults. There's no need to hardcode it; Odoo already must have Postgres connection settings, so let's use those instead. This will make tests testable across different CI infrastructures.
263 lines
9.9 KiB
Python
263 lines
9.9 KiB
Python
# Copyright 2016 LasLabs Inc.
|
|
|
|
import mock
|
|
|
|
from odoo.tests import common
|
|
from odoo.sql_db import connection_info_for
|
|
|
|
from ..exceptions import ConnectionFailedError, ConnectionSuccessError
|
|
|
|
|
|
class TestBaseExternalDbsource(common.TransactionCase):
|
|
|
|
def setUp(self):
|
|
super(TestBaseExternalDbsource, self).setUp()
|
|
# Obtain current odoo instance DB connection settings
|
|
connection_info = connection_info_for(self.env.cr.dbname)[1]
|
|
# Adapt to the format expected by this module
|
|
password = connection_info.get("password", "")
|
|
connection_info["password"] = "%s"
|
|
connection_info["dbname"] = connection_info["database"]
|
|
del connection_info["database"]
|
|
# Create a proper dbsource record to test
|
|
self.dbsource = self.env["base.external.dbsource"].create({
|
|
"conn_string": " ".join(
|
|
"%s='%s'" % item for item in connection_info.items()
|
|
),
|
|
"connector": "postgresql",
|
|
"name": "test postgres with current odoo config",
|
|
"password": password,
|
|
})
|
|
|
|
def _test_adapter_method(
|
|
self, method_name, side_effect=None, return_value=None,
|
|
create=False, args=None, kwargs=None,
|
|
):
|
|
if args is None:
|
|
args = []
|
|
if kwargs is None:
|
|
kwargs = {}
|
|
adapter = '%s_postgresql' % method_name
|
|
with mock.patch.object(self.dbsource,
|
|
adapter, create=create) as adapter:
|
|
if side_effect is not None:
|
|
adapter.side_effect = side_effect
|
|
elif return_value is not None:
|
|
adapter.return_value = return_value
|
|
res = getattr(self.dbsource, method_name)(*args, **kwargs)
|
|
return res, adapter
|
|
|
|
def test_conn_string_full(self):
|
|
""" It should add password if string interpolation not detected """
|
|
self.dbsource.conn_string = 'User=Derp;'
|
|
self.dbsource.password = 'password'
|
|
expect = self.dbsource.conn_string + 'PWD=%s;' % self.dbsource.password
|
|
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 """
|
|
with mock.patch.object(self.dbsource, 'connection_open') as conn:
|
|
conn.side_effect = Exception
|
|
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():
|
|
pass
|
|
close.assert_called_once()
|
|
|
|
def test_connection_close(self):
|
|
""" It should call adapter's close method """
|
|
args = [mock.MagicMock()]
|
|
res, adapter = self._test_adapter_method(
|
|
'connection_close', args=args,
|
|
)
|
|
adapter.assert_called_once_with(args[0])
|
|
|
|
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')
|
|
return_value = 'rows', 'cols'
|
|
res, adapter = self._test_adapter_method(
|
|
'execute', args=expect, return_value=return_value,
|
|
)
|
|
adapter.assert_called_once_with(*expect)
|
|
|
|
def test_execute_return(self):
|
|
""" It should return rows if not metadata """
|
|
expect = (True, True, False)
|
|
return_value = 'rows', 'cols'
|
|
res, adapter = self._test_adapter_method(
|
|
'execute', args=expect, return_value=return_value,
|
|
)
|
|
self.assertEqual(res, return_value[0])
|
|
|
|
def test_execute_return_metadata(self):
|
|
""" It should return rows and cols if metadata """
|
|
expect = (True, True, True)
|
|
return_value = 'rows', 'cols'
|
|
res, adapter = self._test_adapter_method(
|
|
'execute', args=expect, return_value=return_value,
|
|
)
|
|
self.assertEqual(
|
|
res,
|
|
{'rows': return_value[0],
|
|
'cols': return_value[1]},
|
|
)
|
|
|
|
def test_remote_browse(self):
|
|
""" It should call the adapter method with proper args """
|
|
args = [1], 'args'
|
|
kwargs = {'kwargs': True}
|
|
self.dbsource.current_table = 'table'
|
|
res, adapter = self._test_adapter_method(
|
|
'remote_browse', create=True, args=args, kwargs=kwargs,
|
|
)
|
|
adapter.assert_called_once_with(*args, **kwargs)
|
|
self.assertEqual(res, adapter())
|
|
|
|
def test_remote_browse_asserts_current_table(self):
|
|
""" It should raise AssertionError if a table not selected """
|
|
args = [1], 'args'
|
|
kwargs = {'kwargs': True}
|
|
with self.assertRaises(AssertionError):
|
|
res, adapter = self._test_adapter_method(
|
|
'remote_browse', create=True, args=args, kwargs=kwargs,
|
|
)
|
|
|
|
def test_remote_create(self):
|
|
""" It should call the adapter method with proper args """
|
|
args = {'val': 'Value'}, 'args'
|
|
kwargs = {'kwargs': True}
|
|
self.dbsource.current_table = 'table'
|
|
res, adapter = self._test_adapter_method(
|
|
'remote_create', create=True, args=args, kwargs=kwargs,
|
|
)
|
|
adapter.assert_called_once_with(*args, **kwargs)
|
|
self.assertEqual(res, adapter())
|
|
|
|
def test_remote_create_asserts_current_table(self):
|
|
""" It should raise AssertionError if a table not selected """
|
|
args = [1], 'args'
|
|
kwargs = {'kwargs': True}
|
|
with self.assertRaises(AssertionError):
|
|
res, adapter = self._test_adapter_method(
|
|
'remote_create', create=True, args=args, kwargs=kwargs,
|
|
)
|
|
|
|
def test_remote_delete(self):
|
|
""" It should call the adapter method with proper args """
|
|
args = [1], 'args'
|
|
kwargs = {'kwargs': True}
|
|
self.dbsource.current_table = 'table'
|
|
res, adapter = self._test_adapter_method(
|
|
'remote_delete', create=True, args=args, kwargs=kwargs,
|
|
)
|
|
adapter.assert_called_once_with(*args, **kwargs)
|
|
self.assertEqual(res, adapter())
|
|
|
|
def test_remote_delete_asserts_current_table(self):
|
|
""" It should raise AssertionError if a table not selected """
|
|
args = [1], 'args'
|
|
kwargs = {'kwargs': True}
|
|
with self.assertRaises(AssertionError):
|
|
res, adapter = self._test_adapter_method(
|
|
'remote_delete', create=True, args=args, kwargs=kwargs,
|
|
)
|
|
|
|
def test_remote_search(self):
|
|
""" It should call the adapter method with proper args """
|
|
args = {'search': 'query'}, 'args'
|
|
kwargs = {'kwargs': True}
|
|
self.dbsource.current_table = 'table'
|
|
res, adapter = self._test_adapter_method(
|
|
'remote_search', create=True, args=args, kwargs=kwargs,
|
|
)
|
|
adapter.assert_called_once_with(*args, **kwargs)
|
|
self.assertEqual(res, adapter())
|
|
|
|
def test_remote_search_asserts_current_table(self):
|
|
""" It should raise AssertionError if a table not selected """
|
|
args = [1], 'args'
|
|
kwargs = {'kwargs': True}
|
|
with self.assertRaises(AssertionError):
|
|
res, adapter = self._test_adapter_method(
|
|
'remote_search', create=True, args=args, kwargs=kwargs,
|
|
)
|
|
|
|
def test_remote_update(self):
|
|
""" It should call the adapter method with proper args """
|
|
args = [1], {'vals': 'Value'}, 'args'
|
|
kwargs = {'kwargs': True}
|
|
self.dbsource.current_table = 'table'
|
|
res, adapter = self._test_adapter_method(
|
|
'remote_update', create=True, args=args, kwargs=kwargs,
|
|
)
|
|
adapter.assert_called_once_with(*args, **kwargs)
|
|
self.assertEqual(res, adapter())
|
|
|
|
def test_remote_update_asserts_current_table(self):
|
|
""" It should raise AssertionError if a table not selected """
|
|
args = [1], 'args'
|
|
kwargs = {'kwargs': True}
|
|
with self.assertRaises(AssertionError):
|
|
res, adapter = self._test_adapter_method(
|
|
'remote_update', create=True, args=args, kwargs=kwargs,
|
|
)
|
|
|
|
# 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', autospec=True,
|
|
) as execute:
|
|
execute.return_value = 'rows', 'cols'
|
|
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', autospec=True,
|
|
) as psql:
|
|
psql.return_value = 'rows', 'cols'
|
|
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', autospec=True,
|
|
) as connection:
|
|
res = self.dbsource.conn_open()
|
|
self.assertEqual(
|
|
res,
|
|
connection().__enter__(),
|
|
)
|