feat: adds ModelFilter to allow inclusive and exclusive filtering dur…

…ing tasks

Replaces skip_model_list with ModelFilter
Add ModelFilter to existing tasks
Updates CreateCD to use the filter to only process the new compdef

BREAKING CHANGE: skip_model_list in tasks have been replace with ModelFilter

Signed-off-by: Jennifer Power <[email protected]>

pyproject.toml

@@ -21,6 +21,7 @@ repository = 'https://github.com/RedHatProductSecurity/trestle-bot'
 [tool.poetry.scripts]
 trestlebot-autosync = "trestlebot.entrypoints.autosync:main"
 trestlebot-rules-transform = "trestlebot.entrypoints.rule_transform:main"
+trestlebot-create-cd = "trestlebot.entrypoints.create_cd:main"
 
 [tool.poetry.dependencies]
 python = '^3.8.1'

tests/testutils.py

@@ -269,12 +269,3 @@ def replace_string_in_file(file_path: str, old_string: str, new_string: str) ->
     # Write the updated content back to the file
     with open(file_path, "w") as file:
         file.write(updated_content)
-
-
-def args_dict_to_list(args_dict: Dict[str, str]) -> List[str]:
-    args = []
-    for k, v in args_dict.items():
-        args.append(f"--{k}")
-        if v is not None:
-            args.append(v)
-    return args

tests/trestlebot/tasks/test_assemble_task.py

@@ -36,6 +36,7 @@
 from trestlebot.tasks.assemble_task import AssembleTask
 from trestlebot.tasks.authored.base_authored import AuthorObjectBase
 from trestlebot.tasks.authored.types import AuthoredType
+from trestlebot.tasks.base_task import ModelFilter
 
 
 test_prof = "simplified_nist_profile"
@@ -95,11 +96,13 @@ def test_assemble_task_with_skip(tmp_trestle_dir: str, skip_list: List[str]) ->
 
     mock = Mock(spec=AuthorObjectBase)
 
+    filter = ModelFilter(skip_list, ["."])
+
     assemble_task = AssembleTask(
         working_dir=tmp_trestle_dir,
         authored_model=AuthoredType.CATALOG.value,
         markdown_dir=cat_md_dir,
-        skip_model_list=skip_list,
+        filter=filter,
     )
 
     with patch(

tests/trestlebot/tasks/test_base_task.py

@@ -0,0 +1,47 @@
+#!/usr/bin/python
+
+#    Copyright 2023 Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+"""Test workspace filtering logic."""
+
+import pathlib
+from typing import List
+
+import pytest
+
+from trestlebot.tasks.base_task import ModelFilter
+
+
+@pytest.mark.parametrize(
+    "skip_list, include_list, model_name, expected",
+    [
+        [["simplified_nist_catalog"], [], "simplified_nist_catalog", True],
+        [[], ["simplified_nist_catalog"], "simplified_nist_catalog", False],
+        [["simplified*"], ["."], "simplified_nist_catalog", True],
+        [
+            ["simplified_nist_catalog"],
+            ["simplified*"],
+            "simplified_nist_profile",
+            False,
+        ],
+    ],
+)
+def test_is_skipped(
+    skip_list: List[str], include_list: List[str], model_name: str, expected: str
+) -> None:
+    """Test skip logic."""
+    model_path = pathlib.Path(model_name)
+    model_filter = ModelFilter(skip_list, include_list)
+    assert model_filter.is_skipped(model_path) == expected

tests/trestlebot/tasks/test_regenerate_task.py

@@ -28,6 +28,7 @@
 from tests import testutils
 from trestlebot.tasks.authored.base_authored import AuthorObjectBase
 from trestlebot.tasks.authored.types import AuthoredType
+from trestlebot.tasks.base_task import ModelFilter
 from trestlebot.tasks.regenerate_task import RegenerateTask
 
 
@@ -86,11 +87,13 @@ def test_regenerate_task_with_skip(tmp_trestle_dir: str, skip_list: List[str]) -
 
     mock = Mock(spec=AuthorObjectBase)
 
+    filter = ModelFilter(skip_list, ["."])
+
     regenerate_task = RegenerateTask(
         working_dir=tmp_trestle_dir,
         authored_model=AuthoredType.CATALOG.value,
         markdown_dir=cat_md_dir,
-        skip_model_list=skip_list,
+        filter=filter,
     )
 
     with patch(
@@ -119,19 +122,6 @@ def test_catalog_regenerate_task(tmp_trestle_dir: str) -> None:
     assert os.path.exists(os.path.join(tmp_trestle_dir, md_path))
 
 
-def test_catalog_regenerate_task_with_skip(tmp_trestle_dir: str) -> None:
-    """Test catalog regenerate at the task level"""
-    trestle_root = pathlib.Path(tmp_trestle_dir)
-    md_path = os.path.join(cat_md_dir, test_cat)
-    _ = testutils.setup_for_catalog(trestle_root, test_cat, md_path)
-
-    regenerate_task = RegenerateTask(
-        tmp_trestle_dir, AuthoredType.CATALOG.value, cat_md_dir, "", [test_cat]
-    )
-    assert regenerate_task.execute() == 0
-    assert not os.path.exists(os.path.join(tmp_trestle_dir, md_path))
-
-
 def test_profile_regenerate_task(tmp_trestle_dir: str) -> None:
     """Test profile regenerate at the task level"""
     trestle_root = pathlib.Path(tmp_trestle_dir)

tests/trestlebot/tasks/test_rule_transform_task.py

@@ -14,7 +14,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-"""Test for Trestle Bot rule transform task"""
+"""Test for Trestle Bot rule transform task."""
 
 import pathlib
 
@@ -26,7 +26,7 @@
 from trestle.tasks.csv_to_oscal_cd import RULE_DESCRIPTION, RULE_ID
 
 from tests.testutils import setup_rules_view
-from trestlebot.tasks.base_task import TaskException
+from trestlebot.tasks.base_task import ModelFilter, TaskException
 from trestlebot.tasks.rule_transform_task import RuleTransformTask
 from trestlebot.transformers.yaml_transformer import ToRulesYAMLTransformer
 
@@ -116,8 +116,10 @@ def test_rule_transform_task_with_skip(tmp_trestle_dir: str) -> None:
     trestle_root = pathlib.Path(tmp_trestle_dir)
     setup_rules_view(trestle_root, test_comp, test_rules_dir)
     transformer = ToRulesYAMLTransformer()
+
+    filter = ModelFilter([test_comp], [])
     rule_transform_task = RuleTransformTask(
-        tmp_trestle_dir, test_rules_dir, transformer, skip_model_list=[test_comp]
+        tmp_trestle_dir, test_rules_dir, transformer, filter=filter
     )
     return_code = rule_transform_task.execute()
     assert return_code == 0

trestlebot/const.py

@@ -46,3 +46,4 @@
 YAML_EXTENSION = ".yaml"
 
 RULES_VIEW_DIR = "rules"
+RULE_PREFIX = "rule-"

trestlebot/entrypoints/autosync.py

@@ -32,7 +32,7 @@
 from trestlebot.entrypoints.log import set_log_level_from_args
 from trestlebot.tasks.assemble_task import AssembleTask
 from trestlebot.tasks.authored import types
-from trestlebot.tasks.base_task import TaskBase
+from trestlebot.tasks.base_task import ModelFilter, TaskBase
 from trestlebot.tasks.regenerate_task import RegenerateTask
 
 
@@ -113,35 +113,43 @@ def run(self, args: argparse.Namespace) -> None:
                 logger.error("Must set markdown path with oscal model.")
                 sys.exit(const.ERROR_EXIT_CODE)
 
-            if args.oscal_model == "ssp" and args.ssp_index_path == "":
+            if (
+                args.oscal_model == types.AuthoredType.SSP.value
+                and args.ssp_index_path == ""
+            ):
                 logger.error("Must set ssp_index_path when using SSP as oscal model.")
                 sys.exit(const.ERROR_EXIT_CODE)
 
+            filter: ModelFilter = ModelFilter(
+                skip_patterns=comma_sep_to_list(args.skip_items),
+                include_patterns=["."],
+            )
+
             # Assuming an edit has occurred assemble would be run before regenerate.
             # Adding this to the list first
             if not args.skip_assemble:
                 assemble_task = AssembleTask(
-                    args.working_dir,
-                    args.oscal_model,
-                    args.markdown_path,
-                    args.ssp_index_path,
-                    comma_sep_to_list(args.skip_items),
+                    working_dir=args.working_dir,
+                    authored_model=args.oscal_model,
+                    markdown_dir=args.markdown_path,
+                    ssp_index_path=args.ssp_index_path,
+                    filter=filter,
                 )
                 pre_tasks.append(assemble_task)
             else:
-                logger.info("Assemble task skipped")
+                logger.info("Assemble task skipped.")
 
             if not args.skip_regenerate:
                 regenerate_task = RegenerateTask(
-                    args.working_dir,
-                    args.oscal_model,
-                    args.markdown_path,
-                    args.ssp_index_path,
-                    comma_sep_to_list(args.skip_items),
+                    working_dir=args.working_dir,
+                    authored_model=args.oscal_model,
+                    markdown_dir=args.markdown_path,
+                    ssp_index_path=args.ssp_index_path,
+                    filter=filter,
                 )
                 pre_tasks.append(regenerate_task)
             else:
-                logger.info("Regeneration task skipped")
+                logger.info("Regeneration task skipped.")
 
             super().run_base(args, pre_tasks)
 

trestlebot/entrypoints/create_cd.py

@@ -14,97 +14,117 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+"""
+Entrypoint for component definition bootstrapping.
+
+This will create a rules-view directory in the working directory, create a component
+definition in JSON format with the initially generated rules and initial trestle control markdown.
+"""
+
 import argparse
+import logging
 from typing import List
 
-from trestlebot.entrypoints.entrypoint_base import EntrypointBase, comma_sep_to_list
+from trestlebot.const import RULE_PREFIX, RULES_VIEW_DIR
+from trestlebot.entrypoints.entrypoint_base import EntrypointBase
+from trestlebot.entrypoints.log import set_log_level_from_args
 from trestlebot.tasks.authored.compdef import AuthoredComponentDefinition
-from trestlebot.tasks.base_task import TaskBase
+from trestlebot.tasks.authored.types import AuthoredType
+from trestlebot.tasks.base_task import ModelFilter, TaskBase
 from trestlebot.tasks.regenerate_task import RegenerateTask
 from trestlebot.tasks.rule_transform_task import RuleTransformTask
 from trestlebot.transformers.yaml_transformer import ToRulesYAMLTransformer
-from trestlebot.const import RULES_VIEW_DIR
+
+
+logger = logging.getLogger(__name__)
 
 
 class CreateCDEntrypoint(EntrypointBase):
-    """Entrypoint for setting default component fields."""
+    """Entrypoint for component definition bootstrapping."""
 
     def __init__(self, parser: argparse.ArgumentParser) -> None:
         """Initialize."""
         super().__init__(parser)
-        self.setup_set_default_component_fields_arguments()
+        self.setup_create_cd_arguments()
 
-    def setup_set_default_component_fields_arguments(self) -> None:
+    def setup_create_cd_arguments(self) -> None:
         """Setup specific arguments for this entrypoint."""
         self.parser.add_argument(
-            "--profile-name", required=True, help="Name of profile"
+            "--profile-name",
+            required=True,
+            help="Name of profile in the trestle workspace to use with the component definition.",
         )
         self.parser.add_argument(
             "--compdef-name", required=True, help="Name of component definition"
         )
         self.parser.add_argument(
-            "--comp-title", required=True, help="Title of component"
+            "--component-title", required=True, help="Title of initial component"
         )
         self.parser.add_argument(
-            "--comp-description", required=True, help="Description of component"
+            "--component-description",
+            required=True,
+            help="Description of initial component",
         )
         self.parser.add_argument(
-            "--cd-type",
-            required=False,
-            default="service",
-            help="Type of component definition",
+            "--markdown-path",
+            required=True,
+            type=str,
+            help="Path to create markdown files in.",
         )
         self.parser.add_argument(
-            "--generate",
+            "--component-definition-type",
             required=False,
-            default=False,
-            help="Generate markdown and ssp index",
-        )
-        self.parser.add_argument(
-            "--transform-rules",
-            required=False,
-            default=False,
-            help="Transform rules to YAML",
+            type=str,
+            choices=["service", "validation"],
+            default="service",
+            help="Type of component definition",
         )
-
 
     def run(self, args: argparse.Namespace) -> None:
         """Run the entrypoint."""
+
+        set_log_level_from_args(args)
+        pre_tasks: List[TaskBase] = []
+
+        # In this case we only want to do the transformation and generation for this component
+        # definition, so we skip all other component definitions and components.
+        filter: ModelFilter = ModelFilter(
+            [], [args.compdef_name, args.component_title, f"{RULE_PREFIX}*"]
+        )
+
         authored_comp = AuthoredComponentDefinition(args.working_dir)
         authored_comp.create_new_default(
             args.profile_name,
             args.compdef_name,
-            args.comp_title,
-            args.comp_description,
-            args.cd_type,
+            args.component_title,
+            args.component_description,
+            args.component_definition_type,
         )
-        pre_tasks: List[TaskBase] = []
-        if args.transform_rules:
-            transformer: ToRulesYAMLTransformer = ToRulesYAMLTransformer()
-            rule_transform_task: RuleTransformTask = RuleTransformTask(
-                args.working_dir,
-                args.RULES_VIEW_DIR,
-                transformer,
-                comma_sep_to_list(args.skip_items),
-            )
-            pre_tasks: List[TaskBase] = [rule_transform_task]
-        if args.generate:
-            regenerate_task = RegenerateTask(
-                args.working_dir,
-                "compdef",
-                args.markdown_path,
-                args.ssp_index_path,
-                comma_sep_to_list(args.skip_items),
-            )
-            pre_tasks.append(regenerate_task)
+
+        transformer: ToRulesYAMLTransformer = ToRulesYAMLTransformer()
+        rule_transform_task: RuleTransformTask = RuleTransformTask(
+            working_dir=args.working_dir,
+            rules_view_dir=RULES_VIEW_DIR,
+            rule_transformer=transformer,
+            filter=filter,
+        )
+        pre_tasks.append(rule_transform_task)
+
+        regenerate_task = RegenerateTask(
+            working_dir=args.working_dir,
+            authored_model=AuthoredType.COMPDEF.value,
+            markdown_dir=args.markdown_path,
+            filter=filter,
+        )
+        pre_tasks.append(regenerate_task)
 
         super().run_base(args, pre_tasks)
 
 
 def main() -> None:
     """Run the CLI."""
     parser = argparse.ArgumentParser(
-        description="Set default component fields entrypoint for trestle."
+        description="Create new component definition with defaults."
     )
     set_default_component_fields = CreateCDEntrypoint(parser=parser)