Refactor io.py to simplify unit tests (#335)

- Split function into smaller functions
- Add unit test
- Add type hints

Author: phiwuu

lobster/io.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 #
 # LOBSTER - Lightweight Open BMW Software Traceability Evidence Report
-# Copyright (C) 2023 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+# Copyright (C) 2023, 2025 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as
@@ -21,6 +21,7 @@
 import io
 from collections.abc import Iterable
 import json
+from typing import Dict, Sequence, Union
 
 from lobster.errors import Message_Handler
 from lobster.location import File_Reference
@@ -52,7 +53,13 @@ def lobster_write(fd, kind, generator, items):
     fd.write("\n")
 
 
-def lobster_read(mh, filename, level, items, source_info=None):
+def lobster_read(
+        mh,
+        filename,
+        level,
+        items: Dict[str, Union[Activity, Implementation, Requirement]],
+        source_info=None,
+):
     assert isinstance(mh, Message_Handler)
     assert isinstance(filename, str)
     assert isinstance(level, str)
@@ -132,11 +139,24 @@ def lobster_read(mh, filename, level, items, source_info=None):
             else:
                 items[item.tag.key()] = item
 
+    signal_duplicate_items(mh, items, duplicate_items)
+
+
+def signal_duplicate_items(
+        mh: Message_Handler,
+        items,
+        duplicate_items: Sequence[Union[Activity, Implementation, Requirement]],
+):
+    """
+    Report errors for duplicate items to the message handler.
+    If there are any duplicate items, the last one is considered fatal.
+    """
     if duplicate_items:
         for counter, item in enumerate(duplicate_items, start=1):
             mh.error(
-                item.location,
-                f"duplicate definition of {item.tag.key()}, "
-                f"previously defined at {items[item.tag.key()].location.to_string()}",
+                location=item.location,
+                message=f"duplicate definition of {item.tag.key()}, "
+                        f"previously defined at "
+                        f"{items[item.tag.key()].location.to_string()}",
                 fatal=(counter == len(duplicate_items)),
             )

lobster/items.py

@@ -20,12 +20,18 @@
 from enum import Enum, auto
 from abc import ABCMeta
 from hashlib import sha1
+from typing import Optional
 
 from lobster.location import Location
 
 
 class Tracing_Tag:
-    def __init__(self, namespace, tag, version=None):
+    def __init__(
+            self,
+            namespace: str,
+            tag: str,
+            version: Optional[str] = None,
+    ):
         assert isinstance(namespace, str) and " " not in namespace
         assert isinstance(tag, str)
         assert version is None or isinstance(version, (str, int))
@@ -84,7 +90,7 @@ class Tracing_Status(Enum):
 
 
 class Item(metaclass=ABCMeta):
-    def __init__(self, tag, location):
+    def __init__(self, tag: Tracing_Tag, location: Location):
         assert isinstance(tag, Tracing_Tag)
         assert isinstance(location, Location)
 
@@ -220,8 +226,16 @@ def to_json(self):
 
 
 class Requirement(Item):
-    def __init__(self, tag, location, framework, kind, name,
-                 text=None, status=None):
+    def __init__(
+            self,
+            tag: Tracing_Tag,
+            location: Location,
+            framework: str,
+            kind: str,
+            name: str,
+            text: Optional[str] = None,
+            status: Optional[str] = None,
+    ):
         super().__init__(tag, location)
         assert isinstance(framework, str)
         assert isinstance(kind, str)
@@ -270,7 +284,14 @@ def from_json(cls, level, data, schema_version):
 
 
 class Implementation(Item):
-    def __init__(self, tag, location, language, kind, name):
+    def __init__(
+            self,
+            tag: Tracing_Tag,
+            location: Location,
+            language: str,
+            kind: str,
+            name: str,
+    ):
         super().__init__(tag, location)
         assert isinstance(language, str)
         assert isinstance(kind, str)
@@ -303,7 +324,14 @@ def from_json(cls, level, data, schema_version):
 
 
 class Activity(Item):
-    def __init__(self, tag, location, framework, kind, status=None):
+    def __init__(
+            self,
+            tag: Tracing_Tag,
+            location: Location,
+            framework: str,
+            kind: str,
+            status: Optional[str] = None,
+    ):
         super().__init__(tag, location)
         assert isinstance(framework, str)
         assert isinstance(kind, str)

tests-unit/test_io_signal_duplicate_items.py

@@ -0,0 +1,76 @@
+from itertools import islice
+from typing import List
+from unittest import TestCase
+from unittest.mock import Mock
+from lobster.errors import Message_Handler
+from lobster.items import Tracing_Tag, Requirement
+from lobster.io import signal_duplicate_items
+from lobster.location import File_Reference
+
+class SignalDuplicateItemsTest(TestCase):
+    def setUp(self) -> None:
+        self._mh = Mock(spec=Message_Handler)
+        self._items = {}
+        for pizza in ["Capricciosa", "Margherita", "Quattro Stagioni", "Pepperoni"]:
+            req = Requirement(
+                    Tracing_Tag("Pizza", pizza, "tomatoe"),
+                    File_Reference("filename"),
+                    "woodstove",
+                    "delicious",
+                    "name",
+                )
+            self._items[req.tag.key()] = req
+    
+    def _create_duplicates(self, source: List[Requirement], count) -> List[Requirement]:
+        """
+        Create a list of duplicate items based on the existing items.
+        They use the same tracing tag, but a different location.
+        The number of duplicates is determined by the count parameter.
+        """
+        if len(source) < count:
+            self.fail("Count exceeds the number of available items.")
+
+        duplicates = []
+        for item in source[:count]:
+            duplicate = Requirement(
+                Tracing_Tag(item.tag.namespace, item.tag.tag, item.tag.version),
+                File_Reference(f"{item.location.filename}-duplicated"),
+                item.framework,
+                item.kind,
+                item.name,
+            )
+            duplicates.append(duplicate)
+        return duplicates
+
+    def test_no_duplicates(self):
+        signal_duplicate_items(self._mh, self._items, [])
+        self.assertEqual(self._mh.error.call_count, 0)
+
+    def test_few_duplicates(self):
+        num_duplicates = 2
+        duplicate_items = self._create_duplicates(
+            list(self._items.values()),
+            num_duplicates,
+        )
+
+        signal_duplicate_items(self._mh, self._items, duplicate_items)
+
+        self.assertEqual(self._mh.error.call_count, num_duplicates)
+
+        for i, call in enumerate(self._mh.error.call_args_list[:-1]):
+            self.assertFalse(
+                call.kwargs.get("fatal", False),
+                f"Call #{i + 1} did not use fatal=False",
+            )
+            duplicated_key = duplicate_items[i].tag.key()
+            self.assertEqual(
+                call.kwargs.get("message"),
+                f"duplicate definition of {duplicated_key}, "
+                f"previously defined at {self._items[duplicated_key].location.to_string()}",
+            )
+            self.assertIs(duplicate_items[i].location, call.kwargs.get("location"))
+
+        self.assertTrue(
+            self._mh.error.call_args_list[-1].kwargs.get("fatal"),
+            "Last call did not use fatal=True",
+        )