perf: cache plugin_names.json with lru_cache for O(1) lookups

- Add _load_plugin_names() with @lru_cache to read file once at first access
- Replace O(n) linear scan with frozenset O(1) membership check
- Extract shared _tokenize_plugin_name() helper (removes duplicate tokenize())
- Add 15 unit tests covering caching, tokenization, and validation

Fixes #352

Author: sharma-sugurthi

chatbot-core/api/tools/utils.py

@@ -2,6 +2,7 @@
 Utilities for the tools package.
 """
 
+import functools
 import json
 import os
 import re
@@ -193,29 +194,45 @@ def extract_chunks_content(chunks: List[Dict], logger) -> str:
         else retrieval_config["empty_context_message"]
     )
 
+def _tokenize_plugin_name(name: str) -> str:
+    """Normalize a plugin name for case/separator-insensitive comparison."""
+    return name.replace('-', '').replace(' ', '').lower()
+
+
+@functools.lru_cache(maxsize=1)
+def _load_plugin_names() -> frozenset:
+    """
+    Load and cache the set of known plugin names (tokenized) from disk.
+
+    The JSON file is static data that never changes at runtime, so it is
+    read once on first access and kept in memory for O(1) lookups.
+
+    Returns:
+        frozenset: A set of tokenized plugin names.
+    """
+    list_plugin_names_path = os.path.join(
+        os.path.dirname(os.path.abspath(__file__)),
+        "..", "data", "raw", "plugin_names.json"
+    )
+    with open(list_plugin_names_path, "r", encoding="utf-8") as f:
+        list_plugin_names = json.load(f)
+    return frozenset(_tokenize_plugin_name(name) for name in list_plugin_names)
+
+
 def is_valid_plugin(plugin_name: str) -> bool:
     """
     Checks whether the given plugin name exists in the list of known plugin names.
 
+    Uses a cached frozenset for O(1) membership checks instead of
+    re-reading from disk and performing a linear scan on every call.
+
     Args:
         plugin_name (str): The name of the plugin to validate.
 
     Returns:
         bool: True if the plugin exists in the list, False otherwise.
     """
-    def tokenize(item: str) -> str:
-        item = item.replace('-', '')
-        return item.replace(' ', '').lower()
-    list_plugin_names_path = os.path.join(os.path.abspath(__file__),
-                                          "..", "..", "data", "raw", "plugin_names.json")
-    with open(list_plugin_names_path, "r", encoding="utf-8") as f:
-        list_plugin_names = json.load(f)
-
-    for name in list_plugin_names:
-        if tokenize(plugin_name) == tokenize(name):
-            return True
-
-    return False
+    return _tokenize_plugin_name(plugin_name) in _load_plugin_names()
 
 def filter_retrieved_data(
     semantic_data: List[Dict],
@@ -234,14 +251,12 @@ def filter_retrieved_data(
     Returns:
         Tuple[List[Dict], List[Dict]]: Filtered semantic and keyword data.
     """
-    def tokenize(item: str) -> str:
-        item = item.replace('-', '')
-        return item.replace(' ', '').lower()
-
     semantic_filtered_data = [item for item in semantic_data
-                              if tokenize(item["metadata"]["title"]) == tokenize(plugin_name)]
+                              if _tokenize_plugin_name(item["metadata"]["title"])
+                              == _tokenize_plugin_name(plugin_name)]
     keyword_filtered_data = [item for item in keyword_data
-                             if tokenize(item["metadata"]["title"]) == tokenize(plugin_name)]
+                             if _tokenize_plugin_name(item["metadata"]["title"])
+                             == _tokenize_plugin_name(plugin_name)]
 
     return semantic_filtered_data, keyword_filtered_data
 

chatbot-core/tests/unit/tools/test_utils.py

@@ -0,0 +1,194 @@
+"""Unit tests for plugin name caching and validation in tools/utils.py."""
+import json
+import unittest
+from unittest.mock import patch, mock_open
+
+from api.tools.utils import (
+    _tokenize_plugin_name,
+    _load_plugin_names,
+    is_valid_plugin,
+    filter_retrieved_data,
+)
+
+
+class TestTokenizePluginName(unittest.TestCase):
+    """Tests for the _tokenize_plugin_name helper."""
+
+    def test_converts_to_lowercase(self):
+        """Plugin names should be case-insensitive."""
+        self.assertEqual(_tokenize_plugin_name("Git"), "git")
+
+    def test_removes_hyphens(self):
+        """Hyphens should be stripped for comparison."""
+        self.assertEqual(
+            _tokenize_plugin_name("blue-ocean"), "blueocean"
+        )
+
+    def test_removes_spaces(self):
+        """Spaces should be stripped for comparison."""
+        self.assertEqual(
+            _tokenize_plugin_name("Blue Ocean"), "blueocean"
+        )
+
+    def test_removes_hyphens_and_spaces(self):
+        """Both hyphens and spaces should be stripped."""
+        self.assertEqual(
+            _tokenize_plugin_name("my-cool plugin"), "mycoolplugin"
+        )
+
+    def test_empty_string(self):
+        """Empty string should return empty string."""
+        self.assertEqual(_tokenize_plugin_name(""), "")
+
+
+class TestLoadPluginNames(unittest.TestCase):
+    """Tests for the cached _load_plugin_names function."""
+
+    def setUp(self):
+        """Clear lru_cache before each test to ensure isolation."""
+        _load_plugin_names.cache_clear()
+
+    def tearDown(self):
+        """Clear lru_cache after each test."""
+        _load_plugin_names.cache_clear()
+
+    @patch(
+        "builtins.open",
+        mock_open(read_data=json.dumps(["git", "Blue-Ocean", "credentials"])),
+    )
+    @patch("os.path.dirname", return_value="/fake/dir")
+    def test_returns_frozenset(self, _mock_dir):
+        """Result should be a frozenset for O(1) lookups."""
+        result = _load_plugin_names()
+        self.assertIsInstance(result, frozenset)
+
+    @patch(
+        "builtins.open",
+        mock_open(read_data=json.dumps(["git", "Blue-Ocean", "credentials"])),
+    )
+    @patch("os.path.dirname", return_value="/fake/dir")
+    def test_tokenizes_all_names(self, _mock_dir):
+        """All names should be tokenized (lowercased, no hyphens/spaces)."""
+        result = _load_plugin_names()
+        self.assertIn("git", result)
+        self.assertIn("blueocean", result)
+        self.assertIn("credentials", result)
+
+    @patch(
+        "builtins.open",
+        mock_open(read_data=json.dumps(["git", "Blue-Ocean", "credentials"])),
+    )
+    @patch("os.path.dirname", return_value="/fake/dir")
+    def test_caching_reads_file_once(self, _mock_dir):
+        """Calling _load_plugin_names twice should only open the file once."""
+        with patch("builtins.open", mock_open(
+            read_data=json.dumps(["git"])
+        )) as mocked_file:
+            _load_plugin_names()
+            _load_plugin_names()
+            mocked_file.assert_called_once()
+
+
+class TestIsValidPlugin(unittest.TestCase):
+    """Tests for the is_valid_plugin function."""
+
+    def setUp(self):
+        _load_plugin_names.cache_clear()
+
+    def tearDown(self):
+        _load_plugin_names.cache_clear()
+
+    @patch(
+        "builtins.open",
+        mock_open(
+            read_data=json.dumps(
+                ["git", "blue-ocean", "credentials", "github-branch-source"]
+            )
+        ),
+    )
+    @patch("os.path.dirname", return_value="/fake/dir")
+    def test_exact_match(self, _mock_dir):
+        """Exact plugin name should be valid."""
+        self.assertTrue(is_valid_plugin("git"))
+
+    @patch(
+        "builtins.open",
+        mock_open(
+            read_data=json.dumps(
+                ["git", "blue-ocean", "credentials", "github-branch-source"]
+            )
+        ),
+    )
+    @patch("os.path.dirname", return_value="/fake/dir")
+    def test_case_insensitive_match(self, _mock_dir):
+        """Plugin names should match case-insensitively."""
+        self.assertTrue(is_valid_plugin("Git"))
+        self.assertTrue(is_valid_plugin("GIT"))
+
+    @patch(
+        "builtins.open",
+        mock_open(
+            read_data=json.dumps(
+                ["git", "blue-ocean", "credentials", "github-branch-source"]
+            )
+        ),
+    )
+    @patch("os.path.dirname", return_value="/fake/dir")
+    def test_hyphen_insensitive_match(self, _mock_dir):
+        """Hyphens should be ignored during matching."""
+        self.assertTrue(is_valid_plugin("blue ocean"))
+        self.assertTrue(is_valid_plugin("blueocean"))
+        self.assertTrue(is_valid_plugin("Blue-Ocean"))
+
+    @patch(
+        "builtins.open",
+        mock_open(
+            read_data=json.dumps(
+                ["git", "blue-ocean", "credentials", "github-branch-source"]
+            )
+        ),
+    )
+    @patch("os.path.dirname", return_value="/fake/dir")
+    def test_invalid_plugin_returns_false(self, _mock_dir):
+        """Non-existent plugin name should return False."""
+        self.assertFalse(is_valid_plugin("nonexistent-plugin"))
+        self.assertFalse(is_valid_plugin(""))
+
+
+class TestFilterRetrievedData(unittest.TestCase):
+    """Tests for filter_retrieved_data using the shared tokenizer."""
+
+    def test_filters_matching_entries(self):
+        """Only entries whose title matches the plugin name should remain."""
+        semantic_data = [
+            {"metadata": {"title": "blue-ocean"}, "chunk_text": "a"},
+            {"metadata": {"title": "credentials"}, "chunk_text": "b"},
+        ]
+        keyword_data = [
+            {"metadata": {"title": "Blue Ocean"}, "chunk_text": "c"},
+            {"metadata": {"title": "git"}, "chunk_text": "d"},
+        ]
+        sem, kw = filter_retrieved_data(
+            semantic_data, keyword_data, "blue-ocean"
+        )
+        self.assertEqual(len(sem), 1)
+        self.assertEqual(sem[0]["chunk_text"], "a")
+        self.assertEqual(len(kw), 1)
+        self.assertEqual(kw[0]["chunk_text"], "c")
+
+    def test_returns_empty_when_no_match(self):
+        """No results should be returned when nothing matches."""
+        data = [{"metadata": {"title": "git"}, "chunk_text": "x"}]
+        sem, kw = filter_retrieved_data(data, data, "nonexistent")
+        self.assertEqual(len(sem), 0)
+        self.assertEqual(len(kw), 0)
+
+    def test_empty_input(self):
+        """Empty input lists should return empty lists."""
+        sem, kw = filter_retrieved_data([], [], "git")
+        self.assertEqual(sem, [])
+        self.assertEqual(kw, [])
+
+
+if __name__ == "__main__":
+    unittest.main()