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