/ dashboard / tests / test_database.py
test_database.py
 1  """
 2  Database Query Tests
 3  
 4  Test database query functions for correctness and read-only enforcement.
 5  """
 6  
 7  import pytest
 8  import sqlite3
 9  import sys
10  from pathlib import Path
11  
12  # Add parent directory to path
13  sys.path.insert(0, str(Path(__file__).parent.parent.parent))
14  
15  from dashboard.utils import database
16  
17  
18  def test_db_connection_is_readonly():
19      """Ensure database connection is read-only."""
20      conn = database.get_db_connection()
21  
22      with pytest.raises(sqlite3.OperationalError):
23          conn.execute("INSERT INTO sites (domain) VALUES ('test.com')")
24  
25  
26  def test_pipeline_funnel_query():
27      """Test pipeline funnel query returns expected columns."""
28      df = database.get_pipeline_funnel()
29  
30      assert 'status' in df.columns
31      assert 'count' in df.columns
32  
33  
34  def test_error_breakdown_query():
35      """Test error breakdown query."""
36      df = database.get_error_breakdown()
37  
38      assert 'error_message' in df.columns
39      assert 'stage' in df.columns
40      assert 'count' in df.columns
41  
42  
43  def test_response_rates_query():
44      """Test response rates query."""
45      df = database.get_response_rates()
46  
47      assert 'channel' in df.columns
48      assert 'total_sent' in df.columns
49      assert 'responses' in df.columns
50      assert 'response_rate' in df.columns
51  
52  
53  def test_sales_data_query():
54      """Test sales data query."""
55      df = database.get_sales_data()
56  
57      assert 'channel' in df.columns
58      assert 'sales_count' in df.columns
59      assert 'total_revenue' in df.columns
60  
61  
62  def test_cron_job_status_query():
63      """Test cron job status query."""
64      df = database.get_cron_job_status()
65  
66      assert 'job_name' in df.columns
67      assert 'last_run' in df.columns
68  
69  
70  def test_database_health():
71      """Test database health query."""
72      result = database.get_database_health()
73  
74      assert 'size_mb' in result
75      assert 'integrity' in result
76      assert 'table_counts' in result