Updated database connection validation

Author: nullchimp

.github/workflows/deploy-azure.yml

@@ -207,3 +207,29 @@ jobs:
           fi
           exit 1
         fi
+
+    #─────────────── Verify Memgraph connectivity ─────────────────
+    - name: Set up Python for Memgraph verification
+      uses: actions/setup-python@v4
+      with:
+        python-version: '3.9'
+
+    - name: Install dependencies for Memgraph verification
+      run: |
+        python -m pip install --upgrade pip
+        pip install python-dotenv mgclient
+
+    - name: Export Memgraph connection details
+      run: |
+        # Get service IP
+        MEMGRAPH_IP=$(kubectl get service memgraph -o jsonpath='{.status.loadBalancer.ingress[0].ip}')
+        echo "MEMGRAPH_URI=$MEMGRAPH_IP" >> $GITHUB_ENV
+        echo "MEMGRAPH_PORT=7687" >> $GITHUB_ENV
+        echo "MEMGRAPH_USERNAME=${{ env.DB_USERNAME }}" >> $GITHUB_ENV
+        echo "MEMGRAPH_PASSWORD=${{ env.DB_PASSWORD }}" >> $GITHUB_ENV
+        echo "Memgraph connection details exported: $MEMGRAPH_IP:7687"
+
+    - name: Verify Memgraph connection and authentication
+      run: |
+        echo "Verifying Memgraph connection and authentication..."
+        python scripts/verify_memgraph_connection.py

scripts/get_memgraph_connection.py

@@ -0,0 +1,63 @@
+from dotenv import load_dotenv
+import os
+import mgclient
+import sys
+import time
+from typing import Tuple, Optional
+
+def verify_memgraph_connection(max_retries: int = 5, retry_delay: int = 5) -> Tuple[bool, Optional[str]]:
+    host = os.environ.get("MEMGRAPH_URI", "127.0.0.1")
+    port = int(os.environ.get("MEMGRAPH_PORT", "7687"))
+    username = "test" or os.environ.get("MEMGRAPH_USERNAME", "memgraph")
+    password = os.environ.get("MEMGRAPH_PASSWORD", "memgraph")
+    
+    print(f"Verifying connection to Memgraph at {host}:{port}")
+
+    if host == "localhost":
+        host = "127.0.0.1"
+    
+    for attempt in range(1, max_retries + 1):
+        try:
+            print(f"Attempt {attempt}/{max_retries}: Connecting to Memgraph")
+            conn = mgclient.connect(
+                host=host, 
+                port=port,
+                username=username,
+                password=password
+            )
+            
+            conn.autocommit = True
+            cursor = conn.cursor()
+            
+            # Basic connectivity check
+            cursor.execute("RETURN 'Connection successful!' AS message")
+            result = cursor.fetchone()[0]
+            print(f"Success: {result}")
+            
+            # Validate authentication works
+            cursor.execute("CREATE (n:TestNode {name: 'connection_test'}) RETURN n")
+            cursor.execute("MATCH (n:TestNode {name: 'connection_test'}) DELETE n")
+            print("Authentication validation successful")
+            
+            cursor.close()
+            conn.close()
+            return True, None
+        except Exception as e:
+            error_message = f"Connection attempt {attempt} failed: {str(e)}"
+            print(error_message)
+            if attempt < max_retries:
+                print(f"Retrying in {retry_delay} seconds...")
+                time.sleep(retry_delay)
+            else:
+                return False, error_message
+
+if __name__ == "__main__":
+    load_dotenv(override=True)
+    success, error = verify_memgraph_connection()
+    
+    if not success:
+        print(f"Failed to connect to Memgraph: {error}")
+        sys.exit(1)
+        
+    print("Successfully connected to and validated Memgraph")
+    sys.exit(0)

scripts/verify_memgraph_connection.py

@@ -0,0 +1,45 @@
+import unittest
+from unittest.mock import patch, MagicMock
+import sys
+import os
+
+# Add the scripts directory to the path so we can import the module
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from scripts.verify_memgraph_connection import verify_memgraph_connection
+
+class TestVerifyMemgraphConnection(unittest.TestCase):
+    
+    @patch('mgclient.connect')
+    def test_successful_connection(self, mock_connect):
+        # Set up mocks
+        mock_cursor = MagicMock()
+        mock_cursor.fetchone.return_value = ["Connection successful!"]
+        
+        mock_conn = MagicMock()
+        mock_conn.cursor.return_value = mock_cursor
+        mock_connect.return_value = mock_conn
+        
+        # Call the function
+        success, error = verify_memgraph_connection(max_retries=1)
+        
+        # Assertions
+        self.assertTrue(success)
+        self.assertIsNone(error)
+        mock_connect.assert_called_once()
+        self.assertEqual(mock_cursor.execute.call_count, 3)
+        
+    @patch('mgclient.connect')
+    def test_failed_connection(self, mock_connect):
+        # Set up mocks
+        mock_connect.side_effect = Exception("Connection error")
+        
+        # Call the function
+        success, error = verify_memgraph_connection(max_retries=1, retry_delay=0)
+        
+        # Assertions
+        self.assertFalse(success)
+        self.assertIn("Connection error", error)
+        mock_connect.assert_called_once()
+
+if __name__ == '__main__':
+    unittest.main()