healthchecks/hc/api/tests/test_channel_model.py

import json

from hc.api.models import Channel
from hc.test import BaseTestCase


class ChannelModelTestCase(BaseTestCase):
    def test_webhook_spec_handles_plain_single_address(self):
        c = Channel(kind="webhook")
        c.value = "http://example.org"
        self.assertEqual(
            c.down_webhook_spec,
            {"method": "GET", "url": "http://example.org", "body": "", "headers": {}},
        )

        self.assertEqual(
            c.up_webhook_spec, {"method": "GET", "url": "", "body": "", "headers": {}}
        )

    def test_webhook_spec_handles_plain_pair(self):
        c = Channel(kind="webhook")
        c.value = "http://example.org\nhttp://example.com/up/"
        self.assertEqual(
            c.down_webhook_spec,
            {"method": "GET", "url": "http://example.org", "body": "", "headers": {}},
        )

        self.assertEqual(
            c.up_webhook_spec,
            {
                "method": "GET",
                "url": "http://example.com/up/",
                "body": "",
                "headers": {},
            },
        )

    def test_webhook_spec_handles_plain_post(self):
        c = Channel(kind="webhook")
        c.value = "http://example.org\n\nhello world"
        self.assertEqual(
            c.down_webhook_spec,
            {
                "method": "POST",
                "url": "http://example.org",
                "body": "hello world",
                "headers": {},
            },
        )

        self.assertEqual(
            c.up_webhook_spec,
            {"method": "POST", "url": "", "body": "hello world", "headers": {}},
        )

    def test_webhook_spec_handles_legacy_get(self):
        c = Channel(kind="webhook")
        c.value = json.dumps(
            {
                "url_down": "http://example.org",
                "url_up": "http://example.org/up/",
                "headers": {"X-Name": "value"},
                "post_data": "",
            }
        )

        self.assertEqual(
            c.down_webhook_spec,
            {
                "method": "GET",
                "url": "http://example.org",
                "body": "",
                "headers": {"X-Name": "value"},
            },
        )

        self.assertEqual(
            c.up_webhook_spec,
            {
                "method": "GET",
                "url": "http://example.org/up/",
                "body": "",
                "headers": {"X-Name": "value"},
            },
        )

    def test_webhook_spec_handles_legacy_post(self):
        c = Channel(kind="webhook")
        c.value = json.dumps(
            {
                "url_down": "http://example.org",
                "url_up": "http://example.org/up/",
                "headers": {"X-Name": "value"},
                "post_data": "hello world",
            }
        )

        self.assertEqual(
            c.down_webhook_spec,
            {
                "method": "POST",
                "url": "http://example.org",
                "body": "hello world",
                "headers": {"X-Name": "value"},
            },
        )

        self.assertEqual(
            c.up_webhook_spec,
            {
                "method": "POST",
                "url": "http://example.org/up/",
                "body": "hello world",
                "headers": {"X-Name": "value"},
            },
        )

    def test_webhook_spec_handles_mixed(self):
        c = Channel(kind="webhook")
        c.value = json.dumps(
            {
                "method_down": "GET",
                "url_down": "http://example.org",
                "body_down": "",
                "headers_down": {"X-Status": "X"},
                "method_up": "POST",
                "url_up": "http://example.org/up/",
                "body_up": "hello world",
                "headers_up": {"X-Status": "OK"},
            }
        )

        self.assertEqual(
            c.down_webhook_spec,
            {
                "method": "GET",
                "url": "http://example.org",
                "body": "",
                "headers": {"X-Status": "X"},
            },
        )

        self.assertEqual(
            c.up_webhook_spec,
            {
                "method": "POST",
                "url": "http://example.org/up/",
                "body": "hello world",
                "headers": {"X-Status": "OK"},
            },
        )

    def test_it_handles_legacy_opsgenie_value(self):
        c = Channel(kind="opsgenie", value="foo123")
        self.assertEqual(c.opsgenie_key, "foo123")
        self.assertEqual(c.opsgenie_region, "us")

    def test_it_handles_json_opsgenie_value(self):
        c = Channel(kind="opsgenie")
        c.value = json.dumps({"key": "abc", "region": "eu"})
        self.assertEqual(c.opsgenie_key, "abc")
        self.assertEqual(c.opsgenie_region, "eu")