Merge pull request #143 from guyonvarch/document-connman-service

Add initial tests and typings for the Kiosk browser proxy
dividat · Nov 16, 2023 · 399356b · 399356b
2 parents 03e9ac3 + 9c26e9a
commit 399356b
Show file tree

Hide file tree

Showing 8 changed files with 128 additions and 40 deletions.
diff --git a/kiosk/Readme.md b/kiosk/Readme.md
@@ -12,6 +12,10 @@ Then, start the kiosk browser with, for example:
 bin/kiosk-browser http://localhost:8080/play.html http://localhost:3333
 ```
 
+## Testing
+
+        bin/test
+
 ## Developer tools
 
 Run with `QTWEBENGINE_REMOTE_DEBUGGING` equals to a specific port:

diff --git a/kiosk/bin/test b/kiosk/bin/test
@@ -0,0 +1,6 @@
+#!/usr/bin/env bash
+set -euo pipefail
+cd "$(dirname "$0")/.."
+
+mypy kiosk_browser
+python -m pytest
diff --git a/kiosk/default.nix b/kiosk/default.nix
@@ -17,13 +17,15 @@ python3Packages.buildPythonApplication rec {
 
   doCheck = false;
 
-  nativeBuildInputs = [ qt5.wrapQtAppsHook ];
+  nativeBuildInputs = [ qt5.wrapQtAppsHook mypy ];
 
   propagatedBuildInputs = with python3Packages; [
-    pyqtwebengine
-    requests
     dbus-python
     pygobject3
+    pyqtwebengine
+    pytest
+    requests
+    types-requests
   ];
 
   postInstall = ''

diff --git a/kiosk/kiosk_browser/browser_widget.py b/kiosk/kiosk_browser/browser_widget.py
@@ -101,10 +101,10 @@ def _load_finished(self, success):
 
     def _proxy_auth(self, get_current_proxy, url, auth, proxyHost):
         proxy = get_current_proxy()
-        if proxy is not None and proxy.username is not None and proxy.password is not None:
+        if proxy is not None and proxy.credentials is not None:
             logging.info("Authenticating proxy")
-            auth.setUser(proxy.username)
-            auth.setPassword(proxy.password)
+            auth.setUser(proxy.credentials.username)
+            auth.setPassword(proxy.credentials.password)
         else:
             logging.info("Proxy authentication request ignored because credentials are not provided.")
 

diff --git a/kiosk/kiosk_browser/proxy.py b/kiosk/kiosk_browser/proxy.py
@@ -1,54 +1,78 @@
-"""Monitor proxy changes and automatically apply changes in Qt application"""
+"""Monitor proxy changes and automatically apply changes in Qt application.
+"""
 
 import collections
 import dbus
 import logging
 import threading
 import urllib
 from PyQt5.QtNetwork import QNetworkProxy
+from dataclasses import dataclass
 from dbus.mainloop.glib import DBusGMainLoop
 from gi.repository import GLib
 
-ProxyConfig = collections.namedtuple('ProxyConfig', ['hostname', 'port', 'username', 'password'])
+@dataclass
+class Credentials:
+    username: str
+    password: str
+
+@dataclass
+class ProxyConf:
+    hostname: str
+    port: int
+    credentials: Credentials | None
+
+@dataclass
+class Service:
+    state: str
+    proxy: ProxyConf | None
+
+def parse_service(service: dbus.Struct) -> Service | None:
+    if len(service) >= 2 and 'State' in service[1]:
+        return Service(service[1]['State'], extract_manual_proxy(service[1]))
+    else:
+        return None
+
+def extract_manual_proxy(service_conf: dbus.Dictionary) -> ProxyConf | None:
+    if 'Proxy' in service_conf:
+        proxy = service_conf['Proxy']
+        if 'Method' in proxy and 'Servers' in proxy:
+            method = proxy['Method']
+            servers = proxy['Servers']
+            if method == 'manual' and len(servers) >= 1:
+                return parse_proxy_url(servers[0])
+            else:
+                return None
+        else:
+            return None
+    else:
+        return None
 
-def parse_url(url):
+def parse_proxy_url(url: str) -> ProxyConf | None:
     if url.startswith('http://'):
         parsed = urllib.parse.urlparse(url)
     else:
         parsed = urllib.parse.urlparse(f'http://{url}')
 
     if parsed.hostname != None and parsed.port != None:
-        return ProxyConfig(
-            parsed.hostname,
-            parsed.port,
-            (urllib.parse.unquote(parsed.username) if parsed.username != None else None),
-            (urllib.parse.unquote(parsed.password) if parsed.password != None else None)
-        )
-    else:
-        logging.warn(f"Hostname or port missing in proxy url")
-        return None
-
-def extract_manual_proxy(config):
-    """Extract manual proxy from dbus configuration
+        assert isinstance(parsed.hostname, str)
+        assert isinstance(parsed.port, int)
 
-    Example configuration:
+        if parsed.username != None and parsed.password != None:
+            assert isinstance(parsed.username, str)
+            assert isinstance(parsed.password, str)
 
-    dbus.Dictionary({dbus.String('Servers'): dbus.Array([dbus.String('http://localhost:1234')], signature=dbus.Signature('s'), variant_level=1), dbus.String('Excludes'): dbus.Array([], signature=dbus.Signature('s'), variant_level=1), dbus.String('Method'): dbus.String('manual', variant_level=1)}, signature=dbus.Signature('sv'), variant_level=1)"""
-
-    if 'Method' in config:
-        method = config['Method']
-        if method == 'direct':
-            return None
-        elif method == 'manual':
-            if 'Servers' in config:
-                servers = config['Servers']
-                if len(servers) >= 1:
-                    return parse_url(servers[0])
+            username = urllib.parse.unquote(parsed.username)
+            password = urllib.parse.unquote(parsed.password)
+            return ProxyConf(parsed.hostname, parsed.port, Credentials(username, password))
+        else:
+            return ProxyConf(parsed.hostname, parsed.port, None)
     else:
+        logging.warning(f"Hostname or port missing in proxy url")
         return None
 
-def get_current_proxy(bus):
-    """Get current proxy from dbus
+def get_current_proxy(bus) -> ProxyConf | None:
+    """Get current proxy from dbus.
 
     Return the proxy of a connected service preferentially, or of a ready
     service.
@@ -62,16 +86,16 @@ def get_current_proxy(bus):
             'net.connman.Manager')
 
         # List services, each service is a (id, properties) struct
-        services = client.GetServices()
+        services = [parse_service(s) for s in client.GetServices()]
 
         # The service with the default route will always be sorted at the top of
         # the list. (From connman doc/overview-api.txt)
-        default_service = find(
-            lambda s: s[1]['State'] == 'online' or s[1]['State'] == 'ready',
-            services)
+        default_service = find(lambda s: s.state in ['online', 'ready'], services)
 
         if default_service:
-            return extract_manual_proxy(default_service[1]['Proxy'])
+            return default_service.proxy
+        else:
+            return None
 
     except dbus.exceptions.DBusException:
         return None

diff --git a/kiosk/kiosk_browser/test_proxy.py b/kiosk/kiosk_browser/test_proxy.py
@@ -0,0 +1,19 @@
+import dbus
+from kiosk_browser import proxy
+
+# Mock
+
+service = dbus.Struct((dbus.ObjectPath('/net/connman/service/wifi_a0510b58100d_517561657374791f6e657e2d4d7060046065_managed_psk'), dbus.Dictionary({dbus.String('Type'): dbus.String('wifi', variant_level=1), dbus.String('Security'): dbus.Array([dbus.String('psk')], signature=dbus.Signature('s'), variant_level=1), dbus.String('State'): dbus.String('online', variant_level=1), dbus.String('Strength'): dbus.Byte(51, variant_level=1), dbus.String('Favorite'): dbus.Boolean(True, variant_level=1), dbus.String('Immutable'): dbus.Boolean(False, variant_level=1), dbus.String('AutoConnect'): dbus.Boolean(True, variant_level=1), dbus.String('Name'): dbus.String('Super-Network', variant_level=1), dbus.String('Ethernet'): dbus.Dictionary({dbus.String('Method'): dbus.String('auto', variant_level=1), dbus.String('Interface'): dbus.String('wlp2s0', variant_level=1), dbus.String('Address'): dbus.String('B7:71:01:51:10:AD', variant_level=1), dbus.String('MTU'): dbus.UInt16(1500, variant_level=1)}, signature=dbus.Signature('sv'), variant_level=1), dbus.String('IPv4'): dbus.Dictionary({dbus.String('Method'): dbus.String('dhcp', variant_level=1), dbus.String('Address'): dbus.String('192.168.1.1', variant_level=1), dbus.String('Netmask'): dbus.String('255.255.255.0', variant_level=1), dbus.String('Gateway'): dbus.String('192.168.1.254', variant_level=1)}, signature=dbus.Signature('sv'), variant_level=1), dbus.String('IPv4.Configuration'): dbus.Dictionary({dbus.String('Method'): dbus.String('dhcp', variant_level=1)}, signature=dbus.Signature('sv'), variant_level=1), dbus.String('IPv6'): dbus.Dictionary({dbus.String('Method'): dbus.String('auto', variant_level=1), dbus.String('Address'): dbus.String('0a21:e0f:5ba:7390:3304:872c:531c:cbad', variant_level=1), dbus.String('PrefixLength'): dbus.Byte(64, variant_level=1), dbus.String('Privacy'): dbus.String('prefered', variant_level=1)}, signature=dbus.Signature('sv'), variant_level=1), dbus.String('IPv6.Configuration'): dbus.Dictionary({dbus.String('Method'): dbus.String('auto', variant_level=1), dbus.String('Privacy'): dbus.String('prefered', variant_level=1)}, signature=dbus.Signature('sv'), variant_level=1), dbus.String('Nameservers'): dbus.Array([dbus.String('192.168.1.254')], signature=dbus.Signature('s'), variant_level=1), dbus.String('Nameservers.Configuration'): dbus.Array([], signature=dbus.Signature('s'), variant_level=1), dbus.String('Timeservers'): dbus.Array([], signature=dbus.Signature('s'), variant_level=1), dbus.String('Timeservers.Configuration'): dbus.Array([], signature=dbus.Signature('s'), variant_level=1), dbus.String('Domains'): dbus.Array([], signature=dbus.Signature('s'), variant_level=1), dbus.String('Domains.Configuration'): dbus.Array([], signature=dbus.Signature('s'), variant_level=1), dbus.String('Proxy'): dbus.Dictionary({dbus.String('Servers'): dbus.Array([dbus.String('http://proxy.dividat.com:1234')], signature=dbus.Signature('s'), variant_level=1), dbus.String('Excludes'): dbus.Array([], signature=dbus.Signature('s'), variant_level=1), dbus.String('Method'): dbus.String('manual', variant_level=1)}, signature=dbus.Signature('sv'), variant_level=1), dbus.String('Proxy.Configuration'): dbus.Dictionary({dbus.String('Servers'): dbus.Array([dbus.String('http://proxy.dividat.com')], signature=dbus.Signature('s'), variant_level=1), dbus.String('Method'): dbus.String('manual', variant_level=1)}, signature=dbus.Signature('sv'), variant_level=1), dbus.String('mDNS'): dbus.Boolean(False, variant_level=1), dbus.String('mDNS.Configuration'): dbus.Boolean(False, variant_level=1), dbus.String('Provider'): dbus.Dictionary({}, signature=dbus.Signature('sv'), variant_level=1)}, signature=dbus.Signature('sv'))), signature=None)
+
+# Tests
+
+def test_parse_service():
+    s = proxy.parse_service(service)
+    assert s.state == 'online'
+    assert s.proxy == proxy.ProxyConf('proxy.dividat.com', 1234, None)
+
+def test_extract_manual_proxy():
+    assert proxy.parse_proxy_url('') == None
+    assert proxy.parse_proxy_url('dividat') == None
+    assert proxy.parse_proxy_url('dividat:88') == proxy.ProxyConf('dividat', 88, None)
+    assert proxy.parse_proxy_url('user:pw@dividat:987') == proxy.ProxyConf('dividat', 987, proxy.Credentials('user', 'pw'))
diff --git a/kiosk/mypy.ini b/kiosk/mypy.ini
@@ -0,0 +1,25 @@
+[mypy]
+
+[mypy-PyQt5]
+ignore_missing_imports = True
+
+[mypy-PyQt5.QtNetwork]
+ignore_missing_imports = True
+
+[mypy-PyQt5.QtCore]
+ignore_missing_imports = True
+
+[mypy-PyQt5.QtGui]
+ignore_missing_imports = True
+
+[mypy-PyQt5.QtWidgets]
+ignore_missing_imports = True
+
+[mypy-dbus]
+ignore_missing_imports = True
+
+[mypy-dbus.mainloop.glib]
+ignore_missing_imports = True
+
+[mypy-gi.repository]
+ignore_missing_imports = True
diff --git a/testing/run b/testing/run
@@ -1,8 +1,16 @@
 #!/usr/bin/env bash
 set -euxo pipefail
 
+# Integration tests
+
 INTEGRATION_TEST_DIR="$(dirname "$(realpath "$0")")/integration"
 
 for TEST_DEF in $(ls "$INTEGRATION_TEST_DIR"/*.nix); do
   nix-build "$TEST_DEF"
 done
+
+# Kiosk browser
+
+pushd "$(dirname "$(realpath "$0")")/../kiosk"
+nix-shell --run bin/test
+popd