cephadm: address Copilot review on list-networks BGP parsing

Skip default routes in ip -j BGP JSON parsing, fix black line-length
in loopback check, and update list-networks CLI help for --allow-lo-routes
and --allow-bgp-routes.
Signed-off-by: Kobi Ginon <kginon@redhat.com>

diff --git a/src/cephadm/cephadm.py b/src/cephadm/cephadm.py
index b3133088961e3e03be9589b88d81a11d3c7c683b..ba990aed6ad73e223688347fe08c9e8167087a78 100755 (executable)
--- a/src/cephadm/cephadm.py
+++ b/src/cephadm/cephadm.py
@@ -5023,13 +5023,13 @@ def _get_parser():
         '--allow-lo-routes',
         action='store_true',
         default=False,
-        help='Reserved for future filtering of loopback-related routes',
+        help='Include loopback (lo) routes in the listing (default: omit)',
     )
     parser_list_networks.add_argument(
         '--allow-bgp-routes',
         action='store_true',
         default=False,
-        help='Reserved for future filtering of BGP-derived routes',
+        help='Merge BGP routes from ip -j route ls proto bgp (default: omit)',
     )
     parser_list_networks.set_defaults(func=command_list_networks)
 

diff --git a/src/cephadm/cephadmlib/host_facts.py b/src/cephadm/cephadmlib/host_facts.py
index c4acd4a6b69818782d25455c5633d8c34190f9e6..c169632b07209aaedb8ce2fed37f48083ebe6c30 100644 (file)
--- a/src/cephadm/cephadmlib/host_facts.py
+++ b/src/cephadm/cephadmlib/host_facts.py
@@ -909,45 +909,37 @@ def list_rdma(ctx: CephadmContext) -> List[Dict[str, str]]:
     return result
 
 
-_IPV4_LOOPBACK_SPACE = ipaddress.ip_network('127.0.0.0/8')
-_IPV6_LOOPBACK_HOST = ipaddress.ip_network('::1/128')
-
+def _list_ipv4_networks(
+    ctx: CephadmContext,
+    allow_lo_routes: bool = False,
+    allow_bgp_routes: bool = False,
+) -> Dict[str, Dict[str, Set[str]]]:
+    """IPv4 networks from ``ip route ls`` (and optional ``proto bgp`` supplement).
 
-def _is_ipv4_loopback(net_s: str) -> bool:
-    """True if *net_s* is an IPv4 prefix wholly within host loopback 127.0.0.0/8."""
-    try:
-        n = ipaddress.ip_network(net_s, strict=False)
-    except ValueError:
-        return False
-    return (
-        n.version == 4
-        and n.network_address in _IPV4_LOOPBACK_SPACE
-        and n.broadcast_address in _IPV4_LOOPBACK_SPACE
+    When *allow_lo_routes* is false, routes on ``lo`` are ignored by the parsers.
+    When true, ``_parse_ipv4_lo_route`` supplements the base table.
+    """
+    execstr: Optional[str] = find_executable('ip')
+    if not execstr:
+        raise FileNotFoundError("unable to find 'ip' command")
+    out, _, _ = call_throws(
+        ctx,
+        [execstr, 'route', 'ls'],
+        verbosity=CallVerbosity.QUIET_UNLESS_ERROR,
     )
-
-
-def _is_ipv6_loopback(net_s: str) -> bool:
-    """True if *net_s* is exactly the host loopback prefix ::1/128."""
-    try:
-        n = ipaddress.ip_network(net_s, strict=False)
-    except ValueError:
-        return False
-    if n.version != 6:
-        return False
-    return n == _IPV6_LOOPBACK_HOST
-
-
-def _merge_ipv4_network_dicts(
-    dst: Dict[str, Dict[str, Set[str]]], add: Dict[str, Dict[str, Set[str]]]
-) -> None:
-    """Merge *add* into *dst* (nested sets)."""
-    for net, ifaces in add.items():
-        if net not in dst:
-            dst[net] = {}
-        for iface, ips in ifaces.items():
-            if iface not in dst[net]:
-                dst[net][iface] = set()
-            dst[net][iface].update(ips)
+    res = _parse_ipv4_route(out, allow_lo_routes)
+    if allow_lo_routes:
+        _merge_ipv4_network_dicts(res, _parse_ipv4_lo_route(out))
+    if allow_bgp_routes:
+        bgp_out, _, _ = call_throws(
+            ctx,
+            [execstr, '-j', 'route', 'ls', 'proto', 'bgp'],
+            verbosity=CallVerbosity.QUIET_UNLESS_ERROR,
+        )
+        _merge_ipv4_network_dicts(
+            res, _parse_ipv4_bgp_route(bgp_out, allow_lo_routes)
+        )
+    return res
 
 
 def _parse_ipv4_route(
@@ -1056,9 +1048,10 @@ def _parse_bgp_routes_json(
 
     is_v4 = version == 4
     host_plen = 32 if is_v4 else 128
+    is_loopback_net = _is_ipv4_loopback if is_v4 else _is_ipv6_loopback
 
     def put(net: str, iface: str, ip: str) -> None:
-        if _is_ipv4_loopback(net) if is_v4 else _is_ipv6_loopback(net):
+        if is_loopback_net(net):
             return
         if net not in r:
             r[net] = {}
@@ -1070,7 +1063,7 @@ def _parse_bgp_routes_json(
         if not isinstance(route, dict):
             continue
         dst = route.get('dst')
-        if not dst:
+        if not dst or dst == 'default':
             continue
 
         net = _route_dst_to_network(dst, version)
@@ -1127,39 +1120,6 @@ def _parse_ipv6_bgp_route(
     return _parse_bgp_routes_json(out, 6, allow_lo_routes)
 
 
-def _list_ipv4_networks(
-    ctx: CephadmContext,
-    allow_lo_routes: bool = False,
-    allow_bgp_routes: bool = False,
-) -> Dict[str, Dict[str, Set[str]]]:
-    """IPv4 networks from ``ip route ls`` (and optional ``proto bgp`` supplement).
-
-    When *allow_lo_routes* is false, routes on ``lo`` are ignored by the parsers.
-    When true, ``_parse_ipv4_lo_route`` supplements the base table.
-    """
-    execstr: Optional[str] = find_executable('ip')
-    if not execstr:
-        raise FileNotFoundError("unable to find 'ip' command")
-    out, _, _ = call_throws(
-        ctx,
-        [execstr, 'route', 'ls'],
-        verbosity=CallVerbosity.QUIET_UNLESS_ERROR,
-    )
-    res = _parse_ipv4_route(out, allow_lo_routes)
-    if allow_lo_routes:
-        _merge_ipv4_network_dicts(res, _parse_ipv4_lo_route(out))
-    if allow_bgp_routes:
-        bgp_out, _, _ = call_throws(
-            ctx,
-            [execstr, '-j', 'route', 'ls', 'proto', 'bgp'],
-            verbosity=CallVerbosity.QUIET_UNLESS_ERROR,
-        )
-        _merge_ipv4_network_dicts(
-            res, _parse_ipv4_bgp_route(bgp_out, allow_lo_routes)
-        )
-    return res
-
-
 def _list_ipv6_networks(
     ctx: CephadmContext,
     allow_lo_routes: bool = False,
@@ -1247,3 +1207,44 @@ def _parse_ipv6_route(
             r[net[0]][iface].add(ip)
 
     return r
+
+
+_IPV4_LOOPBACK_SPACE = ipaddress.ip_network('127.0.0.0/8')
+_IPV6_LOOPBACK_HOST = ipaddress.ip_network('::1/128')
+
+
+def _is_ipv4_loopback(net_s: str) -> bool:
+    """True if *net_s* is an IPv4 prefix wholly within host loopback 127.0.0.0/8."""
+    try:
+        n = ipaddress.ip_network(net_s, strict=False)
+    except ValueError:
+        return False
+    return (
+        n.version == 4
+        and n.network_address in _IPV4_LOOPBACK_SPACE
+        and n.broadcast_address in _IPV4_LOOPBACK_SPACE
+    )
+
+
+def _is_ipv6_loopback(net_s: str) -> bool:
+    """True if *net_s* is exactly the host loopback prefix ::1/128."""
+    try:
+        n = ipaddress.ip_network(net_s, strict=False)
+    except ValueError:
+        return False
+    if n.version != 6:
+        return False
+    return n == _IPV6_LOOPBACK_HOST
+
+
+def _merge_ipv4_network_dicts(
+    dst: Dict[str, Dict[str, Set[str]]], add: Dict[str, Dict[str, Set[str]]]
+) -> None:
+    """Merge *add* into *dst* (nested sets)."""
+    for net, ifaces in add.items():
+        if net not in dst:
+            dst[net] = {}
+        for iface, ips in ifaces.items():
+            if iface not in dst[net]:
+                dst[net][iface] = set()
+            dst[net][iface].update(ips)

diff --git a/src/cephadm/tests/test_networks.py b/src/cephadm/tests/test_networks.py
index f854dfab1fb7689e01c7db1dbb89126678382930..6401c53675e2a05b96a78c53cd2dc947731d90c7 100644 (file)
--- a/src/cephadm/tests/test_networks.py
+++ b/src/cephadm/tests/test_networks.py
@@ -48,8 +48,6 @@ class TestEndPoint:
 
 
 class TestCommandListNetworks:
-    # fmt: off
-    # Keep table-style parametrization unchanged for reviewable diffs (see PR discussion).
     @pytest.mark.parametrize("test_input, expected", [
         (
             dedent("""
@@ -108,11 +106,9 @@ class TestCommandListNetworks:
             }
         ),
     ])
-    # fmt: on
     def test_parse_ipv4_route(self, test_input, expected):
         assert _parse_ipv4_route(test_input) == expected
 
-    # fmt: off
     @pytest.mark.parametrize("test_routes, test_ips, expected", [
         (
             dedent("""
@@ -240,8 +236,7 @@ class TestCommandListNetworks:
             ::1 dev lo proto kernel metric 256 pref medium
             fe80::/64 dev ceph-brx proto kernel metric 256 pref medium
             fe80::/64 dev brx.0 proto kernel metric 256 pref medium
-            default via fe80::327c:5e00:6487:71e0 dev enp3s0f1 proto ra metric 1024 expires 1790sec hoplimit 64 pref medium
-            """),
+            default via fe80::327c:5e00:6487:71e0 dev enp3s0f1 proto ra metric 1024 expires 1790sec hoplimit 64 pref medium            """),
             dedent("""
             1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536 state UNKNOWN qlen 1000
                 inet6 ::1/128 scope host
@@ -264,7 +259,6 @@ class TestCommandListNetworks:
             }
         ),
     ])
-    # fmt: on
     def test_parse_ipv6_route(self, test_routes, test_ips, expected):
         assert _parse_ipv6_route(test_routes, test_ips) == expected
 
@@ -311,18 +305,8 @@ fe80000000000000505400fffe04c154 03 40 20 80     eth1
 
     @mock.patch('cephadmlib.host_facts.call_throws')
     @mock.patch('cephadmlib.host_facts.find_executable')
-    def test_command_list_networks(
-        self, _find_exe, _call_throws, cephadm_fs, capsys
-    ):
-        _call_throws.side_effect = [
-            (
-                '10.4.0.1 dev tun0 proto kernel scope link src 10.4.0.2 metric 50\n',
-                '',
-                '',
-            ),
-            ('', '', ''),
-            ('', '', ''),
-        ]
+    def test_command_list_networks(self, _find_exe, _call_throws, cephadm_fs, capsys):
+        _call_throws.return_value = ('10.4.0.1 dev tun0 proto kernel scope link src 10.4.0.2 metric 50\n', '', '')
         _find_exe.return_value = 'ip'
         with with_cephadm_ctx([]) as ctx:
             _cephadm.command_list_networks(ctx)

diff --git a/src/cephadm/tests/test_networks_lo_routes.py b/src/cephadm/tests/test_networks_lo_routes.py
index 4e02c972e3562fd25fa9447e3b9e7b16d3cf7ba2..4ce5fa75b3df92e956f8b1ea9b365eb31ef4add0 100644 (file)
--- a/src/cephadm/tests/test_networks_lo_routes.py
+++ b/src/cephadm/tests/test_networks_lo_routes.py
@@ -253,6 +253,43 @@ class TestLoopbackRouteParsing:
             '10.168.100.10/32': {'lo': {'10.168.100.10'}},
         }
 
+    def test_parse_ipv4_bgp_route_skips_default(self):
+        bgp_input = json.dumps(
+            [
+                {
+                    'dst': 'default',
+                    'protocol': 'bgp',
+                    'prefsrc': '192.168.100.11',
+                    'flags': [],
+                    'nexthops': [
+                        {
+                            'gateway': '169.254.0.1',
+                            'dev': 'ens1f0np0',
+                            'weight': 1,
+                            'flags': ['onlink'],
+                        },
+                    ],
+                },
+                {
+                    'dst': '192.168.100.5',
+                    'protocol': 'bgp',
+                    'prefsrc': '192.168.100.11',
+                    'flags': [],
+                    'nexthops': [
+                        {
+                            'gateway': '169.254.0.1',
+                            'dev': 'ens1f0np0',
+                            'weight': 1,
+                            'flags': ['onlink'],
+                        },
+                    ],
+                },
+            ]
+        )
+        assert _parse_ipv4_bgp_route(bgp_input) == {
+            '192.168.100.5/32': {'ens1f0np0': {'192.168.100.11'}},
+        }
+
     def test_parse_ipv4_bgp_route_lo_from_full_table(self):
         # ``ip -j route ls`` entry for ``10.10.10.10 dev lo proto bgp``.
         bgp_input = json.dumps(