Avoid dependency on dbus-python

netplan_cli/cli/state.py

@@ -17,20 +17,20 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 
-from collections import defaultdict, namedtuple
 import ipaddress
 import json
 import logging
 import re
-from socket import inet_ntop, AF_INET, AF_INET6
+import shutil
 import subprocess
 import sys
+from collections import defaultdict, namedtuple
 from io import StringIO
+from socket import AF_INET, AF_INET6, inet_ntop
 from typing import Dict, List, Type, Union
 
 import yaml
 
-import dbus
 import netplan
 
 from . import utils
@@ -579,14 +579,29 @@ def query_resolved(cls) -> tuple:
         addresses = None
         search = None
         try:
-            ipc = dbus.SystemBus()
-            resolve1 = ipc.get_object('org.freedesktop.resolve1', '/org/freedesktop/resolve1')
-            resolve1_if = dbus.Interface(resolve1, 'org.freedesktop.DBus.Properties')
-            res = resolve1_if.GetAll('org.freedesktop.resolve1.Manager')
-            addresses = res['DNS']
-            search = res['Domains']
-        except Exception as e:
-            logging.debug('Cannot query resolved DNS data: {}'.format(str(e)))
+            busctl = shutil.which('busctl')
+            if busctl is None:
+                raise RuntimeError('missing busctl utility')
+            json_out = subprocess.check_output(
+                [busctl, '--json=short', 'call', '--system',
+                 'org.freedesktop.resolve1',  # the service
+                 '/org/freedesktop/resolve1',  # the object
+                 'org.freedesktop.DBus.Properties',  # the interface
+                 'GetAll', 's',  # the method and signature
+                 'org.freedesktop.resolve1.Manager',  # the parameter
+                 ], text=True)
+            res = json.loads(json_out)
+            data = res.get('data', [{}])[0]
+            # make sure the type doesn't change. We expect an array of two
+            # intergers and an array of bytes (IP address)
+            assert data.get('DNS', {}).get('type') == 'a(iiay)', 'DNS address type doesn\'t match'
+            addresses = data.get('DNS', {}).get('data')
+            # make sure the type dosn't change. We expect an array of an integer
+            # a string (DNS search domain) and a boolean
+            assert data.get('Domains', {}).get('type') == 'a(isb)', 'DNS search type doesn\'t match'
+            search = data.get('Domains', {}).get('data')
+        except Exception as err:
+            logging.debug('Cannot query resolved DNS data: %s', str(err))
         return (addresses, search)
 
     @classmethod

tests/cli/test_state.py

@@ -19,35 +19,23 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import copy
+import json
 import os
 import shutil
 import subprocess
 import tempfile
 import unittest
+from unittest.mock import call, mock_open, patch
+
 import yaml
 
-from unittest.mock import patch, call, mock_open
-from netplan_cli.cli.state import Interface, NetplanConfigState, SystemConfigState
+from netplan_cli.cli.state import (Interface, NetplanConfigState,
+                                   SystemConfigState)
+
 from .test_status import (BRIDGE, DNS_ADDRESSES, DNS_IP4, DNS_SEARCH, FAKE_DEV,
                           IPROUTE2, NETWORKD, NMCLI, ROUTE4, ROUTE6)
 
 
-class resolve1_ipc_mock():
-    def get_object(self, _foo, _bar):
-        return {}  # dbus Object
-
-
-class resolve1_iface_mock():
-    def __init__(self, _foo, _bar):
-        pass  # dbus Interface
-
-    def GetAll(self, _):
-        return {
-            'DNS': DNS_ADDRESSES,
-            'Domains': DNS_SEARCH,
-            }
-
-
 class TestSystemState(unittest.TestCase):
     '''Test netplan state module'''
 
@@ -137,11 +125,17 @@ def test_query_routes_fail(self, mock):
             self.assertIsNone(res6)
             self.assertIn('DEBUG:root:Cannot query iproute2 route data:', cm.output[0])
 
-    @patch('dbus.Interface')
-    @patch('dbus.SystemBus')
-    def test_query_resolved(self, mock_ipc, mock_iface):
-        mock_ipc.return_value = resolve1_ipc_mock()
-        mock_iface.return_value = resolve1_iface_mock('foo', 'bar')
+    @patch('subprocess.check_output')
+    def test_query_resolved(self, mock_busctl):
+        mock_busctl.return_value = '''{"data":[{
+        "DNS": {
+            "type": "a(iiay)",
+            "data": '''+json.dumps(DNS_ADDRESSES)+'''
+        },
+        "Domains": {
+            "type": "a(isb)",
+            "data": '''+json.dumps(DNS_SEARCH)+'''
+        }}]}'''
         addresses, search = SystemConfigState.query_resolved()
         self.assertEqual(len(addresses), 4)
         self.assertListEqual([addr[0] for addr in addresses],
@@ -150,15 +144,23 @@ def test_query_resolved(self, mock_ipc, mock_iface):
         self.assertListEqual([s[1] for s in search],
                              ['search.domain', 'search.domain'])
 
-    @patch('dbus.SystemBus')
-    def test_query_resolved_fail(self, mock):
-        mock.return_value = resolve1_ipc_mock()
-        mock.side_effect = Exception(1, '', 'ERR')
+    @patch('subprocess.check_output')
+    def test_query_resolved_fail(self, mock_busctl):
+        mock_busctl.return_value = '{"data":[{"DNS":{"type":"invalid","data":"garbage"}}]}'
+        with self.assertLogs(level='DEBUG') as cm:
+            addresses, search = SystemConfigState.query_resolved()
+            self.assertIsNone(addresses)
+            self.assertIsNone(search)
+            self.assertIn('DEBUG:root:Cannot query resolved DNS data: DNS address type doesn\'t match', cm.output[0])
+
+    @patch('shutil.which')
+    def test_query_resolved_fail_missing_busctl(self, mock):
+        mock.return_value = None
         with self.assertLogs(level='DEBUG') as cm:
             addresses, search = SystemConfigState.query_resolved()
             self.assertIsNone(addresses)
             self.assertIsNone(search)
-            self.assertIn('DEBUG:root:Cannot query resolved DNS data:', cm.output[0])
+            self.assertIn('DEBUG:root:Cannot query resolved DNS data: missing busctl utility', cm.output[0])
 
     def test_query_resolvconf(self):
         with patch('builtins.open', mock_open(read_data='''\

tests/cli/test_status.py

@@ -33,8 +33,8 @@
 NMCLI = 'wlan0:MYCON:b6b7a21d-186e-45e1-b3a6-636da1735563:/run/NetworkManager/system-connections/netplan-NM-b6b7a21d-186e-45e1-b3a6-636da1735563-MYCON.nmconnection:802-11-wireless:yes'  # nopep8
 ROUTE4 = '[{"family":2,"type":"unicast","dst":"default","gateway":"192.168.178.1","dev":"enp0s31f6","table":"main","protocol":"dhcp","scope":"global","prefsrc":"192.168.178.62","metric":100,"flags":[]},{"family":2,"type":"unicast","dst":"default","gateway":"192.168.178.1","dev":"wlan0","table":"main","protocol":"dhcp","scope":"global","metric":600,"flags":[]},{"family":2,"type":"unicast","dst":"10.10.0.0/24","dev":"wg0","table":"main","protocol":"kernel","scope":"link","prefsrc":"10.10.0.2","flags":[]},{"family":2,"type":"unicast","dst":"192.168.178.0/24","dev":"enp0s31f6","table":"main","protocol":"kernel","scope":"link","prefsrc":"192.168.178.62","metric":100,"flags":[]},{"family":2,"type":"unicast","dst":"192.168.178.0/24","dev":"wlan0","table":"main","protocol":"kernel","scope":"link","prefsrc":"192.168.178.142","metric":600,"flags":[]},{"family":2,"type":"unicast","dst":"192.168.178.1","dev":"enp0s31f6","table":"1234","protocol":"dhcp","scope":"link","prefsrc":"192.168.178.62","metric":100,"flags":[]},{"family":2,"type":"broadcast","dst":"192.168.178.255","dev":"enp0s31f6","table":"local","protocol":"kernel","scope":"link","prefsrc":"192.168.178.62","flags":[]}]'  # nopep8
 ROUTE6 = '[{"family":10,"type":"unicast","dst":"::1","dev":"lo","table":"main","protocol":"kernel","scope":"global","metric":256,"flags":[],"pref":"medium"},{"family":10,"type":"unicast","dst":"2001:9e8:a19f:1c00::/64","dev":"enp0s31f6","table":"main","protocol":"ra","scope":"global","metric":100,"flags":[],"expires":7199,"pref":"medium"},{"family":10,"type":"unicast","dst":"2001:9e8:a19f:1c00::/64","dev":"wlan0","table":"main","protocol":"ra","scope":"global","metric":600,"flags":[],"pref":"medium"},{"family":10,"type":"unicast","dst":"2001:9e8:a19f:1c00::/56","gateway":"fe80::cece:1eff:fe3d:c737","dev":"enp0s31f6","table":"main","protocol":"ra","scope":"global","metric":100,"flags":[],"expires":1799,"pref":"medium"},{"family":10,"type":"unicast","dst":"2001:9e8:a19f:1c00::/56","gateway":"fe80::cece:1eff:fe3d:c737","dev":"wlan0","table":"main","protocol":"ra","scope":"global","metric":600,"flags":[],"pref":"medium"},{"family":10,"type":"unicast","dst":"2001:dead:beef::/64","dev":"tun0","table":"main","protocol":"kernel","scope":"global","metric":256,"flags":[],"pref":"medium"},{"family":10,"type":"unicast","dst":"fe80::/64","dev":"enp0s31f6","table":"main","protocol":"kernel","scope":"global","metric":256,"flags":[],"pref":"medium"},{"family":10,"type":"unicast","dst":"fe80::/64","dev":"wlan0","table":"main","protocol":"kernel","scope":"global","metric":1024,"flags":[],"pref":"medium"},{"family":10,"type":"unicast","dst":"default","gateway":"fe80::cece:1eff:fe3d:c737","dev":"enp0s31f6","table":"1234","protocol":"ra","scope":"global","metric":100,"flags":[],"expires":1799,"metrics":[{"mtu":1492}],"pref":"medium"},{"family":10,"type":"unicast","dst":"default","gateway":"fe80::cece:1eff:fe3d:c737","dev":"wlan0","table":"main","protocol":"ra","scope":"global","metric":20600,"flags":[],"pref":"medium"}]'  # nopep8
-DNS_IP4 = bytearray([192, 168, 178, 1])
-DNS_IP6 = bytearray([0xfd, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0xce, 0xce, 0x1e, 0xff, 0xfe, 0x3d, 0xc7, 0x37])
+DNS_IP4 = ([192, 168, 178, 1])
+DNS_IP6 = ([0xfd, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0xce, 0xce, 0x1e, 0xff, 0xfe, 0x3d, 0xc7, 0x37])
 DNS_ADDRESSES = [(5, 2, DNS_IP4), (5, 10, DNS_IP6), (2, 2, DNS_IP4), (2, 10, DNS_IP6)]  # (IFidx, IPfamily, IPbytes)
 DNS_SEARCH = [(5, 'search.domain', False), (2, 'search.domain', False)]
 FAKE_DEV = {'ifindex': 42, 'ifname': 'fakedev0', 'flags': [], 'operstate': 'DOWN'}