test(spp_import_match): increase test coverage to 90%+ with 19 new tests

emjay0921 · emjay0921 · commit eaeb6cd03404 · 2026-03-11T10:22:10.000+08:00
Add test_base_load.py (8 tests) covering Base.load() override paths:
no usable rules, overwrite true/false, no match creates, mixed records,
import_match_ids context filtering, id field append, match counts tracking.

Add test_base_import_methods.py (8 tests) covering SPPBaseImport helpers:
CSV attachment roundtrip, custom options, extract chunks, import_one_chunk
success/error, ImportValidationError, execute_import with match_ids context.

Add 3 tests to test_import_match_model.py: onchange_field_id skips
relational fields, match_find sub_field matching, match_find multiple
rule combinations.
diff --git a/spp_import_match/tests/__init__.py b/spp_import_match/tests/__init__.py
@@ -2,3 +2,5 @@
 from . import test_import_match_model
 from . import test_base_write
 from . import test_queue_job
+from . import test_base_load
+from . import test_base_import_methods
diff --git a/spp_import_match/tests/test_base_import_methods.py b/spp_import_match/tests/test_base_import_methods.py
@@ -0,0 +1,163 @@
+# Part of OpenSPP. See LICENSE file for full copyright and licensing details.
+
+from unittest.mock import patch
+
+from odoo.tests import TransactionCase, tagged
+
+from odoo.addons.queue_job.exception import FailedJobError
+
+from ..models.base_import import ImportValidationError
+
+OPTIONS = {
+    "import_skip_records": [],
+    "import_set_empty_fields": [],
+    "fallback_values": {},
+    "name_create_enabled_fields": {},
+    "encoding": "utf-8",
+    "separator": ",",
+    "quoting": '"',
+    "date_format": "",
+    "datetime_format": "",
+    "float_thousand_separator": ",",
+    "float_decimal_separator": ".",
+    "advanced": True,
+    "has_headers": True,
+    "keep_matches": False,
+    "limit": 2000,
+    "sheets": [],
+    "sheet": "",
+    "skip": 0,
+    "tracking_disable": True,
+}
+
+
+@tagged("post_install", "-at_install")
+class TestBaseImportMethods(TransactionCase):
+    """Test SPPBaseImport helper methods."""
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.csv_options = {
+            "separator": ",",
+            "quoting": '"',
+            "encoding": "utf-8",
+        }
+        cls.import_record = cls.env["base_import.import"].create(
+            {
+                "res_model": "res.partner",
+                "file_name": "test.csv",
+            }
+        )
+
+    def test_csv_attachment_roundtrip(self):
+        """_create_csv_attachment + _read_csv_attachment round-trip."""
+        fields_in = ["name", "email"]
+        data_in = [["Alice", "alice@test.com"], ["Bob", "bob@test.com"]]
+        attachment = self.import_record._create_csv_attachment(fields_in, data_in, self.csv_options, "roundtrip.csv")
+        self.assertTrue(attachment.id)
+        self.assertEqual(attachment.name, "roundtrip.csv")
+        fields_out, data_out = self.import_record._read_csv_attachment(attachment, self.csv_options)
+        self.assertEqual(fields_out, fields_in)
+        self.assertEqual(len(data_out), 2)
+        self.assertEqual(data_out[0], ["Alice", "alice@test.com"])
+        self.assertEqual(data_out[1], ["Bob", "bob@test.com"])
+
+    def test_csv_attachment_custom_options(self):
+        """_create/_read_csv_attachment with custom separator/quoting/encoding."""
+        custom_options = {
+            "separator": ";",
+            "quoting": "'",
+            "encoding": "latin-1",
+        }
+        fields_in = ["name", "email"]
+        data_in = [["TestAccent", "accent@test.com"]]
+        attachment = self.import_record._create_csv_attachment(fields_in, data_in, custom_options, "custom.csv")
+        fields_out, data_out = self.import_record._read_csv_attachment(attachment, custom_options)
+        self.assertEqual(fields_out, fields_in)
+        self.assertEqual(data_out[0][0], "TestAccent")
+
+    def test_extract_chunks_basic(self):
+        """_extract_chunks splits data into chunks >= chunk_size."""
+        model_obj = self.env["res.partner"]
+        fields = ["name", "email"]
+        data = [[f"P{i}", f"p{i}@test.com"] for i in range(7)]
+        chunks = list(self.env["base_import.import"]._extract_chunks(model_obj, fields, data, 3))
+        self.assertTrue(len(chunks) > 1)
+        # All rows should be covered
+        all_rows = set()
+        for start, end in chunks:
+            for r in range(start, end + 1):
+                all_rows.add(r)
+        self.assertEqual(all_rows, set(range(7)))
+
+    def test_extract_chunks_small_data(self):
+        """_extract_chunks with data smaller than chunk_size yields one chunk."""
+        model_obj = self.env["res.partner"]
+        fields = ["name", "email"]
+        data = [["A", "a@test.com"], ["B", "b@test.com"]]
+        chunks = list(self.env["base_import.import"]._extract_chunks(model_obj, fields, data, 100))
+        self.assertEqual(len(chunks), 1)
+        self.assertEqual(chunks[0], (0, 1))
+
+    def test_import_one_chunk_success(self):
+        """_import_one_chunk loads data successfully."""
+        attachment = self.import_record._create_csv_attachment(
+            ["name"], [["ChunkSuccess99xyz"]], self.csv_options, "chunk.csv"
+        )
+        result = self.import_record._import_one_chunk("res.partner", attachment, self.csv_options, {})
+        errors = [m for m in result["messages"] if m.get("type") == "error"]
+        self.assertFalse(errors)
+        partner = self.env["res.partner"].search([("name", "=", "ChunkSuccess99xyz")])
+        self.assertEqual(len(partner), 1)
+
+    def test_import_one_chunk_error(self):
+        """_import_one_chunk raises FailedJobError on load errors."""
+        attachment = self.import_record._create_csv_attachment(["name"], [["ErrTest"]], self.csv_options, "err.csv")
+        error_result = {
+            "messages": [{"type": "error", "message": "Test error"}],
+            "ids": [],
+        }
+        with patch.object(type(self.env["res.partner"]), "load", return_value=error_result):
+            with self.assertRaises(FailedJobError):
+                self.import_record._import_one_chunk("res.partner", attachment, self.csv_options, {})
+
+    def test_import_validation_error_attributes(self):
+        """ImportValidationError stores field, type, and message attrs."""
+        err = ImportValidationError("Bad value", field="name", error_type="warning", field_type="char")
+        self.assertEqual(str(err), "Bad value")
+        self.assertEqual(err.message, "Bad value")
+        self.assertEqual(err.type, "warning")
+        self.assertEqual(err.field_path, ["name"])
+        self.assertEqual(err.field_type, "char")
+        self.assertFalse(err.record)
+        self.assertTrue(err.not_matching_error)
+        # Test defaults
+        err2 = ImportValidationError("Default error")
+        self.assertEqual(err2.type, "error")
+        self.assertFalse(err2.field_path)
+        self.assertIsNone(err2.field_type)
+
+    def test_execute_import_with_match_ids_passes_context(self):
+        """execute_import passes import_match_ids and overwrite_match to context."""
+        res_partner_model = self.env["ir.model"].search([("model", "=", "res.partner")])
+        name_field = self.env["ir.model.fields"].search(
+            [("name", "=", "name"), ("model_id", "=", res_partner_model.id)]
+        )
+        self.env["res.partner"].create({"name": "ExecMatchTest99xyz", "email": "exec@test.com"})
+        match = self.env["spp.import.match"].create({"model_id": res_partner_model.id, "overwrite_match": True})
+        self.env["spp.import.match.fields"].create({"field_id": name_field.id, "match_id": match.id})
+        import_rec = self.env["base_import.import"].create(
+            {
+                "res_model": "res.partner",
+                "file": b"name,email\nExecMatchTest99xyz,updated@test.com",
+                "file_name": "test_exec.csv",
+                "file_type": "csv",
+            }
+        )
+        options = dict(OPTIONS)
+        options["import_match_ids"] = [match.id]
+        options["overwrite_match"] = True
+        result = import_rec.execute_import(["name", "email"], [], options, dryrun=True)
+        self.assertIn("import_match_counts", result)
+        self.assertEqual(result["import_match_counts"]["overwritten"], 1)
diff --git a/spp_import_match/tests/test_base_load.py b/spp_import_match/tests/test_base_load.py
@@ -0,0 +1,144 @@
+# Part of OpenSPP. See LICENSE file for full copyright and licensing details.
+
+from odoo.tests import TransactionCase, tagged
+
+from ..models.base import _import_match_local
+
+
+@tagged("post_install", "-at_install")
+class TestBaseLoad(TransactionCase):
+    """Test the Base.load() override with import matching."""
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.res_partner_model = cls.env["ir.model"].search([("model", "=", "res.partner")])
+        cls.name_field = cls.env["ir.model.fields"].search(
+            [("name", "=", "name"), ("model_id", "=", cls.res_partner_model.id)]
+        )
+        cls.email_field = cls.env["ir.model.fields"].search(
+            [("name", "=", "email"), ("model_id", "=", cls.res_partner_model.id)]
+        )
+
+    def _create_match_rule(self, field_ids_data, overwrite=True):
+        """Helper to create a match rule with fields."""
+        match = self.env["spp.import.match"].create(
+            {
+                "model_id": self.res_partner_model.id,
+                "overwrite_match": overwrite,
+            }
+        )
+        for data in field_ids_data:
+            data["match_id"] = match.id
+            self.env["spp.import.match.fields"].create(data)
+        return match
+
+    def test_load_no_usable_rules(self):
+        """When no usable rules exist, load() passes through to super()."""
+        result = self.env["res.partner"].load(
+            ["name", "email"],
+            [["NoRuleTest99xyz", "norule@test.com"]],
+        )
+        self.assertFalse(result["messages"])
+        partner = self.env["res.partner"].search([("name", "=", "NoRuleTest99xyz")])
+        self.assertEqual(len(partner), 1)
+
+    def test_load_match_overwrite_true(self):
+        """Match found + overwrite=True -> record updated."""
+        partner = self.env["res.partner"].create({"name": "OverwriteTrue99xyz", "email": "old@test.com"})
+        self._create_match_rule([{"field_id": self.name_field.id}], overwrite=True)
+        result = self.env["res.partner"].load(
+            ["name", "email"],
+            [["OverwriteTrue99xyz", "new@test.com"]],
+        )
+        self.assertFalse(result["messages"])
+        partner.invalidate_recordset()
+        self.assertEqual(partner.email, "new@test.com")
+
+    def test_load_match_overwrite_false(self):
+        """Match found + overwrite=False -> record skipped."""
+        partner = self.env["res.partner"].create({"name": "OverwriteFalse99xyz", "email": "original@test.com"})
+        self._create_match_rule([{"field_id": self.name_field.id}], overwrite=False)
+        self.env["res.partner"].load(
+            ["name", "email"],
+            [["OverwriteFalse99xyz", "changed@test.com"]],
+        )
+        partner.invalidate_recordset()
+        self.assertEqual(partner.email, "original@test.com")
+
+    def test_load_no_match_creates_record(self):
+        """No match found -> new record created."""
+        self._create_match_rule([{"field_id": self.name_field.id}])
+        result = self.env["res.partner"].load(
+            ["name", "email"],
+            [["BrandNewPartner99xyz", "brandnew@test.com"]],
+        )
+        self.assertFalse(result["messages"])
+        partner = self.env["res.partner"].search([("name", "=", "BrandNewPartner99xyz")])
+        self.assertEqual(len(partner), 1)
+        self.assertEqual(partner.email, "brandnew@test.com")
+
+    def test_load_multiple_records_mixed(self):
+        """Mix of matched and new records in one import."""
+        self.env["res.partner"].create({"name": "ExistingMixed99xyz", "email": "existing@test.com"})
+        self._create_match_rule([{"field_id": self.name_field.id}], overwrite=True)
+        result = self.env["res.partner"].load(
+            ["name", "email"],
+            [
+                ["ExistingMixed99xyz", "updated@test.com"],
+                ["NewMixed99xyz", "newmixed@test.com"],
+            ],
+        )
+        self.assertFalse(result["messages"])
+        existing = self.env["res.partner"].search([("name", "=", "ExistingMixed99xyz")])
+        self.assertEqual(len(existing), 1)
+        existing.invalidate_recordset()
+        self.assertEqual(existing.email, "updated@test.com")
+        new_partner = self.env["res.partner"].search([("name", "=", "NewMixed99xyz")])
+        self.assertEqual(len(new_partner), 1)
+
+    def test_load_with_import_match_ids_context(self):
+        """Context import_match_ids filters to specific rules."""
+        self._create_match_rule([{"field_id": self.name_field.id}])
+        match2 = self._create_match_rule([{"field_id": self.email_field.id}])
+        partner = self.env["res.partner"].create({"name": "CtxFilter99xyz", "email": "ctxfilter@test.com"})
+        # Only use match2 (email rule); import with matching email but different name
+        result = (
+            self.env["res.partner"]
+            .with_context(import_match_ids=[match2.id])
+            .load(
+                ["name", "email"],
+                [["CtxFilterRenamed99", "ctxfilter@test.com"]],
+            )
+        )
+        self.assertFalse(result["messages"])
+        partner.invalidate_recordset()
+        self.assertEqual(partner.name, "CtxFilterRenamed99")
+
+    def test_load_appends_id_field(self):
+        """Auto-appends 'id' when not in fields list."""
+        self._create_match_rule([{"field_id": self.name_field.id}])
+        fields = ["name", "email"]
+        self.env["res.partner"].load(
+            fields,
+            [["AppendIdTest99xyz", "appendid@test.com"]],
+        )
+        self.assertIn("id", fields)
+
+    def test_load_match_counts_tracking(self):
+        """_import_match_local.counts populated when import_match_ids in context."""
+        self.env["res.partner"].create({"name": "CountsExisting99xyz", "email": "counts@test.com"})
+        match = self._create_match_rule([{"field_id": self.name_field.id}], overwrite=True)
+        _import_match_local.counts = None
+        self.env["res.partner"].with_context(import_match_ids=[match.id]).load(
+            ["name", "email"],
+            [
+                ["CountsExisting99xyz", "updated@test.com"],
+                ["CountsNew99xyz", "new@test.com"],
+            ],
+        )
+        counts = _import_match_local.counts
+        self.assertIsNotNone(counts)
+        self.assertEqual(counts["overwritten"], 1)
+        self.assertEqual(counts["created"], 1)
+        self.assertEqual(counts["skipped"], 0)
diff --git a/spp_import_match/tests/test_import_match_model.py b/spp_import_match/tests/test_import_match_model.py
@@ -162,3 +162,68 @@ def test_field_compute_name_with_sub_field(self):
                 ]
             )
             self.assertEqual(match.field_ids[0].name, "child_ids/name")
+
+    def test_onchange_field_id_skips_relational(self):
+        """_onchange_field_id skips duplicate check for relational fields."""
+        parent_id_field = self.env["ir.model.fields"].search(
+            [("name", "=", "parent_id"), ("model_id", "=", self.res_partner_model.id)],
+            limit=1,
+        )
+        match = self._create_match_rule(
+            [
+                {"field_id": parent_id_field.id},
+                {"field_id": parent_id_field.id},
+            ]
+        )
+        # Should NOT raise ValidationError because parent_id is many2one
+        for field_rec in match.field_ids:
+            field_rec._onchange_field_id()
+        self.assertEqual(len(match.field_ids), 2)
+
+    def test_match_find_sub_field(self):
+        """_match_find handles sub_field matching for relational fields."""
+        child = self.env["res.partner"].create({"name": "SubFieldChild_Uniq99xyz"})
+        parent = self.env["res.partner"].create(
+            {
+                "name": "SubFieldParent_Uniq99xyz",
+                "child_ids": [(4, child.id)],
+            }
+        )
+        child_ids_field = self.env["ir.model.fields"].search(
+            [
+                ("name", "=", "child_ids"),
+                ("model_id", "=", self.res_partner_model.id),
+            ],
+            limit=1,
+        )
+        match = self._create_match_rule(
+            [
+                {
+                    "field_id": child_ids_field.id,
+                    "sub_field_id": self.name_field.id,
+                }
+            ]
+        )
+        result = match._match_find(
+            self.env["res.partner"],
+            {"child_ids": [(0, 0, {"name": "SubFieldChild_Uniq99xyz"})]},
+            {"child_ids/name": "SubFieldChild_Uniq99xyz", "id": None},
+        )
+        self.assertEqual(result, parent)
+
+    def test_match_find_multiple_combinations(self):
+        """_match_find iterates rules; first matching single result wins."""
+        partner = self.env["res.partner"].create({"name": "MultiCombo99xyz", "email": "multicombo@test.com"})
+        # First rule by email (lower sequence -> tried first)
+        match1 = self._create_match_rule([{"field_id": self.email_field.id}])
+        match1.sequence = 1
+        # Second rule by name
+        match2 = self._create_match_rule([{"field_id": self.name_field.id}])
+        match2.sequence = 2
+        # Email won't match, but name will
+        result = self.env["spp.import.match"]._match_find(
+            self.env["res.partner"],
+            {"name": "MultiCombo99xyz", "email": "nomatch@test.com"},
+            {"name": "MultiCombo99xyz", "email": "nomatch@test.com", "id": None},
+        )
+        self.assertEqual(result, partner)
