[19.0][IMP] spreadsheet_oca: improved extensibility + basic tests

spreadsheet_oca/static/src/spreadsheet/bundle/spreadsheet_renderer.esm.js

@@ -117,6 +117,11 @@ export class SpreadsheetRenderer extends Component {
             });
         };
     }
+    getExtraModelCustom() {
+        // Extension point: patch this to inject extra entries into the Model's
+        // `custom` config, which o-spreadsheet passes to every plugin constructor.
+        return {};
+    }
     setup() {
         this.orm = useService("orm");
         this.http = useService("http");
@@ -140,7 +145,12 @@ export class SpreadsheetRenderer extends Component {
         this.spreadsheet_model = new Model(
             load(this.props.record.spreadsheet_raw),
             {
-                custom: {env: this.env, orm: this.orm, odooDataProvider},
+                custom: {
+                    env: this.env,
+                    orm: this.orm,
+                    odooDataProvider,
+                    ...this.getExtraModelCustom(),
+                },
                 defaultCurrency: this.createDefaultCurrency(defaultCurrency),
                 external: {
                     loadCurrencies: this.loadCurrencies,

spreadsheet_oca/tests/__init__.py

@@ -0,0 +1,5 @@
+# Copyright 2026 arielbarreiros96
+# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
+
+from . import test_spreadsheet
+from . import test_import_wizard

spreadsheet_oca/tests/test_import_wizard.py

@@ -0,0 +1,67 @@
+# Copyright 2026 arielbarreiros96
+# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
+
+from odoo.tests.common import TransactionCase
+
+
+class TestImportWizard(TransactionCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.Wizard = cls.env["spreadsheet.spreadsheet.import"]
+        cls.mode_new = cls.env.ref("spreadsheet_oca.spreadsheet_import_mode_new")
+        cls.mode_add = cls.env.ref("spreadsheet_oca.spreadsheet_import_mode_add")
+
+    def test_default_mode_is_set(self):
+        wizard = self.Wizard.create({"name": "X"})
+        self.assertTrue(wizard.mode_id)
+
+    def test_insert_pivot_new_creates_spreadsheet(self):
+        wizard = self.Wizard.create(
+            {
+                "name": "New Sheet",
+                "datasource_name": "My Source",
+                "mode_id": self.mode_new.id,
+                "import_data": {"foo": "bar"},
+            }
+        )
+        action = wizard.insert_pivot()
+        self.assertEqual(action["tag"], "action_spreadsheet_oca")
+        spreadsheet = self.env["spreadsheet.spreadsheet"].browse(
+            action["params"]["spreadsheet_id"]
+        )
+        self.assertEqual(spreadsheet.name, "New Sheet")
+        self.assertEqual(action["params"]["import_data"]["name"], "My Source")
+        self.assertEqual(action["params"]["import_data"]["new"], 1)
+
+    def test_insert_pivot_new_dynamic_rows(self):
+        wizard = self.Wizard.create(
+            {
+                "name": "Dyn Sheet",
+                "datasource_name": "Src",
+                "mode_id": self.mode_new.id,
+                "import_data": {},
+                "dynamic": True,
+                "number_of_rows": 7,
+            }
+        )
+        action = wizard.insert_pivot()
+        self.assertEqual(action["params"]["import_data"]["dyn_number_of_rows"], 7)
+
+    def test_insert_pivot_add_targets_existing(self):
+        spreadsheet = self.env["spreadsheet.spreadsheet"].create({"name": "Existing"})
+        wizard = self.Wizard.create(
+            {
+                "name": "Add",
+                "datasource_name": "Src2",
+                "mode_id": self.mode_add.id,
+                "import_data": {"x": 1},
+                "spreadsheet_id": spreadsheet.id,
+                "dynamic": True,
+                "number_of_rows": 3,
+            }
+        )
+        action = wizard.insert_pivot()
+        self.assertEqual(action["params"]["spreadsheet_id"], spreadsheet.id)
+        self.assertEqual(action["params"]["import_data"]["name"], "Src2")
+        self.assertEqual(action["params"]["import_data"]["dyn_number_of_rows"], 3)

spreadsheet_oca/tests/test_spreadsheet.py

@@ -0,0 +1,217 @@
+# Copyright 2026 arielbarreiros96
+# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
+
+import base64
+import json
+import zipfile
+from io import BytesIO
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from odoo.exceptions import AccessDenied, AccessError
+from odoo.tests.common import TransactionCase, new_test_user
+
+WS_MODULE = "odoo.addons.bus.models.ir_websocket"
+
+
+class TestSpreadsheet(TransactionCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.Spreadsheet = cls.env["spreadsheet.spreadsheet"]
+        cls.spreadsheet = cls.Spreadsheet.create({"name": "Test Spreadsheet"})
+
+    def _message(self, message_type, **extra):
+        return dict(
+            {
+                "type": message_type,
+                "clientId": "client-1",
+                "nextRevisionId": "next-1",
+                "serverRevisionId": "server-1",
+            },
+            **extra,
+        )
+
+    def test_new_spreadsheet_has_empty_workbook(self):
+        raw = self.spreadsheet.spreadsheet_raw
+        self.assertEqual(raw["version"], 1)
+        self.assertEqual(raw["sheets"][0]["id"], "sheet1")
+        self.assertIn("revisionId", raw)
+
+    def test_spreadsheet_raw_roundtrip(self):
+        data = {"version": 1, "sheets": [], "revisionId": "R1"}
+        self.spreadsheet.spreadsheet_raw = data
+        self.assertEqual(self.spreadsheet.spreadsheet_raw, data)
+        decoded = json.loads(
+            base64.decodebytes(self.spreadsheet.spreadsheet_binary_data).decode("UTF-8")
+        )
+        self.assertEqual(decoded, data)
+
+    def test_open_spreadsheet_action(self):
+        action = self.spreadsheet.open_spreadsheet()
+        self.assertEqual(action["type"], "ir.actions.client")
+        self.assertEqual(action["tag"], "action_spreadsheet_oca")
+        self.assertEqual(action["params"]["spreadsheet_id"], self.spreadsheet.id)
+        self.assertEqual(action["params"]["model"], "spreadsheet.spreadsheet")
+
+    def test_get_spreadsheet_data(self):
+        data = self.spreadsheet.get_spreadsheet_data()
+        self.assertEqual(data["name"], self.spreadsheet.name)
+        self.assertEqual(data["mode"], "normal")
+        self.assertEqual(data["revisions"], [])
+        self.assertIn("default_currency", data)
+        self.assertIn("user_locale", data)
+
+    def test_send_revision_message_creates_revision(self):
+        result = self.spreadsheet.send_spreadsheet_message(
+            self._message("REMOTE_REVISION")
+        )
+        self.assertTrue(result)
+        revision = self.spreadsheet.spreadsheet_revision_ids
+        self.assertEqual(len(revision), 1)
+        self.assertEqual(revision.next_revision_id, "next-1")
+        self.assertEqual(revision.server_revision_id, "server-1")
+        commands = json.loads(revision.commands)
+        self.assertNotIn("serverRevisionId", commands)
+        self.assertNotIn("nextRevisionId", commands)
+        self.assertNotIn("clientId", commands)
+        self.assertEqual(commands["type"], "REMOTE_REVISION")
+
+    def test_send_snapshot_message_creates_revision(self):
+        result = self.spreadsheet.send_spreadsheet_message(self._message("SNAPSHOT"))
+        self.assertTrue(result)
+        self.assertEqual(len(self.spreadsheet.spreadsheet_revision_ids), 1)
+
+    def test_send_presence_message_creates_no_revision(self):
+        result = self.spreadsheet.send_spreadsheet_message(
+            self._message("CLIENT_MOVED")
+        )
+        self.assertTrue(result)
+        self.assertFalse(self.spreadsheet.spreadsheet_revision_ids)
+
+    def test_send_unknown_message_returns_false(self):
+        self.assertFalse(
+            self.spreadsheet.send_spreadsheet_message(self._message("SOMETHING"))
+        )
+
+    def test_send_revision_denied_without_write(self):
+        user = new_test_user(
+            self.env, login="ssheet_ro", groups="spreadsheet_oca.group_user"
+        )
+        self.spreadsheet.reader_ids = user
+        with self.assertRaises(AccessError):
+            self.spreadsheet.with_user(user).send_spreadsheet_message(
+                self._message("REMOTE_REVISION")
+            )
+
+    def test_writing_raw_clears_revisions(self):
+        self.spreadsheet.send_spreadsheet_message(self._message("SNAPSHOT"))
+        self.assertTrue(self.spreadsheet.spreadsheet_revision_ids)
+        self.spreadsheet.spreadsheet_raw = {
+            "version": 1,
+            "sheets": [],
+            "revisionId": "X",
+        }
+        self.assertFalse(self.spreadsheet.spreadsheet_revision_ids)
+
+    def test_get_spreadsheet_data_returns_stored_revisions(self):
+        self.spreadsheet.send_spreadsheet_message(self._message("REMOTE_REVISION"))
+        data = self.spreadsheet.get_spreadsheet_data()
+        self.assertEqual(len(data["revisions"]), 1)
+        self.assertEqual(data["revisions"][0]["nextRevisionId"], "next-1")
+        self.assertEqual(data["revisions"][0]["serverRevisionId"], "server-1")
+
+    def test_compute_filename(self):
+        self.spreadsheet.name = "My Sheet"
+        self.assertEqual(self.spreadsheet.filename, "My Sheet.json")
+
+    def test_tag_default_color(self):
+        tag = self.env["spreadsheet.spreadsheet.tag"].create({"name": "A Tag"})
+        self.assertTrue(1 <= tag.color <= 11)
+
+    def _xlsx_attachment(self, name="Book.xlsx"):
+        buffer = BytesIO()
+        with zipfile.ZipFile(buffer, "w") as archive:
+            archive.writestr("xl/worksheets/sheet1.xml", "<worksheet/>")
+            archive.writestr("xl/_rels/workbook.xml.rels", "<Relationships/>")
+        return self.env["ir.attachment"].create(
+            {"name": name, "datas": base64.b64encode(buffer.getvalue())}
+        )
+
+    def test_create_document_from_attachment_single(self):
+        attachment = self._xlsx_attachment()
+        action = self.Spreadsheet.create_document_from_attachment(attachment.ids)
+        self.assertEqual(action["res_model"], "spreadsheet.spreadsheet")
+        self.assertTrue(action.get("res_id"))
+        self.assertFalse(attachment.exists())
+
+    def test_create_document_from_attachment_multiple(self):
+        attachments = self._xlsx_attachment("A.xlsx") | self._xlsx_attachment("B.xlsx")
+        action = self.Spreadsheet.create_document_from_attachment(attachments.ids)
+        self.assertEqual(action["res_model"], "spreadsheet.spreadsheet")
+        self.assertEqual(action["domain"][0][0], "id")
+
+    def test_has_parent_relation(self):
+        IrModel = self.env["ir.model"]
+        self.assertTrue(IrModel.has_parent_relation("res.partner"))
+        self.assertFalse(IrModel.has_parent_relation("spreadsheet.spreadsheet"))
+        self.assertFalse(IrModel.has_parent_relation("no.such.model"))
+
+    def _build_channels(self, channels, user=None):
+        websocket = self.env["ir.websocket"]
+        if user is not None:
+            websocket = websocket.with_user(user)
+        fake_request = SimpleNamespace(session=SimpleNamespace(uid=websocket.env.uid))
+        with (
+            patch(WS_MODULE + ".request", fake_request),
+            patch(WS_MODULE + ".wsrequest", fake_request),
+        ):
+            result = websocket._build_bus_channel_list(list(channels))
+        # Keep only spreadsheet_oca access tuples: the base method also returns
+        # recordset-bearing channels, and comparing those in membership checks
+        # triggers recordset.__eq__ warnings that fail CI.
+        return [
+            channel
+            for channel in result
+            if isinstance(channel, tuple)
+            and channel
+            and isinstance(channel[-1], str)
+            and channel[-1] == "spreadsheet_oca"
+        ]
+
+    def _access_tuple(self, res_id):
+        return (
+            self.env.registry.db_name,
+            "spreadsheet.spreadsheet",
+            res_id,
+            "spreadsheet_oca",
+        )
+
+    def test_websocket_channel_grants_access(self):
+        channel = f"spreadsheet_oca;spreadsheet.spreadsheet;{self.spreadsheet.id}"
+        result = self._build_channels([channel])
+        self.assertIn(self._access_tuple(self.spreadsheet.id), result)
+
+    def test_websocket_ignores_unrelated_channels(self):
+        self.assertEqual(self._build_channels(["some_channel"]), [])
+
+    def test_websocket_skips_missing_document(self):
+        channel = "spreadsheet_oca;spreadsheet.spreadsheet;999999999"
+        result = self._build_channels([channel])
+        self.assertNotIn(self._access_tuple(999999999), result)
+
+    def test_websocket_skips_without_model_access(self):
+        user = new_test_user(self.env, login="ssheet_noacc", groups="base.group_user")
+        channel = f"spreadsheet_oca;spreadsheet.spreadsheet;{self.spreadsheet.id}"
+        result = self._build_channels([channel], user=user)
+        self.assertNotIn(self._access_tuple(self.spreadsheet.id), result)
+
+    def test_websocket_denies_non_internal_user(self):
+        portal = new_test_user(
+            self.env, login="ssheet_portal", groups="base.group_portal"
+        )
+        channel = f"spreadsheet_oca;spreadsheet.spreadsheet;{self.spreadsheet.id}"
+        with self.assertRaises(AccessDenied):
+            self.env["ir.websocket"].with_user(portal)._build_bus_channel_list(
+                [channel]
+            )