E3AARCHAI-23433: Enhance mps-cli-py parser input API to accept individual JAR files, mixed path lists, and regex patterns

mps-cli-py/src/mpscli/model/builder/SSolutionsRepositoryBuilder.py

@@ -4,13 +4,14 @@
 
 import logging
 import os
+import re
 import sys
 import threading
 import warnings
 from timeit import default_timer as timer
 from concurrent.futures import ProcessPoolExecutor
 from pathlib import Path
-from typing import Dict
+from typing import Dict, List, Optional
 
 from mpscli.model.SRepository import SRepository
 from mpscli.model.builder.SLanguageBuilder import SLanguageBuilder
@@ -57,24 +58,60 @@ def __init__(self):
             cache_save_fn=self._disk_cache.save if self._disk_cache else None,
         )
 
-    def build(self, paths):
+    def build(self, paths, jar_filter: Optional[str] = None):
+        # paths cann be:
+        # : a single string or a directory path or an individual .jar file path
+        # : a list of stringsor any mix of directory paths and individual .jar file paths
+        # Also, jar_filter is an optional regex pattern applied to jar filenames when scanning directories
+        # and individual .jar paths passed directly are always included regardless
+        # jar_filter examples:
+        # - builder.build("/path/to/plugins/")
+        # - builder.build("/path/to/specific.jar")
+        # - builder.build(["/path/to/dir", "/path/to/other.jar"])
+        # - builder.build("/path/to/plugins/", jar_filter=r"org\.something123\..*")
         if isinstance(paths, str):
             paths = [paths]
         elif not isinstance(paths, list):
             _log.error("paths should be either a string or a list of strings")
             sys.exit(1)
 
         start = timer()
-        valid_paths = [p for p in paths if self._is_valid_path(p)]
-        if valid_paths:
-            self.collect_solutions_from_jars(valid_paths)
+        dir_paths, jar_paths = self._resolve_paths(paths, jar_filter)
+        if dir_paths or jar_paths:
+            self.collect_solutions_from_jars(
+                dir_paths, extra_jars=jar_paths, jar_filter=jar_filter
+            )
         self.repo.languages = list(SLanguageBuilder.languages.values())
         stop = timer()
         _log.info("duration for parsing modules: %.2f seconds", stop - start)
         if self._parse_cache:
             self._parse_cache.flush()
         return self.repo
 
+    def _resolve_paths(self, paths: List[str], jar_filter: Optional[str]):
+        # split input paths into directories and individual jar files so that directories are validated
+        # and returned for recursive scanning.
+        dir_paths = []
+        jar_paths = []
+        compiled = re.compile(jar_filter) if jar_filter else None
+
+        for path in paths:
+            p = Path(path)
+            if not p.exists():
+                warnings.warn(f"Path not found: {path}")
+                continue
+            if p.is_file():
+                if p.suffix == ".jar":
+                    jar_paths.append(p)
+                else:
+                    _log.error("path %s is not a directory or a .jar file", path)
+            elif p.is_dir():
+                dir_paths.append(path)
+            else:
+                _log.error("path %s is not a directory or a .jar file", path)
+
+        return dir_paths, jar_paths
+
     def collect_solutions_from_sources(
         self, paths, msd_paths=None, preread_bytes=None, workers=1, msd_stats=None
     ):
@@ -116,12 +153,27 @@ def collect_solutions_from_sources(
             "[diag] phase3 disk: %.1fs -- %d msd files", timer() - t0, len(msd_paths)
         )
 
-    def collect_solutions_from_jars(self, paths):
+    def collect_solutions_from_jars(
+        self,
+        paths: List[str],
+        extra_jars: Optional[List[Path]] = None,
+        jar_filter: Optional[str] = None,
+    ):
         workers = self.JAR_THREADS or min(os.cpu_count() or 4, 16)
+        compiled_filter = re.compile(jar_filter) if jar_filter else None
 
         # phase1: scan ZIP central directories.
         t0 = timer()
-        jar_paths = [jp for path in paths for jp in Path(path).rglob("*.jar")]
+        # collect jars from directories applying jar_filter to filenames if provided
+        jar_paths = []
+        for path in paths:
+            for jp in Path(path).rglob("*.jar"):
+                if compiled_filter is None or compiled_filter.search(jp.name):
+                    jar_paths.append(jp)
+        # add individually specified jar files directly so that they bypass jar_filter
+        if extra_jars:
+            jar_paths.extend(extra_jars)
+
         scan = scan_all_jars(jar_paths, workers)
         t1 = timer()
         _log.info(

mps-cli-py/tests/test_solutions_repository_builder.py

@@ -0,0 +1,166 @@
+# tests/test_solutions_repository_builder.py
+#
+# Tests for SSolutionsRepositoryBuilder. This covers the enhanced build() API where individual JAR file paths, 
+# mixed directory + jar lists and jar_filter regex scenarios.
+
+import os
+import unittest
+from pathlib import Path
+
+from mpscli.model.builder.SLanguageBuilder import SLanguageBuilder
+from mpscli.model.builder.SSolutionsRepositoryBuilder import SSolutionsRepositoryBuilder
+
+_BINARY_PROJECT = os.path.abspath(
+    "../mps_test_projects/mps_cli_binary_persistency_generated"
+)
+_BINARY_JAR_DIR = Path(_BINARY_PROJECT)
+
+# known jar filenames in the binary test project
+_JAR_LIBRARY_TOP = str(
+    _BINARY_JAR_DIR / "mps.cli.lanuse.library_top.binary_persistency-src.jar"
+)
+_JAR_LIBRARY_SECOND = str(
+    _BINARY_JAR_DIR / "mps.cli.lanuse.library_second.binary_persistency-src.jar"
+)
+_JAR_LANDEFS_LIBRARY = str(_BINARY_JAR_DIR / "mps.cli.landefs.library-src.jar")
+
+_SOLUTION_LIBRARY_TOP = "mps.cli.lanuse.library_top.binary_persistency"
+_SOLUTION_LIBRARY_SECOND = "mps.cli.lanuse.library_second.binary_persistency"
+
+
+def _reset():
+    SLanguageBuilder.languages = {}
+    SSolutionsRepositoryBuilder.USE_CACHE = False
+
+
+class TestBuildWithDirectory(unittest.TestCase):
+    # baseline is existing directory path behaviour still works correctly
+    def setUp(self):
+        _reset()
+
+    def test_directory_path_parses_all_jars(self):
+        repo = SSolutionsRepositoryBuilder().build(_BINARY_PROJECT)
+        solution_names = {s.name for s in repo.solutions}
+        self.assertIn(_SOLUTION_LIBRARY_TOP, solution_names)
+        self.assertIn(_SOLUTION_LIBRARY_SECOND, solution_names)
+
+    def test_nonexistent_path_warns_and_returns_empty(self):
+        import warnings
+
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
+            repo = SSolutionsRepositoryBuilder().build("/nonexistent/path")
+        self.assertEqual(len(repo.solutions), 0)
+        self.assertTrue(
+            any("not found" in str(warning.message).lower() for warning in w)
+        )
+
+
+class TestBuildWithIndividualJar(unittest.TestCase):
+
+    def setUp(self):
+        _reset()
+
+    def test_single_jar_path_parses_only_that_jar(self):
+        repo = SSolutionsRepositoryBuilder().build(_JAR_LIBRARY_TOP)
+        solution_names = {s.name for s in repo.solutions}
+        # only library_top solution should be present and library_second jar was not passed
+        self.assertIn(_SOLUTION_LIBRARY_TOP, solution_names)
+        self.assertNotIn(_SOLUTION_LIBRARY_SECOND, solution_names)
+
+    def test_two_jar_paths_parse_both(self):
+        repo = SSolutionsRepositoryBuilder().build(
+            [_JAR_LIBRARY_TOP, _JAR_LIBRARY_SECOND]
+        )
+        solution_names = {s.name for s in repo.solutions}
+        self.assertIn(_SOLUTION_LIBRARY_TOP, solution_names)
+        self.assertIn(_SOLUTION_LIBRARY_SECOND, solution_names)
+
+    def test_jar_path_produces_correct_model_count(self):
+        repo = SSolutionsRepositoryBuilder().build(_JAR_LIBRARY_TOP)
+        sol = repo.find_solution_by_name(_SOLUTION_LIBRARY_TOP)
+        self.assertIsNotNone(sol)
+        self.assertEqual(len(sol.models), 2)
+
+    def test_nonexistent_jar_warns_and_returns_empty(self):
+        import warnings
+
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
+            repo = SSolutionsRepositoryBuilder().build("/nonexistent/fake.jar")
+        self.assertEqual(len(repo.solutions), 0)
+        self.assertTrue(
+            any("not found" in str(warning.message).lower() for warning in w)
+        )
+
+    def test_non_jar_file_logs_error_and_skips(self):
+        # a file that exists but is not a .jar should be skipped with an error log
+        msd_file = str(next(_BINARY_JAR_DIR.rglob("*.jar")).parent / "nonexistent.txt")
+        repo = SSolutionsRepositoryBuilder().build([msd_file, _JAR_LIBRARY_TOP])
+        # library_top should still be parsedd even though the non jar was skipped
+        self.assertIsNotNone(repo.find_solution_by_name(_SOLUTION_LIBRARY_TOP))
+
+
+class TestBuildMixedDirectoryAndJar(unittest.TestCase):
+
+    def setUp(self):
+        _reset()
+
+    def test_mixed_list_directory_and_jar(self):
+        # pass directory which contains all jars and a specific jar but no duplicates
+        repo = SSolutionsRepositoryBuilder().build([_BINARY_PROJECT, _JAR_LIBRARY_TOP])
+        solution_names = {s.name for s in repo.solutions}
+        self.assertIn(_SOLUTION_LIBRARY_TOP, solution_names)
+        self.assertIn(_SOLUTION_LIBRARY_SECOND, solution_names)
+        # no duplicate solutions
+        names = [s.name for s in repo.solutions]
+        self.assertEqual(len(names), len(set(names)))
+
+
+class TestBuildWithJarFilter(unittest.TestCase):
+
+    def setUp(self):
+        _reset()
+
+    def test_jar_filter_includes_matching_jars(self):
+        # filter to only library_top jar by filename pattern
+        repo = SSolutionsRepositoryBuilder().build(
+            _BINARY_PROJECT, jar_filter=r"library_top"
+        )
+        solution_names = {s.name for s in repo.solutions}
+        self.assertIn(_SOLUTION_LIBRARY_TOP, solution_names)
+
+    def test_jar_filter_excludes_non_matching_jars(self):
+        # filter to only library_top and library_second should not be parsed
+        repo = SSolutionsRepositoryBuilder().build(
+            _BINARY_PROJECT, jar_filter=r"library_top"
+        )
+        solution_names = {s.name for s in repo.solutions}
+        self.assertNotIn(_SOLUTION_LIBRARY_SECOND, solution_names)
+
+    def test_jar_filter_no_match_returns_empty(self):
+        repo = SSolutionsRepositoryBuilder().build(
+            _BINARY_PROJECT, jar_filter=r"this_pattern_matches_nothing_xyz"
+        )
+        self.assertEqual(len(repo.solutions), 0)
+
+    def test_individual_jar_bypasses_jar_filter(self):
+        # explicitly named jar should always be included regardless of jar_filter
+        repo = SSolutionsRepositoryBuilder().build(
+            [_BINARY_PROJECT, _JAR_LIBRARY_SECOND], jar_filter=r"library_top"
+        )
+        solution_names = {s.name for s in repo.solutions}
+        # library_top matches filter from directory scan
+        self.assertIn(_SOLUTION_LIBRARY_TOP, solution_names)
+        # library_second was explicitly passed so it bypasses the filter
+        self.assertIn(_SOLUTION_LIBRARY_SECOND, solution_names)
+
+    def test_jar_filter_with_regex_pattern(self):
+        # test a proper regex pattern tat matches jars starting with mps.cli.lanuse
+        repo = SSolutionsRepositoryBuilder().build(
+            _BINARY_PROJECT, jar_filter=r"mps\.cli\.lanuse\."
+        )
+        solution_names = {s.name for s in repo.solutions}
+        # both lanuse solutions should be present
+        self.assertIn(_SOLUTION_LIBRARY_TOP, solution_names)
+        self.assertIn(_SOLUTION_LIBRARY_SECOND, solution_names)