Reduce number of locations where warnings are raised. (#278)

docs/source/changes.md

@@ -11,6 +11,8 @@ releases are available on [PyPI](https://pypi.org/project/pytask) and
 - {pull}`277` ignores `DeprecationWarning` and `PendingDeprecationWarning` by default.
   Previously, they were enabled, but they should be shown when testing the project with
   pytest, not after the execution with pytask. Fixes {issue}`269`.
+- {pull}`278` counts multiple occurrences of a warning instead of listing the module or
+  task name again and again. Fixes {issue}`270`.
 
 ## 0.2.2 - 2022-05-14
 

src/_pytask/warnings.py

@@ -249,7 +249,9 @@ def pytask_log_session_footer(session: Session) -> None:
                 grouped_warnings[warning.message].append(location)
             sorted_gw = {k: sorted(v) for k, v in grouped_warnings.items()}
 
-            renderable = MyRenderable(sorted_gw)
+            reduced_gw = _reduce_grouped_warnings(sorted_gw)
+
+            renderable = MyRenderable(reduced_gw)
 
             panel = Panel(renderable, title="Warnings", style="warning")
             console.print(panel)
@@ -271,3 +273,19 @@ def __rich_console__(
             "[bold red]♥[/bold red] "
             + "https://pytask-dev.rtdf.io/en/stable/how_to_guides/capture_warnings.html"
         )
+
+
+def _reduce_grouped_warnings(
+    grouped_warnings: dict[str, list[str]], max_locations: int = 5
+) -> dict[str, list[str]]:
+    """Reduce grouped warnings."""
+    reduced_gw = {}
+    for message, locations in grouped_warnings.items():
+        if len(locations) > max_locations:
+            adjusted_locations = locations[:max_locations]
+            n_more_locations = len(locations[max_locations:])
+            adjusted_locations.append(f"... in {n_more_locations} more locations.")
+        else:
+            adjusted_locations = locations
+        reduced_gw[message] = adjusted_locations
+    return reduced_gw

tests/test_warnings.py

@@ -155,3 +155,24 @@ def warn_now():
     assert result.returncode == ExitCode.OK
     assert "Warnings" not in result.stdout.decode()
     assert "warning!!!" not in result.stdout.decode()
+
+
+def test_multiple_occurrences_of_warning_are_reduced(tmp_path, runner):
+    source = """
+    import warnings
+    import pytask
+
+    for i in range(10):
+
+        @pytask.mark.task
+        def task_example():
+            warnings.warn("warning!!!")
+    """
+    tmp_path.joinpath("task_example.py").write_text(textwrap.dedent(source))
+
+    result = runner.invoke(cli, [tmp_path.as_posix()])
+
+    assert result.exit_code == ExitCode.OK
+    assert "Warnings" in result.output
+    assert "warning!!!" in result.output
+    assert result.output.count("task_example") == 31