Add ability to silence events via WarnErrorOptions

.changes/unreleased/Features-20240419-232030.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add ability to silence warnings via `WarnErrorOptions`
+time: 2024-04-19T23:20:30.014054-07:00
+custom:
+  Author: QMalcolm
+  Issue: "111"

dbt_common/events/functions.py

@@ -114,12 +114,13 @@ def msg_to_dict(msg: EventMsg) -> dict:
 
 
 def warn_or_error(event, node=None) -> None:
-    if WARN_ERROR or WARN_ERROR_OPTIONS.includes(type(event).__name__):
+    event_name = type(event).__name__
+    if WARN_ERROR or WARN_ERROR_OPTIONS.includes(event_name):
         # TODO: resolve this circular import when at top
         from dbt_common.exceptions import EventCompilationError
 
         raise EventCompilationError(event.message(), node)
-    else:
+    elif not WARN_ERROR_OPTIONS.silenced(event_name):
         fire_event(event)
 
 

dbt_common/helper_types.py

@@ -54,7 +54,7 @@ def __post_init__(self):
         if isinstance(self.exclude, list):
             self._validate_items(self.exclude)
 
-    def includes(self, item_name: str):
+    def includes(self, item_name: str) -> bool:
         return (
             item_name in self.include or self.include in self.INCLUDE_ALL
         ) and item_name not in self.exclude
@@ -69,10 +69,22 @@ def __init__(
         include: Union[str, List[str]],
         exclude: Optional[List[str]] = None,
         valid_error_names: Optional[Set[str]] = None,
+        silence: Optional[List[str]] = None,
     ):
+        self.silence = silence or []
         self._valid_error_names: Set[str] = valid_error_names or set()
         super().__init__(include=include, exclude=(exclude or []))
 
+    def __post_init__(self):
+        super().__post_init__()
+        self._validate_items(self.silence)
+
+    def includes(self, item_name: str) -> bool:
+        return super().includes(item_name) and not self.silenced(item_name)
+
+    def silenced(self, item_name: str) -> bool:
+        return item_name in self.silence
+
     def _validate_items(self, items: List[str]):
         for item in items:
             if item not in self._valid_error_names:

tests/unit/test_events.py

@@ -20,7 +20,14 @@
 def get_all_subclasses(cls):
     all_subclasses = []
     for subclass in cls.__subclasses__():
-        if subclass not in [TestLevel, DebugLevel, WarnLevel, InfoLevel, ErrorLevel, DynamicLevel]:
+        if subclass not in [
+            TestLevel,
+            DebugLevel,
+            WarnLevel,
+            InfoLevel,
+            ErrorLevel,
+            DynamicLevel,
+        ] and not subclass.__module__.startswith("test_"):
             all_subclasses.append(subclass)
         all_subclasses.extend(get_all_subclasses(subclass))
     return set(all_subclasses)

tests/unit/test_functions.py

@@ -0,0 +1,79 @@
+import pytest
+
+from dataclasses import dataclass, field
+from dbt_common.events import functions
+from dbt_common.events.base_types import EventLevel, EventMsg, WarnLevel
+from dbt_common.events.event_manager import EventManager
+from dbt_common.events.event_manager_client import ctx_set_event_manager
+from dbt_common.exceptions import EventCompilationError
+from dbt_common.helper_types import WarnErrorOptions
+from typing import List, Set
+
+
+# Re-implementing `Note` event as a warn event for
+# our testing purposes
+class Note(WarnLevel):
+    def code(self) -> str:
+        return "Z050"
+
+    def message(self) -> str:
+        return self.msg
+
+
+@dataclass
+class EventCatcher:
+    caught_events: List[EventMsg] = field(default_factory=list)
+
+    def catch(self, event: EventMsg) -> None:
+        self.caught_events.append(event)
+
+
+@pytest.fixture(scope="function")
+def event_catcher() -> EventCatcher:
+    return EventCatcher()
+
+
+@pytest.fixture(scope="function")
+def set_event_manager_with_catcher(event_catcher: EventCatcher) -> None:
+    event_manager = EventManager()
+    event_manager.callbacks.append(event_catcher.catch)
+    ctx_set_event_manager(event_manager)
+
+
+@pytest.fixture(scope="function")
+def valid_error_names() -> Set[str]:
+    return {Note.__name__}
+
+
+class TestWarnOrError:
+    def test_fires_error(self, valid_error_names: Set[str]):
+        functions.WARN_ERROR_OPTIONS = WarnErrorOptions(
+            include="*", valid_error_names=valid_error_names
+        )
+        with pytest.raises(EventCompilationError):
+            functions.warn_or_error(Note(msg="hi"))
+
+    def test_fires_warning(
+        self,
+        valid_error_names: Set[str],
+        event_catcher: EventCatcher,
+        set_event_manager_with_catcher,
+    ):
+        functions.WARN_ERROR_OPTIONS = WarnErrorOptions(
+            include="*", exclude=list(valid_error_names), valid_error_names=valid_error_names
+        )
+        functions.warn_or_error(Note(msg="hi"))
+        assert len(event_catcher.caught_events) == 1
+        assert event_catcher.caught_events[0].info.level == EventLevel.WARN.value
+
+    def test_silenced(
+        self,
+        valid_error_names: Set[str],
+        event_catcher: EventCatcher,
+        set_event_manager_with_catcher,
+    ):
+        functions.WARN_ERROR_OPTIONS = WarnErrorOptions(
+            include="*", silence=list(valid_error_names), valid_error_names=valid_error_names
+        )
+        functions.warn_or_error(Note(msg="hi"))
+        assert len(event_catcher.caught_events) == 0

tests/unit/test_helper_types.py

@@ -57,3 +57,38 @@ def test_init_valid_error(self):
         )
         assert warn_error_options.include == "*"
         assert warn_error_options.exclude == ["ValidError"]
+
+    def test_init_default_silence(self):
+        my_options = WarnErrorOptions(include="*")
+        assert my_options.silence == []
+
+    def test_init_invalid_silence_event(self):
+        with pytest.raises(ValidationError):
+            WarnErrorOptions(include="*", silence=["InvalidError"])
+
+    def test_init_valid_silence_event(self):
+        all_events = ["MySilencedEvent"]
+        my_options = WarnErrorOptions(
+            include="*", silence=all_events, valid_error_names=all_events
+        )
+        assert my_options.silence == all_events
+
+    @pytest.mark.parametrize(
+        "include,silence,expected_includes",
+        [
+            (["ItemA"], ["ItemA"], False),
+            ("*", ["ItemA"], False),
+            ("*", ["ItemB"], True),
+        ],
+    )
+    def test_includes(self, include, silence, expected_includes):
+        include_exclude = WarnErrorOptions(
+            include=include, silence=silence, valid_error_names={"ItemA", "ItemB"}
+        )
+
+        assert include_exclude.includes("ItemA") == expected_includes
+
+    def test_silenced(self):
+        my_options = WarnErrorOptions(include="*", silence=["ItemA"], valid_error_names={"ItemA"})
+        assert my_options.silenced("ItemA")
+        assert not my_options.silenced("ItemB")