mirror of
https://github.com/OCA/server-backend.git
synced 2025-02-18 09:52:42 +02:00
253 lines
9.7 KiB
Python
253 lines
9.7 KiB
Python
# Copyright 2016 LasLabs Inc.
|
|
|
|
import mock
|
|
|
|
from odoo.sql_db import connection_info_for
|
|
from odoo.tests import common
|
|
|
|
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__())
|