forked from Rockachopa/Timmy-time-dashboard
fix: add graceful degradation for DB errors in tasks.py
Wrap all _get_db() calls with sqlite3.OperationalError handling so endpoints degrade gracefully when the task DB is locked or missing: - Read endpoints (page, partials, list, queue status) return empty data - Write endpoints (create, update, delete) return 503 Fixes #943 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -3,6 +3,99 @@
|
||||
Verifies task CRUD operations and the dashboard page rendering.
|
||||
"""
|
||||
|
||||
import sqlite3
|
||||
from unittest.mock import patch
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# DB error handling tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_DB_ERROR = sqlite3.OperationalError("database is locked")
|
||||
|
||||
|
||||
def test_tasks_page_degrades_on_db_error(client):
|
||||
"""GET /tasks renders empty columns when DB is unavailable."""
|
||||
with patch(
|
||||
"dashboard.routes.tasks._get_db",
|
||||
side_effect=_DB_ERROR,
|
||||
):
|
||||
response = client.get("/tasks")
|
||||
assert response.status_code == 200
|
||||
assert "TASK QUEUE" in response.text
|
||||
|
||||
|
||||
def test_pending_partial_degrades_on_db_error(client):
|
||||
"""GET /tasks/pending returns fallback HTML when DB is unavailable."""
|
||||
with patch(
|
||||
"dashboard.routes.tasks._get_db",
|
||||
side_effect=_DB_ERROR,
|
||||
):
|
||||
response = client.get("/tasks/pending")
|
||||
assert response.status_code == 200
|
||||
assert "Database unavailable" in response.text
|
||||
|
||||
|
||||
def test_active_partial_degrades_on_db_error(client):
|
||||
"""GET /tasks/active returns fallback HTML when DB is unavailable."""
|
||||
with patch(
|
||||
"dashboard.routes.tasks._get_db",
|
||||
side_effect=_DB_ERROR,
|
||||
):
|
||||
response = client.get("/tasks/active")
|
||||
assert response.status_code == 200
|
||||
assert "Database unavailable" in response.text
|
||||
|
||||
|
||||
def test_completed_partial_degrades_on_db_error(client):
|
||||
"""GET /tasks/completed returns fallback HTML when DB is unavailable."""
|
||||
with patch(
|
||||
"dashboard.routes.tasks._get_db",
|
||||
side_effect=_DB_ERROR,
|
||||
):
|
||||
response = client.get("/tasks/completed")
|
||||
assert response.status_code == 200
|
||||
assert "Database unavailable" in response.text
|
||||
|
||||
|
||||
def test_api_create_task_503_on_db_error(client):
|
||||
"""POST /api/tasks returns 503 when DB is unavailable."""
|
||||
with patch(
|
||||
"dashboard.routes.tasks._get_db",
|
||||
side_effect=_DB_ERROR,
|
||||
):
|
||||
response = client.post("/api/tasks", json={"title": "Test"})
|
||||
assert response.status_code == 503
|
||||
|
||||
|
||||
def test_api_list_tasks_empty_on_db_error(client):
|
||||
"""GET /api/tasks returns empty list when DB is unavailable."""
|
||||
with patch(
|
||||
"dashboard.routes.tasks._get_db",
|
||||
side_effect=_DB_ERROR,
|
||||
):
|
||||
response = client.get("/api/tasks")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == []
|
||||
|
||||
|
||||
def test_queue_status_degrades_on_db_error(client):
|
||||
"""GET /api/queue/status returns idle status when DB is unavailable."""
|
||||
with patch(
|
||||
"dashboard.routes.tasks._get_db",
|
||||
side_effect=_DB_ERROR,
|
||||
):
|
||||
response = client.get("/api/queue/status")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["is_working"] is False
|
||||
assert data["current_task"] is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Existing tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_tasks_page_returns_200(client):
|
||||
response = client.get("/tasks")
|
||||
|
||||
Reference in New Issue
Block a user