Merge pull request #82 from OpenSPP/fix/disambiguate-vocab-by-extra-domain

fix(spp_vocabulary): support extra domain for vocabulary_code search of the same display name

Author: gonzalesedwin1123

spp_vocabulary/__manifest__.py

@@ -2,7 +2,7 @@
 {
     "name": "OpenSPP: Vocabulary",
     "category": "OpenSPP",
-    "version": "19.0.2.0.0",
+    "version": "19.0.2.0.1",
     "sequence": 1,
     "author": "OpenSPP.org",
     "website": "https://github.com/OpenSPP/OpenSPP2",

spp_vocabulary/models/__init__.py

@@ -2,6 +2,7 @@
 from . import vocabulary_code
 from . import vocabulary_mapping
 from . import concept_group
+from . import ir_fields_converter
 
 # Temporarily disabled to break circular dependency with spp_registry
 # from . import relationship

spp_vocabulary/models/ir_fields_converter.py

@@ -0,0 +1,48 @@
+import logging
+
+from odoo import api, models
+from odoo.tools.safe_eval import safe_eval
+
+_logger = logging.getLogger(__name__)
+
+
+class IrFieldsConverter(models.AbstractModel):
+    _inherit = "ir.fields.converter"
+
+    @api.model
+    def db_id_for(self, model, field, subfield, value, savepoint):
+        """Override to scope name_search by field domain during CSV import.
+
+        When importing Many2one fields pointing to spp.vocabulary.code,
+        multiple vocabulary codes may share the same display name across
+        different vocabularies. This passes the field's domain as extra
+        search criteria via _import_name_search_domain context key so
+        name_search can disambiguate.
+        """
+        if (
+            subfield is None
+            and getattr(field, "comodel_name", None) == "spp.vocabulary.code"
+            and getattr(field, "domain", None)
+        ):
+            domain = field.domain
+            if isinstance(domain, str):
+                try:
+                    domain = safe_eval(  # nosemgrep: odoo-unsafe-safe-eval
+                        domain, {"context": self.env.context}
+                    )
+                except Exception as e:
+                    _logger.warning(
+                        "Failed to evaluate domain %r for field %s on model %s;"
+                        "skipping domain scoping during import. Error: %r",
+                        domain,
+                        getattr(field, "name", "<unknown>"),
+                        model._name if model else "<unknown>",
+                        e,
+                    )
+                    domain = []
+            if isinstance(domain, list) and domain:
+                return super(
+                    IrFieldsConverter,
+                    self.with_context(_import_name_search_domain=domain),
+                ).db_id_for(model, field, subfield, value, savepoint)
+        return super().db_id_for(model, field, subfield, value, savepoint)

spp_vocabulary/models/vocabulary_code.py

@@ -353,6 +353,17 @@ def name_get(self):
         """
         return [(rec.id, f"{rec.display} ({rec.code})") for rec in self]
 
+    @api.model
+    def name_search(self, name="", domain=None, operator="ilike", limit=None):
+        """Override name_search to add additional domain conditions.
+        Specifically, if display happens to be the same for different
+        namespaces, the namespace should be used to disambiguate.
+        """
+        extra = self.env.context.get("_import_name_search_domain", [])
+        if extra:
+            domain = list(domain or []) + extra
+        return super().name_search(name=name, domain=domain, operator=operator, limit=limit)
+
     @api.model
     @tools.ormcache("namespace_uri", "code")
     def _get_code_id(self, namespace_uri, code):

spp_vocabulary/tests/test_vocabulary_code.py

@@ -1,3 +1,6 @@
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
 from odoo.exceptions import UserError, ValidationError
 from odoo.tests.common import TransactionCase
 
@@ -786,3 +789,85 @@ def test_uri_updates_on_code_change(self):
         # URI should be recomputed
         self.assertEqual(code.uri, "urn:test:user#CHANGED")
         self.assertNotEqual(code.uri, original_uri)
+
+    # Import name_search scoping tests
+    def _create_shared_display_codes(self, display="Active"):
+        """Helper: create a code with the same display in two different vocabularies."""
+        code_user = self.VocabularyCode.create(
+            {"vocabulary_id": self.vocab_user.id, "code": "NS_ACT1", "display": display}
+        )
+        code_system = self.VocabularyCode.with_context(_test_bypass_system_protection=True).create(
+            {"vocabulary_id": self.vocab_system.id, "code": "NS_ACT2", "display": display}
+        )
+        return code_user, code_system
+
+    def test_name_search_without_context_returns_all_matches(self):
+        """Without context domain, name_search returns codes from all vocabularies."""
+        code_user, code_system = self._create_shared_display_codes("SharedDisplay")
+        results = self.VocabularyCode.name_search("SharedDisplay", operator="=")
+        result_ids = [r[0] for r in results]
+        self.assertIn(code_user.id, result_ids)
+        self.assertIn(code_system.id, result_ids)
+
+    def test_name_search_with_context_domain_scopes_to_vocabulary(self):
+        """With _import_name_search_domain in context, name_search is restricted to that vocabulary."""
+        code_user, code_system = self._create_shared_display_codes("ScopedDisplay")
+        domain = [("vocabulary_id", "=", self.vocab_user.id)]
+        results = self.VocabularyCode.with_context(_import_name_search_domain=domain).name_search(
+            "ScopedDisplay", operator="="
+        )
+        result_ids = [r[0] for r in results]
+        self.assertIn(code_user.id, result_ids)
+        self.assertNotIn(code_system.id, result_ids)
+
+    def test_name_search_context_domain_does_not_affect_other_vocabulary(self):
+        """Context domain scoped to system vocab excludes user vocab codes."""
+        code_user, code_system = self._create_shared_display_codes("OtherScopeDisplay")
+        domain = [("vocabulary_id", "=", self.vocab_system.id)]
+        results = self.VocabularyCode.with_context(_import_name_search_domain=domain).name_search(
+            "OtherScopeDisplay", operator="="
+        )
+        result_ids = [r[0] for r in results]
+        self.assertIn(code_system.id, result_ids)
+        self.assertNotIn(code_user.id, result_ids)
+
+    def test_db_id_for_list_domain_scopes_name_search(self):
+        """db_id_for passes a list field domain to name_search, avoiding cross-vocab matches."""
+        code_user, code_system = self._create_shared_display_codes("DbIdDisplay")
+        field = SimpleNamespace(
+            domain=[("vocabulary_id", "=", self.vocab_user.id)],
+            comodel_name="spp.vocabulary.code",
+        )
+        savepoint = MagicMock()
+        converter = self.env["ir.fields.converter"]
+        result_id, warnings = converter.db_id_for(None, field, None, "DbIdDisplay", savepoint)
+        self.assertEqual(result_id, code_user.id)
+        self.assertEqual(warnings, [])
+
+    def test_db_id_for_string_domain_scopes_name_search(self):
+        """db_id_for evaluates a static string domain and applies it to name_search."""
+        code_user, code_system = self._create_shared_display_codes("StrDomainDisplay")
+        domain_str = f"[('vocabulary_id', '=', {self.vocab_user.id})]"
+        field = SimpleNamespace(
+            domain=domain_str,
+            comodel_name="spp.vocabulary.code",
+        )
+        savepoint = MagicMock()
+        converter = self.env["ir.fields.converter"]
+        result_id, warnings = converter.db_id_for(None, field, None, "StrDomainDisplay", savepoint)
+        self.assertEqual(result_id, code_user.id)
+        self.assertEqual(warnings, [])
+
+    def test_db_id_for_no_domain_returns_multiple_match_warning(self):
+        """db_id_for with no field domain falls back to unscoped search, producing a multiple-match warning."""
+        code_user, code_system = self._create_shared_display_codes("NoDomainDisplay")
+        field = SimpleNamespace(
+            domain=[],
+            comodel_name="spp.vocabulary.code",
+        )
+        savepoint = MagicMock()
+        converter = self.env["ir.fields.converter"]
+        result_id, warnings = converter.db_id_for(None, field, None, "NoDomainDisplay", savepoint)
+        # Still resolves (picks first), but warns about multiple matches
+        self.assertIsNotNone(result_id)
+        self.assertTrue(len(warnings) > 0)