test_plugin.py 5.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140
  1. from functools import cached_property
  2. import pytest
  3. import responses
  4. from sentry.shared_integrations.exceptions import ApiError
  5. from sentry.testutils.cases import PluginTestCase
  6. from sentry.utils import json
  7. from sentry_plugins.splunk.plugin import SplunkPlugin
  8. class SplunkPluginTest(PluginTestCase):
  9. @cached_property
  10. def plugin(self):
  11. return SplunkPlugin()
  12. def test_conf_key(self):
  13. assert self.plugin.conf_key == "splunk"
  14. def test_entry_point(self):
  15. self.assertPluginInstalled("splunk", self.plugin)
  16. @responses.activate
  17. def test_simple_notification(self):
  18. responses.add(responses.POST, "https://splunk.example.com:8088/services/collector")
  19. self.plugin.set_option("token", "12345678-1234-1234-1234-1234567890AB", self.project)
  20. self.plugin.set_option("index", "main", self.project)
  21. self.plugin.set_option("instance", "https://splunk.example.com:8088", self.project)
  22. event = self.store_event(
  23. data={"message": "Hello world", "level": "warning"}, project_id=self.project.id
  24. )
  25. with self.options({"system.url-prefix": "http://example.com"}):
  26. self.plugin.post_process(event)
  27. request = responses.calls[0].request
  28. payload = json.loads(request.body)
  29. assert payload == self.plugin.get_event_payload(event)
  30. headers = request.headers
  31. assert headers["Authorization"] == "Splunk 12345678-1234-1234-1234-1234567890AB"
  32. @responses.activate
  33. def test_dont_reraise_error(self):
  34. responses.add(
  35. responses.POST, "https://splunk.example.com:8088/services/collector", status=404
  36. )
  37. self.plugin.set_option("token", "12345678-1234-1234-1234-1234567890AB", self.project)
  38. self.plugin.set_option("index", "main", self.project)
  39. self.plugin.set_option("instance", "https://splunk.example.com:8088", self.project)
  40. event = self.store_event(
  41. data={"message": "Hello world", "level": "warning"}, project_id=self.project.id
  42. )
  43. with self.options({"system.url-prefix": "http://example.com"}):
  44. self.plugin.post_process(event)
  45. resp = responses.calls[0].response
  46. assert resp.status_code == 404
  47. @responses.activate
  48. def test_reraise_error(self):
  49. responses.add(
  50. responses.POST, "https://splunk.example.com:8088/services/collector", status=500
  51. )
  52. self.plugin.set_option("token", "12345678-1234-1234-1234-1234567890AB", self.project)
  53. self.plugin.set_option("index", "main", self.project)
  54. self.plugin.set_option("instance", "https://splunk.example.com:8088", self.project)
  55. event = self.store_event(
  56. data={"message": "Hello world", "level": "warning"}, project_id=self.project.id
  57. )
  58. with self.options({"system.url-prefix": "http://example.com"}):
  59. with pytest.raises(ApiError):
  60. self.plugin.post_process(event)
  61. def test_http_payload(self):
  62. event = self.store_event(
  63. data={
  64. "request": {
  65. "url": "http://example.com",
  66. "method": "POST",
  67. "headers": {"Referer": "http://example.com/foo"},
  68. }
  69. },
  70. project_id=self.project.id,
  71. )
  72. result = self.plugin.get_event_payload(event)
  73. assert result["event"]["request_url"] == "http://example.com/"
  74. assert result["event"]["request_method"] == "POST"
  75. assert result["event"]["request_referer"] == "http://example.com/foo"
  76. def test_error_payload(self):
  77. event = self.store_event(
  78. data={
  79. "exception": {"values": [{"type": "ValueError", "value": "foo bar"}]},
  80. "type": "error",
  81. },
  82. project_id=self.project.id,
  83. )
  84. result = self.plugin.get_event_payload(event)
  85. assert result["event"]["type"] == "error"
  86. assert result["event"]["exception_type"] == "ValueError"
  87. assert result["event"]["exception_value"] == "foo bar"
  88. def test_csp_payload(self):
  89. event = self.store_event(
  90. data={
  91. "csp": {
  92. "document_uri": "http://example.com/",
  93. "violated_directive": "style-src cdn.example.com",
  94. "blocked_uri": "http://example.com/style.css",
  95. "effective_directive": "style-src",
  96. },
  97. "type": "csp",
  98. },
  99. project_id=self.project.id,
  100. )
  101. result = self.plugin.get_event_payload(event)
  102. assert result["event"]["type"] == "csp"
  103. assert result["event"]["csp_document_uri"] == "http://example.com/"
  104. assert result["event"]["csp_violated_directive"] == "style-src cdn.example.com"
  105. assert result["event"]["csp_blocked_uri"] == "http://example.com/style.css"
  106. assert result["event"]["csp_effective_directive"] == "style-src"
  107. def test_user_payload(self):
  108. event = self.store_event(
  109. data={"user": {"id": "1", "email": "foo@example.com", "ip_address": "127.0.0.1"}},
  110. project_id=self.project.id,
  111. )
  112. result = self.plugin.get_event_payload(event)
  113. assert result["event"]["user_id"] == "1"
  114. assert result["event"]["user_email_hash"] == "b48def645758b95537d4424c84d1a9ff"
  115. assert result["event"]["user_ip_trunc"] == "127.0.0.0"