test: 补充 WO-11 页面错误兜底测试

2026-07-03 17:59:13 +08:00
parent f91d2272d8
commit d6f0f7c1b0
2 changed files with 116 additions and 0 deletions
@@ -384,6 +384,102 @@ def test_index_page_shows_database_error_state_when_database_has_io_error(monkey
    assert "请先保留 data 目录" in response.text


+def test_task_detail_page_shows_database_error_state_when_database_has_io_error(monkeypatch):
+    def broken_get_task(_session, _task_id):
+        raise OperationalError("SELECT 1", {}, Exception("disk I/O error"))
+
+    monkeypatch.setattr("app.main.get_task", broken_get_task)
+
+    with make_test_client() as (client, _engine):
+        response = client.get("/tasks/task-io-error")
+
+    assert response.status_code == 503
+    assert "数据库暂时不可用" in response.text
+    assert "请先保留 data 目录" in response.text
+
+
+def test_hotspot_report_page_shows_database_error_state_when_database_has_io_error(monkeypatch):
+    def broken_get_task(_session, _task_id):
+        raise OperationalError("SELECT 1", {}, Exception("disk I/O error"))
+
+    monkeypatch.setattr("app.main.get_task", broken_get_task)
+
+    with make_test_client() as (client, engine):
+        from sqlalchemy.orm import Session
+
+        with Session(engine) as session:
+            session.add(Task(id="task-hot-io", platform="xiaohongshu", status="success"))
+            session.add(Hotspot(id="hot-io", task_id="task-hot-io", platform="xiaohongshu", title="热点", rank=1, raw_data="{}"))
+            session.commit()
+
+        response = client.get("/hotspots/hot-io/report")
+
+    assert response.status_code == 503
+    assert "数据库暂时不可用" in response.text
+
+
+def test_item_detail_page_shows_database_error_state_when_database_has_io_error(monkeypatch):
+    def broken_get_task(_session, _task_id):
+        raise OperationalError("SELECT 1", {}, Exception("disk I/O error"))
+
+    monkeypatch.setattr("app.main.get_task", broken_get_task)
+
+    with make_test_client() as (client, engine):
+        from sqlalchemy.orm import Session
+
+        with Session(engine) as session:
+            session.add(Task(id="task-item-io", platform="xiaohongshu", status="success"))
+            session.add(Hotspot(id="hot-item-io", task_id="task-item-io", platform="xiaohongshu", title="热点", rank=1, raw_data="{}"))
+            session.add(
+                ContentItem(
+                    id="item-io",
+                    task_id="task-item-io",
+                    hotspot_id="hot-item-io",
+                    platform="xiaohongshu",
+                    source_item_id="item-io-source",
+                    item_type="note",
+                    status="success",
+                    raw_data="{}",
+                )
+            )
+            session.commit()
+
+        response = client.get("/items/item-io")
+
+    assert response.status_code == 503
+    assert "数据库暂时不可用" in response.text
+
+
+def test_export_route_returns_service_unavailable_when_database_has_io_error(monkeypatch):
+    def broken_export(_session, _item_id):
+        raise OperationalError("SELECT 1", {}, Exception("disk I/O error"))
+
+    monkeypatch.setattr("app.main.export_item_comments_csv", broken_export)
+
+    with make_test_client() as (client, _engine):
+        response = client.get("/api/export/items/item-io/comments.csv")
+
+    assert response.status_code == 503
+    assert response.json()["detail"] == "数据库暂时不可用，请稍后重试或联系维护者恢复数据。"
+
+
+def test_missing_html_pages_render_friendly_not_found_page():
+    with make_test_client() as (client, _engine):
+        task_response = client.get("/tasks/missing-task")
+        hotspot_response = client.get("/hotspots/missing-hotspot/report")
+        item_response = client.get("/items/missing-item")
+
+    assert task_response.status_code == 404
+    assert "任务不存在" in task_response.text
+    assert "Internal Server Error" not in task_response.text
+    assert hotspot_response.status_code == 404
+    assert "热点不存在" in hotspot_response.text
+    assert "Internal Server Error" not in hotspot_response.text
+    assert item_response.status_code == 404
+    assert "内容条目不存在" in item_response.text
+    assert "Internal Server Error" not in item_response.text
+
+
 def test_task_api_includes_progress_counts_stage_and_demo_flag():
    with make_test_client() as (client, engine):
        seed_result_data(engine)