mock_service.py 3.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116
  1. from __future__ import annotations
  2. import os
  3. import shutil
  4. from collections import defaultdict
  5. from typing import Any
  6. from fixtures.integrations import FIXTURE_DIRECTORY
  7. from fixtures.integrations.stub_service import StubService
  8. from sentry.utils import json
  9. from sentry.utils.numbers import base32_encode
  10. class MockService(StubService):
  11. """
  12. A mock is a service that replicates the functionality of a real software
  13. system by implementing the same interface with simplified business logic.
  14. For example, a mocked random dice_roll function might return `hash(time()) % 6`.
  15. Like stubs, mocks can make tests simpler and more reliable.
  16. """
  17. def __init__(self, mode="memory"):
  18. """
  19. Initialize the mock instance. Wipe the previous instance's data if it exists.
  20. """
  21. super().__init__()
  22. self.mode = mode
  23. self._next_error_code = None
  24. self._next_ids = defaultdict(int)
  25. if self.mode == "file":
  26. path = os.path.join(FIXTURE_DIRECTORY, self.service_name, "data")
  27. if os.path.exists(path):
  28. shutil.rmtree(path)
  29. os.makedirs(path)
  30. else:
  31. self._memory: dict[str, dict[str, Any]] = defaultdict(dict)
  32. def add_project(self, project):
  33. """
  34. Create a new, empty project.
  35. :param project: String name of project
  36. :return: void
  37. """
  38. self._next_ids.get(project) # touch
  39. if self.mode == "file":
  40. self._get_project_path(project)
  41. def remove_project(self, project):
  42. """
  43. Totally wipe out a project.
  44. :param project: String name of project
  45. :return: void
  46. """
  47. del self._next_ids[project]
  48. if self.mode == "file":
  49. path = self._get_project_path(project)
  50. shutil.rmtree(path)
  51. def break_next_api_call(self, error_code=500):
  52. """
  53. Simulate an outage for a single API call.
  54. """
  55. self._next_error_code = error_code
  56. def _throw_if_broken(self, message_option=None):
  57. """
  58. See break_next_api_call.
  59. :param message_option: What should the message be if this raises?
  60. :raises: Generic Exception
  61. """
  62. if self._next_error_code:
  63. self._next_error_code = None
  64. message = message_option or f"{self.service_name} is down"
  65. raise Exception(f"{self._next_error_code}: {message}")
  66. def _get_project_names(self):
  67. return self._next_ids.keys()
  68. def _get_new_ticket_name(self, project):
  69. counter = self._next_ids[project]
  70. self._next_ids[project] = counter + 1
  71. return f"{project}-{base32_encode(counter)}"
  72. def _get_project_path(self, project):
  73. path = os.path.join(FIXTURE_DIRECTORY, self.service_name, "data", project)
  74. os.makedirs(path, exist_ok=True)
  75. return path
  76. def _set_data(self, project, name, data):
  77. if self.mode == "memory":
  78. if not self._memory[project]:
  79. self._memory[project] = defaultdict()
  80. self._memory[project][name] = data
  81. return
  82. path = os.path.join(self._get_project_path(project), f"{name}.json")
  83. with open(path, "w") as f:
  84. f.write(json.dumps(data))
  85. def _get_data(self, project, name):
  86. if self.mode == "memory":
  87. if not self._memory[project]:
  88. self._memory[project] = defaultdict()
  89. return self._memory[project].get(name)
  90. path = os.path.join(self._get_project_path(project), f"{name}.json")
  91. if not os.path.exists(path):
  92. return None
  93. with open(path) as f:
  94. return json.loads(f.read())