重构物流客服模块并完善会话接待工作台

- 拆分应用工厂、业务服务与路由，保留原有 HTTP/SSE 和存储入口 - 分离页面资源，重排客服工作台并完善响应式导航、草稿和附件交互 - 整理维护脚本、请求示例和贡献指南，移除启动脚本中的密钥配置 - 上传文件改为仅本地保留，补充隔离的后端与前端回归验证验证：46 项 Python 测试通过，含 14 个已有子测试；12 个 Node 前端场景通过。
2026-10-29 10:59:29 +08:00
parent 761c629788
commit 805675f8db
79 changed files with 5737 additions and 4625 deletions
@@ -0,0 +1,128 @@
+"""Maintenance script regression tests using temporary SQLite databases."""
+
+import os
+import sqlite3
+import subprocess
+import sys
+import tempfile
+import unittest
+from contextlib import closing
+from pathlib import Path
+from unittest.mock import patch
+
+from scripts import close_active_sessions, inspect_human_sessions
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+class MaintenanceScriptTests(unittest.TestCase):
+    def setUp(self):
+        self.directory = tempfile.TemporaryDirectory()
+        self.addCleanup(self.directory.cleanup)
+        self.sandbox = Path(self.directory.name)
+        self.database = self.sandbox / "history with spaces.db"
+        self.other_directory = self.sandbox / "elsewhere"
+        self.other_directory.mkdir()
+        with closing(sqlite3.connect(self.database)) as connection:
+            connection.executescript(
+                "CREATE TABLE human_sessions(id INTEGER PRIMARY KEY, status TEXT);"
+                "CREATE TABLE human_messages(id INTEGER PRIMARY KEY, content TEXT);"
+                "INSERT INTO human_sessions VALUES"
+                "(1, 'waiting'), (2, 'active'), (3, 'closed'), (4, 'active');"
+            )
+            connection.executemany(
+                "INSERT INTO human_messages VALUES (?, ?)",
+                [(number, f"message {number}") for number in range(1, 9)],
+            )
+            connection.commit()
+
+    def run_script(self, name, database):
+        return subprocess.run(
+            [sys.executable, str(ROOT / "scripts" / name), "--db", str(database)],
+            cwd=self.other_directory,
+            capture_output=True,
+            text=True,
+            encoding="utf-8",
+            timeout=15,
+        )
+
+    def test_inspect_returns_recent_messages_and_open_sessions_without_changes(self):
+        original = self.database.read_bytes()
+        messages, sessions = inspect_human_sessions.inspect_sessions(self.database)
+        self.assertEqual([row[0] for row in messages], [8, 7, 6, 5, 4])
+        self.assertEqual(sessions, [(1, "waiting"), (2, "active"), (4, "active")])
+        self.assertEqual(self.database.read_bytes(), original)
+
+    def test_close_only_changes_active_sessions_and_can_be_repeated(self):
+        self.assertEqual(close_active_sessions.close_active_sessions(self.database), 2)
+        with closing(sqlite3.connect(self.database)) as connection:
+            sessions = connection.execute(
+                "SELECT id, status FROM human_sessions ORDER BY id"
+            ).fetchall()
+            messages = connection.execute("SELECT COUNT(*) FROM human_messages").fetchone()[0]
+        self.assertEqual(sessions, [(1, "waiting"), (2, "closed"), (3, "closed"), (4, "closed")])
+        self.assertEqual(messages, 8)
+        self.assertEqual(close_active_sessions.close_active_sessions(self.database), 0)
+
+    def test_missing_database_is_not_created(self):
+        missing = self.sandbox / "missing.db"
+        for operation in (
+            inspect_human_sessions.inspect_sessions,
+            close_active_sessions.close_active_sessions,
+        ):
+            with self.subTest(operation=operation.__name__):
+                with self.assertRaises(sqlite3.OperationalError):
+                    operation(missing)
+                self.assertFalse(missing.exists())
+
+    def test_default_database_is_independent_of_working_directory(self):
+        original_directory = Path.cwd()
+        try:
+            os.chdir(self.other_directory)
+            with (
+                patch.object(inspect_human_sessions, "DEFAULT_DB", self.database),
+                patch.object(close_active_sessions, "DEFAULT_DB", self.database),
+            ):
+                self.assertEqual(len(inspect_human_sessions.inspect_sessions()[0]), 5)
+                self.assertEqual(close_active_sessions.close_active_sessions(), 2)
+            self.assertFalse((self.other_directory / "history.db").exists())
+        finally:
+            os.chdir(original_directory)
+
+    def test_cli_supports_other_working_directories_and_paths_with_spaces(self):
+        inspection = self.run_script("inspect_human_sessions.py", self.database)
+        self.assertEqual(inspection.returncode, 0, inspection.stderr)
+        self.assertIn("message 8", inspection.stdout)
+        closing_result = self.run_script("close_active_sessions.py", self.database)
+        self.assertEqual(closing_result.returncode, 0, closing_result.stderr)
+        self.assertIn("Closed 2 active session(s).", closing_result.stdout)
+        self.assertFalse((self.other_directory / "history.db").exists())
+
+    def test_cli_reports_missing_database_without_creating_it(self):
+        missing = self.sandbox / "missing.db"
+        for script in ("inspect_human_sessions.py", "close_active_sessions.py"):
+            with self.subTest(script=script):
+                result = self.run_script(script, missing)
+                self.assertEqual(result.returncode, 1)
+                self.assertIn("Database error:", result.stderr)
+                self.assertFalse(missing.exists())
+
+    def test_importing_scripts_has_no_database_side_effects(self):
+        for script in ("inspect_human_sessions.py", "close_active_sessions.py"):
+            with self.subTest(script=script):
+                result = subprocess.run(
+                    [sys.executable, "-c", "import runpy, sys; runpy.run_path(sys.argv[1])",
+                     str(ROOT / "scripts" / script)],
+                    cwd=self.other_directory,
+                    capture_output=True,
+                    text=True,
+                    timeout=15,
+                )
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertEqual(result.stdout, "")
+        self.assertEqual(list(self.other_directory.iterdir()), [])
+
+
+if __name__ == "__main__":
+    unittest.main()