Merge pull request #1 from stackopshq/fix/cleanup-router-network-deps

fix(project-cleanup): robust router + network teardown

Author: kallioli

orca_cli/commands/project.py

@@ -187,6 +187,67 @@ def _before_cutoff(resource: dict, cutoff: datetime | None) -> bool:
         return True
 
 
+# device_owner values for ports attached to a router that need to be detached
+# via remove_router_interface before the router can be deleted. Router gateway
+# ports (network:router_gateway, network:router_centralized_snat) are released
+# by clearing external_gateway_info instead.
+_ROUTER_INTERFACE_OWNERS = (
+    "network:router_interface",
+    "network:router_interface_distributed",
+    "network:ha_router_replicated_interface",
+)
+
+
+def _delete_router(client, rid: str) -> None:
+    """Detach all router interfaces and clear the external gateway, then delete."""
+    net_svc = NetworkService(client)
+
+    # Clear external gateway (releases router_gateway / centralized_snat ports).
+    try:
+        net_svc.update_router(rid, {"external_gateway_info": None})
+    except Exception:
+        pass
+
+    # Detach every router-interface port, regardless of legacy/DVR/HA flavor.
+    try:
+        ports = net_svc.find_ports(params={"device_id": rid})
+    except Exception:
+        ports = []
+    for p in ports:
+        owner = p.get("device_owner", "")
+        if not any(owner.startswith(o) for o in _ROUTER_INTERFACE_OWNERS):
+            continue
+        try:
+            net_svc.remove_router_interface(rid, {"port_id": p["id"]})
+        except Exception:
+            pass
+
+    net_svc.delete_router(rid)
+
+
+def _delete_network(client, nid: str) -> None:
+    """Delete orphan/compute leftover ports, then delete the network."""
+    net_svc = NetworkService(client)
+
+    # Ports owned by Neutron itself (dhcp, router, floatingip, ha_*) are cleaned
+    # up by their respective delete paths; we only remove truly orphaned ports
+    # or stale compute VIFs whose server has already been deleted.
+    try:
+        ports = net_svc.find_ports(params={"network_id": nid})
+    except Exception:
+        ports = []
+    for p in ports:
+        owner = p.get("device_owner", "")
+        if owner and not owner.startswith("compute:"):
+            continue
+        try:
+            net_svc.delete_port(p["id"])
+        except Exception:
+            pass
+
+    net_svc.delete(nid)
+
+
 def _delete_one(client, rtype: str, rid: str, rname: str) -> bool:
     """Delete a single resource by type. Returns True on success."""
     label = f"{rtype} {rname} ({rid})"
@@ -200,23 +261,9 @@ def _delete_one(client, rtype: str, rid: str, rname: str) -> bool:
         elif rtype == "floating-ip":
             NetworkService(client).delete_floating_ip(rid)
         elif rtype == "router":
-            # Detach all subnet interfaces before deleting
-            net_svc = NetworkService(client)
-            try:
-                ports = net_svc.find_ports(params={
-                    "device_id": rid,
-                    "device_owner": "network:router_interface",
-                })
-            except Exception:
-                ports = []
-            for p in ports:
-                try:
-                    net_svc.remove_router_interface(rid, {"port_id": p["id"]})
-                except Exception:
-                    pass
-            net_svc.delete_router(rid)
+            _delete_router(client, rid)
         elif rtype == "network":
-            NetworkService(client).delete(rid)
+            _delete_network(client, rid)
         elif rtype == "security-group":
             NetworkService(client).delete_security_group(rid)
         elif rtype == "volume":

tests/test_project_cleanup.py

@@ -0,0 +1,157 @@
+"""Tests for ``orca project cleanup`` helpers handling router and network deps."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+from orca_cli.commands import project as proj_mod
+
+
+def _make_net_svc(monkeypatch, **attrs):
+    net_svc = MagicMock()
+    for k, v in attrs.items():
+        setattr(net_svc, k, v)
+    monkeypatch.setattr(proj_mod, "NetworkService", lambda _c: net_svc)
+    return net_svc
+
+
+class TestDeleteRouter:
+    def test_clears_gateway_before_detaching_and_deleting(self, monkeypatch):
+        net = _make_net_svc(monkeypatch)
+        net.find_ports.return_value = []
+
+        proj_mod._delete_router(object(), "r-1")
+
+        net.update_router.assert_called_once_with(
+            "r-1", {"external_gateway_info": None}
+        )
+        net.delete_router.assert_called_once_with("r-1")
+
+    def test_detaches_all_router_interface_flavors(self, monkeypatch):
+        net = _make_net_svc(monkeypatch)
+        net.find_ports.return_value = [
+            {"id": "p-legacy", "device_owner": "network:router_interface"},
+            {"id": "p-dvr",
+             "device_owner": "network:router_interface_distributed"},
+            {"id": "p-ha",
+             "device_owner": "network:ha_router_replicated_interface"},
+        ]
+
+        proj_mod._delete_router(object(), "r-1")
+
+        assert net.remove_router_interface.call_count == 3
+        detached = {
+            c.args[1]["port_id"]
+            for c in net.remove_router_interface.call_args_list
+        }
+        assert detached == {"p-legacy", "p-dvr", "p-ha"}
+
+    def test_ignores_non_interface_ports(self, monkeypatch):
+        """Gateway / SNAT / foreign ports must not go through remove_router_interface."""
+        net = _make_net_svc(monkeypatch)
+        net.find_ports.return_value = [
+            {"id": "p-gw", "device_owner": "network:router_gateway"},
+            {"id": "p-snat", "device_owner": "network:router_centralized_snat"},
+            {"id": "p-dhcp", "device_owner": "network:dhcp"},
+            {"id": "p-orphan", "device_owner": ""},
+        ]
+
+        proj_mod._delete_router(object(), "r-1")
+
+        net.remove_router_interface.assert_not_called()
+        net.delete_router.assert_called_once_with("r-1")
+
+    def test_lists_ports_by_device_id_only(self, monkeypatch):
+        """Broader filter: device_owner is applied Python-side, not in the query."""
+        net = _make_net_svc(monkeypatch)
+        net.find_ports.return_value = []
+
+        proj_mod._delete_router(object(), "r-xyz")
+
+        net.find_ports.assert_called_once_with(params={"device_id": "r-xyz"})
+
+    def test_continues_when_update_router_fails(self, monkeypatch):
+        """Some clouds reject gateway clear if no gateway set — must not abort."""
+        net = _make_net_svc(monkeypatch)
+        net.update_router.side_effect = RuntimeError("no gateway")
+        net.find_ports.return_value = [
+            {"id": "p1", "device_owner": "network:router_interface"},
+        ]
+
+        proj_mod._delete_router(object(), "r-1")
+
+        net.remove_router_interface.assert_called_once_with(
+            "r-1", {"port_id": "p1"}
+        )
+        net.delete_router.assert_called_once_with("r-1")
+
+    def test_continues_when_find_ports_fails(self, monkeypatch):
+        net = _make_net_svc(monkeypatch)
+        net.find_ports.side_effect = RuntimeError("neutron down")
+
+        proj_mod._delete_router(object(), "r-1")
+
+        net.remove_router_interface.assert_not_called()
+        net.delete_router.assert_called_once_with("r-1")
+
+    def test_continues_when_remove_interface_fails(self, monkeypatch):
+        net = _make_net_svc(monkeypatch)
+        net.find_ports.return_value = [
+            {"id": "p1", "device_owner": "network:router_interface"},
+            {"id": "p2", "device_owner": "network:router_interface"},
+        ]
+        net.remove_router_interface.side_effect = [RuntimeError("boom"), None]
+
+        proj_mod._delete_router(object(), "r-1")
+
+        assert net.remove_router_interface.call_count == 2
+        net.delete_router.assert_called_once_with("r-1")
+
+
+class TestDeleteNetwork:
+    def test_deletes_orphan_and_stale_compute_ports(self, monkeypatch):
+        net = _make_net_svc(monkeypatch)
+        net.find_ports.return_value = [
+            {"id": "p-orphan", "device_owner": ""},
+            {"id": "p-vif", "device_owner": "compute:nova"},
+            {"id": "p-dhcp", "device_owner": "network:dhcp"},
+            {"id": "p-router", "device_owner": "network:router_interface"},
+            {"id": "p-fip", "device_owner": "network:floatingip"},
+        ]
+
+        proj_mod._delete_network(object(), "n-1")
+
+        deleted = {c.args[0] for c in net.delete_port.call_args_list}
+        assert deleted == {"p-orphan", "p-vif"}
+        net.delete.assert_called_once_with("n-1")
+
+    def test_lists_ports_by_network_id(self, monkeypatch):
+        net = _make_net_svc(monkeypatch)
+        net.find_ports.return_value = []
+
+        proj_mod._delete_network(object(), "n-42")
+
+        net.find_ports.assert_called_once_with(params={"network_id": "n-42"})
+        net.delete.assert_called_once_with("n-42")
+
+    def test_continues_when_port_delete_fails(self, monkeypatch):
+        net = _make_net_svc(monkeypatch)
+        net.find_ports.return_value = [
+            {"id": "p1", "device_owner": ""},
+            {"id": "p2", "device_owner": "compute:nova"},
+        ]
+        net.delete_port.side_effect = [RuntimeError("boom"), None]
+
+        proj_mod._delete_network(object(), "n-1")
+
+        assert net.delete_port.call_count == 2
+        net.delete.assert_called_once_with("n-1")
+
+    def test_continues_when_find_ports_fails(self, monkeypatch):
+        net = _make_net_svc(monkeypatch)
+        net.find_ports.side_effect = RuntimeError("neutron down")
+
+        proj_mod._delete_network(object(), "n-1")
+
+        net.delete_port.assert_not_called()
+        net.delete.assert_called_once_with("n-1")