server: add circuit breaker to rpc network_status

Author: pinheadmz

src/cli/network.py

@@ -122,9 +122,12 @@ def status(network: str):
     assert isinstance(result, list), "Result is not a list"  # Make mypy happy
     for tank in result:
         lightning_status = ""
+        circuitbreaker_status = ""
         if "lightning_status" in tank:
             lightning_status = f"\tLightning: {tank['lightning_status']}"
-        print(f"Tank: {tank['tank_index']} \tBitcoin: {tank['bitcoin_status']}{lightning_status}")
+        if "circuitbreaker_status" in tank:
+            circuitbreaker_status = f"\tCircuit Breaker: {tank['circuitbreaker_status']}"
+        print(f"Tank: {tank['tank_index']} \tBitcoin: {tank['bitcoin_status']}{lightning_status}{circuitbreaker_status}")
 
 
 @network.command()

src/warnet/lnnode.py

@@ -27,6 +27,12 @@ def __str__(self):
     def status(self) -> RunningStatus:
         return self.warnet.container_interface.get_status(self.tank.index, ServiceType.LIGHTNING)
 
+    @property
+    def cb_status(self) -> RunningStatus:
+        if not self.cb:
+            return None
+        return self.warnet.container_interface.get_status(self.tank.index, ServiceType.CIRCUITBREAKER)
+
     @exponential_backoff(max_retries=20, max_delay=300)
     @handle_json
     def lncli(self, cmd) -> dict:

src/warnet/server.py

@@ -565,6 +565,8 @@ def network_status(self, network: str = "warnet") -> list[dict]:
                 status = {"tank_index": tank.index, "bitcoin_status": tank.status.name.lower()}
                 if tank.lnnode is not None:
                     status["lightning_status"] = tank.lnnode.status.name.lower()
+                    if tank.lnnode.cb is not None:
+                        status["circuitbreaker_status"] = tank.lnnode.cb_status.name.lower()
                 stats.append(status)
             return stats
         except Exception as e:

test/test_base.py

@@ -175,6 +175,12 @@ def check_status():
                     if lightning_status not in stats:
                         stats[lightning_status] = 0
                     stats[lightning_status] += 1
+                if "circuitbreaker_status" in tank:
+                    stats["total"] += 1
+                    circuitbreaker_status = tank["circuitbreaker_status"]
+                    if circuitbreaker_status not in stats:
+                        stats[circuitbreaker_status] = 0
+                    stats[circuitbreaker_status] += 1
             print(f"Waiting for all tanks to reach '{target}': {stats}")
             # All tanks are running, proceed
             return target in stats and stats[target] == stats["total"]