fix(ha): clear stale ha_failed when refreshing HA active state

panos/base.py

@@ -4765,6 +4765,8 @@ def refresh_ha_active(self):
         else:
             for fw, state in ((self, self_state), (self.ha_peer, peer_state)):
                 fw._ha_active = state == "active"
+                # Reached this device for its state, so it is no longer failed.
+                fw.ha_failed = False
             return self_state
 
     def synchronize_config(self):

tests/test_firewall.py

@@ -14,6 +14,12 @@
 
 import unittest
 
+# Python 2 has no unittest.mock; fall back to the standalone mock package.
+try:
+    from unittest import mock
+except ImportError:
+    import mock
+
 import panos
 import panos.firewall
 
@@ -51,5 +57,41 @@ def test_id_returns_no_id(self):
         self.assertEqual(expected, ret_val)
 
 
+class TestFirewallHa(unittest.TestCase):
+    def _ha_pair(self):
+        fw = panos.firewall.Firewall("10.0.0.1", "user", "pass")
+        peer = panos.firewall.Firewall("10.0.0.2", "user", "pass")
+        fw.set_ha_peers(peer)
+        return fw, peer
+
+    def test_refresh_ha_active_clears_stale_ha_failed(self):
+        """A live state refresh clears ha_failed on the reachable device."""
+        fw, peer = self._ha_pair()
+        # A transient error failed this device and promoted the peer to active.
+        fw.set_failed()
+        self.assertTrue(fw.ha_failed)
+        # Live HA now reports this device is active again.
+        fw.show_highavailability_state = mock.Mock(return_value=("active", None))
+        peer.show_highavailability_state = mock.Mock(return_value=("passive", None))
+
+        fw.refresh_ha_active()
+
+        self.assertFalse(fw.ha_failed)
+
+    def test_refresh_ha_active_keeps_ha_failed_when_state_not_authoritative(self):
+        """A non-authoritative state (initial/disabled) leaves ha_failed untouched."""
+        fw, peer = self._ha_pair()
+        # A transient error failed this device.
+        fw.set_failed()
+        self.assertTrue(fw.ha_failed)
+        # HA is still initializing, so the reported state is not authoritative.
+        fw.show_highavailability_state = mock.Mock(return_value=("initial", None))
+        peer.show_highavailability_state = mock.Mock(return_value=("active", None))
+
+        fw.refresh_ha_active()
+
+        self.assertTrue(fw.ha_failed)
+
+
 if __name__ == "__main__":
     unittest.main()