首次上传

2026-05-16 22:25:48 +08:00
commit 2b8f902548
88 changed files with 16386 additions and 0 deletions
@@ -0,0 +1,75 @@
+"""执行器列表存储单元测试。"""
+from __future__ import annotations
+
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from app.config import OrderExecutorConfig, Settings, AppConfig, AuthConfig, WeComConfig, GateConfig
+from app import order_executors_store as store
+
+
+def _minimal_settings() -> Settings:
+    return Settings(
+        app=AppConfig(
+            host="127.0.0.1",
+            port=8088,
+            poll_interval_seconds=60,
+            log_file="./runtime/system.log",
+            database_url="sqlite+aiosqlite:///./runtime/t.db",
+            session_secret="x",
+        ),
+        auth=AuthConfig(enabled=False, username="a", password="b"),
+        wecom=WeComConfig(webhook="https://example.com/hook"),
+        gate=GateConfig(),
+        order_executor=OrderExecutorConfig(
+            enabled=True,
+            base_url="http://127.0.0.1:8090",
+            webhook_secret="sec",
+            timeout_seconds=15.0,
+        ),
+    )
+
+
+class TestOrderExecutorsStore(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp = tempfile.TemporaryDirectory()
+        self.path = Path(self.tmp.name) / "order_executors.json"
+        self._patch = patch.object(store, "_STORE_PATH", self.path)
+        self._patch.start()
+
+    def tearDown(self) -> None:
+        self._patch.stop()
+        self.tmp.cleanup()
+
+    def test_migrate_from_settings(self) -> None:
+        s = _minimal_settings()
+        store.ensure_store_initialized(s)
+        snap = store.read_snapshot(s)
+        self.assertTrue(snap["enabled"])
+        self.assertEqual(snap["webhook_secret"], "sec")
+        self.assertEqual(len(snap["executors"]), 1)
+        self.assertEqual(snap["executors"][0]["base_url"], "http://127.0.0.1:8090")
+
+    def test_add_and_active(self) -> None:
+        s = _minimal_settings()
+        store.ensure_store_initialized(s)
+        row = store.add_executor(s, name="b", base_url="http://127.0.0.1:8091", enabled=True)
+        active = store.active_executors(s)
+        urls = {e["base_url"] for e in active}
+        self.assertIn("http://127.0.0.1:8090", urls)
+        self.assertIn("http://127.0.0.1:8091", urls)
+        self.assertEqual(row["name"], "b")
+
+    def test_write_global(self) -> None:
+        s = _minimal_settings()
+        store.ensure_store_initialized(s)
+        snap = store.write_global_settings(s, enabled=False, webhook_secret="new")
+        self.assertFalse(snap["enabled"])
+        self.assertEqual(snap["webhook_secret"], "new")
+
+
+if __name__ == "__main__":
+    unittest.main()