feat(ip): use pythonping to perform a real ping test instead to deter…

…mine the internet connection status instead of opening a socket

CHANGELOG.md

@@ -12,6 +12,7 @@
 - feat(keypad): ability to use key-combinations, set key combinations for screenshot, snapshot and quit
 - feat(web-ui): add `fields` in `InputDescription` with `InputFieldDescription` data structures to describe the fields of an input demand in detail
 - fix(users): avoid setting user as sudoer when it performs a password reset
+- feat(ip): use pythonping to perform a real ping test instead to determine the internet connection status instead of opening a socket
 
 ## Version 1.0.0
 

pyproject.toml

@@ -40,6 +40,7 @@ dependencies = [
   "betterproto [compiler] >=2.0.0b7",
   "gpiozero >=2.0.1 ; platform_machine != 'aarch64'",
   "quart >=0.19.6",
+  "pythonping>=1.1.4",
 ]
 
 [build-system]
@@ -187,7 +188,13 @@ quote-style = "single"
 profile = "black"
 
 [tool.pyright]
-exclude = ["typings", "ubo_app/rpc/generated", ".venv", "setup_scm_schemes.py"]
+exclude = [
+  "typings",
+  "ubo_app/rpc/generated",
+  ".venv",
+  "setup_scm_schemes.py",
+  "dist/",
+]
 disableTaggedHints = true
 
 [[tool.pyright.executionEnvironments]]

tests/fixtures/mock_environment.py

@@ -9,6 +9,8 @@
 
 import pytest
 
+from ubo_app.store.services.ethernet import NetState
+
 originals = {}
 
 
@@ -259,14 +261,6 @@ def _monkeypatch_asyncio_subprocess(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(asyncio, 'create_subprocess_exec', _fake_create_subprocess_exec)
 
 
-def _monkeypatch_asyncio_socket(monkeypatch: pytest.MonkeyPatch) -> None:
-    import asyncio
-
-    from fake import Fake
-
-    monkeypatch.setattr(asyncio, 'open_connection', Fake())
-
-
 @pytest.fixture
 def mock_environment(monkeypatch: pytest.MonkeyPatch) -> None:
     """Mock external resources."""
@@ -280,6 +274,7 @@ def mock_environment(monkeypatch: pytest.MonkeyPatch) -> None:
 
     import ubo_app.constants
     import ubo_app.utils.serializer
+    import ubo_app.utils.server
 
     tracemalloc.start()
 
@@ -289,6 +284,11 @@ def mock_environment(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(ubo_app.constants, 'STORE_GRACE_PERIOD', 0.1)
     monkeypatch.setattr(ubo_app.constants, 'NOTIFICATIONS_FLASH_TIME', 1000)
     monkeypatch.setattr(ubo_app.utils.serializer, 'add_type_field', lambda _, obj: obj)
+    monkeypatch.setattr(
+        ubo_app.utils.server,
+        'send_command',
+        Fake(_Fake__return_value=Fake(_Fake__await_value=NetState.CONNECTED)),
+    )
 
     sys.modules['ubo_app.utils.secrets'] = Fake(
         _Fake__attrs={'read_secret': lambda _: None},
@@ -302,4 +302,3 @@ def mock_environment(monkeypatch: pytest.MonkeyPatch) -> None:
     _monkeypatch_rpi_modules()
     _monkeypatch_subprocess(monkeypatch)
     _monkeypatch_asyncio_subprocess(monkeypatch)
-    _monkeypatch_asyncio_socket(monkeypatch)

ubo_app/services/030-ethernet/ethernet_manager.py

@@ -5,7 +5,7 @@
 import asyncio
 from typing import TYPE_CHECKING, Any, TypeVar
 
-from ubo_app.store.services.ethernet import GlobalEthernetState
+from ubo_app.store.services.ethernet import NetState
 from ubo_app.utils.bus_provider import get_system_bus
 
 if TYPE_CHECKING:
@@ -40,23 +40,23 @@ async def get_ethernet_device() -> NetworkDeviceGeneric | None:
     return None
 
 
-async def get_ethernet_device_state() -> GlobalEthernetState:
+async def get_ethernet_device_state() -> NetState:
     ethernet_device = await get_ethernet_device()
     if ethernet_device is None:
-        return GlobalEthernetState.UNKNOWN
+        return NetState.UNKNOWN
 
     state = await ethernet_device.state
     if state is DeviceState.UNKNOWN:
-        return GlobalEthernetState.UNKNOWN
+        return NetState.UNKNOWN
     if state in (
         DeviceState.DISCONNECTED,
         DeviceState.UNMANAGED,
         DeviceState.UNAVAILABLE,
         DeviceState.FAILED,
     ):
-        return GlobalEthernetState.DISCONNECTED
+        return NetState.DISCONNECTED
     if state in (DeviceState.NEED_AUTH,):
-        return GlobalEthernetState.NEEDS_ATTENTION
+        return NetState.NEEDS_ATTENTION
     if state in (
         DeviceState.DEACTIVATING,
         DeviceState.PREPARE,
@@ -65,8 +65,8 @@ async def get_ethernet_device_state() -> GlobalEthernetState:
         DeviceState.IP_CHECK,
         DeviceState.SECONDARIES,
     ):
-        return GlobalEthernetState.PENDING
+        return NetState.PENDING
     if state == DeviceState.ACTIVATED:
-        return GlobalEthernetState.CONNECTED
+        return NetState.CONNECTED
 
-    return GlobalEthernetState.UNKNOWN
+    return NetState.UNKNOWN

ubo_app/services/030-ethernet/setup.py

@@ -6,7 +6,7 @@
 from ethernet_manager import get_ethernet_device, get_ethernet_device_state
 
 from ubo_app.store.main import store
-from ubo_app.store.services.ethernet import GlobalEthernetState
+from ubo_app.store.services.ethernet import NetState
 from ubo_app.store.status_icons import StatusIconsRegisterAction
 from ubo_app.utils.async_ import create_task
 
@@ -20,11 +20,11 @@ async def update_ethernet_icon() -> None:
     store.dispatch(
         StatusIconsRegisterAction(
             icon={
-                GlobalEthernetState.CONNECTED: '󱊪',
-                GlobalEthernetState.DISCONNECTED: '󰌙',
-                GlobalEthernetState.PENDING: '󰌘',
-                GlobalEthernetState.NEEDS_ATTENTION: '󰌚',
-                GlobalEthernetState.UNKNOWN: '󰈅',
+                NetState.CONNECTED: '󱊪',
+                NetState.DISCONNECTED: '󰌙',
+                NetState.PENDING: '󰌘',
+                NetState.NEEDS_ATTENTION: '󰌚',
+                NetState.UNKNOWN: '󰈅',
             }[state],
             priority=ETHERNET_STATE_ICON_PRIORITY,
             id=ETHERNET_STATE_ICON_ID,

ubo_app/services/030-ip/setup.py

@@ -13,12 +13,14 @@
 
 from ubo_app.store.core import RegisterSettingAppAction, SettingsCategory
 from ubo_app.store.main import store
+from ubo_app.store.services.ethernet import NetState
 from ubo_app.store.services.ip import (
     IpNetworkInterface,
     IpSetIsConnectedAction,
     IpUpdateInterfacesAction,
 )
 from ubo_app.store.status_icons import StatusIconsRegisterAction
+from ubo_app.utils.server import send_command
 
 if TYPE_CHECKING:
     from collections.abc import Sequence
@@ -71,31 +73,10 @@ def load_ip_addresses() -> None:
     )
 
 
-async def is_connected() -> bool:
-    results = await asyncio.gather(
-        *(
-            asyncio.wait_for(asyncio.open_connection(ip, 53), timeout=1)
-            for ip in ('1.1.1.1', '8.8.8.8')
-        ),
-        return_exceptions=True,
-    )
-    is_connected = any(not isinstance(result, Exception) for result in results)
-
-    close_tasks = []
-    for result in results:
-        if isinstance(result, tuple):
-            _, writer = result
-            writer.close()
-            close_tasks.append(writer.wait_closed())
-    await asyncio.gather(*close_tasks)
-
-    return is_connected
-
-
 async def check_connection() -> bool:
     while True:
         load_ip_addresses()
-        if await is_connected():
+        if await send_command('connection', has_output=True) == NetState.CONNECTED:
             store.dispatch(
                 StatusIconsRegisterAction(
                     icon='󰖟',

ubo_app/services/030-wifi/reducer.py

@@ -12,8 +12,8 @@
     ReducerResult,
 )
 
+from ubo_app.store.services.ethernet import NetState
 from ubo_app.store.services.wifi import (
-    GlobalWiFiState,
     WiFiAction,
     WiFiEvent,
     WiFiSetHasVisitedOnboardingAction,
@@ -34,7 +34,7 @@ def reducer(
             return CompleteReducerResult(
                 state=WiFiState(
                     connections=[],
-                    state=GlobalWiFiState.UNKNOWN,
+                    state=NetState.UNKNOWN,
                     current_connection=None,
                 ),
                 actions=[WiFiUpdateRequestAction()],
@@ -64,15 +64,15 @@ def reducer(
             actions=[
                 StatusIconsRegisterAction(
                     icon={
-                        GlobalWiFiState.CONNECTED: get_signal_icon(
+                        NetState.CONNECTED: get_signal_icon(
                             action.current_connection.signal_strength
                             if action.current_connection
                             else 0,
                         ),
-                        GlobalWiFiState.DISCONNECTED: '󰖪',
-                        GlobalWiFiState.PENDING: '󱛇',
-                        GlobalWiFiState.NEEDS_ATTENTION: '󱚵',
-                        GlobalWiFiState.UNKNOWN: '󰈅',
+                        NetState.DISCONNECTED: '󰖪',
+                        NetState.PENDING: '󱛇',
+                        NetState.NEEDS_ATTENTION: '󱚵',
+                        NetState.UNKNOWN: '󰈅',
                     }[action.state],
                     priority=WIFI_STATE_ICON_PRIORITY,
                     id=WIFI_STATE_ICON_ID,

ubo_app/services/030-wifi/wifi_manager.py

@@ -11,18 +11,14 @@
 from ubo_gui.constants import DANGER_COLOR
 
 from ubo_app.store.main import store
+from ubo_app.store.services.ethernet import NetState
 from ubo_app.store.services.notifications import (
     Chime,
     Notification,
     NotificationDisplayType,
     NotificationsAddAction,
 )
-from ubo_app.store.services.wifi import (
-    ConnectionState,
-    GlobalWiFiState,
-    WiFiConnection,
-    WiFiType,
-)
+from ubo_app.store.services.wifi import ConnectionState, WiFiConnection, WiFiType
 from ubo_app.utils.bus_provider import get_system_bus
 
 if TYPE_CHECKING:
@@ -73,23 +69,23 @@ async def get_wifi_device() -> NetworkDeviceWireless | None:
     return None
 
 
-async def get_wifi_device_state() -> GlobalWiFiState:
+async def get_wifi_device_state() -> NetState:
     wifi_device = await get_wifi_device()
     if wifi_device is None:
-        return GlobalWiFiState.UNKNOWN
+        return NetState.UNKNOWN
 
     state = await wifi_device.state
     if state is DeviceState.UNKNOWN:
-        return GlobalWiFiState.UNKNOWN
+        return NetState.UNKNOWN
     if state in (
         DeviceState.DISCONNECTED,
         DeviceState.UNMANAGED,
         DeviceState.UNAVAILABLE,
         DeviceState.FAILED,
     ):
-        return GlobalWiFiState.DISCONNECTED
+        return NetState.DISCONNECTED
     if state in (DeviceState.NEED_AUTH,):
-        return GlobalWiFiState.NEEDS_ATTENTION
+        return NetState.NEEDS_ATTENTION
     if state in (
         DeviceState.DEACTIVATING,
         DeviceState.PREPARE,
@@ -98,11 +94,11 @@ async def get_wifi_device_state() -> GlobalWiFiState:
         DeviceState.IP_CHECK,
         DeviceState.SECONDARIES,
     ):
-        return GlobalWiFiState.PENDING
+        return NetState.PENDING
     if state == DeviceState.ACTIVATED:
-        return GlobalWiFiState.CONNECTED
+        return NetState.CONNECTED
 
-    return GlobalWiFiState.UNKNOWN
+    return NetState.UNKNOWN
 
 
 @debounce(wait=0.5, options=DebounceOptions(trailing=True, time_window=2))

ubo_app/store/services/ethernet.py

@@ -4,7 +4,7 @@
 from enum import StrEnum
 
 
-class GlobalEthernetState(StrEnum):
+class NetState(StrEnum):
     CONNECTED = 'Connected'
     DISCONNECTED = 'Disconnected'
     PENDING = 'Pending'

ubo_app/store/services/wifi.py

@@ -10,6 +10,8 @@
 if TYPE_CHECKING:
     from collections.abc import Sequence
 
+    from ubo_app.store.services.ethernet import NetState
+
 
 class WiFiType(StrEnum):
     WEP = 'WEP'
@@ -25,14 +27,6 @@ class ConnectionState(StrEnum):
     UNKNOWN = 'Unknown'
 
 
-class GlobalWiFiState(StrEnum):
-    CONNECTED = 'Connected'
-    DISCONNECTED = 'Disconnected'
-    PENDING = 'Pending'
-    NEEDS_ATTENTION = 'Needs Attention'
-    UNKNOWN = 'Unknown'
-
-
 class WiFiConnection(Immutable):
     ssid: str
     state: ConnectionState = ConnectionState.UNKNOWN
@@ -51,7 +45,7 @@ class WiFiSetHasVisitedOnboardingAction(WiFiAction):
 
 class WiFiUpdateAction(WiFiAction):
     connections: Sequence[WiFiConnection]
-    state: GlobalWiFiState
+    state: NetState
     current_connection: WiFiConnection | None
 
 
@@ -67,6 +61,6 @@ class WiFiUpdateRequestEvent(WiFiEvent): ...
 
 class WiFiState(Immutable):
     connections: Sequence[WiFiConnection] | None
-    state: GlobalWiFiState
+    state: NetState
     current_connection: WiFiConnection | None
     has_visited_onboarding: bool | None = None