Refactor: Simplify connectivity analysis logic by breaking down into smaller helper functions for improved readability and maintainability

.gitea/workflows/ci.yml

@@ -395,7 +395,7 @@ jobs:
             -Dsonar.projectKey=ansible \
             -Dsonar.sources=. \
             -Dsonar.host.url=${SONAR_HOST_URL} \
-            -Dsonar.login=${SONAR_TOKEN} \
+            -Dsonar.token=${SONAR_TOKEN} \
             -X; then
             echo ""
             echo "❌ SonarScanner analysis failed!"

test_connectivity.py

@@ -141,39 +141,91 @@ class ConnectivityTester:
         return result
     
     def _analyze_connectivity(self, result: Dict) -> Tuple[str, str]:
-        """Analyze connectivity results and provide recommendations."""
+        """Analyze connectivity results and provide recommendations.
-        hostname = result['hostname']
-        primary_ip = result['primary_ip']
-        fallback_ip = result['fallback_ip']
 
-        # Primary IP works perfectly
+        Split into smaller helpers to keep this function's complexity low
-        if result['primary_ping'] and result['primary_ssh']:
+        while preserving the original decision logic.
-            return 'success', f"✓ {hostname} is fully accessible via primary IP {primary_ip}"
+        """
+        for handler in (
+            self._handle_primary_success,
+            self._handle_primary_ping_only,
+            self._handle_fallback_path,
+            self._handle_no_fallback,
+        ):
+            outcome = handler(result)
+            if outcome is not None:
+                return outcome
 
-        # Primary ping works but SSH fails
+        hostname = result["hostname"]
-        if result['primary_ping'] and not result['primary_ssh']:
+        return "unknown", f"? {hostname}: Unknown connectivity state"
-            error = result['primary_ssh_error']
-            if 'Permission denied' in error:
-                return 'ssh_key', f"⚠ {hostname}: SSH key issue on {primary_ip} - run: make copy-ssh-key HOST={hostname}"
-            elif 'Connection refused' in error:
-                return 'ssh_service', f"⚠ {hostname}: SSH service not running on {primary_ip}"
-            else:
-                return 'ssh_error', f"⚠ {hostname}: SSH error on {primary_ip} - {error}"
 
-        # Primary IP fails, test fallback
+    def _handle_primary_success(self, result: Dict) -> Optional[Tuple[str, str]]:
-        if not result['primary_ping'] and fallback_ip:
+        """Handle case where primary IP works perfectly."""
-            if result['fallback_ping'] and result['fallback_ssh']:
+        if result.get("primary_ping") and result.get("primary_ssh"):
-                return 'use_fallback', f"→ {hostname}: Switch to fallback IP {fallback_ip} (primary {primary_ip} failed)"
+            hostname = result["hostname"]
-            elif result['fallback_ping'] and not result['fallback_ssh']:
+            primary_ip = result["primary_ip"]
-                return 'fallback_ssh', f"⚠ {hostname}: Fallback IP {fallback_ip} reachable but SSH failed"
+            return "success", f"✓ {hostname} is fully accessible via primary IP {primary_ip}"
-            else:
+        return None
-                return 'both_failed', f"✗ {hostname}: Both primary {primary_ip} and fallback {fallback_ip} failed"
 
-        # No fallback IP and primary failed
+    def _handle_primary_ping_only(self, result: Dict) -> Optional[Tuple[str, str]]:
-        if not result['primary_ping'] and not fallback_ip:
+        """Handle cases where primary ping works but SSH fails."""
-            return 'no_fallback', f"✗ {hostname}: Primary IP {primary_ip} failed, no fallback available"
+        if result.get("primary_ping") and not result.get("primary_ssh"):
+            hostname = result["hostname"]
+            primary_ip = result["primary_ip"]
+            error = result.get("primary_ssh_error", "")
 
-        return 'unknown', f"? {hostname}: Unknown connectivity state"
+            if "Permission denied" in error:
+                return (
+                    "ssh_key",
+                    f"⚠ {hostname}: SSH key issue on {primary_ip} - run: make copy-ssh-key HOST={hostname}",
+                )
+            if "Connection refused" in error:
+                return "ssh_service", f"⚠ {hostname}: SSH service not running on {primary_ip}"
+            return "ssh_error", f"⚠ {hostname}: SSH error on {primary_ip} - {error}"
+
+        return None
+
+    def _handle_fallback_path(self, result: Dict) -> Optional[Tuple[str, str]]:
+        """Handle cases where primary fails and a fallback IP is defined."""
+        if result.get("primary_ping"):
+            return None
+
+        fallback_ip = result.get("fallback_ip")
+        if not fallback_ip:
+            return None
+
+        hostname = result["hostname"]
+        primary_ip = result["primary_ip"]
+
+        if result.get("fallback_ping") and result.get("fallback_ssh"):
+            return (
+                "use_fallback",
+                f"→ {hostname}: Switch to fallback IP {fallback_ip} (primary {primary_ip} failed)",
+            )
+
+        if result.get("fallback_ping") and not result.get("fallback_ssh"):
+            return (
+                "fallback_ssh",
+                f"⚠ {hostname}: Fallback IP {fallback_ip} reachable but SSH failed",
+            )
+
+        return (
+            "both_failed",
+            f"✗ {hostname}: Both primary {primary_ip} and fallback {fallback_ip} failed",
+        )
+
+    def _handle_no_fallback(self, result: Dict) -> Optional[Tuple[str, str]]:
+        """Handle cases where primary failed and no fallback IP is available."""
+        if result.get("primary_ping"):
+            return None
+
+        fallback_ip = result.get("fallback_ip")
+        if fallback_ip:
+            return None
+
+        hostname = result["hostname"]
+        primary_ip = result["primary_ip"]
+        return "no_fallback", f"✗ {hostname}: Primary IP {primary_ip} failed, no fallback available"
     
     def run_tests(self) -> List[Dict]:
         """Run connectivity tests for all hosts."""
@@ -264,8 +316,8 @@ class ConnectivityTester:
         
         # Auto-fallback suggestion
         if fallback_needed:
-            print(f"\n🤖 Or run auto-fallback to fix automatically:")
+            print("\n🤖 Or run auto-fallback to fix automatically:")
-            print(f"  make auto-fallback")
+            print("  make auto-fallback")
     
     def export_json(self, results: List[Dict], output_file: str):
         """Export results to JSON file."""