[MIG] base_external_dbsource: Migration to v13.0

base_external_dbsource/tests/test_base_external_dbsource.py

@@ -2,14 +2,13 @@
 
 import mock
 
-from odoo.tests import common
 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
@@ -20,26 +19,32 @@ class TestBaseExternalDbsource(common.TransactionCase):
         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,
-        })
+        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,
+        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:
+        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:
@@ -49,12 +54,10 @@ class TestBaseExternalDbsource(common.TransactionCase):
 
     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,
-        )
+        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
 
@@ -65,16 +68,14 @@ class TestBaseExternalDbsource(common.TransactionCase):
 
     def test_connection_fail(self):
         """ It should raise for failed/invalid connection """
-        with mock.patch.object(self.dbsource, 'connection_open') as conn:
+        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 mock.patch.object(self.dbsource, "connection_close") as close:
             with self.dbsource.connection_open():
                 pass
             close.assert_called_once()
@@ -82,9 +83,7 @@ class TestBaseExternalDbsource(common.TransactionCase):
     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,
-        )
+        res, adapter = self._test_adapter_method("connection_close", args=args)
         adapter.assert_called_once_with(args[0])
 
     def test_execute_asserts_query_arg(self):
@@ -94,144 +93,140 @@ class TestBaseExternalDbsource(common.TransactionCase):
 
     def test_execute_calls_adapter(self):
         """ It should call the adapter methods with proper args """
-        expect = ('query', 'execute', 'metadata')
-        return_value = 'rows', 'cols'
+        expect = ("query", "execute", "metadata")
+        return_value = "rows", "cols"
         res, adapter = self._test_adapter_method(
-            'execute', args=expect, return_value=return_value,
+            "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'
+        return_value = "rows", "cols"
         res, adapter = self._test_adapter_method(
-            'execute', args=expect, return_value=return_value,
+            "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'
+        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]},
+            "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'
+        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,
+            "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}
+        args = [1], "args"
+        kwargs = {"kwargs": True}
         with self.assertRaises(AssertionError):
             res, adapter = self._test_adapter_method(
-                'remote_browse', create=True, args=args, kwargs=kwargs,
+                "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'
+        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,
+            "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}
+        args = [1], "args"
+        kwargs = {"kwargs": True}
         with self.assertRaises(AssertionError):
             res, adapter = self._test_adapter_method(
-                'remote_create', create=True, args=args, kwargs=kwargs,
+                "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'
+        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,
+            "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}
+        args = [1], "args"
+        kwargs = {"kwargs": True}
         with self.assertRaises(AssertionError):
             res, adapter = self._test_adapter_method(
-                'remote_delete', create=True, args=args, kwargs=kwargs,
+                "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'
+        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,
+            "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}
+        args = [1], "args"
+        kwargs = {"kwargs": True}
         with self.assertRaises(AssertionError):
             res, adapter = self._test_adapter_method(
-                'remote_search', create=True, args=args, kwargs=kwargs,
+                "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'
+        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,
+            "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}
+        args = [1], "args"
+        kwargs = {"kwargs": True}
         with self.assertRaises(AssertionError):
             res, adapter = self._test_adapter_method(
-                'remote_update', create=True, args=args, kwargs=kwargs,
+                "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')
+        expect = ("query", "execute", "metadata")
         with mock.patch.object(
-            self.dbsource, '_execute_generic', autospec=True,
+            self.dbsource, "_execute_generic", autospec=True
         ) as execute:
-            execute.return_value = 'rows', 'cols'
+            execute.return_value = "rows", "cols"
             self.dbsource.execute(*expect)
             execute.assert_called_once_with(*expect)
 
@@ -239,24 +234,19 @@ class TestBaseExternalDbsource(common.TransactionCase):
 
     def test_execute_calls_adapter_old_api(self):
         """ It should call the adapter correctly if old kwargs provided """
-        expect = [None, None, 'metadata']
+        expect = [None, None, "metadata"]
         with mock.patch.object(
-            self.dbsource, 'execute_postgresql', autospec=True,
+            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.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,
+            self.dbsource, "connection_open", autospec=True
         ) as connection:
             res = self.dbsource.conn_open()
-            self.assertEqual(
-                res,
-                connection().__enter__(),
-            )
+            self.assertEqual(res, connection().__enter__())