test collection filter (#24)

Author: cle-b

.flake8

@@ -1,3 +1,3 @@
 [flake8]
 max-line-length = 88
-extend-ignore = E203,E501,E701
+extend-ignore = E203,E266,E501,E701

lymbo/__init__.py

@@ -6,7 +6,7 @@
 from lymbo.resource_manager import scope_global
 from lymbo.resource_manager import scope_module
 
-__version__ = "0.0.2"
+__version__ = "0.1.0"
 
 __all__ = [
     "args",

lymbo/__main__.py

@@ -28,7 +28,7 @@ def lymbo_entry_point():
     )
 
     print("==== collecting tests")
-    test_plan = collect_tests(config.paths, config.groupby)
+    test_plan = collect_tests(config.paths, config.groupby, config.filter)
 
     if config.collect:
         test_plan_to_print, _ = test_plan.test_plan(show_status=False)

lymbo/collect.py

@@ -9,13 +9,55 @@
 
 from lymbo.config import GroupBy
 from lymbo.env import LYMBO_TEST_COLLECTION
+from lymbo.exception import LymboExceptionFilter
 from lymbo.item import TestItem
 from lymbo.item import TestPlan
 from lymbo.log import trace_call
+from lymbo.log import logger
+
 
 from lymbo.cm import args
 
 
+@trace_call
+def collect_tests(
+    paths: list[Path], group_by: GroupBy, filter_by_path: str = ""
+) -> TestPlan:
+    """Collect all the functions/methods decorated with @lymbo.test."""
+
+    tests = []
+    filtered_tests = []
+
+    os.environ[LYMBO_TEST_COLLECTION] = "1"
+
+    for path in list_python_files(paths):
+
+        tests += list_tests_from_file(path, group_by)
+
+    del os.environ[LYMBO_TEST_COLLECTION]
+
+    if filter_by_path:
+        for group in tests:
+            new_group = []
+            for test in group:
+                if match_filter(str(test), filter_by_path):
+                    new_group.append(test)
+                else:
+                    logger().debug(f"collect_tests - {str(test)} has been filtered.")
+
+            if new_group:
+                filtered_tests.append(new_group)
+    else:
+        filtered_tests = tests
+
+    test_plan = TestPlan(filtered_tests, group_by)
+
+    return test_plan
+
+
+## List tests
+
+
 @trace_call
 def list_python_files(paths: list[Path]) -> list[Path]:
     """Walk into all directories and subdirectories to list the Python files."""
@@ -138,25 +180,6 @@ def parse_body(
     return collected_tests
 
 
-@trace_call
-def collect_tests(paths: list[Path], group_by: GroupBy) -> TestPlan:
-    """Collect all the functions/methods decorated with @lymbo.test."""
-
-    tests = []
-
-    os.environ[LYMBO_TEST_COLLECTION] = "1"
-
-    for path in list_python_files(paths):
-
-        tests += list_tests_from_file(path, group_by)
-
-    del os.environ[LYMBO_TEST_COLLECTION]
-
-    test_plan = TestPlan(tests, group_by)
-
-    return test_plan
-
-
 def eval_ast_call(call_node, global_vars, local_vars):
 
     # local_vars and global_vars should be passed in the context
@@ -241,3 +264,45 @@ def dynamic_import_modules(imports: list[tuple[str, str]]) -> dict[str, str]:
             global_vars[alias] = importlib.import_module(full_name)
 
     return global_vars
+
+
+## Filter tests
+
+
+def extract_words_from_filter(filter: str) -> list[str]:
+    """Parse a filter and returns only the words of the expression.
+    Example: filter = "abc and not (def or ghi)"
+             output = ["abc", "def", "ghi"]
+    """
+
+    # remove the stop words
+    stop_words = ["(", ")", "and", "or", "not"]
+    for stop_word in stop_words:
+        filter = filter.replace(stop_word, " ")
+
+    # extract words
+    words = set(filter.split(" "))
+
+    return [word for word in words if word != ""]
+
+
+def match_filter(item: str, filter: str) -> bool:
+    """Indicate if this item match with the filter"""
+    words = {word: False for word in extract_words_from_filter(filter)}
+
+    for word in words:
+        words[word] = word in item
+
+    # We sort the list of words to first replace the longer words to have partial
+    # replacement if for example we have these two words: abc and abcdef.
+    for word in sorted(words, key=len, reverse=True):
+        filter = filter.replace(word, str(words[word]))
+
+    try:
+        match = eval(filter)
+    except Exception as ex:
+        raise LymboExceptionFilter(
+            f'The filter ["{filter}"] is broken. exception=[{str(ex)}]'
+        )
+
+    return match

lymbo/config.py

@@ -61,7 +61,14 @@ def parse_args() -> argparse.Namespace:
         "--workers",
         type=int,
         default=None,
-        help="The number of workers in parrallel (default = number of CPU)",
+        help="The number of workers in parrallel (default = number of CPU).",
+    )
+
+    parser.add_argument(
+        "--filter",
+        type=str,
+        default="",
+        help="Select only the tests that match this filter (include full path and parameters).",
     )
 
     return parser.parse_args()

lymbo/exception.py

@@ -8,3 +8,9 @@ class LymboExceptionScopeNested(Exception):
     """A shared resource is initialized from another shared resource."""
 
     pass
+
+
+class LymboExceptionFilter(Exception):
+    """The test collection filter is broken."""
+
+    pass

tests/test_collect.py

@@ -3,6 +3,9 @@
 import unittest
 
 from lymbo.collect import collect_tests
+from lymbo.collect import extract_words_from_filter
+from lymbo.collect import match_filter
+from lymbo.exception import LymboExceptionFilter
 from lymbo.item import GroupBy
 
 dir = os.path.dirname(os.path.abspath(__file__))
@@ -82,6 +85,82 @@ def test_collect_tests_two_files(self):
 
         self.assertEqual(test_plan.count, (11, 11))
 
+    # filters
+
+    def test_extract_words(self):
+
+        params = {
+            "abc": ["abc"],
+            "not abc": ["abc"],
+            "not (abc)": ["abc"],
+            "(not abc)": ["abc"],
+            "abc or abc": ["abc"],
+            "abc or abcdef": ["abc", "abcdef"],
+            "abc or def": ["abc", "def"],
+            "abc and def": ["abc", "def"],
+            "abc or (def and ijk)": ["abc", "def", "ijk"],
+            "abc or not (def and ijk)": ["abc", "def", "ijk"],
+            "(  abc or ((def and (ijk))    )": ["abc", "def", "ijk"],
+        }
+
+        for filter, words in params.items():
+            with self.subTest(f'filter="{filter}" words={words}'):
+                self.assertListEqual(
+                    sorted(extract_words_from_filter(filter)), sorted(words)
+                )
+
+    def test_match_filter(self):
+
+        params = {
+            "abc": "abc",
+            "abc/def/ijk.py::func1": "abc",
+            "abc/def/ijk.py::func2": "abc or abcdef",
+            "abc/def/ijk.py::func3": "abcdef or abc",
+            "abc/def/ijk.py::func4": "def",
+            "abc/def/ijk.py::func5": "abc and def",
+            "abc/def/ijk.py::func6": "abc and def and not ABC",
+        }
+
+        for item, filter in params.items():
+            with self.subTest(f'item="{item}" filter="{filter}"'):
+                self.assertTrue(match_filter(item, filter))
+
+    def test_match_filter_not_match(self):
+
+        params = {
+            "abc": "def",
+            "abc/def/ijk.py::func1": "ABC",
+            "abc/def/ijk.py::func2": "abc and abcdef",
+            "abc/def/ijk.py::func3": "abcdef and abc",
+            "abc/def/ijk.py::func4": "not def",
+            "abc/def/ijk.py::func5": "abc and not def",
+            "abc/def/ijk.py::func6": "not abc or (DEF and not ABC)",
+        }
+
+        for item, filter in params.items():
+            with self.subTest(f'item="{item}" filter="{filter}"'):
+                self.assertFalse(match_filter(item, filter))
+
+    def test_collect_tests_filter(self):
+
+        test_plan = collect_tests(
+            [Path(os.path.join(dir, "data_collect"))],
+            GroupBy.MODULE,
+            "second and not ((p=4) or (p=5))",
+        )
+
+        self.assertEqual(test_plan.count, (5, 2))
+
+    def test_collect_tests_broken(self):
+
+        try:
+            _ = collect_tests(
+                [Path(os.path.join(dir, "data_collect"))], GroupBy.MODULE, "second )"
+            )
+            self.assertFalse(True, "No exception has been raised.")
+        except Exception as ex:
+            self.assertIsInstance(ex, LymboExceptionFilter)
+
 
 if __name__ == "__main__":
     unittest.main()