[16.0][ADD] base_external_system_odoorpc

Author: ntsirintanis

base_external_system_odoorpc/__init__.py

@@ -0,0 +1 @@
+from . import models

base_external_system_odoorpc/__manifest__.py

@@ -0,0 +1,21 @@
+# Copyright 2026 Therp BV <https://therp.nl>
+# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
+
+{
+    "name": "Base External System Odoo-rpc",
+    "summary": "Connect to a remote Odoo instance via the odoorpc library.",
+    "version": "16.0.1.0.0",
+    "category": "Base",
+    "website": "https://github.com/OCA/server-backend",
+    "author": "Therp BV, " "Odoo Community Association (OCA)",
+    "license": "AGPL-3",
+    "application": False,
+    "installable": True,
+    "depends": ["base_external_system"],
+    "external_dependencies": {"python": ["odoorpc"]},
+    "data": [
+        "demo/external_system_odoo_demo.xml",
+        "security/ir.model.access.csv",
+        "views/external_system_views.xml",
+    ],
+}

base_external_system_odoorpc/demo/external_system_odoo_demo.xml

@@ -0,0 +1,18 @@
+<?xml version="1.0" encoding="utf-8" ?>
+<!--
+    Copyright 2026 Therp BV <https://therp.nl>
+    License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
+-->
+<odoo>
+    <record id="external_system_rpc" model="external.system">
+        <field name="name">External connection via RPC</field>
+        <field name="system_type">external.system.odoo</field>
+        <field name="host">localhost</field>
+        <field name="port">8069</field>
+        <field name="db_name">odoo16</field>
+        <field name="username">admin</field>
+        <field name="password">admin</field>
+        <field name="is_ssl">0</field>
+        <field name="company_ids" eval="[(5, 0), (4, ref('base.main_company'))]" />
+    </record>
+</odoo>

base_external_system_odoorpc/models/__init__.py

@@ -0,0 +1,2 @@
+from . import external_system
+from . import external_system_odoo

base_external_system_odoorpc/models/external_system.py

@@ -0,0 +1,19 @@
+# Copyright 2026 Therp BV <https://therp.nl>.
+# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
+
+from odoo import fields, models
+
+
+class ExternalSystem(models.Model):
+    """Extend base external system"""
+
+    _inherit = "external.system"
+
+    db_name = fields.Char(
+        string="Database",
+        help="Input database name",
+    )
+    is_ssl = fields.Boolean(
+        string="SSL",
+        help="Change protocol if SSL",
+    )

base_external_system_odoorpc/models/external_system_odoo.py

@@ -0,0 +1,75 @@
+# Copyright 2026 Therp BV <https://therp.nl>
+# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
+
+from urllib.error import URLError
+
+import odoorpc
+
+from odoo import _, models
+from odoo.exceptions import ValidationError
+
+
+class ExternalSystemOdoo(models.Model):
+    """This is an Interface implementing the RPC module."""
+
+    _name = "external.system.odoo"
+    _inherit = "external.system.adapter"
+    _description = "External System RPC"
+
+    def external_get_client(self):
+        """Return a usable client representing the remote system."""
+        self.ensure_one()
+        return self._connect()
+
+    def external_destroy_client(self, client):
+        """Cleanup the client connection"""
+        self.ensure_one()
+        try:
+            if client:
+                client.logout()
+        except Exception:
+            pass
+        return super().external_destroy_client(client)
+
+    def external_test_connection(self):
+        """Test connection in the UI."""
+        self.ensure_one()
+        try:
+            with self.client() as odoo:
+                user_model = odoo.env["res.users"]
+                ids = user_model.search([("login", "=", "admin")])
+                if not ids:
+                    raise ValidationError(
+                        _("Connected, but could not find admin user.")
+                    )
+                user_model.read([ids[0]], ["name"])[0]
+        except ValidationError:
+            raise
+        except Exception as e:
+            raise ValidationError(_("Connection failed.\n\nDETAIL: %s") % e) from e
+
+        return super().external_test_connection()
+
+    def _connect(self):
+        """Return connection object"""
+        self.ensure_one()
+        if not all([self.host, self.port, self.db_name, self.username, self.password]):
+            raise ValidationError(
+                _(
+                    "Connection failed. Please make sure that all fields "
+                    "are filled: Database, Host, Port, Username, Password."
+                )
+            )
+        try:
+            odoo = odoorpc.ODOO(
+                self.host,
+                port=self.port,
+                protocol="jsonrpc+ssl" if self.is_ssl else "jsonrpc",
+            )
+        except URLError as exc:
+            raise ValidationError(
+                _("Could not connect the Odoo server at %(host)s:%(port)s")
+                % {"host": self.host, "port": self.port}
+            ) from exc
+        odoo.login(self.db_name, self.username, self.password)
+        return odoo

base_external_system_odoorpc/security/ir.model.access.csv

@@ -0,0 +1,2 @@
+id,name,model_id:id,group_id:id,perm_read,perm_write,perm_create,perm_unlink
+access_external_system_odoo_admin,access_external_system_odoo_admin,model_external_system_odoo,base.group_system,1,1,1,1

base_external_system_odoorpc/tests/__init__.py

@@ -0,0 +1 @@
+from . import test_odoorpc_adapter

base_external_system_odoorpc/tests/test_odoorpc_adapter.py

@@ -0,0 +1,187 @@
+# Copyright 2023 Therp BV
+# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
+
+from odoo.exceptions import UserError, ValidationError
+from odoo.tests.common import TransactionCase
+
+
+class _FakeModel:
+    def __init__(self, name):
+        self._name = name
+        self.search_calls = []
+        self.read_calls = []
+
+    def search(self, domain, limit=None, order=None):
+        self.search_calls.append((domain, limit, order))
+        # Simulate "admin" found
+        return [1]
+
+    def read(self, ids, fields=None):
+        self.read_calls.append((ids, fields))
+        return [{"id": ids[0], "name": "Admin"}]
+
+
+class _FakeEnv:
+    def __init__(self):
+        self._models = {}
+
+    def __getitem__(self, model_name):
+        if model_name not in self._models:
+            self._models[model_name] = _FakeModel(model_name)
+        return self._models[model_name]
+
+
+class _FakeODOOClient:
+    def __init__(self):
+        self.logged_in = False
+        self.logged_out = False
+        self.login_calls = []
+        self.env = _FakeEnv()
+
+    def login(self, db, username, password):
+        self.login_calls.append((db, username, password))
+        # simulate successful login
+        self.logged_in = True
+
+    def logout(self):
+        self.logged_out = True
+
+
+class _FakeOdooRPCModule:
+    """Fake `odoorpc` module replacement with an ODOO constructor."""
+
+    def __init__(self):
+        self.created = []
+        self.last_client = None
+
+    def ODOO(self, host, port=None, protocol=None):
+        self.created.append((host, port, protocol))
+        self.last_client = _FakeODOOClient()
+        return self.last_client
+
+
+class TestBaseExternalSystemOdooRPC(TransactionCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.env = cls.env(su=True)
+
+    def _make_system(self, **overrides):
+        vals = {
+            "name": "Roetz 8",
+            "system_type": "external.system.odoo",
+            "host": "roetz-test.therp1.nl",
+            "port": 443,
+            "db_name": "roetz-testdatabase",
+            "username": "admin",
+            "password": "admin",
+            "is_ssl": True,
+            "company_ids": [(6, 0, [self.env.company.id])],
+        }
+        vals.update(overrides)
+        return self.env["external.system"].create(vals)
+
+    def _patch_odoorpc(self):
+        """
+        Patch the `odoorpc` module reference imported inside the adapter file.
+
+        Adapter file imports odoorpc at module import time, so we override that
+        module-level name with our fake.
+        """
+        import odoo.addons.base_external_system_odoorpc.models.external_system_odoo as mod
+
+        fake = _FakeOdooRPCModule()
+        mod.odoorpc = fake
+        return fake
+
+    def test_interface_is_created_and_points_to_adapter(self):
+        sys = self._make_system()
+        self.assertTrue(sys.interface, "System should have an auto-created interface")
+        self.assertEqual(
+            sys.interface._name,
+            "external.system.odoo",
+            "Interface should be the OdooRPC adapter model",
+        )
+        self.assertEqual(
+            sys.interface.system_id.id,
+            sys.id,
+            "Interface should point back to the external.system via system_id",
+        )
+
+    def test_external_get_client_success_ssl(self):
+        sys = self._make_system(is_ssl=True)
+        fake = self._patch_odoorpc()
+
+        client = sys.interface.external_get_client()
+        # constructor called with jsonrpc+ssl
+        self.assertEqual(fake.created[-1], (sys.host, sys.port, "jsonrpc+ssl"))
+        # login called
+        self.assertTrue(client.logged_in)
+        self.assertEqual(
+            client.login_calls[-1],
+            (sys.db_name, sys.username, sys.password),
+        )
+
+    def test_external_get_client_success_no_ssl(self):
+        sys = self._make_system(is_ssl=False)
+        fake = self._patch_odoorpc()
+
+        client = sys.interface.external_get_client()
+        self.assertEqual(fake.created[-1], (sys.host, sys.port, "jsonrpc"))
+        self.assertTrue(client.logged_in)
+
+    def test_external_get_client_missing_params_raises_validationerror(self):
+        sys = self._make_system(db_name=False)
+        self._patch_odoorpc()
+
+        with self.assertRaises(ValidationError):
+            sys.interface.external_get_client()
+
+    def test_client_context_manager_logs_out(self):
+        sys = self._make_system()
+        fake = self._patch_odoorpc()
+
+        with sys.client() as client:
+            self.assertTrue(client.logged_in)
+            self.assertFalse(client.logged_out)
+
+        # after context, destroy_client should have logged out
+        self.assertTrue(fake.last_client.logged_out)
+
+    def test_interface_connect_helper_calls_external_get_client(self):
+        """
+        Your adapter keeps a legacy `_connect()` helper.
+        This ensures it returns the same object and logs in.
+        """
+        sys = self._make_system()
+        fake = self._patch_odoorpc()
+
+        client = sys.interface._connect()
+        self.assertIs(client, fake.last_client)
+        self.assertTrue(client.logged_in)
+
+    def test_external_test_connection_raises_success_usererror(self):
+        """
+        base_external_system's default external_test_connection raises UserError
+        with a success message. Our adapter should:
+          - open a client
+          - call res.users search/read probe
+          - then call super() which raises UserError
+          - and logout on exit
+        """
+        sys = self._make_system()
+        fake = self._patch_odoorpc()
+
+        with self.assertRaises(UserError):
+            sys.interface.external_test_connection()
+
+        self.assertIsNotNone(fake.last_client)
+
+        users = fake.last_client.env["res.users"]
+        self.assertEqual(len(users.search_calls), 1)
+        self.assertEqual(users.search_calls[0][0], [("login", "=", "admin")])
+        self.assertEqual(len(users.read_calls), 1)
+        self.assertEqual(users.read_calls[0][1], ["name"])
+
+        # And should have logged out due to adapter.client() finally block
+        self.assertTrue(fake.last_client.logged_out)

base_external_system_odoorpc/views/external_system_views.xml

@@ -0,0 +1,32 @@
+<?xml version="1.0" encoding="utf-8" ?>
+<odoo>
+
+    <record id="external_system_view_form_odoo" model="ir.ui.view">
+        <field name="name">external.system.form.odoo</field>
+        <field name="model">external.system</field>
+        <field name="inherit_id" ref="base_external_system.external_system_view_form" />
+        <field name="arch" type="xml">
+
+            <!-- Show Odoo-specific fields -->
+            <xpath expr="//field[@name='remote_path']" position="after">
+                <field
+                    name="db_name"
+                    attrs="{'invisible': [('system_type', '!=', 'external.system.odoo')]}"
+                />
+                <field
+                    name="is_ssl"
+                    attrs="{'invisible': [('system_type', '!=', 'external.system.odoo')]}"
+                />
+            </xpath>
+
+            <!-- Hide remote_path for Odoo RPC -->
+            <xpath expr="//field[@name='remote_path']" position="attributes">
+                <attribute name="attrs">
+                    {'invisible': [('system_type', '=', 'external.system.odoo')]}
+                </attribute>
+            </xpath>
+
+        </field>
+    </record>
+
+</odoo>