import json import unittest import uuid from unittest.mock import patch from tests._import_app import clear_login_attempts, import_web_app_module class ErrorAndTraceTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.module = import_web_app_module() cls.app = cls.module.app def setUp(self): # 每个测试前清理登录限制记录，避免测试间互相影响 with self.app.app_context(): clear_login_attempts() def _login(self, client, password: str = "testpass123"): resp = client.post("/login", json={"password": password}) self.assertEqual(resp.status_code, 200) return resp def test_healthz(self): client = self.app.test_client() resp = client.get("/healthz") self.assertEqual(resp.status_code, 200) data = resp.get_json() self.assertEqual(data.get("status"), "ok") def test_login_required_returns_structured_error(self): client = self.app.test_client() resp = client.get("/api/system/health") self.assertEqual(resp.status_code, 401) data = resp.get_json() self.assertEqual(data.get("success"), False) self.assertEqual(data.get("need_login"), True) self.assertIsInstance(data.get("error"), dict) self.assertEqual(data["error"].get("code"), "AUTH_REQUIRED") self.assertEqual(data["error"].get("status"), 401) self.assertTrue(data["error"].get("trace_id")) def test_api_404_has_structured_error_and_trace_id(self): client = self.app.test_client() resp = client.get("/api/__not_found__") self.assertEqual(resp.status_code, 404) data = resp.get_json() self.assertIsInstance(data, dict) self.assertEqual(data.get("success"), False) self.assertIsInstance(data.get("error"), dict) self.assertEqual(data["error"].get("code"), "HTTP_ERROR") self.assertEqual(data["error"].get("status"), 404) self.assertTrue(data["error"].get("trace_id")) self.assertEqual(resp.headers.get("X-Trace-Id"), data["error"].get("trace_id")) def test_trace_id_can_be_propagated_from_request(self): client = self.app.test_client() resp = client.get("/api/__not_found__", headers={"X-Trace-Id": "trace_from_test"}) data = resp.get_json() self.assertEqual(resp.headers.get("X-Trace-Id"), "trace_from_test") self.assertEqual(data["error"].get("trace_id"), "trace_from_test") def test_sanitize_error_details_masks_tokens(self): sanitized = self.module.sanitize_error_details('Bearer abcdefg refresh_token=xyz password: "123456"') self.assertIn("Bearer ***", sanitized) self.assertIn("refresh_token=***", sanitized) self.assertIn('"123456"', 'Bearer abcdefg refresh_token=xyz password: "123456"') # sanity self.assertNotIn("123456", sanitized) def test_build_error_payload_sanitizes_message_and_details(self): payload = self.module.build_error_payload( "TEST", 'Bearer abcdefg refresh_token=xyz password: "123456"', "TestError", 400, 'Bearer abcdefg refresh_token=xyz password: "123456"', ) self.assertIn("Bearer ***", payload.get("message", "")) self.assertIn("refresh_token=***", payload.get("message", "")) self.assertNotIn("abcdefg", payload.get("message", "")) self.assertNotIn("123456", payload.get("message", "")) self.assertIn("Bearer ***", payload.get("details", "")) self.assertIn("refresh_token=***", payload.get("details", "")) self.assertNotIn("abcdefg", payload.get("details", "")) self.assertNotIn("123456", payload.get("details", "")) def test_delete_emails_all_methods_fail_returns_aggregated_error(self): client = self.app.test_client() login = client.post("/login", json={"password": "testpass123"}) self.assertEqual(login.status_code, 200) email_addr = "user@example.com" graph_error = { "code": "EMAIL_DELETE_FAILED", "message": "Graph 删除失败", "type": "GraphAPIError", "status": 502, "details": "graph_failed", "trace_id": "", } # 控制器现在使用 repositories 和 services，需要 mock 这些模块 from outlook_web.repositories import accounts as accounts_repo from outlook_web.services import graph as graph_service from outlook_web.services import imap as imap_service with ( patch.object( accounts_repo, "get_account_by_email", return_value={ "email": email_addr, "client_id": "cid", "refresh_token": "rt", "group_id": None, }, ), patch.object( graph_service, "delete_emails_graph", return_value={ "success": False, "error": graph_error, "success_count": 0, "failed_count": 2, "errors": ["e1"], }, ), patch.object( imap_service, "delete_emails_imap", side_effect=[ {"success": False, "error": "imap_new_failed"}, {"success": False, "error": "imap_old_failed"}, ], ), ): resp = client.post( "/api/emails/delete", json={"email": email_addr, "ids": ["m1", "m2"]}, ) self.assertEqual(resp.status_code, 502) data = resp.get_json() self.assertEqual(data.get("success"), False) self.assertIsInstance(data.get("error"), dict) self.assertEqual(data["error"].get("code"), "EMAIL_DELETE_ALL_METHODS_FAILED") self.assertEqual(data["error"].get("status"), 502) self.assertTrue(data["error"].get("trace_id")) self.assertEqual(resp.headers.get("X-Trace-Id"), data["error"].get("trace_id")) details_text = data["error"].get("details") or "" self.assertIn("Graph API", details_text) self.assertIn("IMAP（新服务器）", details_text) self.assertIn("IMAP（旧服务器）", details_text) def test_delete_emails_proxy_error_does_not_fallback_to_imap(self): client = self.app.test_client() login = client.post("/login", json={"password": "testpass123"}) self.assertEqual(login.status_code, 200) email_addr = "user@example.com" graph_res = { "success": False, "error": { "code": "EMAIL_DELETE_FAILED", "message": "代理连接失败", "type": "ProxyError", "status": 502, "details": "proxy_failed", "trace_id": "", }, "success_count": 0, "failed_count": 2, "errors": ["ProxyError: connect failed"], } # 控制器现在使用 repositories 和 services，需要 mock 这些模块 from outlook_web.repositories import accounts as accounts_repo from outlook_web.services import graph as graph_service from outlook_web.services import imap as imap_service with ( patch.object( accounts_repo, "get_account_by_email", return_value={ "email": email_addr, "client_id": "cid", "refresh_token": "rt", "group_id": None, }, ), patch.object( graph_service, "delete_emails_graph", return_value=graph_res, ), patch.object( imap_service, "delete_emails_imap", ) as imap_mock, ): resp = client.post( "/api/emails/delete", json={"email": email_addr, "ids": ["m1", "m2"]}, ) imap_mock.assert_not_called() self.assertEqual(resp.status_code, 502) data = resp.get_json() self.assertEqual(data.get("success"), False) self.assertIsInstance(data.get("error"), dict) self.assertEqual(data["error"].get("type"), "ProxyError") self.assertTrue(data["error"].get("trace_id")) self.assertEqual(resp.headers.get("X-Trace-Id"), data["error"].get("trace_id")) def test_legacy_error_string_is_normalized_to_structured_error(self): client = self.app.test_client() # 登录 login = client.post("/login", json={"password": "testpass123"}) self.assertEqual(login.status_code, 200) self.assertEqual(login.get_json().get("success"), True) # 触发一个 legacy 的字符串错误（现在应返回正确的 400 状态码） resp = client.get("/api/groups/999999") self.assertEqual(resp.status_code, 400) data = resp.get_json() self.assertEqual(data.get("success"), False) self.assertIsInstance(data.get("error"), dict) self.assertEqual(data["error"].get("code"), "LEGACY_ERROR") self.assertEqual(data["error"].get("message"), "分组不存在") self.assertTrue(data["error"].get("trace_id")) # 响应体应是 JSON（确保 after_request 的 set_data 没破坏格式） json.loads(resp.get_data(as_text=True)) def test_scheduler_status_endpoint_is_accessible_after_login(self): client = self.app.test_client() login = client.post("/login", json={"password": "testpass123"}) self.assertEqual(login.status_code, 200) resp = client.get("/api/scheduler/status") self.assertEqual(resp.status_code, 200) data = resp.get_json() self.assertEqual(data.get("success"), True) self.assertIn("scheduler", data) self.assertIn("refresh", data) def test_validate_cron_endpoint(self): client = self.app.test_client() login = client.post("/login", json={"password": "testpass123"}) self.assertEqual(login.status_code, 200) resp = client.post("/api/settings/validate-cron", json={"cron_expression": "0 2 * * *"}) data = resp.get_json() self.assertEqual(resp.status_code, 200) self.assertEqual(data.get("success"), True) self.assertEqual(data.get("valid"), True) self.assertIsInstance(data.get("future_runs"), list) self.assertGreaterEqual(len(data.get("future_runs")), 1) def test_system_diagnostics_includes_schema(self): client = self.app.test_client() login = client.post("/login", json={"password": "testpass123"}) self.assertEqual(login.status_code, 200) resp = client.get("/api/system/diagnostics") self.assertEqual(resp.status_code, 200) data = resp.get_json() self.assertEqual(data.get("success"), True) diagnostics = data.get("diagnostics") or {} self.assertIn("schema", diagnostics) self.assertIn("version", diagnostics["schema"]) self.assertIn("target_version", diagnostics["schema"]) self.assertIn("up_to_date", diagnostics["schema"]) def test_system_upgrade_status_endpoint(self): client = self.app.test_client() login = client.post("/login", json={"password": "testpass123"}) self.assertEqual(login.status_code, 200) resp = client.get("/api/system/upgrade-status") self.assertEqual(resp.status_code, 200) data = resp.get_json() self.assertEqual(data.get("success"), True) upgrade = data.get("upgrade") or {} self.assertIn("schema_version", upgrade) self.assertIn("target_version", upgrade) self.assertIn("up_to_date", upgrade) self.assertIn("backup_hint", upgrade) def test_login_rate_limiting_locks_after_max_attempts(self): client = self.app.test_client() ip = "10.0.0.99" for _ in range(self.module.MAX_LOGIN_ATTEMPTS): resp = client.post( "/login", json={"password": "wrong_password"}, headers={"X-Forwarded-For": ip}, ) self.assertEqual(resp.status_code, 401) data = resp.get_json() self.assertEqual(data.get("success"), False) self.assertIsInstance(data.get("error"), dict) self.assertEqual(data["error"].get("code"), "LOGIN_INVALID_PASSWORD") resp = client.post( "/login", json={"password": "wrong_password"}, headers={"X-Forwarded-For": ip}, ) self.assertEqual(resp.status_code, 429) data = resp.get_json() self.assertEqual(data.get("success"), False) self.assertIsInstance(data.get("error"), dict) self.assertEqual(data["error"].get("code"), "LOGIN_RATE_LIMITED") self.assertEqual(data["error"].get("status"), 429) self.assertTrue(data["error"].get("trace_id")) def test_export_requires_verify_token(self): client = self.app.test_client() login = client.post("/login", json={"password": "testpass123"}) self.assertEqual(login.status_code, 200) resp = client.post( "/api/accounts/export-selected", json={"group_ids": [1], "verify_token": "invalid_token"}, ) self.assertEqual(resp.status_code, 401) data = resp.get_json() self.assertEqual(data.get("success"), False) self.assertEqual(data.get("need_verify"), True) self.assertIsInstance(data.get("error"), dict) self.assertEqual(data["error"].get("status"), 401) self.assertTrue(data["error"].get("trace_id")) def test_accounts_list_and_search_include_tags_and_last_refresh(self): client = self.app.test_client() login = client.post("/login", json={"password": "testpass123"}) self.assertEqual(login.status_code, 200) unique = uuid.uuid4().hex email_addr = f"test_{unique}@example.com" client_id = "client_id_" + unique refresh_token = "rt_" + unique tag_name = "tag_" + unique conn = self.module.create_sqlite_connection() try: # 插入账号 cur = conn.execute( """ INSERT INTO accounts (email, password, client_id, refresh_token, group_id, remark, status) VALUES (?, ?, ?, ?, ?, ?, ?) """, (email_addr, "", client_id, refresh_token, 1, "remark", "active"), ) account_id = cur.lastrowid # 插入标签并关联 cur = conn.execute( "INSERT INTO tags (name, color) VALUES (?, ?)", (tag_name, "#111111"), ) tag_id = cur.lastrowid conn.execute( "INSERT OR IGNORE INTO account_tags (account_id, tag_id) VALUES (?, ?)", (account_id, tag_id), ) # 插入刷新日志（用于 last_refresh_status/error） conn.execute( """ INSERT INTO account_refresh_logs (account_id, account_email, refresh_type, status, error_message) VALUES (?, ?, ?, ?, ?) """, (account_id, email_addr, "manual", "failed", "network_error"), ) conn.commit() finally: conn.close() # 列表接口：应返回 tags + last_refresh_* resp = client.get("/api/accounts") self.assertEqual(resp.status_code, 200) data = resp.get_json() self.assertEqual(data.get("success"), True) accounts = data.get("accounts") or [] target = next((a for a in accounts if a.get("email") == email_addr), None) self.assertIsNotNone(target) self.assertTrue(target.get("client_id", "").startswith(client_id[:8])) self.assertEqual(target.get("last_refresh_status"), "failed") self.assertEqual(target.get("last_refresh_error"), "network_error") self.assertIsInstance(target.get("tags"), list) self.assertIn(tag_name, [t.get("name") for t in target.get("tags")]) # 搜索接口：按标签名应可搜索到 resp = client.get(f"/api/accounts/search?q={tag_name}") self.assertEqual(resp.status_code, 200) data = resp.get_json() self.assertEqual(data.get("success"), True) accounts = data.get("accounts") or [] target = next((a for a in accounts if a.get("email") == email_addr), None) self.assertIsNotNone(target) self.assertEqual(target.get("last_refresh_status"), "failed") self.assertIn(tag_name, [t.get("name") for t in target.get("tags")]) def test_system_group_cannot_be_deleted(self): client = self.app.test_client() login = client.post("/login", json={"password": "testpass123"}) self.assertEqual(login.status_code, 200) conn = self.module.create_sqlite_connection() try: row = conn.execute("SELECT id FROM groups WHERE is_system = 1 LIMIT 1").fetchone() self.assertIsNotNone(row) system_group_id = row["id"] finally: conn.close() resp = client.delete(f"/api/groups/{system_group_id}") self.assertEqual(resp.status_code, 403) data = resp.get_json() self.assertEqual(data.get("success"), False) self.assertIsInstance(data.get("error"), dict) self.assertEqual(data["error"].get("code"), "SYSTEM_GROUP_PROTECTED") self.assertEqual(data["error"].get("status"), 403) def test_account_cannot_be_moved_to_system_group_via_update(self): client = self.app.test_client() login = client.post("/login", json={"password": "testpass123"}) self.assertEqual(login.status_code, 200) unique = uuid.uuid4().hex email_addr = f"move_{unique}@example.com" conn = self.module.create_sqlite_connection() try: system_row = conn.execute("SELECT id FROM groups WHERE is_system = 1 LIMIT 1").fetchone() self.assertIsNotNone(system_row) system_group_id = system_row["id"] cur = conn.execute( """ INSERT INTO accounts (email, password, client_id, refresh_token, group_id, remark, status) VALUES (?, ?, ?, ?, ?, ?, ?) """, (email_addr, "", "client", "rt", 1, "", "active"), ) account_id = cur.lastrowid conn.commit() finally: conn.close() resp = client.put( f"/api/accounts/{account_id}", json={"email": email_addr, "group_id": system_group_id}, ) self.assertEqual(resp.status_code, 403) data = resp.get_json() self.assertEqual(data.get("success"), False) self.assertIsInstance(data.get("error"), dict) self.assertEqual(data["error"].get("code"), "SYSTEM_GROUP_PROTECTED") self.assertEqual(data["error"].get("status"), 403)