add channels to find

grafana_wtf/core.py

@@ -121,6 +121,7 @@ def start_progressbar(self, total):
     def scan_common(self):
         self.scan_dashboards()
         self.scan_datasources()
+        self.scan_notifications()
 
     def scan_all(self):
         self.scan_common()
@@ -155,7 +156,8 @@ def scan_snapshots(self):
         self.data.snapshots = self.grafana.snapshots.get_dashboard_snapshots()
 
     def scan_notifications(self):
-        self.data.notifications = self.grafana.notifications.lookup_channels()
+        self.data.notifications = munchify(self.grafana.notifications.lookup_channels())
+        return self.data.notifications
 
     def scan_datasources(self):
         log.info("Scanning datasources")
@@ -326,7 +328,7 @@ def dashboard_details(self):
     def search(self, expression):
         log.info('Searching Grafana at "{}" for expression "{}"'.format(self.grafana_url, expression))
 
-        results = Munch(datasources=[], dashboard_list=[], dashboards=[])
+        results = Munch(datasources=[], dashboard_list=[], dashboards=[], notifications=[])
 
         # Check datasources
         log.info("Searching data sources")
@@ -336,6 +338,10 @@ def search(self, expression):
         log.info("Searching dashboards")
         self.search_items(expression, self.data.dashboards, results.dashboards)
 
+        # Check notification channels
+        log.info("Searching notification channels")
+        self.search_items(expression, self.data.notifications, results.notifications)
+
         return results
 
     def replace(self, expression, replacement, dry_run: bool = False):
@@ -562,6 +568,7 @@ def __init__(self, engine: GrafanaWtf):
         # Gather all data.
         self.dashboards = self.engine.scan_dashboards()
         self.datasources = self.engine.scan_datasources()
+        self.notifications = self.engine.scan_notifications()
 
         # Invoke indexer.
         self.index()

grafana_wtf/report/data.py

@@ -44,5 +44,7 @@ def display(self, expression, result):
             ),
             datasources=self.get_output_items("Datasource", result.datasources, self.compute_url_datasource),
             dashboards=self.get_output_items("Dashboard", result.dashboards, self.compute_url_dashboard),
+            notifications=self.get_output_items("Notifications channel", result.notifications,
+                                                self.compute_url_notifications),
         )
         output_results(self.format, output)

grafana_wtf/report/textual.py

@@ -23,9 +23,10 @@ def display(self, expression, result):
         print('Searching for expression "{}" at Grafana instance {}'.format(_m(expression), self.grafana_url))
         self.output_items("Data Sources", result.datasources, self.compute_url_datasource)
         self.output_items("Dashboards", result.dashboards, self.compute_url_dashboard)
+        self.output_items("Notification channels", result.notifications, self.compute_url_notifications)
 
     def output_items(self, label, items, url_callback):
-        # Output section name (data source vs. dashboard).
+        # Output section name (data source, dashboard or notification channel).
         hits = len(items)
         print("=" * 42)
         print(f"{_s(label)}: {_m(hits)} hits.")
@@ -42,7 +43,7 @@ def output_items(self, label, items, url_callback):
 
             # Output match title / entity name.
             name = self.get_item_name(item)
-            section = f"Dashboard »{name}«"
+            section = f"{_s(label)[:-1]} »{name}«"
             print(_ssb(section))
             print("=" * len(section))
 
@@ -168,11 +169,14 @@ def format_where(self, item):
         return answer
 
     def compute_url_datasource(self, datasource):
-        return urljoin(self.grafana_url, "/datasources/edit/{}".format(datasource.data.id))
+        return urljoin(self.grafana_url, f"/datasources/edit/{datasource.data.id}")
 
     def compute_url_dashboard(self, dashboard):
         return urljoin(self.grafana_url, dashboard.data.meta.url)
 
+    def compute_url_notifications(self, notifications):
+        return urljoin(self.grafana_url, f"/alerting/notification/{notifications.data.id}/edit")
+
     def experimental(self):
         # print(match)
         # print(dir(match.context))

tests/test_commands.py

@@ -157,7 +157,7 @@ def test_find_format_json(ldi_resources, capsys):
 
     # Verify output.
     data = json.loads(captured.out)
-    assert len(data) == 3
+    assert len(data) == 4
 
 
 def test_find_format_yaml(ldi_resources, capsys):
@@ -171,7 +171,7 @@ def test_find_format_yaml(ldi_resources, capsys):
 
     # Verify output.
     data = yaml.safe_load(captured.out)
-    assert len(data) == 3
+    assert len(data) == 4
 
 
 def test_replace_dashboard_success(ldi_resources, capsys):