audrius/server-backend
1
0
Fork 0
mirror of https://github.com/OCA/server-backend.git synced 2025-02-18 09:52:42 +02:00
Code Issues Packages Projects Releases Wiki Activity

[IMP] update pre-commit

Browse Source
This commit is contained in:
JordiMForgeFlow
2023-07-06 16:50:56 +02:00
parent ddd0ff6f37
commit 6c496c114a
16 changed files with 92 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
base_external_dbsource/models/base_external_dbsource.py
View File

@@ -15,7 +15,7 @@ _logger = logging.getLogger(__name__)
class BaseExternalDbsource(models.Model):
""" It provides logic for connection to an external data source
"""It provides logic for connection to an external data source
Classes implementing this interface must provide the following methods
suffixed with the adapter type. See the method definitions and examples
@@ -96,11 +96,11 @@ class BaseExternalDbsource(models.Model):
# Interface
def change_table(self, name):
""" Change the table that is used for CRUD operations """
"""Change the table that is used for CRUD operations"""
self.current_table = name
def connection_close(self, connection):
""" It closes the connection to the data source.
"""It closes the connection to the data source.
This method calls adapter method of this same name, suffixed with
the adapter type.
@@ -111,7 +111,7 @@ class BaseExternalDbsource(models.Model):
@contextmanager
def connection_open(self):
""" It provides a context manager for the data source.
"""It provides a context manager for the data source.
This method calls adapter method of this same name, suffixed with
the adapter type.
@@ -128,23 +128,23 @@ class BaseExternalDbsource(models.Model):
_logger.exception("Connection close failure.")
def execute(self, query=None, execute_params=None, metadata=False, **kwargs):
""" Executes a query and returns a list of rows.
"""Executes a query and returns a list of rows.
"execute_params" can be a dict of values, that can be referenced
in the SQL statement using "%(key)s" or, in the case of Oracle,
":key".
Example:
query = "SELECT * FROM mytable WHERE city = %(city)s AND
date > %(dt)s"
execute_params = {
'city': 'Lisbon',
'dt': datetime.datetime(2000, 12, 31),
}
"execute_params" can be a dict of values, that can be referenced
in the SQL statement using "%(key)s" or, in the case of Oracle,
":key".
Example:
query = "SELECT * FROM mytable WHERE city = %(city)s AND
date > %(dt)s"
execute_params = {
'city': 'Lisbon',
'dt': datetime.datetime(2000, 12, 31),
}
If metadata=True, it will instead return a dict containing the
rows list and the columns list, in the format:
{ 'cols': [ 'col_a', 'col_b', ...]
, 'rows': [ (a0, b0, ...), (a1, b1, ...), ...] }
If metadata=True, it will instead return a dict containing the
rows list and the columns list, in the format:
{ 'cols': [ 'col_a', 'col_b', ...]
, 'rows': [ (a0, b0, ...), (a1, b1, ...), ...] }
"""
# Old API compatibility
@@ -168,7 +168,7 @@ class BaseExternalDbsource(models.Model):
return rows
def connection_test(self):
""" It tests the connection
"""It tests the connection
Raises:
ConnectionSuccessError: On connection success
@@ -187,7 +187,7 @@ class BaseExternalDbsource(models.Model):
)
def remote_browse(self, record_ids, *args, **kwargs):
""" It browses for and returns the records from remote by ID
"""It browses for and returns the records from remote by ID
This method calls adapter method of this same name, suffixed with
the adapter type.
@@ -205,7 +205,7 @@ class BaseExternalDbsource(models.Model):
return method(record_ids, *args, **kwargs)
def remote_create(self, vals, *args, **kwargs):
""" It creates a record on the remote data source.
"""It creates a record on the remote data source.
This method calls adapter method of this same name, suffixed with
the adapter type.
@@ -223,7 +223,7 @@ class BaseExternalDbsource(models.Model):
return method(vals, *args, **kwargs)
def remote_delete(self, record_ids, *args, **kwargs):
""" It deletes records by ID on remote
"""It deletes records by ID on remote
This method calls adapter method of this same name, suffixed with
the adapter type.
@@ -241,7 +241,7 @@ class BaseExternalDbsource(models.Model):
return method(record_ids, *args, **kwargs)
def remote_search(self, query, *args, **kwargs):
""" It searches the remote for the query.
"""It searches the remote for the query.
This method calls adapter method of this same name, suffixed with
the adapter type.
@@ -259,7 +259,7 @@ class BaseExternalDbsource(models.Model):
return method(query, *args, **kwargs)
def remote_update(self, record_ids, vals, *args, **kwargs):
""" It updates the remote records with the vals
"""It updates the remote records with the vals
This method calls adapter method of this same name, suffixed with
the adapter type.
@@ -300,7 +300,7 @@ class BaseExternalDbsource(models.Model):
# Compatibility & Private
def conn_open(self):
""" It opens and returns a connection to the remote data source.
"""It opens and returns a connection to the remote data source.
This method calls adapter method of this same name, suffixed with
the adapter type.
@@ -313,7 +313,7 @@ class BaseExternalDbsource(models.Model):
return connection
def _get_adapter_method(self, method_prefix):
""" It returns the connector adapter method for ``method_prefix``.
"""It returns the connector adapter method for ``method_prefix``.
Args:
method_prefix: (str) Prefix of adapter method (such as

44
base_external_dbsource/tests/test_base_external_dbsource.py
View File

@@ -53,7 +53,7 @@ class TestBaseExternalDbsource(common.TransactionCase):
return res, adapter
def test_conn_string_full(self):
""" It should add password if string interpolation not detected """
"""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
@@ -62,37 +62,37 @@ class TestBaseExternalDbsource(common.TransactionCase):
# Interface
def test_connection_success(self):
""" It should raise for successful connection """
"""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 """
"""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 """
"""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 """
"""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 """
"""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 """
"""It should call the adapter methods with proper args"""
expect = ("query", "execute", "metadata")
return_value = "rows", "cols"
res, adapter = self._test_adapter_method(
@@ -101,7 +101,7 @@ class TestBaseExternalDbsource(common.TransactionCase):
adapter.assert_called_once_with(*expect)
def test_execute_return(self):
""" It should return rows if not metadata """
"""It should return rows if not metadata"""
expect = (True, True, False)
return_value = "rows", "cols"
res, adapter = self._test_adapter_method(
@@ -110,7 +110,7 @@ class TestBaseExternalDbsource(common.TransactionCase):
self.assertEqual(res, return_value[0])
def test_execute_return_metadata(self):
""" It should return rows and cols if metadata """
"""It should return rows and cols if metadata"""
expect = (True, True, True)
return_value = "rows", "cols"
res, adapter = self._test_adapter_method(
@@ -119,7 +119,7 @@ class TestBaseExternalDbsource(common.TransactionCase):
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 """
"""It should call the adapter method with proper args"""
args = [1], "args"
kwargs = {"kwargs": True}
self.dbsource.current_table = "table"
@@ -130,7 +130,7 @@ class TestBaseExternalDbsource(common.TransactionCase):
self.assertEqual(res, adapter())
def test_remote_browse_asserts_current_table(self):
""" It should raise AssertionError if a table not selected """
"""It should raise AssertionError if a table not selected"""
args = [1], "args"
kwargs = {"kwargs": True}
with self.assertRaises(AssertionError):
@@ -139,7 +139,7 @@ class TestBaseExternalDbsource(common.TransactionCase):
)
def test_remote_create(self):
""" It should call the adapter method with proper args """
"""It should call the adapter method with proper args"""
args = {"val": "Value"}, "args"
kwargs = {"kwargs": True}
self.dbsource.current_table = "table"
@@ -150,7 +150,7 @@ class TestBaseExternalDbsource(common.TransactionCase):
self.assertEqual(res, adapter())
def test_remote_create_asserts_current_table(self):
""" It should raise AssertionError if a table not selected """
"""It should raise AssertionError if a table not selected"""
args = [1], "args"
kwargs = {"kwargs": True}
with self.assertRaises(AssertionError):
@@ -159,7 +159,7 @@ class TestBaseExternalDbsource(common.TransactionCase):
)
def test_remote_delete(self):
""" It should call the adapter method with proper args """
"""It should call the adapter method with proper args"""
args = [1], "args"
kwargs = {"kwargs": True}
self.dbsource.current_table = "table"
@@ -170,7 +170,7 @@ class TestBaseExternalDbsource(common.TransactionCase):
self.assertEqual(res, adapter())
def test_remote_delete_asserts_current_table(self):
""" It should raise AssertionError if a table not selected """
"""It should raise AssertionError if a table not selected"""
args = [1], "args"
kwargs = {"kwargs": True}
with self.assertRaises(AssertionError):
@@ -179,7 +179,7 @@ class TestBaseExternalDbsource(common.TransactionCase):
)
def test_remote_search(self):
""" It should call the adapter method with proper args """
"""It should call the adapter method with proper args"""
args = {"search": "query"}, "args"
kwargs = {"kwargs": True}
self.dbsource.current_table = "table"
@@ -190,7 +190,7 @@ class TestBaseExternalDbsource(common.TransactionCase):
self.assertEqual(res, adapter())
def test_remote_search_asserts_current_table(self):
""" It should raise AssertionError if a table not selected """
"""It should raise AssertionError if a table not selected"""
args = [1], "args"
kwargs = {"kwargs": True}
with self.assertRaises(AssertionError):
@@ -199,7 +199,7 @@ class TestBaseExternalDbsource(common.TransactionCase):
)
def test_remote_update(self):
""" It should call the adapter method with proper args """
"""It should call the adapter method with proper args"""
args = [1], {"vals": "Value"}, "args"
kwargs = {"kwargs": True}
self.dbsource.current_table = "table"
@@ -210,7 +210,7 @@ class TestBaseExternalDbsource(common.TransactionCase):
self.assertEqual(res, adapter())
def test_remote_update_asserts_current_table(self):
""" It should raise AssertionError if a table not selected """
"""It should raise AssertionError if a table not selected"""
args = [1], "args"
kwargs = {"kwargs": True}
with self.assertRaises(AssertionError):
@@ -221,7 +221,7 @@ class TestBaseExternalDbsource(common.TransactionCase):
# Postgres
def test_execute_postgresql(self):
""" It should call generic executor with proper args """
"""It should call generic executor with proper args"""
expect = ("query", "execute", "metadata")
with mock.patch.object(
self.dbsource, "_execute_generic", autospec=True
@@ -233,7 +233,7 @@ class TestBaseExternalDbsource(common.TransactionCase):
# Old API Compat
def test_execute_calls_adapter_old_api(self):
""" It should call the adapter correctly if old kwargs provided """
"""It should call the adapter correctly if old kwargs provided"""
expect = [None, None, "metadata"]
with mock.patch.object(
self.dbsource, "execute_postgresql", autospec=True
@@ -244,7 +244,7 @@ class TestBaseExternalDbsource(common.TransactionCase):
psql.assert_called_once_with(*expect)
def test_conn_open(self):
""" It should return open connection for use """
"""It should return open connection for use"""
with mock.patch.object(
self.dbsource, "connection_open", autospec=True
) as connection: