adding support to victorops

.pre-commit-config.yaml

@@ -7,6 +7,9 @@ repos:
       - id: pretty-format-json
         args: ['--autofix']
       - id: trailing-whitespace
+  - repo: https://gitlab.com/pycqa/flake8
+    rev: 3.9.2
+    hooks:
       - id: flake8
         args: ['--max-line-length=120']
         additional_dependencies: ['flake8-mutable', 'flake8-comprehensions']

notifiers/providers/__init__.py

@@ -13,6 +13,7 @@
 from . import statuspage
 from . import telegram
 from . import twilio
+from . import victorops
 from . import zulip
 
 _all_providers = {
@@ -32,4 +33,5 @@
     "mailgun": mailgun.MailGun,
     "popcornnotify": popcornnotify.PopcornNotify,
     "statuspage": statuspage.Statuspage,
+    "victorops": victorops.VictorOps,
 }

notifiers/providers/victorops.py

@@ -0,0 +1,90 @@
+from ..core import Provider
+from ..core import Response
+from ..utils import requests
+
+
+class VictorOps(Provider):
+    """Send VictorOps webhook notifications"""
+
+    base_url = "https://portal.victorops.com/ui/{ORGANIZATION_ID}/incidents"
+    site_url = "https://portal.victorops.com/dash/{ORGANIZATION_ID}#/advanced/rest"
+    name = "victorops"
+
+    _required = {
+        "required": [
+            "rest_url",
+            "message_type",
+            "entity_id",
+            "entity_display_name",
+            "message",
+        ]
+    }
+    _schema = {
+        "type": "object",
+        "properties": {
+            "rest_url": {
+                "type": "string",
+                "format": "uri",
+                "title": "the REST URL to use with routing_key. create one in victorops `integrations` tab.",
+            },
+            "message_type": {
+                "type": "string",
+                "title": "severity level can be: "
+                "- critical or warning: Triggers an incident "
+                "- acknowledgement: sends Acknowledgment to an incident "
+                "- info: Creates a timeline event but doesn't trigger an incident "
+                "- recovery or ok: Resolves an incident",
+                "enum": [
+                    "critical",
+                    "warning",
+                    "acknowledgement",
+                    "info",
+                    "recovery",
+                    "ok",
+                ],
+            },
+            "entity_id": {
+                "type": "string",
+                "title": "Unique id for the incident for aggregation ,Acknowledging, or resolving.",
+            },
+            "entity_display_name": {
+                "type": "string",
+                "title": "Display Name in the UI and Notifications.",
+            },
+            "message": {
+                "type": "string",
+                "title": "This is the description that will be posted in the incident.",
+            },
+            "annotations": {
+                "type": "object",
+                "patternProperties": {
+                    "vo_annotate.u.": {"type": "string"},
+                    "vo_annotate.s.": {"type": "string"},
+                    "vo_annotate.i.": {"type": "string"},
+                },
+                "minProperties": 1,
+                "title": "annotations can be of three types vo_annotate.u.Runbook, vo_annotate.s.Note, "
+                "vo_annotate.i.image.",
+            },
+            "additional_keys": {
+                "type": "object",
+                "title": "any additional keys that can be passed in the body",
+            },
+        },
+        "additionalProperties": False,
+    }
+
+    def _prepare_data(self, data: dict) -> dict:
+        annotations = data.pop("annotations", {})
+        for annotation, value in annotations.items():
+            data[annotation] = value
+
+        additional_keys = data.pop("additional_keys", {})
+        for additional_key, value in additional_keys.items():
+            data[additional_key] = value
+        return data
+
+    def _send_notification(self, data: dict) -> Response:
+        url = data.pop("rest_url")
+        response, errors = requests.post(url, json=data)
+        return self.create_response(data, response, errors)

source/CLI.rst

@@ -34,14 +34,15 @@ To view the main help just enter ``notifiers`` or ``notifiers --help``:
       slack       Options for 'slack'
       telegram    Options for 'telegram'
       zulip       Options for 'zulip'
+      victorops       Options for 'victorops'
 
 
 To view all providers use the ``providers`` command like so:
 
 .. code-block:: console
 
      $ notifiers providers
-     pushover, simplepush, slack, email, gmail, telegram, gitter, pushbullet, join, hipchat, zulip
+     pushover, simplepush, slack, email, gmail, telegram, gitter, pushbullet, join, hipchat, zulip, victorops
 
 This will return all available provider names
 

source/api/providers.rst

@@ -66,3 +66,8 @@ API documentation for the different providers.
    :members:
    :undoc-members:
 
+.. automodule:: notifiers.providers.victorops
+   :members:
+   :undoc-members:
+
+

source/providers/victorops.rst

@@ -0,0 +1,72 @@
+VictorOps (REST)
+--------------------
+
+Send `VictorOps <https://alert.victorops.com/integrations/generic>`_ rest integration notifications.
+
+Minimal example:
+
+.. code-block:: python
+
+    >>> from notifiers import get_notifier
+    >>> victorops = get_notifier('victorops')
+    >>> victorops.notify(rest_url='https://alert.victorops.com/integrations/generic/20104876/alert/f7dc2eeb-ms9k-43b8-kd89-0f00000f4ec2/$routing_key',
+                         message_type='CRITICAL',
+                         entity_id='foo testing',
+                         entity_display_name="bla test title text",
+                         message="bla message description")
+
+Full schema:
+
+.. code-block:: yaml
+
+    additionalProperties: false
+    properties:
+      rest_url:
+          type: string
+          format: uri
+          title: the REST URL to use with routing_key, create one in victorops integrations tab.
+
+      message_type:
+          type: string
+          title: severity level can be:
+                   - critical or warning: Triggers an incident
+                   - acknowledgement: Acks an incident
+                   - info: Creates a timeline event but doesn't trigger an incident
+                   - recovery or ok: Resolves an incident
+
+      entity_id:
+          type: string
+          title: Unique id for the incident for aggregation acking or resolving.
+
+      entity_display_name:
+          type: string
+          title: Display Name in the UI and Notifications.
+
+      message:
+          type: string
+          title: This is the description that will be posted in the incident.
+
+      annotations:
+          type: object
+          format:
+              vo_annotate.s.Note: annotation
+              vo_annotate.u.Runbook: annotation
+              vo_annotate.i.Graph: annotation
+          title: annotations can be of three types vo_annotate.u.Runbook vo_annotate.s.Note vo_annotate.i.image.
+
+      additional_keys:
+          type: object
+          format:
+              key: value
+              key: value
+              key: value
+          title: any additional keys that ca be passed in the body
+
+    required:
+      - rest_url
+      - message_type
+      - entity_id
+      - entity_display_name
+      - message
+    type: object
+

tests/providers/test_gitter.py

@@ -112,7 +112,7 @@ def test_gitter_rooms_positive(self, cli_runner):
 
     @pytest.mark.online
     def test_gitter_rooms_with_query(self, cli_runner):
-        cmd = f"gitter rooms --filter notifiers/testing".split()
+        cmd = "gitter rooms --filter notifiers/testing".split()
         result = cli_runner(cmd)
         assert not result.exit_code
         assert "notifiers/testing" in result.output

tests/providers/test_join.py

@@ -77,7 +77,7 @@ def test_join_devices_negative(self, cli_runner):
     @pytest.mark.skip("tests fail due to no device connected")
     @pytest.mark.online
     def test_join_updates_positive(self, cli_runner):
-        cmd = f"join devices".split()
+        cmd = "join devices".split()
         result = cli_runner(cmd)
         assert not result.exit_code
         replies = ["You have no devices associated with this apikey", "Device name: "]

tests/providers/test_telegram.py

@@ -100,7 +100,7 @@ def test_telegram_updates_negative(self, cli_runner):
     @pytest.mark.online
     @retry(AssertionError, tries=3, delay=10)
     def test_telegram_updates_positive(self, cli_runner):
-        cmd = f"telegram updates".split()
+        cmd = "telegram updates".split()
         result = cli_runner(cmd)
         assert not result.exit_code
         reply = json.loads(result.output)

tests/providers/test_victorops.py

@@ -0,0 +1,36 @@
+import os
+
+import pytest
+
+provider = "victorops"
+
+
+class TestVicrotops:
+    """
+    Victorops rest alert tests
+    Online test rely on setting the env variable VICTOROPS_REST_URL
+    """
+
+    @pytest.mark.online
+    def test_all_options(self, provider):
+        VICTOROPS_REST_URL = os.getenv("VICTOROPS_REST_URL")
+        data = {
+            "rest_url": VICTOROPS_REST_URL,
+            "message_type": "info",
+            "entity_id": "BA tesing",
+            "entity_display_name": "message test header",
+            "message": "text in body",
+            "annotations": {
+                "vo_annotate.i.Graph": "https://shorturl.at/dAQ28",
+                "vo_annotate.s.Note": "'You can't have everything. Where would you put it?' Steven Wright",
+                "vo_annotate.u.Runbook": "https://giphy.com/gifs/win-xNBcChLQt7s9a/fullscreen",
+            },
+            "additional_keys": {
+                "foo": "this is a custom fields",
+                "monitoring_tool": "this is an official victorops fields",
+            },
+        }
+        rsp = provider.notify(**data)
+        assert rsp.ok
+        assert rsp.status == "Success"
+        assert rsp.errors is None

tests/test_cli.py

@@ -93,7 +93,7 @@ def test_piping_input(self, cli_runner):
         ],
     )
     def test_environ(self, prefix, command, monkeypatch, cli_runner):
-        """Test provider environ usage """
+        """Test provider environ usage"""
         prefix = prefix if prefix else "NOTIFIERS_"
         monkeypatch.setenv(f"{prefix}MOCK_PROVIDER_REQUIRED", "foo")
         monkeypatch.setenv(f"{prefix}MOCK_PROVIDER_MESSAGE", "foo")

tests/test_core.py

@@ -129,7 +129,7 @@ def test_bad_integration(self, bad_provider):
 
     def test_environs(self, mock_provider, monkeypatch):
         """Test environs usage"""
-        prefix = f"mock_"
+        prefix = "mock_"
         monkeypatch.setenv(f"{prefix}{mock_provider.name}_required".upper(), "foo")
         rsp = mock_provider.notify(env_prefix=prefix)
         assert rsp.status == SUCCESS_STATUS
@@ -139,7 +139,7 @@ def test_provided_data_takes_precedence_over_environ(
         self, mock_provider, monkeypatch
     ):
         """Verify that given data overrides environ"""
-        prefix = f"mock_"
+        prefix = "mock_"
         monkeypatch.setenv(f"{prefix}{mock_provider.name}_required".upper(), "foo")
         rsp = mock_provider.notify(required="bar", env_prefix=prefix)
         assert rsp.status == SUCCESS_STATUS