You've already forked MicroPythonOS
mirror of
https://github.com/m5stack/MicroPythonOS.git
synced 2026-05-20 11:51:27 -07:00
Move internal_filesystem/lib/mpos/wifi.py to internal_filesystem/lib/mpos/net/wifi_service.py
This commit is contained in:
Thomas Farstrike
@@ -10,7 +10,7 @@ from mpos.ui.keyboard import MposKeyboard
|
||||
import mpos.config
|
||||
import mpos.ui.anim
|
||||
import mpos.ui.theme
|
||||
import mpos.wifi
|
||||
from mpos.net.wifi_service import WifiService
|
||||
|
||||
have_network = True
|
||||
try:
|
||||
@@ -69,8 +69,8 @@ class WiFi(Activity):
|
||||
global access_points
|
||||
access_points = mpos.config.SharedPreferences("com.micropythonos.system.wifiservice").get_dict("access_points")
|
||||
if len(self.ssids) == 0:
|
||||
if mpos.wifi.WifiService.wifi_busy == False:
|
||||
mpos.wifi.WifiService.wifi_busy = True
|
||||
if WifiService.wifi_busy == False:
|
||||
WifiService.wifi_busy = True
|
||||
self.start_scan_networks()
|
||||
else:
|
||||
self.show_error("Wifi is busy, please try again later.")
|
||||
@@ -107,7 +107,7 @@ class WiFi(Activity):
|
||||
self.show_error("Wi-Fi scan failed")
|
||||
# scan done:
|
||||
self.busy_scanning = False
|
||||
mpos.wifi.WifiService.wifi_busy = False
|
||||
WifiService.wifi_busy = False
|
||||
self.update_ui_threadsafe_if_foreground(self.scan_button_label.set_text,self.scan_button_scan_text)
|
||||
self.update_ui_threadsafe_if_foreground(self.scan_button.remove_state, lv.STATE.DISABLED)
|
||||
self.update_ui_threadsafe_if_foreground(self.refresh_list)
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
# mpos.net module - Networking utilities for MicroPythonOS
|
||||
@@ -0,0 +1,318 @@
|
||||
"""
|
||||
WiFi Service for MicroPythonOS.
|
||||
|
||||
Manages WiFi connections including:
|
||||
- Auto-connect to saved networks on boot
|
||||
- Network scanning
|
||||
- Connection management with saved credentials
|
||||
- Concurrent access locking
|
||||
|
||||
This service works alongside ConnectivityManager which monitors connection status.
|
||||
"""
|
||||
|
||||
import ujson
|
||||
import os
|
||||
import time
|
||||
|
||||
import mpos.config
|
||||
import mpos.time
|
||||
|
||||
# Try to import network module (not available on desktop)
|
||||
HAS_NETWORK_MODULE = False
|
||||
try:
|
||||
import network
|
||||
HAS_NETWORK_MODULE = True
|
||||
except ImportError:
|
||||
print("WifiService: network module not available (desktop mode)")
|
||||
|
||||
|
||||
class WifiService:
|
||||
"""
|
||||
Service for managing WiFi connections.
|
||||
|
||||
This class handles connecting to saved WiFi networks and managing
|
||||
the WiFi hardware state. It's typically started in a background thread
|
||||
on boot to auto-connect to known networks.
|
||||
"""
|
||||
|
||||
# Class-level lock to prevent concurrent WiFi operations
|
||||
# Used by WiFi app when scanning to avoid conflicts with connection attempts
|
||||
wifi_busy = False
|
||||
|
||||
# Dictionary of saved access points {ssid: {password: "..."}}
|
||||
access_points = {}
|
||||
|
||||
@staticmethod
|
||||
def connect(network_module=None):
|
||||
"""
|
||||
Scan for available networks and connect to the first saved network found.
|
||||
|
||||
Args:
|
||||
network_module: Network module for dependency injection (testing)
|
||||
|
||||
Returns:
|
||||
bool: True if successfully connected, False otherwise
|
||||
"""
|
||||
net = network_module if network_module else network
|
||||
wlan = net.WLAN(net.STA_IF)
|
||||
|
||||
# Restart WiFi hardware in case it's in a bad state
|
||||
wlan.active(False)
|
||||
wlan.active(True)
|
||||
|
||||
# Scan for available networks
|
||||
networks = wlan.scan()
|
||||
|
||||
for n in networks:
|
||||
ssid = n[0].decode()
|
||||
print(f"WifiService: Found network '{ssid}'")
|
||||
|
||||
if ssid in WifiService.access_points:
|
||||
password = WifiService.access_points.get(ssid).get("password")
|
||||
print(f"WifiService: Attempting to connect to saved network '{ssid}'")
|
||||
|
||||
if WifiService.attempt_connecting(ssid, password, network_module=network_module):
|
||||
print(f"WifiService: Connected to '{ssid}'")
|
||||
return True
|
||||
else:
|
||||
print(f"WifiService: Failed to connect to '{ssid}'")
|
||||
else:
|
||||
print(f"WifiService: Skipping '{ssid}' (not configured)")
|
||||
|
||||
print("WifiService: No saved networks found or connected")
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def attempt_connecting(ssid, password, network_module=None, time_module=None):
|
||||
"""
|
||||
Attempt to connect to a specific WiFi network.
|
||||
|
||||
Args:
|
||||
ssid: Network SSID to connect to
|
||||
password: Network password
|
||||
network_module: Network module for dependency injection (testing)
|
||||
time_module: Time module for dependency injection (testing)
|
||||
|
||||
Returns:
|
||||
bool: True if successfully connected, False otherwise
|
||||
"""
|
||||
print(f"WifiService: Connecting to SSID: {ssid}")
|
||||
|
||||
net = network_module if network_module else network
|
||||
time_mod = time_module if time_module else time
|
||||
|
||||
try:
|
||||
wlan = net.WLAN(net.STA_IF)
|
||||
wlan.connect(ssid, password)
|
||||
|
||||
# Wait up to 10 seconds for connection
|
||||
for i in range(10):
|
||||
if wlan.isconnected():
|
||||
print(f"WifiService: Connected to '{ssid}' after {i+1} seconds")
|
||||
|
||||
# Sync time from NTP server if possible
|
||||
try:
|
||||
mpos.time.sync_time()
|
||||
except Exception as e:
|
||||
print(f"WifiService: Could not sync time: {e}")
|
||||
|
||||
return True
|
||||
|
||||
elif not wlan.active():
|
||||
# WiFi was disabled during connection attempt
|
||||
print("WifiService: WiFi disabled during connection, aborting")
|
||||
return False
|
||||
|
||||
print(f"WifiService: Waiting for connection, attempt {i+1}/10")
|
||||
time_mod.sleep(1)
|
||||
|
||||
print(f"WifiService: Connection timeout for '{ssid}'")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"WifiService: Connection error: {e}")
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def auto_connect(network_module=None, time_module=None):
|
||||
"""
|
||||
Auto-connect to a saved WiFi network on boot.
|
||||
|
||||
This is typically called in a background thread from main.py.
|
||||
It loads saved networks and attempts to connect to the first one found.
|
||||
|
||||
Args:
|
||||
network_module: Network module for dependency injection (testing)
|
||||
time_module: Time module for dependency injection (testing)
|
||||
"""
|
||||
print("WifiService: Auto-connect thread starting")
|
||||
|
||||
# Load saved access points from config
|
||||
WifiService.access_points = mpos.config.SharedPreferences(
|
||||
"com.micropythonos.system.wifiservice"
|
||||
).get_dict("access_points")
|
||||
|
||||
if not len(WifiService.access_points):
|
||||
print("WifiService: No access points configured, exiting")
|
||||
return
|
||||
|
||||
# Check if WiFi is busy (e.g., WiFi app is scanning)
|
||||
if WifiService.wifi_busy:
|
||||
print("WifiService: WiFi busy, auto-connect aborted")
|
||||
return
|
||||
|
||||
WifiService.wifi_busy = True
|
||||
|
||||
try:
|
||||
if not HAS_NETWORK_MODULE and network_module is None:
|
||||
# Desktop mode - simulate connection delay
|
||||
print("WifiService: Desktop mode, simulating connection...")
|
||||
time_mod = time_module if time_module else time
|
||||
time_mod.sleep(2)
|
||||
print("WifiService: Simulated connection complete")
|
||||
else:
|
||||
# Attempt to connect to saved networks
|
||||
if WifiService.connect(network_module=network_module):
|
||||
print("WifiService: Auto-connect successful")
|
||||
else:
|
||||
print("WifiService: Auto-connect failed")
|
||||
|
||||
# Disable WiFi to conserve power if connection failed
|
||||
if network_module:
|
||||
net = network_module
|
||||
else:
|
||||
net = network
|
||||
|
||||
wlan = net.WLAN(net.STA_IF)
|
||||
wlan.active(False)
|
||||
print("WifiService: WiFi disabled to conserve power")
|
||||
|
||||
finally:
|
||||
WifiService.wifi_busy = False
|
||||
print("WifiService: Auto-connect thread finished")
|
||||
|
||||
@staticmethod
|
||||
def is_connected(network_module=None):
|
||||
"""
|
||||
Check if WiFi is currently connected.
|
||||
|
||||
This is a simple connection check. For comprehensive connectivity
|
||||
monitoring with callbacks, use ConnectivityManager instead.
|
||||
|
||||
Args:
|
||||
network_module: Network module for dependency injection (testing)
|
||||
|
||||
Returns:
|
||||
bool: True if connected, False otherwise
|
||||
"""
|
||||
# If WiFi operations are in progress, report not connected
|
||||
if WifiService.wifi_busy:
|
||||
return False
|
||||
|
||||
# Desktop mode - always report connected
|
||||
if not HAS_NETWORK_MODULE and network_module is None:
|
||||
return True
|
||||
|
||||
# Check actual connection status
|
||||
try:
|
||||
net = network_module if network_module else network
|
||||
wlan = net.WLAN(net.STA_IF)
|
||||
return wlan.isconnected()
|
||||
except Exception as e:
|
||||
print(f"WifiService: Error checking connection: {e}")
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def disconnect(network_module=None):
|
||||
"""
|
||||
Disconnect from current WiFi network and disable WiFi.
|
||||
|
||||
Args:
|
||||
network_module: Network module for dependency injection (testing)
|
||||
"""
|
||||
if not HAS_NETWORK_MODULE and network_module is None:
|
||||
print("WifiService: Desktop mode, cannot disconnect")
|
||||
return
|
||||
|
||||
try:
|
||||
net = network_module if network_module else network
|
||||
wlan = net.WLAN(net.STA_IF)
|
||||
wlan.disconnect()
|
||||
wlan.active(False)
|
||||
print("WifiService: Disconnected and WiFi disabled")
|
||||
except Exception as e:
|
||||
print(f"WifiService: Error disconnecting: {e}")
|
||||
|
||||
@staticmethod
|
||||
def get_saved_networks():
|
||||
"""
|
||||
Get list of saved network SSIDs.
|
||||
|
||||
Returns:
|
||||
list: List of saved SSIDs
|
||||
"""
|
||||
if not WifiService.access_points:
|
||||
WifiService.access_points = mpos.config.SharedPreferences(
|
||||
"com.micropythonos.system.wifiservice"
|
||||
).get_dict("access_points")
|
||||
|
||||
return list(WifiService.access_points.keys())
|
||||
|
||||
@staticmethod
|
||||
def save_network(ssid, password):
|
||||
"""
|
||||
Save a new WiFi network credential.
|
||||
|
||||
Args:
|
||||
ssid: Network SSID
|
||||
password: Network password
|
||||
"""
|
||||
# Load current saved networks
|
||||
prefs = mpos.config.SharedPreferences("com.micropythonos.system.wifiservice")
|
||||
access_points = prefs.get_dict("access_points")
|
||||
|
||||
# Add or update the network
|
||||
access_points[ssid] = {"password": password}
|
||||
|
||||
# Save back to config
|
||||
editor = prefs.edit()
|
||||
editor.put_dict("access_points", access_points)
|
||||
editor.commit()
|
||||
|
||||
# Update class-level cache
|
||||
WifiService.access_points = access_points
|
||||
|
||||
print(f"WifiService: Saved network '{ssid}'")
|
||||
|
||||
@staticmethod
|
||||
def forget_network(ssid):
|
||||
"""
|
||||
Remove a saved WiFi network.
|
||||
|
||||
Args:
|
||||
ssid: Network SSID to forget
|
||||
|
||||
Returns:
|
||||
bool: True if network was found and removed, False otherwise
|
||||
"""
|
||||
# Load current saved networks
|
||||
prefs = mpos.config.SharedPreferences("com.micropythonos.system.wifiservice")
|
||||
access_points = prefs.get_dict("access_points")
|
||||
|
||||
# Remove the network if it exists
|
||||
if ssid in access_points:
|
||||
del access_points[ssid]
|
||||
|
||||
# Save back to config
|
||||
editor = prefs.edit()
|
||||
editor.put_dict("access_points", access_points)
|
||||
editor.commit()
|
||||
|
||||
# Update class-level cache
|
||||
WifiService.access_points = access_points
|
||||
|
||||
print(f"WifiService: Forgot network '{ssid}'")
|
||||
return True
|
||||
else:
|
||||
print(f"WifiService: Network '{ssid}' not found in saved networks")
|
||||
return False
|
||||
@@ -152,7 +152,8 @@ def create_notification_bar():
|
||||
update_battery_icon() # run it immediately instead of waiting for the timer
|
||||
|
||||
def update_wifi_icon(timer):
|
||||
if mpos.wifi.WifiService.is_connected():
|
||||
from mpos.net.wifi_service import WifiService
|
||||
if WifiService.is_connected():
|
||||
wifi_icon.remove_flag(lv.obj.FLAG.HIDDEN)
|
||||
else:
|
||||
wifi_icon.add_flag(lv.obj.FLAG.HIDDEN)
|
||||
|
||||
@@ -1,101 +0,0 @@
|
||||
# Automatically connect to the WiFi, based on the saved networks
|
||||
# Manage concurrent accesses to the wifi (scan while connect, connect while scan etc)
|
||||
# Manage saved networks
|
||||
# This gets started in a new thread, does an autoconnect, and exits.
|
||||
|
||||
import ujson
|
||||
import os
|
||||
import time
|
||||
|
||||
import mpos.config
|
||||
import mpos.time
|
||||
|
||||
have_network = False
|
||||
try:
|
||||
import network
|
||||
have_network = True
|
||||
except Exception as e:
|
||||
print("Could not import network, have_network=False")
|
||||
|
||||
class WifiService():
|
||||
|
||||
wifi_busy = False # crude lock on wifi
|
||||
access_points = {}
|
||||
|
||||
@staticmethod
|
||||
def connect():
|
||||
wlan=network.WLAN(network.STA_IF)
|
||||
wlan.active(False) # restart WiFi hardware in case it's in a bad state
|
||||
wlan.active(True)
|
||||
networks = wlan.scan()
|
||||
for n in networks:
|
||||
ssid = n[0].decode()
|
||||
print(f"auto_connect: checking ssid '{ssid}'")
|
||||
if ssid in WifiService.access_points:
|
||||
password = WifiService.access_points.get(ssid).get("password")
|
||||
print(f"auto_connect: attempting to connect to saved network {ssid} with password {password}")
|
||||
if WifiService.attempt_connecting(ssid,password):
|
||||
print(f"auto_connect: Connected to {ssid}")
|
||||
return True
|
||||
else:
|
||||
print(f"auto_connect: failed to connect to {ssid}")
|
||||
else:
|
||||
print(f"auto_connect: not trying {ssid} because it hasn't been configured")
|
||||
print("auto_connect: no known networks connected")
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def attempt_connecting(ssid,password):
|
||||
print(f"auto_connect.py attempt_connecting: Attempting to connect to SSID: {ssid}")
|
||||
try:
|
||||
wlan=network.WLAN(network.STA_IF)
|
||||
wlan.connect(ssid,password)
|
||||
for i in range(10):
|
||||
if wlan.isconnected():
|
||||
print(f"auto_connect.py attempt_connecting: Connected to {ssid} after {i+1} seconds")
|
||||
mpos.time.sync_time()
|
||||
return True
|
||||
elif not wlan.active(): # wificonf app or others might stop the wifi, no point in continuing then
|
||||
print("auto_connect.py attempt_connecting: Someone disabled wifi, bailing out...")
|
||||
return False
|
||||
print(f"auto_connect.py attempt_connecting: Waiting for connection, attempt {i+1}/10")
|
||||
time.sleep(1)
|
||||
print(f"auto_connect.py attempt_connecting: Failed to connect to {ssid}")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"auto_connect.py attempt_connecting: Connection error: {e}")
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def auto_connect():
|
||||
print("auto_connect thread running")
|
||||
|
||||
# load config:
|
||||
WifiService.access_points = mpos.config.SharedPreferences("com.micropythonos.system.wifiservice").get_dict("access_points")
|
||||
if not len(WifiService.access_points):
|
||||
print("WifiService.py: not access points configured, exiting...")
|
||||
return
|
||||
|
||||
if not WifiService.wifi_busy:
|
||||
WifiService.wifi_busy = True
|
||||
if not have_network:
|
||||
print("auto_connect: no network module found, waiting to simulate connection...")
|
||||
time.sleep(10)
|
||||
print("auto_connect: wifi connect simulation done")
|
||||
else:
|
||||
if WifiService.connect():
|
||||
print("WifiService.py managed to connect.")
|
||||
else:
|
||||
print("WifiService.py did not manage to connect.")
|
||||
wlan=network.WLAN(network.STA_IF)
|
||||
wlan.active(False) # disable to conserve power
|
||||
WifiService.wifi_busy = False
|
||||
|
||||
@staticmethod
|
||||
def is_connected():
|
||||
if WifiService.wifi_busy:
|
||||
return False
|
||||
elif not have_network:
|
||||
return True
|
||||
else:
|
||||
return network.WLAN(network.STA_IF).isconnected()
|
||||
@@ -51,11 +51,11 @@ except Exception as e:
|
||||
print("main.py: WARNING: could not import/run freezefs_mount_builtin: ", e)
|
||||
|
||||
try:
|
||||
import mpos.wifi
|
||||
from mpos.net.wifi_service import WifiService
|
||||
_thread.stack_size(mpos.apps.good_stack_size())
|
||||
_thread.start_new_thread(mpos.wifi.WifiService.auto_connect, ())
|
||||
_thread.start_new_thread(WifiService.auto_connect, ())
|
||||
except Exception as e:
|
||||
print(f"Couldn't start mpos.wifi.WifiService.auto_connect thread because: {e}")
|
||||
print(f"Couldn't start WifiService.auto_connect thread because: {e}")
|
||||
|
||||
# Start launcher so it's always at bottom of stack
|
||||
launcher_app = PackageManager.get_launcher()
|
||||
|
||||
@@ -42,7 +42,9 @@ class MockNetwork:
|
||||
def __init__(self, interface, connected=True):
|
||||
self.interface = interface
|
||||
self._connected = connected
|
||||
self._active = True
|
||||
self._config = {}
|
||||
self._scan_results = [] # Can be configured for testing
|
||||
|
||||
def isconnected(self):
|
||||
"""Return whether the WLAN is connected."""
|
||||
@@ -51,7 +53,7 @@ class MockNetwork:
|
||||
def active(self, is_active=None):
|
||||
"""Get/set whether the interface is active."""
|
||||
if is_active is None:
|
||||
return self._connected
|
||||
return self._active
|
||||
self._active = is_active
|
||||
|
||||
def connect(self, ssid, password):
|
||||
@@ -73,6 +75,10 @@ class MockNetwork:
|
||||
return ('192.168.1.100', '255.255.255.0', '192.168.1.1', '8.8.8.8')
|
||||
return ('0.0.0.0', '0.0.0.0', '0.0.0.0', '0.0.0.0')
|
||||
|
||||
def scan(self):
|
||||
"""Scan for available networks."""
|
||||
return self._scan_results
|
||||
|
||||
def __init__(self, connected=True):
|
||||
"""
|
||||
Initialize mock network module.
|
||||
|
||||
@@ -0,0 +1,459 @@
|
||||
import unittest
|
||||
import sys
|
||||
|
||||
# Add tests directory to path for network_test_helper
|
||||
sys.path.insert(0, '../tests')
|
||||
|
||||
# Import network test helpers
|
||||
from network_test_helper import MockNetwork, MockTime
|
||||
|
||||
# Mock config classes
|
||||
class MockSharedPreferences:
|
||||
"""Mock SharedPreferences for testing."""
|
||||
_all_data = {} # Class-level storage
|
||||
|
||||
def __init__(self, app_id):
|
||||
self.app_id = app_id
|
||||
if app_id not in MockSharedPreferences._all_data:
|
||||
MockSharedPreferences._all_data[app_id] = {}
|
||||
|
||||
def get_dict(self, key):
|
||||
return MockSharedPreferences._all_data.get(self.app_id, {}).get(key, {})
|
||||
|
||||
def edit(self):
|
||||
return MockEditor(self)
|
||||
|
||||
@classmethod
|
||||
def reset_all(cls):
|
||||
cls._all_data = {}
|
||||
|
||||
|
||||
class MockEditor:
|
||||
"""Mock editor for SharedPreferences."""
|
||||
|
||||
def __init__(self, prefs):
|
||||
self.prefs = prefs
|
||||
self.pending = {}
|
||||
|
||||
def put_dict(self, key, value):
|
||||
self.pending[key] = value
|
||||
|
||||
def commit(self):
|
||||
if self.prefs.app_id not in MockSharedPreferences._all_data:
|
||||
MockSharedPreferences._all_data[self.prefs.app_id] = {}
|
||||
MockSharedPreferences._all_data[self.prefs.app_id].update(self.pending)
|
||||
|
||||
|
||||
# Create mock mpos module
|
||||
class MockMpos:
|
||||
"""Mock mpos module with config and time."""
|
||||
|
||||
class config:
|
||||
@staticmethod
|
||||
def SharedPreferences(app_id):
|
||||
return MockSharedPreferences(app_id)
|
||||
|
||||
class time:
|
||||
@staticmethod
|
||||
def sync_time():
|
||||
pass # No-op for testing
|
||||
|
||||
|
||||
# Inject mocks before importing WifiService
|
||||
sys.modules['mpos'] = MockMpos
|
||||
sys.modules['mpos.config'] = MockMpos.config
|
||||
sys.modules['mpos.time'] = MockMpos.time
|
||||
|
||||
# Add path to wifi_service.py
|
||||
sys.path.append('lib/mpos/net')
|
||||
|
||||
# Import WifiService
|
||||
from wifi_service import WifiService
|
||||
|
||||
|
||||
class TestWifiServiceConnect(unittest.TestCase):
|
||||
"""Test WifiService.connect() method."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up test fixtures."""
|
||||
MockSharedPreferences.reset_all()
|
||||
WifiService.access_points = {}
|
||||
WifiService.wifi_busy = False
|
||||
|
||||
def tearDown(self):
|
||||
"""Clean up after test."""
|
||||
WifiService.access_points = {}
|
||||
WifiService.wifi_busy = False
|
||||
|
||||
def test_connect_to_saved_network(self):
|
||||
"""Test connecting to a saved network."""
|
||||
mock_network = MockNetwork(connected=False)
|
||||
WifiService.access_points = {
|
||||
"TestNetwork": {"password": "testpass123"}
|
||||
}
|
||||
|
||||
# Configure mock scan results
|
||||
mock_wlan = mock_network.WLAN(mock_network.STA_IF)
|
||||
mock_wlan._scan_results = [(b"TestNetwork", -50, 1, 3, b"", 0)]
|
||||
|
||||
# Mock connect to succeed immediately
|
||||
def mock_connect(ssid, password):
|
||||
mock_wlan._connected = True
|
||||
|
||||
mock_wlan.connect = mock_connect
|
||||
|
||||
result = WifiService.connect(network_module=mock_network)
|
||||
|
||||
self.assertTrue(result)
|
||||
|
||||
def test_connect_with_no_saved_networks(self):
|
||||
"""Test connecting when no networks are saved."""
|
||||
mock_network = MockNetwork(connected=False)
|
||||
WifiService.access_points = {}
|
||||
|
||||
mock_wlan = mock_network.WLAN(mock_network.STA_IF)
|
||||
mock_wlan._scan_results = [(b"UnsavedNetwork", -50, 1, 3, b"", 0)]
|
||||
|
||||
result = WifiService.connect(network_module=mock_network)
|
||||
|
||||
self.assertFalse(result)
|
||||
|
||||
def test_connect_when_no_saved_networks_available(self):
|
||||
"""Test connecting when saved networks aren't in range."""
|
||||
mock_network = MockNetwork(connected=False)
|
||||
WifiService.access_points = {
|
||||
"SavedNetwork": {"password": "password123"}
|
||||
}
|
||||
|
||||
mock_wlan = mock_network.WLAN(mock_network.STA_IF)
|
||||
mock_wlan._scan_results = [(b"DifferentNetwork", -50, 1, 3, b"", 0)]
|
||||
|
||||
result = WifiService.connect(network_module=mock_network)
|
||||
|
||||
self.assertFalse(result)
|
||||
|
||||
|
||||
class TestWifiServiceAttemptConnecting(unittest.TestCase):
|
||||
"""Test WifiService.attempt_connecting() method."""
|
||||
|
||||
def test_successful_connection(self):
|
||||
"""Test successful WiFi connection."""
|
||||
mock_network = MockNetwork(connected=False)
|
||||
mock_time = MockTime()
|
||||
|
||||
mock_wlan = mock_network.WLAN(mock_network.STA_IF)
|
||||
|
||||
# Mock connect to succeed immediately
|
||||
call_count = [0]
|
||||
|
||||
def mock_connect(ssid, password):
|
||||
pass # Don't set connected yet
|
||||
|
||||
def mock_isconnected():
|
||||
call_count[0] += 1
|
||||
if call_count[0] >= 1:
|
||||
return True
|
||||
return False
|
||||
|
||||
mock_wlan.connect = mock_connect
|
||||
mock_wlan.isconnected = mock_isconnected
|
||||
|
||||
result = WifiService.attempt_connecting(
|
||||
"TestSSID",
|
||||
"testpass",
|
||||
network_module=mock_network,
|
||||
time_module=mock_time
|
||||
)
|
||||
|
||||
self.assertTrue(result)
|
||||
|
||||
def test_connection_timeout(self):
|
||||
"""Test connection timeout after 10 attempts."""
|
||||
mock_network = MockNetwork(connected=False)
|
||||
mock_time = MockTime()
|
||||
|
||||
mock_wlan = mock_network.WLAN(mock_network.STA_IF)
|
||||
|
||||
# Connection never succeeds
|
||||
def mock_isconnected():
|
||||
return False
|
||||
|
||||
mock_wlan.isconnected = mock_isconnected
|
||||
|
||||
result = WifiService.attempt_connecting(
|
||||
"TestSSID",
|
||||
"testpass",
|
||||
network_module=mock_network,
|
||||
time_module=mock_time
|
||||
)
|
||||
|
||||
self.assertFalse(result)
|
||||
# Should have slept 10 times
|
||||
self.assertEqual(len(mock_time.get_sleep_calls()), 10)
|
||||
|
||||
def test_connection_aborted_when_wifi_disabled(self):
|
||||
"""Test connection aborts if WiFi is disabled during attempt."""
|
||||
mock_network = MockNetwork(connected=False)
|
||||
mock_time = MockTime()
|
||||
|
||||
mock_wlan = mock_network.WLAN(mock_network.STA_IF)
|
||||
|
||||
# Never connected
|
||||
def mock_isconnected():
|
||||
return False
|
||||
|
||||
# WiFi becomes inactive on 3rd check
|
||||
check_count = [0]
|
||||
|
||||
def mock_active(state=None):
|
||||
if state is not None:
|
||||
mock_wlan._active = state
|
||||
return None
|
||||
check_count[0] += 1
|
||||
if check_count[0] >= 3:
|
||||
return False
|
||||
return True
|
||||
|
||||
mock_wlan.isconnected = mock_isconnected
|
||||
mock_wlan.active = mock_active
|
||||
|
||||
result = WifiService.attempt_connecting(
|
||||
"TestSSID",
|
||||
"testpass",
|
||||
network_module=mock_network,
|
||||
time_module=mock_time
|
||||
)
|
||||
|
||||
self.assertFalse(result)
|
||||
# Should have checked less than 10 times (aborted early)
|
||||
self.assertTrue(check_count[0] < 10)
|
||||
|
||||
def test_connection_error_handling(self):
|
||||
"""Test handling of connection errors."""
|
||||
mock_network = MockNetwork(connected=False)
|
||||
mock_time = MockTime()
|
||||
|
||||
mock_wlan = mock_network.WLAN(mock_network.STA_IF)
|
||||
|
||||
def raise_error(ssid, password):
|
||||
raise Exception("Connection failed")
|
||||
|
||||
mock_wlan.connect = raise_error
|
||||
|
||||
result = WifiService.attempt_connecting(
|
||||
"TestSSID",
|
||||
"testpass",
|
||||
network_module=mock_network,
|
||||
time_module=mock_time
|
||||
)
|
||||
|
||||
self.assertFalse(result)
|
||||
|
||||
|
||||
class TestWifiServiceAutoConnect(unittest.TestCase):
|
||||
"""Test WifiService.auto_connect() method."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up test fixtures."""
|
||||
MockSharedPreferences.reset_all()
|
||||
WifiService.access_points = {}
|
||||
WifiService.wifi_busy = False
|
||||
|
||||
def tearDown(self):
|
||||
"""Clean up after test."""
|
||||
WifiService.access_points = {}
|
||||
WifiService.wifi_busy = False
|
||||
MockSharedPreferences.reset_all()
|
||||
|
||||
def test_auto_connect_with_no_saved_networks(self):
|
||||
"""Test auto_connect when no networks are saved."""
|
||||
WifiService.auto_connect()
|
||||
|
||||
# Should exit early
|
||||
self.assertEqual(len(WifiService.access_points), 0)
|
||||
|
||||
def test_auto_connect_when_wifi_busy(self):
|
||||
"""Test auto_connect aborts when WiFi is busy."""
|
||||
# Save a network
|
||||
prefs = MockSharedPreferences("com.micropythonos.system.wifiservice")
|
||||
editor = prefs.edit()
|
||||
editor.put_dict("access_points", {"TestNet": {"password": "pass"}})
|
||||
editor.commit()
|
||||
|
||||
# Set WiFi as busy
|
||||
WifiService.wifi_busy = True
|
||||
|
||||
WifiService.auto_connect()
|
||||
|
||||
# Should still be busy (not changed)
|
||||
self.assertTrue(WifiService.wifi_busy)
|
||||
|
||||
def test_auto_connect_desktop_mode(self):
|
||||
"""Test auto_connect in desktop mode (no network module)."""
|
||||
mock_time = MockTime()
|
||||
|
||||
# Save a network
|
||||
prefs = MockSharedPreferences("com.micropythonos.system.wifiservice")
|
||||
editor = prefs.edit()
|
||||
editor.put_dict("access_points", {"TestNet": {"password": "pass"}})
|
||||
editor.commit()
|
||||
|
||||
WifiService.auto_connect(network_module=None, time_module=mock_time)
|
||||
|
||||
# Should have "slept" to simulate connection
|
||||
self.assertTrue(len(mock_time.get_sleep_calls()) > 0)
|
||||
# Should clear wifi_busy flag
|
||||
self.assertFalse(WifiService.wifi_busy)
|
||||
|
||||
|
||||
class TestWifiServiceIsConnected(unittest.TestCase):
|
||||
"""Test WifiService.is_connected() method."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up test fixtures."""
|
||||
WifiService.wifi_busy = False
|
||||
|
||||
def tearDown(self):
|
||||
"""Clean up after test."""
|
||||
WifiService.wifi_busy = False
|
||||
|
||||
def test_is_connected_when_connected(self):
|
||||
"""Test is_connected returns True when WiFi is connected."""
|
||||
mock_network = MockNetwork(connected=True)
|
||||
|
||||
result = WifiService.is_connected(network_module=mock_network)
|
||||
|
||||
self.assertTrue(result)
|
||||
|
||||
def test_is_connected_when_disconnected(self):
|
||||
"""Test is_connected returns False when WiFi is disconnected."""
|
||||
mock_network = MockNetwork(connected=False)
|
||||
|
||||
result = WifiService.is_connected(network_module=mock_network)
|
||||
|
||||
self.assertFalse(result)
|
||||
|
||||
def test_is_connected_when_wifi_busy(self):
|
||||
"""Test is_connected returns False when WiFi is busy."""
|
||||
mock_network = MockNetwork(connected=True)
|
||||
WifiService.wifi_busy = True
|
||||
|
||||
result = WifiService.is_connected(network_module=mock_network)
|
||||
|
||||
# Should return False even though connected
|
||||
self.assertFalse(result)
|
||||
|
||||
def test_is_connected_desktop_mode(self):
|
||||
"""Test is_connected in desktop mode."""
|
||||
result = WifiService.is_connected(network_module=None)
|
||||
|
||||
# Desktop mode always returns True
|
||||
self.assertTrue(result)
|
||||
|
||||
|
||||
class TestWifiServiceNetworkManagement(unittest.TestCase):
|
||||
"""Test network save/forget functionality."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up test fixtures."""
|
||||
MockSharedPreferences.reset_all()
|
||||
WifiService.access_points = {}
|
||||
|
||||
def tearDown(self):
|
||||
"""Clean up after test."""
|
||||
WifiService.access_points = {}
|
||||
MockSharedPreferences.reset_all()
|
||||
|
||||
def test_save_network(self):
|
||||
"""Test saving a network."""
|
||||
WifiService.save_network("MyNetwork", "mypassword123")
|
||||
|
||||
# Should be in class-level cache
|
||||
self.assertTrue("MyNetwork" in WifiService.access_points)
|
||||
self.assertEqual(WifiService.access_points["MyNetwork"]["password"], "mypassword123")
|
||||
|
||||
# Should be persisted
|
||||
prefs = MockSharedPreferences("com.micropythonos.system.wifiservice")
|
||||
saved = prefs.get_dict("access_points")
|
||||
self.assertTrue("MyNetwork" in saved)
|
||||
|
||||
def test_save_network_updates_existing(self):
|
||||
"""Test updating an existing saved network."""
|
||||
WifiService.save_network("MyNetwork", "oldpassword")
|
||||
WifiService.save_network("MyNetwork", "newpassword")
|
||||
|
||||
# Should have new password
|
||||
self.assertEqual(WifiService.access_points["MyNetwork"]["password"], "newpassword")
|
||||
|
||||
def test_forget_network(self):
|
||||
"""Test forgetting a saved network."""
|
||||
WifiService.save_network("MyNetwork", "mypassword")
|
||||
|
||||
result = WifiService.forget_network("MyNetwork")
|
||||
|
||||
self.assertTrue(result)
|
||||
self.assertFalse("MyNetwork" in WifiService.access_points)
|
||||
|
||||
def test_forget_nonexistent_network(self):
|
||||
"""Test forgetting a network that doesn't exist."""
|
||||
result = WifiService.forget_network("NonExistent")
|
||||
|
||||
self.assertFalse(result)
|
||||
|
||||
def test_get_saved_networks(self):
|
||||
"""Test getting list of saved networks."""
|
||||
WifiService.save_network("Network1", "pass1")
|
||||
WifiService.save_network("Network2", "pass2")
|
||||
WifiService.save_network("Network3", "pass3")
|
||||
|
||||
saved = WifiService.get_saved_networks()
|
||||
|
||||
self.assertEqual(len(saved), 3)
|
||||
self.assertTrue("Network1" in saved)
|
||||
self.assertTrue("Network2" in saved)
|
||||
self.assertTrue("Network3" in saved)
|
||||
|
||||
def test_get_saved_networks_empty(self):
|
||||
"""Test getting saved networks when none exist."""
|
||||
saved = WifiService.get_saved_networks()
|
||||
|
||||
self.assertEqual(len(saved), 0)
|
||||
|
||||
|
||||
class TestWifiServiceDisconnect(unittest.TestCase):
|
||||
"""Test WifiService.disconnect() method."""
|
||||
|
||||
def test_disconnect(self):
|
||||
"""Test disconnecting from WiFi."""
|
||||
mock_network = MockNetwork(connected=True)
|
||||
mock_wlan = mock_network.WLAN(mock_network.STA_IF)
|
||||
|
||||
# Track calls
|
||||
disconnect_called = [False]
|
||||
active_false_called = [False]
|
||||
|
||||
def mock_disconnect():
|
||||
disconnect_called[0] = True
|
||||
|
||||
def mock_active(state=None):
|
||||
if state is False:
|
||||
active_false_called[0] = True
|
||||
return True if state is None else None
|
||||
|
||||
mock_wlan.disconnect = mock_disconnect
|
||||
mock_wlan.active = mock_active
|
||||
|
||||
WifiService.disconnect(network_module=mock_network)
|
||||
|
||||
# Should have called both
|
||||
self.assertTrue(disconnect_called[0])
|
||||
self.assertTrue(active_false_called[0])
|
||||
|
||||
def test_disconnect_desktop_mode(self):
|
||||
"""Test disconnect in desktop mode."""
|
||||
# Should not raise an error
|
||||
WifiService.disconnect(network_module=None)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user