diff --git a/spreadsheet_oca/static/src/spreadsheet/bundle/spreadsheet_renderer.esm.js b/spreadsheet_oca/static/src/spreadsheet/bundle/spreadsheet_renderer.esm.js index 55bbd414..624277de 100644 --- a/spreadsheet_oca/static/src/spreadsheet/bundle/spreadsheet_renderer.esm.js +++ b/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, diff --git a/spreadsheet_oca/tests/__init__.py b/spreadsheet_oca/tests/__init__.py new file mode 100644 index 00000000..832bbee8 --- /dev/null +++ b/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 diff --git a/spreadsheet_oca/tests/test_import_wizard.py b/spreadsheet_oca/tests/test_import_wizard.py new file mode 100644 index 00000000..f2e8e64e --- /dev/null +++ b/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) diff --git a/spreadsheet_oca/tests/test_spreadsheet.py b/spreadsheet_oca/tests/test_spreadsheet.py new file mode 100644 index 00000000..c05d921b --- /dev/null +++ b/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", "") + archive.writestr("xl/_rels/workbook.xml.rels", "") + 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] + )