merge: Merge pull request #23 from DSD-DBS/force-mode

Add `force` mode to CLI

.github/workflows/docs.yml

@@ -5,7 +5,7 @@ name: Docs
 
 on:
   push:
-    branches: ["add-docs"]
+    branches: ["master"]
 
 jobs:
   sphinx:

capella_rm_bridge/__main__.py

@@ -19,7 +19,8 @@
 
 from . import auditing
 
-CHANGE_PATH = pathlib.Path("change-set.yaml")
+CHANGE_FOLDER_PATH = pathlib.Path("change-sets")
+CHANGE_FILENAME = "change-set.yaml"
 CHANGE_HISTORY_PATH = pathlib.Path("change.history")
 COMMIT_MSG_PATH = pathlib.Path("commit-message.txt")
 ERROR_PATH = pathlib.Path("change-errors.txt")
@@ -31,6 +32,19 @@ def create_errors_statement(errors: cabc.Iterable[str]) -> str:
     return "\n".join(errors)
 
 
+def write_change_set(change: str, module: dict[str, t.Any]) -> pathlib.Path:
+    """Create a change-set.yaml underneath the change-sets folder."""
+    CHANGE_FOLDER_PATH.mkdir(parents=True, exist_ok=True)
+    mid = module["id"].replace("/", "~")
+    path = CHANGE_FOLDER_PATH / f"{mid}-{CHANGE_FILENAME}"
+    if path.is_file():
+        path.unlink(missing_ok=True)
+
+    path.write_text(change, encoding="utf8")
+    LOGGER.info("Change-set file %s written.", str(path))
+    return path
+
+
 @click.command()
 @click.option(
     "-c",
@@ -69,11 +83,12 @@ def create_errors_statement(errors: cabc.Iterable[str]) -> str:
     help="Pull the latest changes from remote.",
 )
 @click.option(
-    "--no-safe-mode",
+    "--force",
     is_flag=True,
     default=False,
-    help="Modifications are still done to a RequirementModule if an error in "
-    "another, independent module in the snapshot was identified.",
+    help="If a non RequirementModule-error was encountered only the object "
+    "and all related objects will be skipped. Intact objects will still be "
+    "synchronized",
 )
 @click.option(
     "--gather-logs/--no-gather-logs",
@@ -104,7 +119,7 @@ def main(
     dry_run: bool,
     push: bool,
     pull: bool,
-    no_safe_mode: bool,
+    force: bool,
     gather_logs: bool,
     save_change_history: bool,
     save_error_log: bool,
@@ -144,34 +159,35 @@ def main(
             model,
             tconfig,
             module,
-            safe_mode=not no_safe_mode,
+            force=force,
             gather_logs=gather_logs,
         )
 
         if change_set:
             change = decl.dump(change_set)
-            CHANGE_PATH.write_text(change, encoding="utf8")
+            change_path = write_change_set(change, module)
             with auditing.ChangeAuditor(model) as changed_objs:
-                decl.apply(model, CHANGE_PATH)
+                decl.apply(model, change_path)
 
             reporter.store_change(
                 changed_objs, module["id"], module["category"]
             )
 
+    if force or not errors:
+        commit_message = reporter.create_commit_message(snapshot["metadata"])
+        print(commit_message)
+        if reporter.store and not dry_run:
+            model.save(
+                push=push, commit_msg=commit_message, push_options=["skip.ci"]
+            )
+
     if errors:
         error_statement = create_errors_statement(errors)
         print(error_statement)
         if save_error_log:
             ERROR_PATH.write_text(error_statement, encoding="utf8")
             LOGGER.info("Change-errors file %s written.", ERROR_PATH)
-
         sys.exit(1)
-    else:
-        commit_message = reporter.create_commit_message(snapshot["metadata"])
-        COMMIT_MSG_PATH.write_text(commit_message, encoding="utf8")
-        print(commit_message)
-        if reporter.store and not dry_run:
-            model.save(push=push, commit_msg=commit_message)
 
     report = reporter.get_change_report()
     if report and save_change_history:

capella_rm_bridge/auditing.py

@@ -232,14 +232,14 @@ def __audit(self, event: str, args: tuple[t.Any, ...]) -> None:
                 oval = getattr(obj, attr_name)
                 nrepr = self._get_value_repr(value)
                 orepr = self._get_value_repr(oval)
-                params = (obj.uuid, attr_name, nrepr, orepr)
+                events = [EventType(obj.uuid, attr_name, nrepr, orepr)]
             elif event.endswith("setitem"):
                 assert len(args) == 4
                 obj, attr_name, index, value = args
                 nrepr = self._get_value_repr(value)
                 oval = getattr(obj, attr_name)[index]
                 orepr = self._get_value_repr(oval)
-                params = (obj.uuid, attr_name, nrepr, orepr)
+                events = [EventType(obj.uuid, attr_name, nrepr, orepr)]
             elif event.endswith("delete"):
                 assert len(args) == 3
                 obj, attr_name, index = args
@@ -248,22 +248,34 @@ def __audit(self, event: str, args: tuple[t.Any, ...]) -> None:
                 if index is not None:
                     oval = oval[index]
 
-                assert isinstance(oval, common.GenericElement)
-                orepr = self._get_value_repr(oval)
-                params = (obj.uuid, attr_name, orepr, oval.uuid)
+                if not isinstance(oval, common.GenericElement):
+                    assert isinstance(oval, common.ElementList)
+                    events = []
+                    assert EventType is Deletion
+                    for elt in oval:
+                        event_type = EventType(
+                            obj.uuid,
+                            attr_name,
+                            self._get_value_repr(elt),
+                            elt.uuid,
+                        )
+                        events.append(event_type)
+                else:
+                    orepr = self._get_value_repr(oval)
+                    events = [EventType(obj.uuid, attr_name, orepr, oval.uuid)]
             elif event.endswith("insert"):
                 assert len(args) == 4
                 obj, attr_name, _, value = args
                 nrepr = self._get_value_repr(value)
                 assert isinstance(value, common.GenericElement)
-                params = (obj.uuid, attr_name, nrepr, value.uuid)
+                events = [EventType(obj.uuid, attr_name, nrepr, value.uuid)]
             elif event.endswith("create"):
                 assert len(args) == 3
                 obj, attr_name, value = args
                 repr = self._get_value_repr(value)
-                params = (obj.uuid, attr_name, repr, value.uuid)
+                events = [EventType(obj.uuid, attr_name, repr, value.uuid)]
 
-            self.context.append(EventType(*params))
+            self.context.extend(events)
 
     def _get_value_repr(self, value: t.Any) -> str | t.Any:
         if hasattr(value, "_short_repr_"):
@@ -363,7 +375,7 @@ def store_change(
     def _assign_module(self, change: Change) -> LiveDocID | TrackerID | None:
         try:
             obj = self.model.by_uuid(change.parent)
-            while not isinstance(obj, reqif.RequirementsModule):
+            while not isinstance(obj, reqif.CapellaModule):
                 obj = obj.parent
             return obj.identifier
         except (KeyError, AttributeError):
@@ -408,7 +420,7 @@ def create_commit_message(self, tool_metadata: dict[str, str]) -> str:
             main_message = generate_main_message(self.categories.items())
             main = "\n".join((main_message, "\n".join(list_lines) + "\n"))
 
-        summary = f"{summary} from rev.{tool_metadata['revision']} [skip ci]\n"
+        summary = f"{summary} from rev.{tool_metadata['revision']}\n"
         rm_bridge_dependencies = get_dependencies()
         dependencies = "\n".join(
             (
@@ -439,20 +451,20 @@ def _count_changes(
         return ext_count, mod_count, del_count, type_count
 
     def _is_reqtype_change(self, change: Change) -> bool:
-        if isinstance(change, Modification):
+        if isinstance(change, (Modification, Deletion)):
             obj = self.model.by_uuid(change.parent)
-        else:
+        elif isinstance(change, Extension):
             obj = self.model.by_uuid(change.uuid)
 
         return type(obj) in {
             reqif.AttributeDefinition,
             reqif.AttributeDefinitionEnumeration,
             reqif.DataTypeDefinition,
-            reqif.EnumDataTypeDefinition,
+            reqif.EnumerationDataTypeDefinition,
             reqif.EnumValue,
             reqif.ModuleType,
             reqif.RelationType,
-            reqif.RequirementsTypesFolder,
+            reqif.CapellaTypesFolder,
             reqif.RequirementType,
         }
 

capella_rm_bridge/changeset/__init__.py

@@ -24,8 +24,16 @@
 ERROR_MESSAGE_PREFIX = "Skipping module: {module_id}"
 
 
-def _wrap_errors(module_id: int | str, errors: cabc.Sequence[str]) -> str:
-    start = ERROR_MESSAGE_PREFIX.format(module_id=module_id)
+def _wrap_errors(
+    module_id: int | str,
+    errors: cabc.Sequence[str],
+    include_start: bool = True,
+) -> str:
+    if include_start:
+        start = ERROR_MESSAGE_PREFIX.format(module_id=module_id)
+    else:
+        start = f"Encountered error(s) in {module_id!r}"
+
     first_sep = len(start) * "="
     last_sep = len(errors[-1]) * "="
     return "\n".join((start, first_sep, *errors, last_sep))
@@ -35,6 +43,7 @@ def calculate_change_set(
     model: capellambse.MelodyModel,
     config: actiontypes.TrackerConfig,
     snapshot: actiontypes.TrackerSnapshot,
+    force: bool = False,
     safe_mode: bool = True,
     gather_logs: bool = True,
 ) -> tuple[list[dict[str, t.Any]], list[str]]:
@@ -54,6 +63,10 @@ def calculate_change_set(
         calculated ``ChangeSet``.
     snapshot
         A snapshot of a tracker or live-document from the external tool.
+    force
+        If ``True`` a ``ChangeSet`` will be rendered even if an error
+        occurred during the change-calculation loop. All related objects
+        will be skipped.
     safe_mode
         If ``True`` no ``ChangeSet`` will be rendered iff atleast one
         error occurred during the change-calculation loop. If ``False``
@@ -83,14 +96,21 @@ def calculate_change_set(
             snapshot, model, config, gather_logs=gather_logs
         )
         if tchange.errors:
-            message = _wrap_errors(module_id, tchange.errors)
+            message = _wrap_errors(
+                module_id, tchange.errors, include_start=not force
+            )
             errors.append(message)
         else:
             actions.extend(tchange.actions)
+
+        if force and tchange.actions:
+            for action in tchange.actions:
+                if action not in actions:
+                    actions.append(action)
     except (
         actiontypes.InvalidTrackerConfig,
         actiontypes.InvalidSnapshotModule,
-        change.MissingRequirementsModule,
+        change.MissingCapellaModule,
     ) as error:
         if gather_logs:
             message = _wrap_errors(module_id, [error.args[0]])
@@ -99,6 +119,8 @@ def calculate_change_set(
             prefix = ERROR_MESSAGE_PREFIX.format(module_id=module_id)
             LOGGER.error("%s. %s", prefix, error.args[0])
 
+    if force:
+        safe_mode = False
     if safe_mode and any(errors):
         actions = []
     return actions, errors