conftest.py 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140
  1. from __future__ import annotations
  2. import importlib
  3. import sys
  4. import types
  5. from pathlib import Path
  6. from types import SimpleNamespace
  7. import pytest
  8. from mongomock_motor import AsyncMongoMockClient
  9. from pyrogram.enums import ChatMemberStatus
  10. ROOT = Path(__file__).resolve().parents[1]
  11. class FakeTelegramApp:
  12. def __init__(self) -> None:
  13. self.members: dict[tuple[int, int], object] = {}
  14. self.chats: dict[int, object] = {}
  15. self.histories: dict[int, list[object]] = {}
  16. self.sent_messages: list[tuple[int, str]] = []
  17. self.sent_photos: list[tuple[int, object, str]] = []
  18. self.edited_messages: list[tuple[int, int, str]] = []
  19. async def get_chat_member(self, chat_id: int, user_id: int):
  20. member = self.members.get((int(chat_id), int(user_id)))
  21. if member is not None:
  22. return member
  23. return SimpleNamespace(
  24. status=ChatMemberStatus.MEMBER,
  25. privileges=None,
  26. user=SimpleNamespace(
  27. id=int(user_id),
  28. username=None,
  29. first_name=str(user_id),
  30. last_name=None,
  31. is_bot=False,
  32. is_deleted=False,
  33. ),
  34. custom_title=None,
  35. until_date=None,
  36. )
  37. async def get_chat(self, chat_id: int):
  38. chat = self.chats.get(int(chat_id))
  39. if chat is None:
  40. raise ValueError("chat not found")
  41. return chat
  42. async def get_chat_history(self, chat_id: int, limit: int = 100):
  43. for message in self.histories.get(int(chat_id), [])[:limit]:
  44. yield message
  45. async def send_message(self, chat_id: int, text: str, **_kwargs):
  46. self.sent_messages.append((int(chat_id), text))
  47. return SimpleNamespace(chat=SimpleNamespace(id=int(chat_id)), id=len(self.sent_messages))
  48. async def send_photo(self, chat_id: int, photo: object, caption: str = "", **_kwargs):
  49. self.sent_photos.append((int(chat_id), photo, caption))
  50. return SimpleNamespace(chat=SimpleNamespace(id=int(chat_id)), id=len(self.sent_photos))
  51. async def edit_message_text(
  52. self, chat_id: int, message_id: int, text: str, **_kwargs
  53. ):
  54. self.edited_messages.append((int(chat_id), int(message_id), text))
  55. return SimpleNamespace(chat=SimpleNamespace(id=int(chat_id)), id=int(message_id))
  56. def on_message(self, *_args, **_kwargs):
  57. return lambda function: function
  58. def on_callback_query(self, *_args, **_kwargs):
  59. return lambda function: function
  60. def on_edited_message(self, *_args, **_kwargs):
  61. return lambda function: function
  62. def on_deleted_messages(self, *_args, **_kwargs):
  63. return lambda function: function
  64. def on_chat_member_updated(self, *_args, **_kwargs):
  65. return lambda function: function
  66. @pytest.fixture
  67. def app_modules(tmp_path):
  68. previous = {
  69. name: module
  70. for name, module in sys.modules.items()
  71. if name == "wbb" or name.startswith("wbb.")
  72. }
  73. for name in list(previous):
  74. sys.modules.pop(name, None)
  75. fake_app = FakeTelegramApp()
  76. fake_wbb = types.ModuleType("wbb")
  77. fake_wbb.__path__ = [str(ROOT / "wbb")]
  78. fake_wbb.db = AsyncMongoMockClient().wbb
  79. fake_wbb.control_db = fake_wbb.db
  80. fake_wbb.app = fake_app
  81. fake_wbb.log = SimpleNamespace(info=lambda _message: None, error=lambda _message: None)
  82. fake_wbb.BOT_ID = 999
  83. fake_wbb.BOT_NAME = "Test Bot"
  84. fake_wbb.BOT_USERNAME = "test_bot"
  85. fake_wbb.BOT_PROFILE_ID = "primary"
  86. fake_wbb.BOT_PERMISSIONS = {"*"}
  87. fake_wbb.LOG_GROUP_ID = 0
  88. fake_wbb.TELEGRAM_CONNECTED = True
  89. fake_wbb.SUPERVISOR_MODE = False
  90. fake_wbb.ADMIN_BOOTSTRAP_MODE = False
  91. fake_wbb.BOT_SUPERVISOR = None
  92. fake_wbb.SUDOERS = set()
  93. fake_wbb.MOD_LOAD = []
  94. fake_wbb.MOD_NOLOAD = []
  95. fake_wbb.MESSAGE_DUMP_CHAT = 0
  96. fake_wbb.ADMIN_WEB_USERNAME = "admin"
  97. fake_wbb.ADMIN_WEB_INITIAL_PASSWORD = "qwe0.123456"
  98. fake_wbb.ADMIN_WEB_SESSION_HOURS = 12
  99. fake_wbb.ADMIN_WEB_COOKIE_SECURE = False
  100. fake_wbb.ADMIN_WEB_UPLOAD_MAX_MB = 20
  101. fake_wbb.ADMIN_WEB_HOST = "127.0.0.1"
  102. fake_wbb.ADMIN_WEB_PORT = 8088
  103. fake_wbb.SERVICE_ADDRESS_ENCRYPTION_KEY = (
  104. "MDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDA="
  105. )
  106. fake_wbb.SERVICE_ADDRESS_RETENTION_DAYS = 7
  107. fake_wbb.SERVICE_REVIEW_QR_EXPIRY_HOURS = 24
  108. fake_wbb.SERVICE_CUSTOMER_MAX_OPEN_ORDERS = 3
  109. fake_wbb.SERVICE_CUSTOMER_DAILY_REQUEST_LIMIT = 10
  110. fake_wbb.BOT_PROFILES_PATH = str(tmp_path / "bot_profiles.json")
  111. sys.modules["wbb"] = fake_wbb
  112. def load(name: str):
  113. return importlib.import_module(name)
  114. yield SimpleNamespace(wbb=fake_wbb, app=fake_app, load=load)
  115. for name in list(sys.modules):
  116. if name == "wbb" or name.startswith("wbb."):
  117. sys.modules.pop(name, None)
  118. sys.modules.update(previous)