Repository: WxboySuper/Productivity-App
Issue category: Feature Request

Issue:
Create Basic Health Check in server.py 



Hints:

Patch:
diff --git a/requirements.txt b/requirements.txt
index 74936d4..daa8aa2 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,2 +1,3 @@
 flask==3.1.0
-coverage==7.6.10
\ No newline at end of file
+coverage==7.6.10
+psutil==6.1.1
\ No newline at end of file
diff --git a/src/python/server.py b/src/python/server.py
index a67cb0c..ca58da3 100644
--- a/src/python/server.py
+++ b/src/python/server.py
@@ -4,6 +4,11 @@
 import signal
 import sys
 from python.logging_config import setup_logging
+import psutil
+import time
+import sqlite3
+from typing import Dict, Any
+import concurrent.futures
 
 os.makedirs("logs", exist_ok=True)
 
@@ -15,11 +20,14 @@
 app.config.update(
     PORT=5000,
     ENV='development',
-    DEBUG=True
+    DEBUG=os.environ.get("FLASK_DEBUG", False),
+    START_TIME=time.time(),  # Add startup time to track uptime
+    DB_TIMEOUT=os.environ.get("DB_TIMEOUT", 1.0)  # Add configurable timeout
 )
 
 class AppContext:
     """Context manager for Flask app cleanup"""
+
     def __init__(self, flask_app):
         self.app = flask_app
         self.cleanup_handlers = []
@@ -65,11 +73,135 @@ def before_request():
         app.config['handled_first_request'] = True
 
 
+def check_database_health(timeout: float = None) -> Dict[str, Any]:
+    """Check database connection health with timeout"""
+    if timeout is None:
+        timeout = float(app.config.get('DB_TIMEOUT', 1.0))
+
+    if timeout <= 0:
+        raise ValueError("Timeout must be a positive number")
+
+    db_health = {
+        "status": "disconnected",
+        "response_time": None,
+        "error": None
+    }
+
+    def check_connection():
+        """Attempt to establish and verify database connection.
+
+        Establishes connection to SQLite database, executes a simple query
+        to verify connection is working, and properly closes resources.
+
+        Raises:
+            sqlite3.OperationalError: If database connection fails
+            sqlite3.Error: For other SQLite-related errors
+            Exception: For unexpected errors during connection attempt
+        """
+        conn = None
+        try:
+            # Cap connection timeout at 5 seconds as a safety measure
+            conn = sqlite3.connect('productivity.db', timeout=min(timeout, 5.0))
+            cursor = conn.cursor()
+            cursor.execute('SELECT 1')
+            cursor.fetchone()
+            cursor.close()
+        finally:
+            if conn:
+                conn.close()
+
+    try:
+        start_time = time.time()
+        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
+            future = executor.submit(check_connection)
+            future.result(timeout=timeout)  # This will raise TimeoutError if it takes too long
+
+        response_time = time.time() - start_time
+        db_health.update({
+            "status": "connected",
+            "response_time": round(response_time * 1000, 2),  # Convert to milliseconds
+            "error": None
+        })
+    except concurrent.futures.TimeoutError:
+        db_health["error"] = "Connection timed out"
+        log.error("Database connection timed out")
+    except sqlite3.OperationalError as e:
+        db_health["error"] = f"Database error: {str(e)}"
+        log.error("SQLite operational error: %s", e)
+    except Exception as e:
+        db_health["error"] = f"Unexpected error: {str(e)}"
+        log.error("Unexpected error during database health check")
+
+    return db_health
+
+
 @app.route('/health')
 def health_check():
-    """Basic health check endpoint"""
+    """Comprehensive health check endpoint
+
+    Returns:
+        tuple: A tuple containing:
+            - dict: Health check data with the following structure:
+                - status (str): 'healthy' or 'degraded'
+                - uptime_seconds (float): Server uptime in seconds
+                - memory (dict): Memory statistics
+                    - total (int): Total memory in bytes
+                    - available (int): Available memory in bytes
+                    - percent (float): Memory usage percentage
+                - system_load (dict): System load averages
+                    - 1min (float): 1-minute load average
+                    - 5min (float): 5-minute load average
+                    - 15min (float): 15-minute load average
+                - database (dict): Database health information
+                    - status (str): 'connected' or 'disconnected'
+                    - response_time (float|None): Response time in milliseconds
+                    - error (str|None): Error message if any
+            - int: HTTP status code (200 for healthy, 503 for degraded)
+    """
     log.debug("Health check requested")
-    return {'status': 'healthy'}, 200
+
+    # Calculate uptime
+    uptime = time.time() - app.config['START_TIME']
+
+    # Get system metrics
+    memory = psutil.virtual_memory()
+    load = psutil.getloadavg()
+
+    # Perform database health check
+    db_health = check_database_health()
+
+    health_data = {
+        'status': 'healthy',
+        'uptime_seconds': round(uptime, 2),
+        'memory': {
+            'total': memory.total,
+            'available': memory.available,
+            'percent': memory.percent
+        },
+        'system_load': {
+            '1min': load[0],
+            '5min': load[1],
+            '15min': load[2]
+        },
+        'database': db_health
+    }
+
+    # Get thresholds from config with defaults
+    memory_threshold = float(app.config.get('HEALTH_CHECK_MEMORY_THRESHOLD', 90))
+    load_threshold = float(app.config.get('HEALTH_CHECK_LOAD_THRESHOLD', 5))
+    log.debug("Health check thresholds - Memory: %s%%, Load: %s", memory_threshold, load_threshold)
+
+    # Set response status based on metrics
+    is_healthy = (
+        memory.percent < memory_threshold and  # Changed back to < for strict comparison
+        load[0] < load_threshold and  # Changed back to < for strict comparison
+        db_health['status'] == "connected"
+    )
+
+    health_data['status'] = 'healthy' if is_healthy else 'degraded'
+    status_code = 200 if is_healthy else 503
+
+    return health_data, status_code
 
 
 if __name__ == '__main__':  # pragma: no cover
diff --git a/src/python/todo_bridge.py b/src/python/todo_bridge.py
index 41c339d..9030381 100644
--- a/src/python/todo_bridge.py
+++ b/src/python/todo_bridge.py
@@ -74,7 +74,7 @@ def handle_command(cmd, payload):
         raise
 
 
-if __name__ == "__main__":
+if __name__ == "__main__":  # pragma: no cover
     try:
         request_id = generate_request_id()
         log.info("Bridge script started [RequestID: %s]", request_id)


Test patch:
diff --git a/src/tests/test_server.py b/src/tests/test_server.py
index b507fb4..9ee0d1a 100644
--- a/src/tests/test_server.py
+++ b/src/tests/test_server.py
@@ -1,6 +1,8 @@
-from python.server import app, AppContext, signal_handler
+from python.server import app, signal_handler, AppContext, signal
 import unittest
-import signal
+import time
+import sqlite3
+import os
 
 class TestServer(unittest.TestCase):
     """Test suite for server configuration."""
@@ -12,6 +14,22 @@ def setUp(self):
         self.app.config['PORT'] = 5000
         self.app.config['DEBUG'] = False
         self.client = self.app.test_client()
+        
+        # Register database cleanup
+        self.addCleanup(self.cleanup_test_database)
+
+    @staticmethod
+    def cleanup_test_database():
+        """Remove test database if it exists."""
+        try:
+            if os.path.exists('productivity.db'):
+                os.remove('productivity.db')
+        except OSError as e:
+            print(f"Warning: Could not remove test database: {e}")
+
+    def tearDown(self):
+        """Clean up test environment."""
+        self.cleanup_test_database()
 
     def test_app_exists(self):
         """Verify Flask app exists."""
@@ -26,6 +44,232 @@ def test_server_port(self):
         self.assertIn('PORT', self.app.config)
         self.assertEqual(5000, self.app.config['PORT'])
 
+    def test_health_check_structure(self):
+        """Test health check response structure."""
+        response = self.client.get('/health')
+        data = response.get_json()
+
+        self.assertIn('status', data)
+        self.assertIn('uptime_seconds', data)
+        self.assertIn('memory', data)
+        self.assertIn('system_load', data)
+        self.assertIn('database', data)
+
+        # Test memory structure
+        self.assertIn('total', data['memory'])
+        self.assertIn('available', data['memory'])
+        self.assertIn('percent', data['memory'])
+
+        # Test load structure
+        self.assertIn('1min', data['system_load'])
+        self.assertIn('5min', data['system_load'])
+        self.assertIn('15min', data['system_load'])
+
+    def test_health_check_values(self):
+        """Test health check returns valid values."""
+        response = self.client.get('/health')
+        data = response.get_json()
+
+        # Test value types
+        self.assertIsInstance(data['uptime_seconds'], (int, float))
+        self.assertIsInstance(data['memory']['total'], (int, float))
+        self.assertIsInstance(data['memory']['available'], (int, float))
+        self.assertIsInstance(data['memory']['percent'], (int, float))
+        self.assertIsInstance(data['system_load']['1min'], (int, float))
+
+    @unittest.mock.patch('sqlite3.connect')
+    def test_database_health_success(self, mock_connect):
+        """Test successful database health check"""
+        mock_cursor = unittest.mock.MagicMock()
+        mock_cursor.fetchone.return_value = (1,)
+        mock_connect.return_value.cursor.return_value = mock_cursor
+        
+        response = self.client.get('/health')
+        data = response.get_json()
+        
+        self.assertEqual(data['database']['status'], 'connected')
+        self.assertIsNone(data['database']['error'])
+        self.assertIsInstance(data['database']['response_time'], float)
+        self.assertEqual(response.status_code, 200)
+
+    @unittest.mock.patch('sqlite3.connect')
+    def test_database_health_timeout(self, mock_connect):
+        """Test database health check timeout"""
+        def slow_connection(*args, **kwargs):
+            time.sleep(0.2)  # Simulate slow connection
+            return unittest.mock.MagicMock()
+            
+        mock_connect.side_effect = slow_connection
+        
+        # Set a very short timeout for the test
+        self.app.config['DB_TIMEOUT'] = 0.1
+        
+        response = self.client.get('/health')
+        data = response.get_json()
+        
+        self.assertEqual(data['database']['status'], 'disconnected')
+        self.assertEqual(data['database']['error'], 'Connection timed out')
+        self.assertEqual(response.status_code, 503)
+
+    @unittest.mock.patch('sqlite3.connect')
+    def test_database_health_error(self, mock_connect):
+        """Test database connection error handling"""
+        mock_connect.side_effect = sqlite3.OperationalError('test error')
+        
+        response = self.client.get('/health')
+        data = response.get_json()
+        
+        self.assertEqual(data['database']['status'], 'disconnected')
+        self.assertEqual(data['database']['error'], 'Database error: test error')
+        self.assertEqual(response.status_code, 503)
+
+    @unittest.mock.patch('sqlite3.connect')
+    def test_database_response_time(self, mock_connect):
+        """Test database response time measurement"""
+        mock_cursor = unittest.mock.MagicMock()
+        mock_cursor.fetchone.return_value = (1,)
+        mock_connect.return_value.cursor.return_value = mock_cursor
+        
+        response = self.client.get('/health')
+        data = response.get_json()
+        
+        self.assertIsNotNone(data['database']['response_time'])
+        self.assertGreater(data['database']['response_time'], 0)
+        self.assertLess(data['database']['response_time'], 1000)  # Should be less than 1 second
+
+    @unittest.mock.patch('psutil.virtual_memory')
+    @unittest.mock.patch('psutil.getloadavg')
+    def test_health_check_degraded(self, mock_load, mock_memory):
+        """Test health check degraded status."""
+        # Mock high memory usage
+        mock_memory.return_value = unittest.mock.Mock(
+            total=16000000000,
+            available=1000000000,
+            percent=95.0
+        )
+        # Mock high system load
+        mock_load.return_value = (6.0, 5.0, 4.0)
+
+        response = self.client.get('/health')
+        data = response.get_json()
+
+        self.assertEqual(response.status_code, 503)
+        self.assertEqual(data['status'], 'degraded')
+
+    @unittest.mock.patch('psutil.virtual_memory')
+    @unittest.mock.patch('psutil.getloadavg')
+    def test_health_check_edge_memory(self, mock_load, mock_memory):
+        """Test health check with edge case memory values."""
+        # Test 100% memory usage
+        mock_memory.return_value = unittest.mock.Mock(
+            total=16000000000,
+            available=0,
+            percent=100.0
+        )
+        mock_load.return_value = (1.0, 1.0, 1.0)  # Normal load
+        
+        response = self.client.get('/health')
+        data = response.get_json()
+        self.assertEqual(data['status'], 'degraded')
+        self.assertEqual(response.status_code, 503)
+
+        # Test 0% memory usage
+        mock_memory.return_value = unittest.mock.Mock(
+            total=16000000000,
+            available=16000000000,
+            percent=0.0
+        )
+        
+        response = self.client.get('/health')
+        data = response.get_json()
+        self.assertEqual(data['status'], 'healthy')
+        self.assertEqual(response.status_code, 200)
+
+    @unittest.mock.patch('psutil.virtual_memory')
+    @unittest.mock.patch('psutil.getloadavg')
+    def test_health_check_edge_load(self, mock_load, mock_memory):
+        """Test health check with edge case load values."""
+        mock_memory.return_value = unittest.mock.Mock(
+            total=16000000000,
+            available=8000000000,
+            percent=50.0
+        )
+
+        # Test negative load (shouldn't happen in reality, but should handle gracefully)
+        mock_load.return_value = (-1.0, -1.0, -1.0)
+        response = self.client.get('/health')
+        data = response.get_json()
+        self.assertEqual(data['status'], 'healthy')
+        self.assertEqual(response.status_code, 200)
+
+        # Test extremely high load
+        mock_load.return_value = (999.9, 999.9, 999.9)
+        response = self.client.get('/health')
+        data = response.get_json()
+        self.assertEqual(data['status'], 'degraded')
+        self.assertEqual(response.status_code, 503)
+
+    @unittest.mock.patch('psutil.virtual_memory')
+    @unittest.mock.patch('psutil.getloadavg')
+    def test_health_check_custom_thresholds(self, mock_load, mock_memory):
+        """Test health check with custom threshold configuration."""
+        mock_memory.return_value = unittest.mock.Mock(
+            total=16000000000,
+            available=8000000000,
+            percent=75.0
+        )
+        mock_load.return_value = (3.0, 3.0, 3.0)
+
+        # Test with custom thresholds
+        self.app.config['HEALTH_CHECK_MEMORY_THRESHOLD'] = 80
+        self.app.config['HEALTH_CHECK_LOAD_THRESHOLD'] = 4
+        
+        response = self.client.get('/health')
+        data = response.get_json()
+        self.assertEqual(data['status'], 'healthy')
+        
+        # Adjust thresholds to trigger degradation
+        self.app.config['HEALTH_CHECK_MEMORY_THRESHOLD'] = 70
+        self.app.config['HEALTH_CHECK_LOAD_THRESHOLD'] = 2
+        
+        response = self.client.get('/health')
+        data = response.get_json()
+        self.assertEqual(data['status'], 'degraded')
+
+    @unittest.mock.patch('psutil.virtual_memory')
+    @unittest.mock.patch('psutil.getloadavg')
+    @unittest.mock.patch('sqlite3.connect')
+    def test_health_check_multiple_degradation(self, mock_connect, mock_load, mock_memory):
+        """Test health check with multiple degradation conditions."""
+        # Set up all metrics to trigger degradation
+        mock_memory.return_value = unittest.mock.Mock(
+            total=16000000000,
+            available=1000000000,
+            percent=95.0
+        )
+        mock_load.return_value = (10.0, 8.0, 6.0)
+        mock_connect.side_effect = sqlite3.OperationalError("test error")
+
+        response = self.client.get('/health')
+        data = response.get_json()
+        
+        self.assertEqual(data['status'], 'degraded')
+        self.assertEqual(response.status_code, 503)
+        self.assertTrue(data['memory']['percent'] > 90)
+        self.assertTrue(data['system_load']['1min'] > 5)
+        self.assertEqual(data['database']['status'], 'disconnected')
+
+    def test_health_check_uptime(self):
+        """Test health check uptime calculation."""
+        # Ensure START_TIME is set
+        self.app.config['START_TIME'] = time.time() - 10  # 10 seconds ago
+        
+        response = self.client.get('/health')
+        data = response.get_json()
+
+        self.assertGreaterEqual(data['uptime_seconds'], 10)
+        self.assertLess(data['uptime_seconds'], 11)  # Allow for small execution time
+
     @staticmethod
     def test_server_not_run():
         """Test server doesn't run when not main module."""