mcp-dbutils

"""Unit tests for resource monitoring functionality""" from datetime import datetime from mcp_dbutils.stats import ResourceStats def test_connection_tracking(): """Test connection tracking functionality""" stats = ResourceStats() # Test initial state assert stats.active_connections == 0 assert stats.total_connections == 0 assert stats.connection_start_time is None # Test connection start stats.record_connection_start() assert stats.active_connections == 1 assert stats.total_connections == 1 assert isinstance(stats.connection_start_time, datetime) # Test multiple connections stats.record_connection_start() assert stats.active_connections == 2 assert stats.total_connections == 2 # Test connection end stats.record_connection_end() assert stats.active_connections == 1 assert stats.total_connections == 2 # Total should not decrease # Test connection cleanup stats.record_connection_end() assert stats.active_connections == 0 assert stats.total_connections == 2 def test_query_tracking(): """Test query execution tracking""" stats = ResourceStats() # Test initial state assert stats.query_count == 0 assert stats.last_query_time is None # Test query recording stats.record_query() assert stats.query_count == 1 assert isinstance(stats.last_query_time, datetime) # Test multiple queries stats.record_query() assert stats.query_count == 2 def test_error_tracking(): """Test error tracking functionality""" stats = ResourceStats() # Test initial state assert stats.error_count == 0 assert stats.last_error_time is None assert stats.error_types == {} # Test error recording stats.record_error("DatabaseError") assert stats.error_count == 1 assert isinstance(stats.last_error_time, datetime) assert stats.error_types["DatabaseError"] == 1 # Test multiple errors stats.record_error("ConfigurationError") stats.record_error("DatabaseError") assert stats.error_count == 3 assert stats.error_types["DatabaseError"] == 2 assert stats.error_types["ConfigurationError"] == 1 def test_stats_serialization(): """Test statistics serialization to dict""" stats = ResourceStats() # Setup some data stats.record_connection_start() stats.record_query() stats.record_error("TestError") # Test serialization data = stats.to_dict() assert isinstance(data, dict) assert data["active_connections"] == 1 assert data["total_connections"] == 1 assert data["query_count"] == 1 assert data["error_count"] == 1 assert isinstance(data["connection_duration"], (int, float)) assert data["error_types"]["TestError"] == 1