diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index aa5b249..76ca744 100644 --- a/.gitea/workflows/ci.yml +++ b/.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!" diff --git a/test_connectivity.py b/test_connectivity.py index feeb4f3..1f82db0 100644 --- a/test_connectivity.py +++ b/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.""" - hostname = result['hostname'] - primary_ip = result['primary_ip'] - fallback_ip = result['fallback_ip'] - - # Primary IP works perfectly - if result['primary_ping'] and result['primary_ssh']: - return 'success', f"✓ {hostname} is fully accessible via primary IP {primary_ip}" - - # Primary ping works but SSH fails - if result['primary_ping'] and not result['primary_ssh']: - 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 - if not result['primary_ping'] and fallback_ip: - if result['fallback_ping'] and result['fallback_ssh']: - return 'use_fallback', f"→ {hostname}: Switch to fallback IP {fallback_ip} (primary {primary_ip} failed)" - elif result['fallback_ping'] and not result['fallback_ssh']: - return 'fallback_ssh', f"⚠ {hostname}: Fallback IP {fallback_ip} reachable but SSH failed" - else: - return 'both_failed', f"✗ {hostname}: Both primary {primary_ip} and fallback {fallback_ip} failed" - - # No fallback IP and primary failed - if not result['primary_ping'] and not fallback_ip: - return 'no_fallback', f"✗ {hostname}: Primary IP {primary_ip} failed, no fallback available" - - return 'unknown', f"? {hostname}: Unknown connectivity state" + """Analyze connectivity results and provide recommendations. + + Split into smaller helpers to keep this function's complexity low + while preserving the original decision logic. + """ + 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 + + hostname = result["hostname"] + return "unknown", f"? {hostname}: Unknown connectivity state" + + def _handle_primary_success(self, result: Dict) -> Optional[Tuple[str, str]]: + """Handle case where primary IP works perfectly.""" + if result.get("primary_ping") and result.get("primary_ssh"): + hostname = result["hostname"] + primary_ip = result["primary_ip"] + return "success", f"✓ {hostname} is fully accessible via primary IP {primary_ip}" + return None + + def _handle_primary_ping_only(self, result: Dict) -> Optional[Tuple[str, str]]: + """Handle cases where primary ping works but SSH fails.""" + 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", "") + + 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(f" make auto-fallback") + print("\n🤖 Or run auto-fallback to fix automatically:") + print(" make auto-fallback") def export_json(self, results: List[Dict], output_file: str): """Export results to JSON file."""