Files
zulip/zerver/webhooks/semaphore/tests.py
Abhinav 41fc7b2ae1 webhooks/semaphore: Add support for Semaphore 2.0 notifications.
Semaphore has currently has two different versions of their product -
Classic and 2.0. This commit adds support for Semaphore 2.0, along side
Semaphore Classic, using the same webhook. This would let the integration
work seamlessly for users who have already configured a Zulip integration in
their Semaphore 2.0 projects.

Semaphore 2.0 currently only supports GitHub and their payloads do not
contain URLs for common entities like commits, pull requests and tags. We
construct URLs for them using templates, but also try to support other
services by providing notifications without URLs.

Closes #14171

Co-authored-by: Puneeth Chaganti <punchagan@muse-amuse.in>
2020-04-09 12:41:40 -07:00

123 lines
6.4 KiB
Python

# -*- coding: utf-8 -*-
import ujson
from mock import patch
from zerver.lib.test_classes import WebhookTestCase
class SemaphoreHookTests(WebhookTestCase):
STREAM_NAME = 'semaphore'
URL_TEMPLATE = "/api/v1/external/semaphore?stream={stream}&api_key={api_key}"
# Messages are generated by Semaphore on git push. The subject lines below
# contain information on the repo and branch, and the message has links and
# details about the build, deploy, server, author, and commit
# Tests for Semaphore Classic
def test_semaphore_build(self) -> None:
expected_topic = u"knighthood/master" # repo/branch
expected_message = """
[Build 314](https://semaphoreci.com/donquixote/knighthood/branches/master/builds/314) passed:
* **Commit**: [a490b8d: Create user account for Rocinante](https://github.com/donquixote/knighthood/commit/a490b8d508ebbdab1d77a5c2aefa35ceb2d62daf)
* **Author**: don@lamancha.com
""".strip()
self.send_and_test_stream_message('build', expected_topic, expected_message,
content_type="application/x-www-form-urlencoded")
def test_semaphore_deploy(self) -> None:
expected_topic = u"knighthood/master"
expected_message = """
[Deploy 17](https://semaphoreci.com/donquixote/knighthood/servers/lamancha-271/deploys/17) of [build 314](https://semaphoreci.com/donquixote/knighthood/branches/master/builds/314) passed:
* **Commit**: [a490b8d: Create user account for Rocinante](https://github.com/donquixote/knighthood/commit/a490b8d508ebbdab1d77a5c2aefa35ceb2d62daf)
* **Author**: don@lamancha.com
* **Server**: lamancha-271
""".strip()
self.send_and_test_stream_message('deploy', expected_topic, expected_message,
content_type="application/x-www-form-urlencoded")
# Tests For Semaphore 2.0
def test_semaphore2_push(self) -> None:
expected_topic = u"notifications/rw/webhook_impl" # repo/branch
expected_message = """
[Notifications](https://semaphore.semaphoreci.com/workflows/acabe58e-4bcc-4d39-be06-e98d71917703) pipeline **stopped**:
* **Commit**: [(2d9f5fc)](https://github.com/renderedtext/notifications/commit/2d9f5fcec1ca7c68fa7bd44dd58ec4ff65814563) Implement webhooks for SemaphoreCI
* **Branch**: rw/webhook_impl
* **Author**: [radwo](https://github.com/radwo)
""".strip()
self.send_and_test_stream_message('push', expected_topic, expected_message,
content_type="application/json")
def test_semaphore2_push_non_gh_repo(self) -> None:
expected_topic = u"notifications/rw/webhook_impl" # repo/branch
expected_message = """
[Notifications](https://semaphore.semaphoreci.com/workflows/acabe58e-4bcc-4d39-be06-e98d71917703) pipeline **stopped**:
* **Commit**: (2d9f5fc) Implement webhooks for SemaphoreCI
* **Branch**: rw/webhook_impl
* **Author**: radwo
""".strip()
with patch('zerver.webhooks.semaphore.view.is_github_repo', return_value=False):
self.send_and_test_stream_message('push', expected_topic, expected_message,
content_type="application/json")
def test_semaphore_pull_request(self) -> None:
expected_topic = u"notifications/test-notifications"
expected_message = """
[Notifications](https://semaphore.semaphoreci.com/workflows/84383f37-d025-4811-b719-61c6acc92a1e) pipeline **failed**:
* **Pull Request**: [Testing PR notifications](https://github.com/renderedtext/notifications/pull/3)
* **Branch**: test-notifications
* **Author**: [radwo](https://github.com/radwo)
""".strip()
self.send_and_test_stream_message('pull_request', expected_topic, expected_message,
content_type="application/json")
def test_semaphore_pull_request_non_gh_repo(self) -> None:
expected_topic = u"notifications/test-notifications"
expected_message = """
[Notifications](https://semaphore.semaphoreci.com/workflows/84383f37-d025-4811-b719-61c6acc92a1e) pipeline **failed**:
* **Pull Request**: Testing PR notifications (#3)
* **Branch**: test-notifications
* **Author**: radwo
""".strip()
with patch('zerver.webhooks.semaphore.view.is_github_repo', return_value=False):
self.send_and_test_stream_message('pull_request', expected_topic, expected_message,
content_type="application/json")
def test_semaphore_tag(self) -> None:
expected_topic = u"notifications"
expected_message = """
[Notifications](https://semaphore.semaphoreci.com/workflows/a8704319-2422-4828-9b11-6b2afa3554e6) pipeline **stopped**:
* **Tag**: [v1.0.1](https://github.com/renderedtext/notifications/tree/v1.0.1)
* **Author**: [radwo](https://github.com/radwo)
""".strip()
self.send_and_test_stream_message('tag', expected_topic, expected_message,
content_type="application/json")
def test_semaphore_tag_non_gh_repo(self) -> None:
expected_topic = u"notifications"
expected_message = """
[Notifications](https://semaphore.semaphoreci.com/workflows/a8704319-2422-4828-9b11-6b2afa3554e6) pipeline **stopped**:
* **Tag**: v1.0.1
* **Author**: radwo
""".strip()
with patch('zerver.webhooks.semaphore.view.is_github_repo', return_value=False):
self.send_and_test_stream_message('tag', expected_topic, expected_message,
content_type="application/json")
def test_semaphore_unknown_event(self) -> None:
expected_topic = u"notifications"
expected_message = """
[Notifications](https://semaphore.semaphoreci.com/workflows/a8704319-2422-4828-9b11-6b2afa3554e6) pipeline **stopped** for unknown event
""".strip()
with patch('zerver.webhooks.semaphore.tests.SemaphoreHookTests.get_body', self.get_unknown_event):
self.send_and_test_stream_message('tag', expected_topic, expected_message,
content_type="application/json")
def get_body(self, fixture_name: str) -> str:
return self.webhook_fixture_data("semaphore", fixture_name, file_type="json")
def get_unknown_event(self, fixture_name: str) -> str:
"""Return modified payload with revision.reference_type changed"""
fixture_data = ujson.loads(self.webhook_fixture_data("semaphore", fixture_name, file_type="json"))
fixture_data['revision']['reference_type'] = 'unknown'
return fixture_data