Adding tests for dhcp client and dhcp6 client dbus state interface

diff --git a/test/test-network/systemd-networkd-tests.py b/test/test-network/systemd-networkd-tests.py
index bc60e9385045699bbaf9a5bf3f2522ba11033ec7..58bf23af501c2b85ab506eb259f027e3af8240d6 100755 (executable)
--- a/test/test-network/systemd-networkd-tests.py
+++ b/test/test-network/systemd-networkd-tests.py
@@ -5071,6 +5071,45 @@ class NetworkdDHCPClientTests(unittest.TestCase, Utilities):
         self.assertIn('DHCPREPLY(veth-peer)', output)
         self.assertNotIn('rapid-commit', output)
 
+    def test_dhcp_client_ipv6_dbus_status(self):
+        def get_dbus_dhcp6_client_state(IF):
+            out = subprocess.check_output(['busctl', 'call', 'org.freedesktop.network1',
+                                           '/org/freedesktop/network1', 'org.freedesktop.network1.Manager',
+                                           'GetLinkByName', 's', IF])
+
+            assert out.startswith(b'io ')
+            out = out.strip()
+            assert out.endswith(b'"')
+            out = out.decode()
+            linkPath = out[:-1].split('"')[1]
+
+            print(f"Found {IF} link path: {linkPath}")
+
+            out = subprocess.check_output(['busctl', 'get-property', 'org.freedesktop.network1',
+                                           linkPath, 'org.freedesktop.network1.DHCPv6Client', 'State'])
+            assert out.startswith(b's "')
+            out = out.strip()
+            assert out.endswith(b'"')
+            return out[3:-1].decode()
+
+        copy_network_unit('25-veth.netdev', '25-dhcp-server-veth-peer.network', '25-dhcp-client-ipv6-only.network')
+
+        start_networkd()
+        self.wait_online(['veth-peer:carrier'])
+
+        # Note that at this point the DHCPv6 client has not been started because no RA (with managed
+        # bit set) has yet been recieved and the configuration does not include WithoutRA=true
+        state = get_dbus_dhcp6_client_state('veth99')
+        print(f"State = {state}")
+        self.assertEqual(state, 'stopped')
+
+        start_dnsmasq()
+        self.wait_online(['veth99:routable', 'veth-peer:routable'])
+
+        state = get_dbus_dhcp6_client_state('veth99')
+        print(f"State = {state}")
+        self.assertEqual(state, 'bound')
+
     def test_dhcp_client_ipv6_only_with_custom_client_identifier(self):
         copy_network_unit('25-veth.netdev', '25-dhcp-server-veth-peer.network', '25-dhcp-client-ipv6-only-custom-client-identifier.network')
 
@@ -5224,6 +5263,47 @@ class NetworkdDHCPClientTests(unittest.TestCase, Utilities):
 
         self.teardown_nftset('addr4', 'network4', 'ifindex')
 
+    def test_dhcp_client_ipv4_dbus_status(self):
+        def get_dbus_dhcp4_client_state(IF):
+            out = subprocess.check_output(['busctl', 'call', 'org.freedesktop.network1',
+                                           '/org/freedesktop/network1', 'org.freedesktop.network1.Manager',
+                                           'GetLinkByName', 's', IF])
+
+            assert out.startswith(b'io ')
+            out = out.strip()
+            assert out.endswith(b'"')
+            out = out.decode()
+            linkPath = out[:-1].split('"')[1]
+
+            print(f"Found {IF} link path: {linkPath}")
+
+            out = subprocess.check_output(['busctl', 'get-property', 'org.freedesktop.network1',
+                                           linkPath, 'org.freedesktop.network1.DHCPv4Client', 'State'])
+            assert out.startswith(b's "')
+            out = out.strip()
+            assert out.endswith(b'"')
+            return out[3:-1].decode()
+
+        copy_network_unit('25-veth.netdev', '25-dhcp-server-veth-peer.network', '25-dhcp-client-ipv4-only.network')
+
+        start_networkd()
+        self.wait_online(['veth-peer:carrier'])
+
+        state = get_dbus_dhcp4_client_state('veth99')
+        print(f"State = {state}")
+        self.assertEqual(state, 'selecting')
+
+        start_dnsmasq('--dhcp-option=option:dns-server,192.168.5.6,192.168.5.7',
+                      '--dhcp-option=option:domain-search,example.com',
+                      '--dhcp-alternate-port=67,5555',
+                      ipv4_range='192.168.5.110,192.168.5.119')
+        self.wait_online(['veth99:routable', 'veth-peer:routable'])
+        self.wait_address('veth99', r'inet 192.168.5.11[0-9]*/24', ipv='-4')
+
+        state = get_dbus_dhcp4_client_state('veth99')
+        print(f"State = {state}")
+        self.assertEqual(state, 'bound')
+
     def test_dhcp_client_ipv4_use_routes_gateway(self):
         first = True
         for (routes, gateway, dns_and_ntp_routes, classless) in itertools.product([True, False], repeat=4):