当 Redis 无法连接时，使用 fakeredis 模拟 并且 完善 测试 (#51)

* 当 Redis 无法连接时，使用 fakeredis 模拟

* 完善 测试

model/apihelper/artifact.py

@@ -32,7 +32,7 @@ def get_comment(get_rate_num):
     return choice(data[str(data_ // 20 + 1)])
 
 
-class ArtifactORCRate:
+class ArtifactOcrRate:
     OCR_URL = "https://api.genshin.pub/api/v1/app/ocr"
     RATE_URL = "https://api.genshin.pub/api/v1/relic/rate"
     HEADERS = {

plugins/artifact_rate.py

@@ -5,7 +5,7 @@ from telegram.ext import CallbackContext, ConversationHandler, CommandHandler, C
 from telegram.helpers import escape_markdown
 
 from logger import Log
-from model.apihelper.artifact import ArtifactORCRate, get_comment, get_format_sub_item
+from model.apihelper.artifact import ArtifactOcrRate, get_comment, get_format_sub_item
 from plugins.base import BasePlugins
 from plugins.errorhandler import conversation_error_handler
 from service import BaseService
@@ -19,17 +19,17 @@ class ArtifactRate(BasePlugins):
     """
     STAR_KEYBOARD = [[
         InlineKeyboardButton(
-            f"{i}", callback_data=f"artifact_orc_rate_data|star|{i}") for i in range(1, 6)
+            f"{i}", callback_data=f"artifact_ocr_rate_data|star|{i}") for i in range(1, 6)
     ]]
 
     LEVEL_KEYBOARD = [[
         InlineKeyboardButton(
-            f"{i * 5 + j}", callback_data=f"artifact_orc_rate_data|level|{i * 5 + j}") for j in range(1, 6)
+            f"{i * 5 + j}", callback_data=f"artifact_ocr_rate_data|level|{i * 5 + j}") for j in range(1, 6)
     ] for i in range(0, 4)]
 
     def __init__(self, service: BaseService):
         super().__init__(service)
-        self.artifact_rate = ArtifactORCRate()
+        self.artifact_rate = ArtifactOcrRate()
 
     @staticmethod
     def create_conversation_handler(service: BaseService):

requirements.txt

@@ -14,3 +14,4 @@ beautifulsoup4>=4.11.1
 python-telegram-bot==20.0a0
 pyppeteer
 lxml>=4.9.0
+fakeredis==1.8.1

test/model/apihelper/test_artifact.py

@@ -0,0 +1,26 @@
+import unittest
+from unittest import IsolatedAsyncioTestCase
+
+from model.apihelper.artifact import ArtifactOcrRate
+
+
+class TestArtifact(IsolatedAsyncioTestCase):
+    def setUp(self):
+        self.artifact_rate = ArtifactOcrRate()
+
+    async def test_get_artifact_attr(self):
+        await self.artifact_rate.get_artifact_attr(b"")
+
+    async def test_rate_artifact(self):
+        artifact_attr = {
+            'name': '翠绿的猎人之冠', 'pos': '理之冠', 'star': 5, 'level': 20,
+            'main_item': {'type': 'cr', 'name': '暴击率', 'value': '31.1%'},
+            'sub_item': [{'type': 'hp', 'name': '生命值', 'value': '9.3%'},
+                         {'type': 'df', 'name': '防御力', 'value': '46'},
+                         {'type': 'atk', 'name': '攻击力', 'value': '49'},
+                         {'type': 'cd', 'name': '暴击伤害', 'value': '10.9%'}]}
+        await self.artifact_rate.rate_artifact(artifact_attr)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/run.py

@@ -0,0 +1,18 @@
+import os
+import unittest
+
+main_suite = unittest.TestSuite()
+
+for parent, dirs, _ in os.walk("."):
+    for dirname in dirs:
+        if dirname == "__pycache__":
+            continue
+        discover = unittest.defaultTestLoader.discover(
+            start_dir=parent + os.sep + dirname, pattern='test_*.py',
+            top_level_dir=parent + os.sep + dirname)
+        main_suite.addTest(discover)
+
+
+if __name__ == "__main__":
+    runner = unittest.TextTestRunner(verbosity=2)
+    runner.run(main_suite)

utils/redisdb.py

@@ -1,4 +1,5 @@
 import asyncio
+import fakeredis.aioredis
 
 from logger import Log
 from redis import asyncio as aioredis
@@ -22,8 +23,9 @@ class RedisDB:
         except (KeyboardInterrupt, SystemExit):
             pass
         except Exception as exc:
-            Log.error("尝试连接Redis失败")
-            raise exc
+            Log.warning("尝试连接Redis失败，使用 fakeredis 模拟")
+            self.client = fakeredis.aioredis.FakeRedis()
+            self._loop.run_until_complete(self.ping())
 
     async def ping(self):
         if await self.client.ping():