test_plugin.py 5.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130
  1. from functools import cached_property
  2. from urllib.parse import parse_qs
  3. import responses
  4. from django.urls import reverse
  5. from sentry.models import Rule
  6. from sentry.plugins.base import Notification
  7. from sentry.testutils.cases import PluginTestCase
  8. from sentry.utils import json
  9. from sentry_plugins.pushover.plugin import PushoverPlugin
  10. SUCCESS = """{"status":1,"request":"e460545a8b333d0da2f3602aff3133d6"}"""
  11. class PushoverPluginTest(PluginTestCase):
  12. @cached_property
  13. def plugin(self):
  14. return PushoverPlugin()
  15. def test_conf_key(self):
  16. assert self.plugin.conf_key == "pushover"
  17. def test_entry_point(self):
  18. self.assertPluginInstalled("pushover", self.plugin)
  19. def test_is_configured(self):
  20. assert self.plugin.is_configured(self.project) is False
  21. self.plugin.set_option("apikey", "abcdef", self.project)
  22. assert self.plugin.is_configured(self.project) is False
  23. self.plugin.set_option("userkey", "abcdef", self.project)
  24. assert self.plugin.is_configured(self.project) is True
  25. @responses.activate
  26. def test_simple_notification(self):
  27. responses.add("POST", "https://api.pushover.net/1/messages.json", body=SUCCESS)
  28. self.plugin.set_option("userkey", "abcdef", self.project)
  29. self.plugin.set_option("apikey", "ghijkl", self.project)
  30. event = self.store_event(
  31. data={"message": "Hello world", "level": "warning"}, project_id=self.project.id
  32. )
  33. assert event.group is not None
  34. group = event.group
  35. rule = Rule.objects.create(project=self.project, label="my rule")
  36. notification = Notification(event=event, rule=rule)
  37. with self.options({"system.url-prefix": "http://example.com"}):
  38. self.plugin.notify(notification)
  39. request = responses.calls[0].request
  40. payload = parse_qs(request.body)
  41. assert payload == {
  42. "message": [f"{event.title}\n\nTags: level=warning"],
  43. "title": ["Bar: Hello world"],
  44. "url": [
  45. f"http://example.com/organizations/baz/issues/{group.id}/?referrer=pushover_plugin"
  46. ],
  47. "url_title": ["Issue Details"],
  48. "priority": ["0"],
  49. "user": ["abcdef"],
  50. "token": ["ghijkl"],
  51. "expire": ["90"],
  52. "retry": ["30"],
  53. }
  54. @responses.activate
  55. def test_emergency_notification(self):
  56. responses.add("POST", "https://api.pushover.net/1/messages.json", body=SUCCESS)
  57. self.plugin.set_option("userkey", "abcdef", self.project)
  58. self.plugin.set_option("apikey", "ghijkl", self.project)
  59. self.plugin.set_option("priority", "2", self.project)
  60. self.plugin.set_option("expire", 90, self.project)
  61. self.plugin.set_option("retry", 30, self.project)
  62. event = self.store_event(
  63. data={"message": "Hello world", "level": "warning"}, project_id=self.project.id
  64. )
  65. assert event.group is not None
  66. group = event.group
  67. rule = Rule.objects.create(project=self.project, label="my rule")
  68. notification = Notification(event=event, rule=rule)
  69. with self.options({"system.url-prefix": "http://example.com"}):
  70. self.plugin.notify(notification)
  71. request = responses.calls[0].request
  72. payload = parse_qs(request.body)
  73. assert payload == {
  74. "message": [f"{event.title}\n\nTags: level=warning"],
  75. "title": ["Bar: Hello world"],
  76. "url": [
  77. f"http://example.com/organizations/baz/issues/{group.id}/?referrer=pushover_plugin"
  78. ],
  79. "url_title": ["Issue Details"],
  80. "priority": ["2"],
  81. "user": ["abcdef"],
  82. "token": ["ghijkl"],
  83. "expire": ["90"],
  84. "retry": ["30"],
  85. }
  86. def test_no_secrets(self):
  87. self.user = self.create_user("foo@example.com")
  88. self.org = self.create_organization(owner=self.user, name="Rowdy Tiger")
  89. self.team = self.create_team(organization=self.org, name="Mariachi Band")
  90. self.project = self.create_project(organization=self.org, teams=[self.team], name="Bengal")
  91. self.login_as(self.user)
  92. self.plugin.set_option("userkey", "abcdef", self.project)
  93. self.plugin.set_option("apikey", "abcdef", self.project)
  94. url = reverse(
  95. "sentry-api-0-project-plugin-details",
  96. args=[self.org.slug, self.project.slug, "pushover"],
  97. )
  98. res = self.client.get(url)
  99. config = json.loads(res.content)["config"]
  100. userkey_config = [item for item in config if item["name"] == "userkey"][0]
  101. apikey_config = [item for item in config if item["name"] == "apikey"][0]
  102. assert userkey_config.get("type") == "secret"
  103. assert userkey_config.get("value") is None
  104. assert userkey_config.get("hasSavedValue") is True
  105. assert userkey_config.get("prefix") == "abcd"
  106. assert apikey_config.get("type") == "secret"
  107. assert apikey_config.get("value") is None
  108. assert apikey_config.get("hasSavedValue") is True
  109. assert apikey_config.get("prefix") == "abcd"