Extend tests for interactive mode

florentx · florentx · commit 6cfdeed6f3f8 · 2025-11-11T07:47:56.000+01:00
diff --git a/odooly.py b/odooly.py
@@ -2415,11 +2415,8 @@ def excepthook(exc_type, exc, tb):
         print(msg.strip())
     sys.excepthook = excepthook
 
-    class Usage:
-        def __call__(self):
-            print(usage)
-        __repr__ = lambda s: usage
-    builtins.usage = Usage()
+    builtins.usage = type('Usage', (), {'__call__': lambda s: print(usage),
+                                        '__repr__': lambda s: usage})()
 
     try:
         import readline as rl
diff --git a/tests/_common.py b/tests/_common.py
@@ -1,11 +1,10 @@
 from io import BytesIO
-from urllib.request import HTTPError
+from urllib.request import HTTPError, urljoin
 from unittest import mock, TestCase
-from unittest.mock import call, sentinel
+from unittest.mock import ANY, call, sentinel
 
 import odooly
 
-sample_context = {'lang': 'en_US', 'tz': 'Europe/Zurich'}
 type_call = type(call)
 
 
@@ -20,65 +19,42 @@ def popvalue(self):
 
 def OBJ(model, method, *params, **kw):
     if 'context' not in kw:
-        kw['context'] = sample_context
+        kw['context'] = {**OdooTestCase.user_context}
     elif kw['context'] is None:
         del kw['context']
     return ('object.execute_kw', sentinel.AUTH, model, method, params) + ((kw,) if kw else ())
 
 
-class XmlRpcTestCase(TestCase):
+class OdooTestCase(TestCase):
     server_version = None
     server = "http://192.0.2.199:9999"
     database = user = password = uid = None
+    user_context = {'lang': 'en_US', 'tz': 'Europe/Zurich'}
     maxDiff = None
 
     def setUp(self):
         self.addCleanup(mock.patch.stopall)
         self.stdout = mock.patch('sys.stdout', new=PseudoFile()).start()
         self.stderr = mock.patch('sys.stderr', new=PseudoFile()).start()
+        self.http_request = self._patch_http_request()
 
         # Clear the login cache
         mock.patch.dict('odooly.Env._cache', clear=True).start()
 
         # Avoid hanging on getpass
         mock.patch('odooly.getpass', side_effect=RuntimeError).start()
 
-        self.service = self._patch_service()
-        if self.server and self.database:
-            # create the client
-            self.client = odooly.Client(
-                self.server, self.database, self.user, self.password)
-            self.env = self.client.env
-            # reset the mock
-            self.service.reset_mock()
-
-    def _patch_http_request(self, uid=None, context=sample_context):
+    def _patch_http_request(self, uid=None, context=None):
         def func(url, *, method='POST', data=None, json=None, headers=None):
             if url.endswith("/web/session/authenticate"):
-                result = {'uid': uid or self.uid, 'user_context': context}
+                result = {'uid': uid or self.uid, 'user_context': context or self.user_context}
             else:
                 with HTTPError(url, 404, 'Not Found', headers, BytesIO()) as not_found:
                     raise not_found
             return {'result': result}
         return mock.patch('odooly.HTTPSession.request', side_effect=func).start()
 
-    def _patch_service(self):
-        def get_svc(server, name, *args, **kwargs):
-            return getattr(svcs, name)
-        patcher = mock.patch('odooly.Service', side_effect=get_svc)
-        svcs = patcher.start()
-        svcs.stop = patcher.stop
-        for svc_name in 'db common object wizard report'.split():
-            svcs.attach_mock(mock.Mock(name=svc_name), svc_name)
-        # Default values
-        svcs.db.server_version.return_value = self.server_version
-        svcs.db.list.return_value = [self.database]
-        svcs.common.login.return_value = self.uid
-        # env['res.users'].context_get()
-        svcs.object.execute_kw.return_value = sample_context
-        return svcs
-
-    def _assertCalls(self, mock_, expected_calls):
+    def assertMockCalls(self, mock_, expected_calls):
         for idx, expected in enumerate(expected_calls):
             if isinstance(expected, str):
                 if expected[:4] == 'call':
@@ -88,17 +64,22 @@ def _assertCalls(self, mock_, expected_calls):
         self.assertSequenceEqual(mock_.mock_calls, expected_calls)
         mock_.reset_mock()
 
-    def assertServiceCalls(self, *expected_args):
+    def assertRequests(self, *expected_args):
+        server = urljoin(self.server, '/').rstrip('/')
         expected_calls = list(expected_args)
         for idx, expected in enumerate(expected_calls):
-            if not isinstance(expected, type_call) and isinstance(expected, tuple):
-                rpcmethod = expected[0]
-                if len(expected) > 1 and expected[1] == sentinel.AUTH:
-                    args = (self.database, self.uid, self.password) + expected[2:]
-                else:
-                    args = expected[1:]
-                expected_calls[idx] = getattr(call, rpcmethod)(*args)
-        self._assertCalls(self.service, expected_calls)
+            if isinstance(expected, tuple):
+                if expected[0].startswith('/json/2/'):
+                    headers = {
+                        'Authorization': f'Bearer {self.password}',
+                        'Content-Type': 'application/json',
+                        'X-Odoo-Database': self.database,
+                    }
+                    expected_calls[idx] = call(f"{server}{expected[0]}", json=expected[1], headers=headers)
+                elif expected[0].startswith('/web/'):
+                    jsonrpc_params = {'jsonrpc': '2.0', 'method': 'call', 'params': expected[1], 'id': ANY}
+                    expected_calls[idx] = call(f"{server}{expected[0]}", json=jsonrpc_params)
+        self.assertMockCalls(self.http_request, expected_calls)
 
     def assertOutput(self, stdout='', stderr='', startswith=False):
         # compare with ANY to make sure output is not empty
@@ -117,5 +98,47 @@ def assertOutput(self, stdout='', stderr='', startswith=False):
                 stdout_value = stdout_value[:len(stdout)]
             self.assertMultiLineEqual(stdout_value, stdout)
 
+
+class XmlRpcTestCase(OdooTestCase):
+
+    def setUp(self):
+        super().setUp()
+        self.service = self._patch_service()
+        if self.server and self.database:
+            # create the client
+            self.client = odooly.Client(
+                self.server, self.database, self.user, self.password)
+            self.env = self.client.env
+            # reset the mock
+            self.service.reset_mock()
+
+    def _patch_service(self):
+        def get_svc(server, name, *args, **kwargs):
+            return getattr(svcs, name)
+        patcher = mock.patch('odooly.Service', side_effect=get_svc)
+        svcs = patcher.start()
+        svcs.stop = patcher.stop
+        for svc_name in 'db common object wizard report'.split():
+            svcs.attach_mock(mock.Mock(name=svc_name), svc_name)
+        # Default values
+        svcs.db.server_version.return_value = self.server_version
+        svcs.db.list.return_value = [self.database]
+        svcs.common.login.return_value = self.uid
+        # env['res.users'].context_get()
+        svcs.object.execute_kw.return_value = self.user_context
+        return svcs
+
+    def assertServiceCalls(self, *expected_args):
+        expected_calls = list(expected_args)
+        for idx, expected in enumerate(expected_calls):
+            if not isinstance(expected, type_call) and isinstance(expected, tuple):
+                rpcmethod = expected[0]
+                if len(expected) > 1 and expected[1] == sentinel.AUTH:
+                    args = (self.database, self.uid, self.password) + expected[2:]
+                else:
+                    args = expected[1:]
+                expected_calls[idx] = getattr(call, rpcmethod)(*args)
+        self.assertMockCalls(self.service, expected_calls)
+
     # Legacy
     assertCalls = assertServiceCalls
diff --git a/tests/test_client.py b/tests/test_client.py
@@ -757,10 +757,6 @@ class TestClientApi19(TestClientApi):
     test_exec_workflow = test_wizard = _skip_test
     test_report = test_render_report = test_report_get = _skip_test
 
-    def _patch_service(self):
-        self.auth_http = self._patch_http_request()
-        return super()._patch_service()
-
     def test_obsolete_methods(self):
         self.assertRaises(AttributeError, getattr, self.env, 'exec_workflow')
         self.assertRaises(AttributeError, getattr, self.env, 'render_report')
diff --git a/tests/test_interact.py b/tests/test_interact.py
@@ -4,12 +4,35 @@
 from unittest.mock import call, ANY
 
 import odooly
-from ._common import XmlRpcTestCase
+from ._common import OdooTestCase, XmlRpcTestCase
 
 
-class TestInteract(XmlRpcTestCase):
+class _TestInteract(OdooTestCase):
+
+    def setUp(self):
+        super().setUp()
+        # Hide readline module
+        mock.patch.dict('sys.modules', {'readline': None}).start()
+        mock.patch('odooly.Client._globals', None).start()
+        mock.patch('odooly.Client._set_interactive', wraps=odooly.Client._set_interactive).start()
+        self.interact = mock.patch('odooly._interact', wraps=odooly._interact).start()
+        self.infunc = mock.patch('code.InteractiveConsole.raw_input').start()
+        mock.patch('odooly.main.__defaults__', (self.interact,)).start()
+
+    def _resp_version_info(self):
+        [major, minor] = self.server_version.split('.')
+        return {
+            'server_version': self.server_version,
+            'server_version_info': [int(major), int(minor), 0, 'final', 0, ''],
+            'server_serie': self.server_version,
+            'protocol_version': 1,
+        }
+
+
+class TestInteractXmlRpc(XmlRpcTestCase, _TestInteract):
+    """Test interactive mode with OpenERP 6.1."""
     server_version = '6.1'
-    server = f"{XmlRpcTestCase.server}/xmlrpc"
+    server = f"{OdooTestCase.server}/xmlrpc"
     startup_calls = (
         call(ANY, 'db', ANY),
         'db.server_version',
@@ -21,16 +44,6 @@ class TestInteract(XmlRpcTestCase):
         'db.list',
     )
 
-    def setUp(self):
-        super().setUp()
-        # Hide readline module
-        mock.patch.dict('sys.modules', {'readline': None}).start()
-        mock.patch('odooly.Client._globals', None).start()
-        mock.patch('odooly.Client._set_interactive', wraps=odooly.Client._set_interactive).start()
-        self.interact = mock.patch('odooly._interact', wraps=odooly._interact).start()
-        self.infunc = mock.patch('code.InteractiveConsole.raw_input').start()
-        mock.patch('odooly.main.__defaults__', (self.interact,)).start()
-
     def test_main(self):
         env_tuple = (self.server, 'database', 'usr', None, None)
         mock.patch('sys.argv', new=['odooly', '--env', 'demo']).start()
@@ -147,3 +160,70 @@ def usr17(model, method, *params):
             'Model not found: res.company',
         ])
         self.assertOutput(stderr=ANY)
+
+
+class TestInteract19(_TestInteract):
+    """Test interactive mode with Odoo 19."""
+    server_version = '19.0'
+    server = f"{OdooTestCase.server}/"
+    database, user, password, uid = 'database', 'usr', 'password', 17
+    startup_calls = (
+        ('/web/webclient/version_info', {}),
+        ('/web/database/list', {}),
+        ('/web/session/authenticate', {'db': database, 'login': user, 'password': password}),
+        ('/json/2/res.users/context_get', {}),
+    )
+
+    def test_main(self):
+        env_tuple = (self.server, self.database, self.user, None, None)
+        mock.patch('sys.argv', new=['odooly', '--env', 'demo']).start()
+        read_config = mock.patch('odooly.Client.get_config',
+                                 return_value=env_tuple).start()
+        getpass = mock.patch('odooly.getpass',
+                             return_value=self.password).start()
+        self.http_request.side_effect = [
+            {'result': self._resp_version_info()},
+            [],
+            {'result': {'uid': 17,  'user_context': self.user_context}},
+            {'uid': self.uid, **self.user_context},
+            #
+            {'result': {'uid': 51,  'user_context': self.user_context}},
+            OSError,
+            {'result': {'uid': 51,  'user_context': self.user_context}},
+        ]
+
+        # Launch interactive
+        self.infunc.side_effect = [
+            "client\n",
+            "env\n",
+            "env.sudo('gaspard')\n",
+            "client.login('gaspard')\n",
+            "23 + 19\n",
+            EOFError('Finished')]
+        odooly.main()
+
+        self.assertEqual(sys.ps1, 'demo >>> ')
+        self.assertEqual(sys.ps2, '     ... ')
+        expected_calls = self.startup_calls + (
+            ('/web/session/authenticate', {'db': 'database', 'login': 'gaspard', 'password': 'password'}),
+            ('/json/2/res.users/context_get', {}),
+            ('/web/session/authenticate', {'db': 'database', 'login': 'gaspard', 'password': 'password'}),
+        )
+        self.assertRequests(*expected_calls)
+        self.assertEqual(getpass.call_count, 2)
+        self.assertEqual(read_config.call_count, 1)
+        self.assertEqual(self.interact.call_count, 1)
+        outlines = self.stdout.popvalue().splitlines()
+        self.assertSequenceEqual(outlines[-6:], [
+            "Logged in as 'usr'",
+            f"<Client '{self.server}web?db=database'>",
+            "<Env 'usr@database'>",
+            "<Env 'gaspard@database'>",
+            "Logged in as 'gaspard'",
+            "42",
+        ])
+        self.assertOutput(stderr='\x1b[A\n\n', startswith=True)
+
+    # TODO
+    test_no_database = None
+    test_invalid_user_password = None
diff --git a/tests/test_model.py b/tests/test_model.py
@@ -1,6 +1,5 @@
 from functools import partial
-from unittest.mock import call, sentinel, ANY
-from urllib.request import urljoin
+from unittest.mock import sentinel, ANY
 
 import odooly
 from ._common import XmlRpcTestCase, OBJ
@@ -120,6 +119,7 @@ def setUp(self):
         self.service.object.execute_kw.side_effect = self.obj_exec
         # preload 'foo.bar'
         self.env['foo.bar']
+        self.http_request.reset_mock()
         self.service.reset_mock()
 
 
@@ -1605,23 +1605,6 @@ class TestRecord18(TestRecord):
 class TestModel19(TestModel):
     server_version = '19.0'
 
-    def _patch_service(self):
-        self.auth_http = self._patch_http_request()
-        return super()._patch_service()
-
-    def test_auth_http(self):
-        headers = {
-            'Authorization': 'Bearer passwd',
-            'Content-Type': 'application/json',
-            'X-Odoo-Database': 'database',
-        }
-        test_url = urljoin(self.server, '/json/2/res.users/context_get')
-        self.assertEqual(self.auth_http.mock_calls, [call(test_url, json={}, headers=headers)])
-
 
 class TestRecord19(TestRecord):
     server_version = '19.0'
-
-    def _patch_service(self):
-        self.auth_http = self._patch_http_request()
-        return super()._patch_service()
diff --git a/tests/test_util.py b/tests/test_util.py
@@ -130,7 +130,6 @@ def test_searchargs_invalid(self):
         self.assertRaises(ValueError, searchargs, (['some_id child_off'],))
         self.assertRaises(ValueError, searchargs, (['someth like3'],))
 
-
     def test_readfmt(self):
         dummy = object.__new__(Model)
         readfmt = partial(dummy._parse_format, browse=False)
