import json import time from pathlib import Path import pytest from app.config.store import ConfigStore from app.display.service import DisplayService from app.display.wifi_indicator import render_wifi_indicator from app.network.manager import MockNetworkManager, NmcliNetworkManager from app.network.service import WifiNetworkService, validate_wifi_settings from app.network.store import WifiConfigError, WifiConfigStore @pytest.mark.parametrize('raw_dns', ['192.0.2.1,192.0.2.2', '192.0.2.1\n192.0.2.2', '192.0.2.1, 192.0.2.2\n192.0.2.1']) def test_nmcli_multiple_dns_can_roundtrip_without_password(monkeypatch, raw_dns): backend = NmcliNetworkManager() values = {'802-11-wireless.ssid': 'test-network', 'ipv4.method': 'manual', 'ipv4.addresses': '192.0.2.20/24', 'ipv4.gateway': '192.0.2.1', 'ipv4.dns': raw_dns, '802-11-wireless-security.key-mgmt': 'wpa-psk'} monkeypatch.setattr(backend, '_connection_value', lambda uuid, field, **kwargs: values.get(field, '')) saved = backend.read_saved('test-profile') assert saved['dns_servers'] == ['192.0.2.1', '192.0.2.2'] checked = validate_wifi_settings(saved, previous=saved) assert checked['dns_servers'] == saved['dns_servers'] assert checked['password'] is None UUID = "12345678-1234-5678-1234-567812345678" def saved_profile(**updates): value = { "connection_uuid": UUID, "ssid": "test-hotspot", "password": "secret123", "password_configured": True, "ipv4_mode": "dhcp", "address": None, "prefix": None, "gateway": None, "dns_servers": [], } value.update(updates) return value def active_profile(**updates): value = { "connected": True, "connection_uuid": UUID, "ssid": "test-hotspot", "ipv4_address": "192.168.1.93", "control_url": "http://192.168.1.93:8080/", } value.update(updates) return value def test_wifi_store_creates_strict_v1_and_preserves_boot_session(tmp_path): data = tmp_path / "data" runtime = tmp_path / "run" first = WifiConfigStore(data, runtime, boot_id="boot-a", monotonic=lambda: 100.0) assert first.config == { "managed_connection_uuid": None, "prompt_delay_seconds": 30, } assert first.session["deadline_monotonic"] == 130.0 first.update({"managed_connection_uuid": UUID, "prompt_delay_seconds": 45}) first.dismiss_for_boot() same = WifiConfigStore(data, runtime, boot_id="boot-a", monotonic=lambda: 110.0) assert same.is_new_boot_session is False assert same.session["deadline_monotonic"] == 130.0 assert same.session["dismissed"] is True assert same.config["prompt_delay_seconds"] == 45 next_boot = WifiConfigStore(data, runtime, boot_id="boot-b", monotonic=lambda: 200.0) assert next_boot.is_new_boot_session is True assert next_boot.session["deadline_monotonic"] == 245.0 assert next_boot.session["dismissed"] is False persisted = json.loads((data / "wifi_config.json").read_text(encoding="utf-8")) assert persisted["schema_version"] == 1 @pytest.mark.parametrize( "document", [ {"schema_version": 2, "managed_connection_uuid": None, "prompt_delay_seconds": 30}, {"schema_version": 1, "managed_connection_uuid": None, "prompt_delay_seconds": True}, {"schema_version": 1, "managed_connection_uuid": None, "prompt_delay_seconds": 30, "extra": 1}, ], ) def test_invalid_wifi_store_is_rejected_without_rewrite(tmp_path, document): data = tmp_path / "data" data.mkdir() path = data / "wifi_config.json" original = (json.dumps(document, ensure_ascii=False) + "\n").encode("utf-8") path.write_bytes(original) with pytest.raises(WifiConfigError): WifiConfigStore(data, tmp_path / "run", boot_id="boot") assert path.read_bytes() == original def test_wifi_validation_defaults_optional_manual_fields(): checked = validate_wifi_settings( { "ssid": "phone", "password": "12345678", "ipv4_mode": "manual", "address": "192.168.7.50", "gateway": "192.168.7.1", "prefix": None, "dns_servers": [], }, previous=None, ) assert checked["prefix"] == 24 assert checked["dns_servers"] == ["192.168.7.1"] with pytest.raises(WifiConfigError, match="gateway"): validate_wifi_settings( { "ssid": "phone", "password": "12345678", "ipv4_mode": "manual", "address": "192.168.7.50", "gateway": "10.0.0.1", }, previous=None, ) def test_network_service_binds_active_profile_and_separates_activation_modes(tmp_path): config = ConfigStore(project_root=tmp_path) display = DisplayService(config, driver_kind="mock", startup_boot_id="boot") backend = MockNetworkManager(saved_profile(), active_profile()) wifi_store = WifiConfigStore(config.data_dir, config.runtime_dir, boot_id="boot") service = WifiNetworkService(wifi_store, backend, display) request = { "ssid": "test-hotspot", "password": None, "ipv4_mode": "manual", "address": "192.168.1.94", "gateway": "192.168.1.1", "prefix": None, "dns_servers": [], "prompt_delay_seconds": 30, "activation": "next_boot", } try: assert wifi_store.config["managed_connection_uuid"] == UUID assert "password" not in service.get_status()["saved"] assert service.get_status(include_secret=True)["saved"]["password"] == "secret123" saved = service.save_settings(request) assert saved["activation"] == "next_boot" assert backend.activations == [] assert backend.saved["address"] == "192.168.1.94" assert backend.saved["password"] == "secret123" deadline = wifi_store.session["deadline_monotonic"] network_snapshot = dict(backend.saved) delay_result = service.save_prompt_delay(45) assert delay_result["prompt_delay_seconds"] == 45 assert wifi_store.config["prompt_delay_seconds"] == 45 assert wifi_store.session["deadline_monotonic"] == deadline assert backend.saved == network_snapshot assert backend.activations == [] request["activation"] = "immediate" scheduled = service.save_settings(request) service.activate_saved(scheduled["operation_id"]) assert backend.activations == [UUID] assert wifi_store.session["operation"]["state"] == "succeeded" finally: service.close() display.close() def test_wifi_indicator_frames_and_display_overlay_restore_user_state(tmp_path): connected = render_wifi_indicator( "SSID: demo IP: http://192.168.1.2:8080/", offset=0, connected=True, ) disconnected = render_wifi_indicator( "SSID: demo 密码: 12345678", offset=0, connected=False, ) assert connected.mode == disconnected.mode == "RGB" assert connected.size == disconnected.size == (64, 64) assert disconnected.getbbox() is not None connected_text = render_wifi_indicator("IP: 192.168", offset=64, connected=True) text_pixels = [ pixel for y in range(26, 42) for x in range(64) if (pixel := connected_text.getpixel((x, y))) != (0, 0, 0) ] assert text_pixels assert set(text_pixels) == {(255, 255, 255)} disconnected_text = render_wifi_indicator("SSID: demo", offset=64, connected=False) lower_pixels = [ pixel for y in range(42, 64) for x in range(64) if (pixel := disconnected_text.getpixel((x, y))) != (0, 0, 0) ] assert lower_pixels assert set(lower_pixels) == {(255, 255, 255)} config = ConfigStore(project_root=tmp_path) config.update({"brightness": 27, "last_mode": "composition"}) display = DisplayService(config, driver_kind="mock", startup_boot_id="boot-wifi") try: display.fill((12, 34, 56)) before_revision = display.get_status()["state"]["revision"] display.show_wifi_indicator(connected=True, message="SSID: demo IP: http://192.168.1.2:8080/") status = display.get_status()["state"] assert status["mode"] == "wifi_connected_indicator" assert status["effective_brightness"] == 50 assert status["wifi_indicator_active"] is True assert status["revision"] == before_revision assert config.config["last_mode"] == "fill" display.dismiss_wifi_indicator() final = display.get_status()["state"] assert final["mode"] == "fill" assert final["effective_brightness"] == 27 assert display.get_current_frame().getpixel((0, 0)) == (12, 34, 56) finally: display.close() def test_network_cached_status_does_not_repeat_backend_queries(tmp_path): class CountingNetworkManager(MockNetworkManager): def __init__(self): super().__init__(saved_profile(), active_profile()) self.saved_reads = 0 self.active_reads = 0 def read_saved(self, connection_uuid, *, include_secret=False): self.saved_reads += 1 return super().read_saved(connection_uuid, include_secret=include_secret) def read_active(self): self.active_reads += 1 return super().read_active() config = ConfigStore(project_root=tmp_path) display = DisplayService(config, driver_kind="mock", startup_boot_id="boot") backend = CountingNetworkManager() wifi_store = WifiConfigStore(config.data_dir, config.runtime_dir, boot_id="boot") service = WifiNetworkService(wifi_store, backend, display) try: live = service.get_status() reads = (backend.saved_reads, backend.active_reads) backend.active = active_profile(connected=False) cached = service.get_cached_status() assert cached["active"] == live["active"] assert (backend.saved_reads, backend.active_reads) == reads finally: service.close() display.close() def test_dismissed_network_prompt_does_not_reappear_after_disconnect(tmp_path): config = ConfigStore(project_root=tmp_path) display = DisplayService(config, driver_kind="mock", startup_boot_id="boot") backend = MockNetworkManager(saved_profile(), active_profile()) wifi_store = WifiConfigStore(config.data_dir, config.runtime_dir, boot_id="boot") service = WifiNetworkService(wifi_store, backend, display, poll_seconds=0.01) try: service._refresh_indicator() assert display.get_status()["state"]["wifi_indicator_active"] is True service.dismiss_for_boot() backend.active = active_profile( connected=False, ssid=None, ipv4_address=None, control_url=None, ) time.sleep(0.02) assert wifi_store.session["dismissed"] is True assert display.get_status()["state"]["wifi_indicator_active"] is False finally: service.close() display.close()