feat: compare all AHB props to handle row insertion at arb. positions (#37)

* Rename `test_[...]_multiple_entries_per_section_name` to `multiple_rows`

* Add test case covering row insertion at arbitrary positions

* Enhance comparison to check for matching rows considering all properties

Author: OLILHR

src/ahlbatross/core/ahb_comparison.py

@@ -18,10 +18,10 @@ def _compare_ahb_rows(previous_ahb_row: AhbRow, subsequent_ahb_row: AhbRow) -> A
 
     # consider all AHB properties except `section_name` (Segmentname) and `formatversion`
     for entry in AHB_PROPERTIES:
-        previous_ahb_entry = getattr(previous_ahb_row, entry, "") or ""
-        subsequent_ahb_entry = getattr(subsequent_ahb_row, entry, "") or ""
+        previous_ahb_entry = normalize_entries(getattr(previous_ahb_row, entry, "") or "")
+        subsequent_ahb_entry = normalize_entries(getattr(subsequent_ahb_row, entry, "") or "")
 
-        if (previous_ahb_entry.strip() or subsequent_ahb_entry.strip()) and previous_ahb_entry != subsequent_ahb_entry:
+        if (previous_ahb_entry or subsequent_ahb_entry) and previous_ahb_entry != subsequent_ahb_entry:
             changed_entries.extend(
                 [f"{entry}_{previous_ahb_row.formatversion}", f"{entry}_{subsequent_ahb_row.formatversion}"]
             )
@@ -50,16 +50,36 @@ def _add_empty_row(formatversion: str) -> AhbRow:
 
 
 def _find_matching_subsequent_row(
-    current_ahb_row: AhbRow, subsequent_ahb_rows: List[AhbRow], start_idx: int
+    current_ahb_row: AhbRow, subsequent_ahb_rows: List[AhbRow], start_idx: int, duplicate_indices: set[int]
 ) -> Tuple[int, AhbRow | None]:
     """
-    Find matching row in subsequent version starting from given index.
+    Find matching row in subsequent version starting from given index by consider all AHB properties
+    within the same `section_name` group.
     """
     normalized_current = normalize_entries(current_ahb_row.section_name)
+    current_key = current_ahb_row.get_key()
 
     for idx, row in enumerate(subsequent_ahb_rows[start_idx:], start_idx):
+        if idx in duplicate_indices:
+            continue
+
+        if (
+            normalize_entries(row.section_name) == normalized_current
+            and row.get_key() == current_key
+            and row.segment_id == current_ahb_row.segment_id
+        ):
+            return idx, row
+
+    # in case no match was found, continue by aligning `Segmentname` entries
+    for idx, row in enumerate(subsequent_ahb_rows[start_idx:], start_idx):
+        if idx in duplicate_indices:
+            continue
+
         if normalize_entries(row.section_name) == normalized_current:
+            if row.ahb_expression is not None and current_ahb_row.ahb_expression is None:
+                continue
             return idx, row
+
     return -1, None
 
 
@@ -70,68 +90,74 @@ def align_ahb_rows(previous_ahb_rows: List[AhbRow], subsequent_ahb_rows: List[Ah
     result = []
     i = 0
     j = 0
+    duplicate_indices: set[int] = set()
 
     while i < len(previous_ahb_rows) or j < len(subsequent_ahb_rows):
         if i >= len(previous_ahb_rows):
-            row = subsequent_ahb_rows[j]
-            result.append(
-                AhbRowComparison(
-                    previous_formatversion=_add_empty_row(row.formatversion),
-                    # label remaining rows of subsequent AHB as NEW
-                    diff=AhbRowDiff(diff_type=DiffType.ADDED),
-                    subsequent_formatversion=row,
+            # add remaining rows as "new" if not already used
+            if j not in duplicate_indices:
+                result.append(
+                    AhbRowComparison(
+                        previous_formatversion=_add_empty_row(subsequent_ahb_rows[j].formatversion),
+                        diff=AhbRowDiff(diff_type=DiffType.ADDED),
+                        subsequent_formatversion=subsequent_ahb_rows[j],
+                    )
                 )
-            )
             j += 1
+            continue
 
-        elif j >= len(subsequent_ahb_rows):
-            row = previous_ahb_rows[i]
+        if j >= len(subsequent_ahb_rows):
             result.append(
                 AhbRowComparison(
-                    previous_formatversion=row,
+                    previous_formatversion=previous_ahb_rows[i],
                     # label remaining rows of previous AHB as REMOVED
                     diff=AhbRowDiff(diff_type=DiffType.REMOVED),
-                    subsequent_formatversion=_add_empty_row(row.formatversion),
+                    subsequent_formatversion=_add_empty_row(previous_ahb_rows[i].formatversion),
                 )
             )
             i += 1
+            continue
 
-        else:
-            current_row = previous_ahb_rows[i]
-            next_match_idx, matching_row = _find_matching_subsequent_row(current_row, subsequent_ahb_rows, j)
+        current_row = previous_ahb_rows[i]
+        next_match_idx, matching_row = _find_matching_subsequent_row(
+            current_row, subsequent_ahb_rows, j, duplicate_indices
+        )
 
-            if next_match_idx >= 0 and matching_row is not None:
-                # add new rows until `section_name` (Segmentname) matches
-                for k in range(j, next_match_idx):
-                    new_row = subsequent_ahb_rows[k]
+        if next_match_idx >= 0 and matching_row is not None:
+            # add new rows until `section_name` (Segmentname) matches
+            while j < next_match_idx:
+                if j not in duplicate_indices:
                     result.append(
                         AhbRowComparison(
-                            previous_formatversion=_add_empty_row(new_row.formatversion),
+                            previous_formatversion=_add_empty_row(subsequent_ahb_rows[j].formatversion),
                             diff=AhbRowDiff(diff_type=DiffType.ADDED),
-                            subsequent_formatversion=new_row,
+                            subsequent_formatversion=subsequent_ahb_rows[j],
                         )
                     )
+                j += 1
 
-                # add matching rows with comparison
-                diff = _compare_ahb_rows(current_row, matching_row)
-                result.append(
-                    AhbRowComparison(
-                        previous_formatversion=current_row, diff=diff, subsequent_formatversion=matching_row
-                    )
+            # add matching rows with comparison
+            diff = _compare_ahb_rows(current_row, matching_row)
+            result.append(
+                AhbRowComparison(
+                    previous_formatversion=current_row,
+                    diff=diff,
+                    subsequent_formatversion=matching_row,
                 )
+            )
+            duplicate_indices.add(next_match_idx)
+            i += 1
+            j = next_match_idx + 1
 
-                i += 1
-                j = next_match_idx + 1
-
-            else:
-                # if no match found - label as REMOVED
-                result.append(
-                    AhbRowComparison(
-                        previous_formatversion=current_row,
-                        diff=AhbRowDiff(diff_type=DiffType.REMOVED),
-                        subsequent_formatversion=_add_empty_row(current_row.formatversion),
-                    )
+        else:
+            # if no match found - label as REMOVED
+            result.append(
+                AhbRowComparison(
+                    previous_formatversion=current_row,
+                    diff=AhbRowDiff(diff_type=DiffType.REMOVED),
+                    subsequent_formatversion=_add_empty_row(current_row.formatversion),
                 )
-                i += 1
+            )
+            i += 1
 
     return result

unittests/test_ahb_comparison.py

@@ -871,7 +871,7 @@ def test_align_rows(self) -> None:
         assert "segment_group_key" in str(result[8].diff.changed_entries)
         assert "segment_group_key" in str(result[9].diff.changed_entries)
 
-    def test_align_rows_multiple_entries_per_section_name(self) -> None:
+    def test_align_rows_multiple_rows_per_section_name(self) -> None:
         previous_ahb_rows = [
             AhbRow(
                 formatversion=self.formatversions.previous_formatversion,
@@ -883,21 +883,21 @@ def test_align_rows_multiple_entries_per_section_name(self) -> None:
             AhbRow(
                 formatversion=self.formatversions.previous_formatversion,
                 section_name="1",
-                segment_group_key="a",
+                segment_group_key="b",
                 value_pool_entry=None,
                 name=None,
             ),
             AhbRow(
                 formatversion=self.formatversions.previous_formatversion,
                 section_name="1",
-                segment_group_key="a",
+                segment_group_key="c",
                 value_pool_entry=None,
                 name=None,
             ),
             AhbRow(
                 formatversion=self.formatversions.previous_formatversion,
                 section_name="2",
-                segment_group_key="b",
+                segment_group_key="a",
                 value_pool_entry=None,
                 name=None,
             ),
@@ -911,7 +911,7 @@ def test_align_rows_multiple_entries_per_section_name(self) -> None:
             AhbRow(
                 formatversion=self.formatversions.previous_formatversion,
                 section_name="2",
-                segment_group_key="b",
+                segment_group_key="c",
                 value_pool_entry=None,
                 name=None,
             ),
@@ -927,14 +927,14 @@ def test_align_rows_multiple_entries_per_section_name(self) -> None:
             AhbRow(
                 formatversion=self.formatversions.subsequent_formatversion,
                 section_name="1",
-                segment_group_key="a",
+                segment_group_key="b",
                 value_pool_entry=None,
                 name=None,
             ),
             AhbRow(
                 formatversion=self.formatversions.subsequent_formatversion,
                 section_name="1",
-                segment_group_key="x",
+                segment_group_key="c",
                 value_pool_entry=None,
                 name=None,
             ),
@@ -955,7 +955,7 @@ def test_align_rows_multiple_entries_per_section_name(self) -> None:
             AhbRow(
                 formatversion=self.formatversions.subsequent_formatversion,
                 section_name="2",
-                segment_group_key="x",
+                segment_group_key="c",
                 value_pool_entry=None,
                 name=None,
             ),
@@ -966,17 +966,122 @@ def test_align_rows_multiple_entries_per_section_name(self) -> None:
         assert len(result) == 6
         assert result[0].diff.diff_type == DiffType.MODIFIED
         assert result[1].diff.diff_type == DiffType.UNCHANGED
-        assert result[2].diff.diff_type == DiffType.MODIFIED
+        assert result[2].diff.diff_type == DiffType.UNCHANGED
         assert result[3].diff.diff_type == DiffType.MODIFIED
         assert result[4].diff.diff_type == DiffType.UNCHANGED
-        assert result[5].diff.diff_type == DiffType.MODIFIED
+        assert result[5].diff.diff_type == DiffType.UNCHANGED
 
         assert "segment_group_key" in str(result[0].diff.changed_entries)
         assert not result[1].diff.changed_entries
-        assert "segment_group_key" in str(result[2].diff.changed_entries)
+        assert not result[2].diff.changed_entries
         assert "segment_group_key" in str(result[3].diff.changed_entries)
         assert not result[4].diff.changed_entries
-        assert "segment_group_key" in str(result[5].diff.changed_entries)
+        assert not result[5].diff.changed_entries
+
+    def test_align_rows_all_ahb_properties_within_section_name(self) -> None:
+        # to handle rows that have been inserted at arbitrary positions within the same `section_name` group
+        # for example FV2504 UTILMD 55078 (SG3 Ansprechpartner):
+        # `previous_ahb_rows` should add an empty row at the top (NEW) for AhbRow's to align properly
+        previous_ahb_rows = [
+            AhbRow(  # equivalent to second AhbRow of `subsequent_ahb_rows`
+                formatversion=self.formatversions.previous_formatversion,
+                section_name="Ansprechpartner",
+                segment_group_key="SG3",
+                segment_code="CTA",
+                data_element=None,
+                segment_id="00009",
+                value_pool_entry=None,
+                name=None,
+                ahb_expression="Muss",
+                conditions=None,
+            ),
+            AhbRow(
+                formatversion=self.formatversions.previous_formatversion,
+                section_name="Ansprechpartner",
+                segment_group_key="SG3",
+                segment_code="CTA",
+                data_element="3139",
+                segment_id="00009",
+                value_pool_entry="IC",
+                name="Informationskontakt",
+                ahb_expression="X",
+                conditions=None,
+            ),
+            AhbRow(
+                formatversion=self.formatversions.previous_formatversion,
+                section_name="Ansprechpartner",
+                segment_group_key="SG3",
+                segment_code="CTA",
+                data_element="3412",
+                segment_id="00009",
+                value_pool_entry=None,
+                name="Name vom Ansprechpartner",
+                ahb_expression="X",
+                conditions=None,
+            ),
+        ]
+        subsequent_ahb_rows = [
+            AhbRow(  # NEW
+                formatversion=self.formatversions.subsequent_formatversion,
+                section_name="Ansprechpartner",
+                segment_group_key="SG3",
+                segment_code=None,
+                data_element=None,
+                segment_id=None,
+                value_pool_entry=None,
+                name=None,
+                ahb_expression="Kann",
+                conditions=None,
+            ),
+            AhbRow(  # equivalent to first AhbRow of `previous_ahb_rows`
+                formatversion=self.formatversions.subsequent_formatversion,
+                section_name="Ansprechpartner",
+                segment_group_key="SG3",
+                segment_code="CTA",
+                data_element=None,
+                segment_id="00009",
+                value_pool_entry=None,
+                name=None,
+                ahb_expression="Muss",
+                conditions=None,
+            ),
+            AhbRow(
+                formatversion=self.formatversions.subsequent_formatversion,
+                section_name="Ansprechpartner",
+                segment_group_key="SG3",
+                segment_code="CTA",
+                data_element="3139",
+                segment_id="00009",
+                value_pool_entry="IC",
+                name="Informationskontakt",
+                ahb_expression="X",
+                conditions=None,
+            ),
+            AhbRow(
+                formatversion=self.formatversions.subsequent_formatversion,
+                section_name="Ansprechpartner",
+                segment_group_key="SG3",
+                segment_code="CTA",
+                data_element="3412",
+                segment_id="00009",
+                value_pool_entry=None,
+                name="Name vom Ansprechpartner",
+                ahb_expression="X",
+                conditions=None,
+            ),
+        ]
+
+        result = align_ahb_rows(previous_ahb_rows, subsequent_ahb_rows)
+
+        assert len(result) == 4
+        assert result[0].diff.diff_type == DiffType.ADDED
+        assert result[1].diff.diff_type == DiffType.UNCHANGED
+        assert result[2].diff.diff_type == DiffType.UNCHANGED
+        assert result[3].diff.diff_type == DiffType.UNCHANGED
+
+        assert not result[1].diff.changed_entries
+        assert not result[2].diff.changed_entries
+        assert not result[3].diff.changed_entries
 
     def test_align_rows_different_column_sets(self) -> None:
         previous_ahb_rows = [
@@ -1019,12 +1124,10 @@ def test_align_rows_different_column_sets(self) -> None:
         result = align_ahb_rows(previous_ahb_rows, subsequent_ahb_rows)
 
         assert len(result) == 3
-        # Check sequence: removed, modified, added
         assert result[0].diff.diff_type == DiffType.REMOVED
         assert result[1].diff.diff_type == DiffType.MODIFIED
         assert result[2].diff.diff_type == DiffType.ADDED
 
-        # Verify specific changes in the modified row
         changed_entries = str(result[1].diff.changed_entries)
         assert "data_element" in changed_entries
         assert "value_pool_entry" in changed_entries