add initial implementation notifications squashing

taiga/projects/notifications/squashing.py

@@ -0,0 +1,81 @@
+from collections import namedtuple
+
+
+HistoryEntry = namedtuple('HistoryEntry', 'comment values_diff')
+
+
+# These fields are ignored
+
+EXCLUDED_FIELDS = (
+    'description',
+    'description_html',
+    'blocked_note',
+    'blocked_note_html',
+    'content',
+    'content_html',
+    'epics_order',
+    'backlog_order',
+    'kanban_order',
+    'sprint_order',
+    'taskboard_order',
+    'us_order',
+    'custom_attributes',
+    'tribe_gig',
+)
+
+# These fields can't be squashed because we don't have
+# a squashing algorithm yet.
+
+NON_SQUASHABLE_FIELDS = (
+    'points',
+    'attachments',
+    'tags',
+    'watchers',
+    'description_diff',
+    'content_diff',
+    'blocked_note_diff',
+    'custom_attributes',
+)
+
+
+def is_squashable(field):
+    return field not in EXCLUDED_FIELDS and field not in NON_SQUASHABLE_FIELDS
+
+
+def summary(field, entries):
+    """
+    Given an iterable of HistoryEntry of the same type return a summarized list.
+    """
+    if len(entries) <= 1:
+        return entries
+
+    initial = entries[0].values_diff[field]
+    final = entries[-1].values_diff[field]
+    from_, to = initial[0], final[1]
+    return [] if from_ == to else [HistoryEntry('', {field: [from_, to]})]
+
+
+def squash_history_entries(history_entries):
+    """
+    Given an iterable of HistoryEntry, squash them summarizing entries that have
+    a squashable algorithm available.
+    """
+    history_entries = (HistoryEntry(entry.comment, entry.values_diff) for entry in history_entries)
+    from collections import OrderedDict
+    grouped = OrderedDict()
+    for entry in history_entries:
+        if entry.comment:
+            yield entry
+            continue
+
+        for field, diff in entry.values_diff.items():
+            if is_squashable(field):
+                grouped.setdefault(field, [])
+                grouped[field].append(HistoryEntry('', {field: diff}))
+            else:
+                yield HistoryEntry('', {field: diff})
+
+    for field, entries in grouped.items():
+        squashed = summary(field, entries)
+        for entry in squashed:
+            yield entry

tests/unit/test_notifications_squashing.py

@@ -0,0 +1,86 @@
+from taiga.projects.notifications import squashing
+
+
+def assert_(expected, squashed, *, ordered=True):
+    """
+    Check if expected entries are the same as the squashed.
+
+    Allow to specify if they must maintain the order or conversely they can
+    appear in any order.
+    """
+    squashed = list(squashed)
+    assert len(expected) == len(squashed)
+    if ordered:
+        assert expected == squashed
+    else:
+        # Can't use a set, just check all of the squashed entries
+        # are in the expected ones.
+        for entry in squashed:
+            assert entry in expected
+
+
+def test_squash_omits_comments():
+    history_entries = [
+        squashing.HistoryEntry(comment='A', values_diff={'status': ['A', 'B']}),
+        squashing.HistoryEntry(comment='B', values_diff={'status': ['B', 'C']}),
+        squashing.HistoryEntry(comment='C', values_diff={'status': ['C', 'B']}),
+    ]
+    squashed = squashing.squash_history_entries(history_entries)
+    assert_(history_entries, squashed)
+
+
+def test_squash_allowed_grouped_at_the_end():
+    history_entries = [
+        squashing.HistoryEntry(comment='A', values_diff={}),
+        squashing.HistoryEntry(comment='', values_diff={'status': ['A', 'B']}),
+        squashing.HistoryEntry(comment='', values_diff={'status': ['B', 'C']}),
+        squashing.HistoryEntry(comment='', values_diff={'status': ['C', 'D']}),
+        squashing.HistoryEntry(comment='', values_diff={'status': ['D', 'C']}),
+        squashing.HistoryEntry(comment='Z', values_diff={}),
+    ]
+    expected = [
+        squashing.HistoryEntry(comment='A', values_diff={}),
+        squashing.HistoryEntry(comment='Z', values_diff={}),
+        squashing.HistoryEntry(comment='', values_diff={'status': ['A', 'C']}),
+    ]
+
+    squashed = squashing.squash_history_entries(history_entries)
+    assert_(expected, squashed)
+
+
+def test_squash_remove_noop_changes():
+    history_entries = [
+        squashing.HistoryEntry(comment='', values_diff={'status': ['A', 'B']}),
+        squashing.HistoryEntry(comment='', values_diff={'status': ['B', 'A']}),
+    ]
+    expected = []
+
+    squashed = squashing.squash_history_entries(history_entries)
+    assert_(expected, squashed)
+
+
+def test_squash_remove_noop_changes_but_maintain_others():
+    history_entries = [
+        squashing.HistoryEntry(comment='', values_diff={'status': ['A', 'B'], 'type': ['1', '2']}),
+        squashing.HistoryEntry(comment='', values_diff={'status': ['B', 'A']}),
+    ]
+    expected = [
+        squashing.HistoryEntry(comment='', values_diff={'type': ['1', '2']}),
+    ]
+
+    squashed = squashing.squash_history_entries(history_entries)
+    assert_(expected, squashed)
+
+
+def test_squash_values_diff_with_multiple_fields():
+    history_entries = [
+        squashing.HistoryEntry(comment='', values_diff={'status': ['A', 'B'], 'type': ['1', '2']}),
+        squashing.HistoryEntry(comment='', values_diff={'status': ['B', 'C']}),
+    ]
+    expected = [
+        squashing.HistoryEntry(comment='', values_diff={'type': ['1', '2']}),
+        squashing.HistoryEntry(comment='', values_diff={'status': ['A', 'C']}),
+    ]
+
+    squashed = squashing.squash_history_entries(history_entries)
+    assert_(expected, squashed, ordered=False)