Improve companion setup and USB session recovery
This commit is contained in:
mikecarper committed
1 parent
ecaa5ac2a0
commit
cf70478c51
34 files changed
+6283
-2147
No files matched your search
@@ -0,0 +1,106 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from pathlib import Path
|
||||
import re
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
class CompanionWiFiStatusTest(unittest.TestCase):
|
||||
def test_runtime_fallback_classification(self):
|
||||
source = r'''
|
||||
#include <helpers/CompanionWiFiStatus.h>
|
||||
|
||||
using mesh::wifi::CompanionWiFiFallbackState;
|
||||
using mesh::wifi::classifyCompanionWiFiFallback;
|
||||
|
||||
int main() {
|
||||
if (classifyCompanionWiFiFallback({false, false, false}) !=
|
||||
CompanionWiFiFallbackState::OffDisabled) return 1;
|
||||
if (classifyCompanionWiFiFallback({true, false, false}) !=
|
||||
CompanionWiFiFallbackState::Inactive) return 2;
|
||||
if (classifyCompanionWiFiFallback({true, true, true}) !=
|
||||
CompanionWiFiFallbackState::ReconnectScheduled) return 3;
|
||||
if (classifyCompanionWiFiFallback({true, true, false}) !=
|
||||
CompanionWiFiFallbackState::ConnectingOrRetrying) return 4;
|
||||
return 0;
|
||||
}
|
||||
'''
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
executable = Path(temp_dir) / "companion_wifi_status"
|
||||
compiled = subprocess.run(
|
||||
[
|
||||
"c++",
|
||||
"-std=c++11",
|
||||
f"-I{ROOT / 'src'}",
|
||||
"-x",
|
||||
"c++",
|
||||
"-",
|
||||
"-o",
|
||||
str(executable),
|
||||
],
|
||||
input=source,
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
self.assertEqual(compiled.returncode, 0, compiled.stderr)
|
||||
result = subprocess.run(
|
||||
[str(executable)], capture_output=True, text=True, check=False
|
||||
)
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
|
||||
def test_wifi_status_command_uses_runtime_formatter(self):
|
||||
mesh = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text()
|
||||
local_start = mesh.index("bool MyMesh::handleLocalControlCommand(")
|
||||
local_end = mesh.index(
|
||||
"\n#if COMPANION_FEATURE_TEMP_RADIO\nvoid MyMesh::serviceTempRadio()",
|
||||
local_start,
|
||||
)
|
||||
local = mesh[local_start:local_end]
|
||||
status_case = re.search(
|
||||
r"case mesh::cli::StandaloneWiFiKey::Status:(.*?)return true;",
|
||||
local,
|
||||
re.DOTALL,
|
||||
)
|
||||
self.assertIsNotNone(status_case)
|
||||
self.assertIn(
|
||||
"formatCompanionWiFiStatus(reply, reply_size);",
|
||||
status_case.group(1),
|
||||
)
|
||||
self.assertNotIn(
|
||||
"WebConfigServer::formatWiFiStatus", status_case.group(1)
|
||||
)
|
||||
|
||||
def test_active_runtime_fallback_is_not_off(self):
|
||||
server = (ROOT / "src/helpers/esp32/WebConfigServer.cpp").read_text()
|
||||
start = server.index("bool WebConfigServer::formatWiFiStatus(")
|
||||
end = server.index("\nbool WebConfigServer::startSetupMode", start)
|
||||
formatter = server[start:end]
|
||||
|
||||
self.assertIn("classifyCompanionWiFiFallback", formatter)
|
||||
self.assertIn(
|
||||
'> off, WiFi disabled; configured SSID: %s', formatter
|
||||
)
|
||||
self.assertIn(
|
||||
'> inactive, WiFi enabled but services stopped;', formatter
|
||||
)
|
||||
self.assertIn('> reconnect scheduled, SSID: %s', formatter)
|
||||
self.assertIn('> connecting/retrying, SSID: %s', formatter)
|
||||
# Existing rich setup/connecting and connected reports retain priority.
|
||||
self.assertLess(
|
||||
formatter.index("active->_mode == MODE_SETUP"),
|
||||
formatter.index("classifyCompanionWiFiFallback"),
|
||||
)
|
||||
self.assertLess(
|
||||
formatter.index("if (status == WL_CONNECTED)"),
|
||||
formatter.index("classifyCompanionWiFiFallback"),
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -45,6 +45,271 @@ class Esp32UsbSerialHygieneTest(unittest.TestCase):
|
||||
self.assertIn('"-DARDUINO_USB_CDC_ON_BOOT=1"', board)
|
||||
self.assertIn('"-DARDUINO_USB_MODE=1"', board)
|
||||
|
||||
def test_hwcdc_reply_route_has_completed_frame_grace(self):
|
||||
main = source("examples/companion_radio/main.cpp")
|
||||
start = main.index(
|
||||
"// The ESP32 USB-Serial-JTAG peripheral (HWCDC)"
|
||||
)
|
||||
hwcdc = main[start : main.index("#elif", start)]
|
||||
|
||||
serial_begin = main.index("Serial.begin(115200);")
|
||||
tx_buffer = main.index("Serial.setTxBufferSize(candidate);")
|
||||
logging_begin = main.index("mesh::beginUsbLoggingPort();")
|
||||
self.assertLess(tx_buffer, serial_begin)
|
||||
self.assertLess(serial_begin, logging_begin)
|
||||
|
||||
self.assertIn("board.isUsbHostConnected()", hwcdc)
|
||||
self.assertIn("usb_serial_interface.hasReceivedFrame()", hwcdc)
|
||||
self.assertIn("USB_FRAME_REPLY_GRACE_MS", hwcdc)
|
||||
self.assertIn("USB_HOST_LOSS_EDGE_MS", hwcdc)
|
||||
self.assertIn("USB_HOST_LOSS_GRACE_MS", hwcdc)
|
||||
self.assertIn("USB_CLIENT_IDLE_TIMEOUT", hwcdc)
|
||||
self.assertIn("getCompletedFrameCount()", hwcdc)
|
||||
self.assertIn(
|
||||
"the_mesh.hasFiniteDelayedReplyForRoute(&usb_serial_interface)",
|
||||
hwcdc,
|
||||
)
|
||||
self.assertNotIn("hasSerialOperationForRoute", hwcdc)
|
||||
self.assertNotIn("isReplyRouteLockedFor", hwcdc)
|
||||
self.assertNotIn("hasPendingWork()", hwcdc)
|
||||
self.assertNotIn("_iter_started", hwcdc)
|
||||
self.assertNotRegex(hwcdc, r"\(bool\)\s*Serial")
|
||||
|
||||
def test_finite_usb_lease_covers_only_bounded_reply_families(self):
|
||||
mesh = source("examples/companion_radio/MyMesh.cpp")
|
||||
start = mesh.index("bool MyMesh::hasFiniteDelayedReplyForRoute(")
|
||||
body = mesh[start : mesh.index("void MyMesh::servicePendingSerialReply()", start)]
|
||||
self.assertIn("pending_serial_reply_route == route", body)
|
||||
self.assertIn("command_radio_reply_route == route", body)
|
||||
self.assertIn("binary_trace_reply_route == route", body)
|
||||
self.assertIn("expected_ack_table[i].reply_route == route", body)
|
||||
self.assertNotIn("_iter_started", body)
|
||||
self.assertNotIn("lockReplyRoute", body)
|
||||
self.assertNotIn("hasPendingWork", body)
|
||||
self.assertNotIn("sign_data_reply_route", body)
|
||||
|
||||
def test_hwcdc_sustained_loss_resets_session_state(self):
|
||||
main = source("examples/companion_radio/main.cpp")
|
||||
start = main.index("static void serviceUsbTerminalHostSessionReset()")
|
||||
service = main[start : main.index("static void serviceUsbTerminal()", start)]
|
||||
reset_start = main.index("static void resetUsbTerminalHostSession(")
|
||||
reset = main[reset_start:start]
|
||||
|
||||
self.assertIn("takeSustainedHostLoss()", service)
|
||||
self.assertIn("takeHostLossEdge()", service)
|
||||
self.assertIn("mesh::takeUsbTerminalSessionReset()", service)
|
||||
self.assertIn(
|
||||
"hardware_bus_reset || physical_host_loss || sustained_host_loss",
|
||||
service,
|
||||
)
|
||||
self.assertIn("usb_terminal_host_reset_completion_pending", service)
|
||||
self.assertIn("USB_TRANSPORT_RESET_RETRY_MS", service)
|
||||
self.assertIn("resetUsbTerminalHostSession(true);", service)
|
||||
self.assertIn("cancelUsbSerialOperations();", reset)
|
||||
self.assertIn("usb_serial_interface.resetSessionState();", reset)
|
||||
self.assertIn("if (!mesh::resetUsbCompanionTransport())", reset)
|
||||
self.assertIn("usb_terminal_host_reset_completion_pending = true", reset)
|
||||
self.assertNotIn("board.reboot()", reset)
|
||||
self.assertIn("if (preserve_ascii_terminal)", reset)
|
||||
self.assertIn("the_mesh.resetTerminalSession();", reset)
|
||||
|
||||
logging = source("src/helpers/UsbLogging.cpp")
|
||||
helper_start = logging.index("static bool detachEsp32HwcdcPads()")
|
||||
purge_start = logging.index("bool resetUsbCompanionTransport()")
|
||||
helper = logging[helper_start:purge_start]
|
||||
purge = logging[purge_start : logging.index(
|
||||
"Stream& usbLoggingPort()", purge_start
|
||||
)]
|
||||
self.assertIn(
|
||||
"esp32_hwcdc_access_generation.fetch_add(", purge
|
||||
)
|
||||
self.assertIn("setPlatformDebugOutputEnabled(false);", purge)
|
||||
self.assertIn("detachEsp32HwcdcPads()", purge)
|
||||
self.assertIn("delay(10);", purge)
|
||||
self.assertIn("tryRunExclusive(", purge)
|
||||
self.assertIn("Serial.flush();", helper)
|
||||
self.assertIn("flush_attempts >= 2", helper)
|
||||
self.assertIn("Serial.availableForWrite()", helper)
|
||||
self.assertIn("esp32_hwcdc_tx_buffer_capacity.load(", helper)
|
||||
self.assertNotIn("Serial.setTxBufferSize(", helper)
|
||||
self.assertIn("while (Serial.read() >= 0)", helper)
|
||||
self.assertIn(
|
||||
"setPlatformDebugOutputEnabled(isUsbLoggingEnabled());", purge
|
||||
)
|
||||
self.assertIn(
|
||||
"esp32_hwcdc_self_reset_guard.expectSelfResetBurst()", purge
|
||||
)
|
||||
self.assertLess(
|
||||
purge.index("esp32_hwcdc_self_reset_guard.expectSelfResetBurst()"),
|
||||
purge.index("restoreEsp32HwcdcPads("),
|
||||
)
|
||||
self.assertIn(
|
||||
"esp32_hwcdc_allowed_generation.store(", purge
|
||||
)
|
||||
self.assertNotIn("setUsbLoggingEnabled(false);", purge)
|
||||
self.assertNotIn("Serial.end();", purge)
|
||||
self.assertNotIn("Serial.begin(", purge)
|
||||
|
||||
setup = main[main.index("void setup()") : main.index("board.begin();")]
|
||||
self.assertIn("static const size_t usb_tx_sizes[]", setup)
|
||||
self.assertIn("Serial.setTxBufferSize(candidate)", setup)
|
||||
self.assertIn("usb_tx_capacity = Serial.availableForWrite();", setup)
|
||||
self.assertIn(
|
||||
"mesh::setUsbCompanionTxBufferCapacity(usb_tx_capacity)", setup
|
||||
)
|
||||
|
||||
debug_start = logging.index(
|
||||
"static void setPlatformDebugOutputEnabled(bool enabled)"
|
||||
)
|
||||
debug = logging[debug_start : logging.index(
|
||||
"#if defined(MESH_DUAL_CDC_LOGGING)", debug_start
|
||||
)]
|
||||
self.assertIn("MESH_ESP32_HWCDC_SESSION_GUARD", debug)
|
||||
self.assertIn("Serial.setDebugOutput(false);", debug)
|
||||
self.assertNotIn("&& canAccessEsp32Hwcdc", debug)
|
||||
|
||||
companion_start = logging.index("Stream& usbCompanionPort()")
|
||||
companion = logging[companion_start : logging.index(
|
||||
"Stream& usbMotaPort()", companion_start
|
||||
)]
|
||||
self.assertIn("return guarded_esp32_hwcdc_port;", companion)
|
||||
mota_port_start = logging.index("Stream& usbMotaPort()")
|
||||
mota_port = logging[mota_port_start : logging.index(
|
||||
"Stream& usbTerminalPort()", mota_port_start
|
||||
)]
|
||||
self.assertIn(
|
||||
"return guarded_esp32_hwcdc_mota_port;", mota_port
|
||||
)
|
||||
self.assertIn(
|
||||
"AtomicWholeRecordNonBlockingStream<11>", logging
|
||||
)
|
||||
|
||||
ota_context = source("src/helpers/ota/OtaContext.h")
|
||||
ota_stream = ota_context[
|
||||
ota_context.index("#ifndef OTA_FOLDER_SERIAL_STREAM") :
|
||||
ota_context.index("#ifndef OTA_FOLDER_SERIAL_BAUD")
|
||||
]
|
||||
self.assertIn("defined(ESP32)", ota_stream)
|
||||
self.assertIn("ARDUINO_USB_MODE == 1", ota_stream)
|
||||
self.assertIn("defined(ENABLE_USB_INTERFACE)", ota_stream)
|
||||
self.assertIn("::mesh::usbMotaPort()", ota_stream)
|
||||
|
||||
self.assertIn(
|
||||
"Serial.onEvent(ARDUINO_HW_CDC_ANY_EVENT,", logging
|
||||
)
|
||||
self.assertIn(
|
||||
"esp32_hwcdc_bus_reset_generation.fetch_add(", logging
|
||||
)
|
||||
self.assertIn(
|
||||
"esp32_hwcdc_access_generation.fetch_add(", logging
|
||||
)
|
||||
self.assertIn(
|
||||
"esp32_hwcdc_self_reset_guard.shouldIgnoreBusReset()", logging
|
||||
)
|
||||
self.assertNotIn("esp32_hwcdc_expected_self_reset.exchange(", logging)
|
||||
self.assertIn(
|
||||
"esp32_hwcdc_self_reset_guard.notePostCleanActivity()", logging
|
||||
)
|
||||
take_start = logging.index("bool takeUsbTerminalSessionReset()")
|
||||
take = logging[take_start : logging.index(
|
||||
"bool tryCompleteUsbTerminalSessionReset()", take_start
|
||||
)]
|
||||
self.assertIn("MESH_ESP32_HWCDC_SESSION_GUARD", take)
|
||||
self.assertIn("esp32_hwcdc_bus_reset_generation.load(", take)
|
||||
|
||||
mota_start = main.index("static void serviceUsbMota()")
|
||||
mota_end = main.index(
|
||||
"static bool usb_terminal_host_reset_completion_pending", mota_start
|
||||
)
|
||||
mota = main[mota_start:mota_end]
|
||||
self.assertIn("#if !(defined(ESP32)", mota)
|
||||
self.assertIn("isUsbTerminalDataConnected()", mota)
|
||||
|
||||
def test_hwcdc_pad_detach_supports_idf4_and_idf5(self):
|
||||
logging = source("src/helpers/UsbLogging.cpp")
|
||||
wrappers = logging[
|
||||
logging.index("static bool detachEsp32HwcdcPads()") :
|
||||
logging.index("static void purgeEsp32HwcdcQueues(")
|
||||
]
|
||||
self.assertIn('#include "esp_idf_version.h"', logging)
|
||||
self.assertEqual(wrappers.count("#if ESP_IDF_VERSION_MAJOR >= 5"), 2)
|
||||
self.assertIn("usb_serial_jtag_ll_phy_is_pad_enabled()", wrappers)
|
||||
self.assertIn("usb_serial_jtag_ll_phy_enable_pad(false)", wrappers)
|
||||
self.assertIn("usb_serial_jtag_ll_phy_enable_pad(enabled)", wrappers)
|
||||
self.assertIn("usb_serial_jtag_ll_pad_backup_and_disable()", wrappers)
|
||||
self.assertIn("usb_serial_jtag_ll_enable_pad(enabled)", wrappers)
|
||||
|
||||
reset_start = logging.index("bool resetUsbCompanionTransport()")
|
||||
reset = logging[reset_start : logging.index(
|
||||
"Stream& usbLoggingPort()", reset_start
|
||||
)]
|
||||
self.assertNotIn("usb_serial_jtag_ll_pad_backup_and_disable()", reset)
|
||||
self.assertNotIn("usb_serial_jtag_ll_phy_enable_pad(", reset)
|
||||
|
||||
c6 = source("variants/m5stack_unit_c6l/platformio.ini")
|
||||
usb = c6[c6.index("[env:M5Stack_Unit_C6L_companion_radio_usb]") :]
|
||||
self.assertIn("-D ARDUINO_USB_MODE=1", usb)
|
||||
self.assertIn("-D ENABLE_USB_INTERFACE", usb)
|
||||
|
||||
def test_hwcdc_host_presence_uses_sof_signal(self):
|
||||
board = source("src/helpers/ESP32Board.h")
|
||||
start = board.index("bool isUsbHostConnected() override")
|
||||
host_check = board[start : board.index("void setInhibitSleep", start)]
|
||||
|
||||
self.assertIn("ARDUINO_USB_MODE", host_check)
|
||||
self.assertIn("return Serial.isPlugged();", host_check)
|
||||
|
||||
def test_hwcdc_retries_tx_kick_after_transient_sof_loss(self):
|
||||
logging = source("src/helpers/UsbLogging.cpp")
|
||||
self.assertIn(
|
||||
"static std::atomic<bool> esp32_hwcdc_tx_kick_pending{false};",
|
||||
logging,
|
||||
)
|
||||
write_start = logging.index(
|
||||
"size_t write(const uint8_t* data, size_t size) override"
|
||||
)
|
||||
write_end = logging.index("private:", write_start)
|
||||
guarded_write = logging[write_start:write_end]
|
||||
self.assertIn(
|
||||
"esp32_hwcdc_tx_kick_pending.store(true", guarded_write
|
||||
)
|
||||
self.assertIn("return Serial.write(data, size);", guarded_write)
|
||||
|
||||
kick_start = logging.index(
|
||||
"static void serviceEsp32HwcdcTxKickExclusive"
|
||||
)
|
||||
kick_end = logging.index("#endif", kick_start)
|
||||
kick = logging[kick_start:kick_end]
|
||||
self.assertIn("Serial.availableForWrite()", kick)
|
||||
self.assertIn("esp32_hwcdc_tx_buffer_capacity.load(", kick)
|
||||
self.assertIn("canAccessEsp32Hwcdc(nullptr)", kick)
|
||||
self.assertIn("Serial.isPlugged()", kick)
|
||||
self.assertIn("usb_serial_jtag_ll_txfifo_flush();", kick)
|
||||
self.assertIn("USB_SERIAL_JTAG_INTR_SERIAL_IN_EMPTY", kick)
|
||||
self.assertIn("portENTER_CRITICAL(&esp32_hwcdc_session_mux)", kick)
|
||||
self.assertIn("tryRunExclusive(", kick)
|
||||
|
||||
event_start = logging.index("static void handleEsp32HwcdcEvent")
|
||||
event_end = logging.index(
|
||||
"class Esp32HwcdcSessionStream", event_start
|
||||
)
|
||||
event = logging[event_start:event_end]
|
||||
self.assertIn(
|
||||
"esp32_hwcdc_tx_kick_pending.store(false", event
|
||||
)
|
||||
self.assertIn("usb_serial_jtag_ll_disable_intr_mask(", event)
|
||||
self.assertNotIn(
|
||||
"if (event_id == ARDUINO_HW_CDC_TX_EVENT) {", event
|
||||
)
|
||||
|
||||
service_start = logging.index("void serviceUsbTerminalPort()")
|
||||
service_end = logging.index(
|
||||
"void discardUsbTerminalOutput()", service_start
|
||||
)
|
||||
service = logging[service_start:service_end]
|
||||
self.assertIn("MESH_ESP32_HWCDC_SESSION_GUARD", service)
|
||||
self.assertIn("serviceEsp32HwcdcTxKick();", service)
|
||||
|
||||
def test_mqtt_ntp_detail_never_bypasses_logging_mode(self):
|
||||
text = source("src/helpers/bridges/MQTTBridge.cpp")
|
||||
self.assertIn(
|
||||
|
||||
@@ -0,0 +1,142 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from pathlib import Path
|
||||
import unittest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
PROFILE = ROOT / "variants" / "heltec_v4" / "platformio.ini"
|
||||
R8_PROFILE = ROOT / "variants" / "heltec_v4_r8" / "platformio.ini"
|
||||
UI = ROOT / "examples" / "companion_radio" / "ui-new" / "UITask.cpp"
|
||||
WIFI = ROOT / "examples" / "companion_radio" / "CompanionWiFi.h"
|
||||
MAIN = ROOT / "examples" / "companion_radio" / "main.cpp"
|
||||
|
||||
|
||||
def ini_section(source: str, name: str) -> str:
|
||||
start = source.index(f"[env:{name}]")
|
||||
end = source.find("\n[", start + 1)
|
||||
return source[start:] if end < 0 else source[start:end]
|
||||
|
||||
|
||||
class HeltecV4WiFiSetupPageTest(unittest.TestCase):
|
||||
def test_only_main_v4_oled_full_profiles_enable_the_page(self):
|
||||
profile = PROFILE.read_text(encoding="utf-8")
|
||||
full_profiles = (
|
||||
"heltec_v4_2_v4_3_companion_radio_full_femon",
|
||||
"heltec_v4_3_companion_radio_full_femoff",
|
||||
)
|
||||
for name in full_profiles:
|
||||
section = ini_section(profile, name)
|
||||
self.assertIn("-D UI_WIFI_SETUP_HOME_PAGE=1", section)
|
||||
self.assertIn("-D MESHCORE_REQUIRES_COMPANION_RADIO_FULL=1", section)
|
||||
|
||||
self.assertEqual(profile.count("-D UI_WIFI_SETUP_HOME_PAGE=1"), 2)
|
||||
self.assertNotIn(
|
||||
"UI_WIFI_SETUP_HOME_PAGE",
|
||||
ini_section(profile, "heltec_v4_companion_radio_wifi_femon"),
|
||||
)
|
||||
self.assertNotIn(
|
||||
"UI_WIFI_SETUP_HOME_PAGE",
|
||||
ini_section(profile, "heltec_v4_tft_companion_radio_full_femon"),
|
||||
)
|
||||
self.assertNotIn("UI_WIFI_SETUP_HOME_PAGE", R8_PROFILE.read_text())
|
||||
|
||||
def test_compact_page_fits_128_by_64_without_qr_or_cli_copy(self):
|
||||
ui = UI.read_text(encoding="utf-8")
|
||||
start = ui.index("static bool drawCompactCompanionWiFiSetupPage")
|
||||
end = ui.index("\nstatic void drawCompanionWiFiSetupPage", start)
|
||||
compact = ui[start:end]
|
||||
|
||||
self.assertIn("display.width() > 128 || display.height() > 64", compact)
|
||||
self.assertIn("display.setTextSize(1);", compact)
|
||||
self.assertIn('"SETUP AP ACTIVE" : "SETUP AP INACTIVE"', compact)
|
||||
self.assertIn('"HOLD STOP AP" : "HOLD START AP"', compact)
|
||||
self.assertIn('snprintf(open_ip, sizeof(open_ip), "OPEN %s", setup_ip)', compact)
|
||||
self.assertIn("display.drawTextEllipsized", compact)
|
||||
self.assertNotIn("drawQrCode", compact)
|
||||
self.assertNotIn("buildWiFiSetupQrPayload", compact)
|
||||
self.assertNotIn("start webconfig", compact.lower())
|
||||
|
||||
# Default size-one glyphs are six by eight logical pixels.
|
||||
for text in (
|
||||
"SETUP AP INACTIVE",
|
||||
"WIFI NOT CONFIGURED",
|
||||
"OPEN 255.255.255.255",
|
||||
"HOLD START AP",
|
||||
):
|
||||
self.assertLessEqual(len(text) * 6, 128)
|
||||
for y in (20, 31, 42, 53):
|
||||
self.assertLessEqual(y + 8, 64)
|
||||
|
||||
page_start = ui.index("static void drawCompanionWiFiSetupPage")
|
||||
page_end = ui.index("\n}\n#endif\n\n#include \"icons.h\"", page_start)
|
||||
page = ui[page_start:page_end]
|
||||
self.assertLess(
|
||||
page.index("drawCompactCompanionWiFiSetupPage("),
|
||||
page.index("buildWiFiSetupQrPayload("),
|
||||
)
|
||||
|
||||
def test_short_click_leaves_page_and_hold_toggles_session_ap(self):
|
||||
ui = UI.read_text(encoding="utf-8")
|
||||
home_handler_start = ui.index(" bool handleInput(char c) override {")
|
||||
home_handler_end = ui.index("\n};", home_handler_start)
|
||||
handler = ui[home_handler_start:home_handler_end]
|
||||
|
||||
next_page = handler.index("if (c == KEY_NEXT || c == KEY_RIGHT)")
|
||||
setup_action = handler.index(
|
||||
"if (c == KEY_ENTER && _page == HomePage::WIFI_SETUP)"
|
||||
)
|
||||
self.assertLess(next_page, setup_action)
|
||||
self.assertIn("_page = (_page + 1) % HomePage::Count;", handler)
|
||||
|
||||
setup_handler = handler[setup_action:]
|
||||
self.assertIn("WebConfigServer::getSetupInfo(nullptr, 0, nullptr, 0)", setup_handler)
|
||||
self.assertIn("requestCompanionWiFiSetupStop();", setup_handler)
|
||||
self.assertIn("requestCompanionWiFiSetup();", setup_handler)
|
||||
self.assertNotIn("!isCompanionWiFiConnected()", setup_handler)
|
||||
|
||||
loop = ui[ui.index("void UITask::loop()") : ui.index(
|
||||
"char UITask::checkDisplayOn", ui.index("void UITask::loop()")
|
||||
)]
|
||||
self.assertIn(
|
||||
"if (ev == BUTTON_EVENT_CLICK) {\n"
|
||||
" c = checkDisplayOn(KEY_NEXT);",
|
||||
loop,
|
||||
)
|
||||
self.assertIn(": handleLongPress(KEY_ENTER);", loop)
|
||||
|
||||
long_press = ui[ui.index("char UITask::handleLongPress") :]
|
||||
long_press = long_press[:long_press.index("\nchar UITask::handleDoubleClick")]
|
||||
page_bypass = long_press.index("isWiFiSetupPage()")
|
||||
cli_rescue = long_press.index("the_mesh.enterCLIRescue()")
|
||||
self.assertLess(page_bypass, cli_rescue)
|
||||
self.assertIn("return c;", long_press[page_bypass:cli_rescue])
|
||||
|
||||
def test_deferred_toggle_does_not_change_saved_preferences(self):
|
||||
wifi = WIFI.read_text(encoding="utf-8")
|
||||
self.assertIn("void requestCompanionWiFiSetupStop();", wifi)
|
||||
self.assertIn("current boot session", wifi)
|
||||
|
||||
main = MAIN.read_text(encoding="utf-8")
|
||||
requests_start = main.index("void requestCompanionWiFiSetup()")
|
||||
requests_end = main.index("bool toggleCompanionWiFi()", requests_start)
|
||||
requests = main[requests_start:requests_end]
|
||||
self.assertIn("companion_wifi_setup_requested = true;", requests)
|
||||
self.assertIn("companion_wifi_setup_stop_requested = true;", requests)
|
||||
self.assertNotIn("savePrefs", requests)
|
||||
self.assertNotIn("saveEnabled", requests)
|
||||
|
||||
loop = main[main.index("\nvoid loop()") :]
|
||||
stop = loop.index("if (companion_wifi_setup_stop_requested)")
|
||||
start = loop.index("if (companion_wifi_setup_requested)", stop)
|
||||
state_service = loop.index("serviceCompanionWiFiState();", start)
|
||||
self.assertLess(stop, start)
|
||||
self.assertLess(start, state_service)
|
||||
self.assertIn("the_mesh.stopWebConfig();", loop[stop:start])
|
||||
self.assertIn("the_mesh.startWebConfig(true, web_reply)", loop[start:state_service])
|
||||
self.assertNotIn("savePrefs", loop[stop:state_service])
|
||||
self.assertNotIn("saveEnabled", loop[stop:state_service])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -88,8 +88,12 @@ struct SingleAttemptContext {
|
||||
size_t nested_result = 99;
|
||||
bool reenter = false;
|
||||
bool accessible = true;
|
||||
bool try_exclusive_during_write = false;
|
||||
bool nested_exclusive_result = true;
|
||||
};
|
||||
|
||||
void noOpExclusive(void*) {}
|
||||
|
||||
size_t tryWriteOnce(void* opaque, const uint8_t*, size_t size) {
|
||||
SingleAttemptContext* context =
|
||||
static_cast<SingleAttemptContext*>(opaque);
|
||||
@@ -100,6 +104,10 @@ size_t tryWriteOnce(void* opaque, const uint8_t*, size_t size) {
|
||||
const uint8_t nested = 'n';
|
||||
context->nested_result = context->stream->write(&nested, 1);
|
||||
}
|
||||
if (context->try_exclusive_during_write && context->stream != nullptr) {
|
||||
context->nested_exclusive_result =
|
||||
context->stream->tryRunExclusive(noOpExclusive);
|
||||
}
|
||||
return context->result;
|
||||
}
|
||||
|
||||
@@ -366,6 +374,21 @@ TEST_F(Nrf52DebugOutputTest, SingleAttemptFacadeDropsAReentrantWriter) {
|
||||
EXPECT_EQ(context.nested_result, 0U);
|
||||
}
|
||||
|
||||
TEST_F(Nrf52DebugOutputTest,
|
||||
SingleAttemptFacadeExcludesAPurgeWhileAWriterIsActive) {
|
||||
SingleAttemptContext context;
|
||||
context.result = 1;
|
||||
context.try_exclusive_during_write = true;
|
||||
mesh::SingleAttemptNonBlockingStream port(logging_port, tryWriteOnce,
|
||||
&context, canAccess);
|
||||
context.stream = &port;
|
||||
const uint8_t value = 'x';
|
||||
|
||||
EXPECT_EQ(port.write(&value, 1), 1U);
|
||||
EXPECT_FALSE(context.nested_exclusive_result);
|
||||
EXPECT_TRUE(port.tryRunExclusive(noOpExclusive));
|
||||
}
|
||||
|
||||
TEST_F(Nrf52DebugOutputTest,
|
||||
BufferedFacadeRetainsACompleteMultiLineReplyAcrossShortDrains) {
|
||||
logging_port.reset(0);
|
||||
|
||||
@@ -493,19 +493,28 @@ class Nrf52UsbLoggingContractTest(unittest.TestCase):
|
||||
self.assertIn("bool tryCompleteUsbTerminalSessionReset();", header)
|
||||
self.assertIn("bool takeUsbTerminalSessionReset()", source)
|
||||
self.assertIn("bool tryCompleteUsbTerminalSessionReset()", source)
|
||||
reset = main[
|
||||
reset_helper = main[
|
||||
main.index("static void resetUsbTerminalHostSession(") :
|
||||
main.index("static void serviceUsbTerminalHostSessionReset()")
|
||||
]
|
||||
reset_service = main[
|
||||
main.index("static void serviceUsbTerminalHostSessionReset()") :
|
||||
main.index("static void serviceUsbTerminal()")
|
||||
]
|
||||
self.assertIn("mesh::takeUsbTerminalSessionReset()", reset)
|
||||
self.assertIn("mesh::tryCompleteUsbTerminalSessionReset()", reset)
|
||||
self.assertIn("leaveUsbMotaMode(false);", reset)
|
||||
self.assertIn("leaveUsbTerminalMode(false);", reset)
|
||||
self.assertIn("usb_serial_interface.setPassthroughMode(false);", reset)
|
||||
self.assertIn("clearUsbTerminalLine();", reset)
|
||||
self.assertIn("usb_binary_startup_probe.cancel();", reset)
|
||||
self.assertIn("usb_terminal_host_reset_completion_pending", reset)
|
||||
self.assertIn("cancelUsbSerialOperations();", reset)
|
||||
self.assertIn("mesh::takeUsbTerminalSessionReset()", reset_service)
|
||||
self.assertIn("mesh::tryCompleteUsbTerminalSessionReset()", reset_service)
|
||||
self.assertIn("resetUsbTerminalHostSession(false);", reset_service)
|
||||
self.assertLess(
|
||||
reset_service.index("resetUsbTerminalHostSession(false);"),
|
||||
reset_service.index("mesh::tryCompleteUsbTerminalSessionReset()"),
|
||||
)
|
||||
self.assertIn("leaveUsbMotaMode(false);", reset_helper)
|
||||
self.assertIn("leaveUsbTerminalMode(false);", reset_helper)
|
||||
self.assertIn("usb_serial_interface.setPassthroughMode(false);", reset_helper)
|
||||
self.assertIn("clearUsbTerminalLine();", reset_helper)
|
||||
self.assertIn("usb_binary_startup_probe.cancel();", reset_helper)
|
||||
self.assertIn("usb_terminal_host_reset_completion_pending", reset_service)
|
||||
self.assertIn("cancelUsbSerialOperations();", reset_helper)
|
||||
usb_cancel = main[
|
||||
main.index("static void cancelUsbSerialOperations()"):
|
||||
main.index("static void enterUsbTerminalMode()")
|
||||
@@ -525,7 +534,7 @@ class Nrf52UsbLoggingContractTest(unittest.TestCase):
|
||||
self.assertLess(route_check, route_cancel)
|
||||
self.assertLess(route_cancel, delayed_cancel)
|
||||
self.assertLess(delayed_cancel, route_forget)
|
||||
self.assertIn("the_mesh.resetUsbHostSessionInput();", reset)
|
||||
self.assertIn("the_mesh.resetUsbHostSessionInput();", reset_helper)
|
||||
self.assertNotRegex(main, r"Serial\.(?:available|read|peek)\(")
|
||||
mesh_source = (
|
||||
ROOT / "examples/companion_radio/MyMesh.cpp"
|
||||
|
||||
@@ -52,6 +52,206 @@ public:
|
||||
static const char START_TOKEN[] = "+++MESHCORE-TERM-START";
|
||||
static const char SEEDER_TOKEN[] = "ota folder on";
|
||||
|
||||
TEST(UsbHwcdcSession, IgnoresWholeExpectedSelfResetBurstUntilActivity) {
|
||||
mesh::UsbSelfResetBurstGuard guard;
|
||||
EXPECT_FALSE(guard.shouldIgnoreBusReset());
|
||||
|
||||
guard.expectSelfResetBurst();
|
||||
EXPECT_TRUE(guard.shouldIgnoreBusReset());
|
||||
EXPECT_TRUE(guard.shouldIgnoreBusReset());
|
||||
|
||||
guard.notePostCleanActivity();
|
||||
EXPECT_FALSE(guard.shouldIgnoreBusReset());
|
||||
}
|
||||
|
||||
TEST(UsbHwcdcSession, CompletedFrameGetsImmediateReplyLease) {
|
||||
mesh::UsbHostPresenceDebouncer tracker;
|
||||
EXPECT_TRUE(tracker.isClientConnected(
|
||||
false, 100, true, 100, 1, 2000, 600000, 100, 2000, false));
|
||||
EXPECT_TRUE(tracker.isClientConnected(
|
||||
false, 2099, true, 100, 1, 2000, 600000, 100, 2000, false));
|
||||
EXPECT_FALSE(tracker.isClientConnected(
|
||||
false, 2100, true, 100, 1, 2000, 600000, 100, 2000, false));
|
||||
EXPECT_TRUE(tracker.takeSustainedHostLoss());
|
||||
}
|
||||
|
||||
TEST(UsbHwcdcSession, DebouncesTransientHostLossAndReportsSustainedLossOnce) {
|
||||
mesh::UsbHostPresenceDebouncer tracker;
|
||||
EXPECT_TRUE(tracker.observeHost(true, 100, 100, 2000));
|
||||
// A long interval without sampling is not itself evidence of continuous
|
||||
// loss. The debounce window starts at the first observed false sample.
|
||||
EXPECT_TRUE(tracker.observeHost(false, 5000, 100, 2000));
|
||||
EXPECT_TRUE(tracker.observeHost(false, 6999, 100, 2000));
|
||||
EXPECT_FALSE(tracker.takeSustainedHostLoss());
|
||||
EXPECT_FALSE(tracker.observeHost(false, 7000, 100, 2000));
|
||||
EXPECT_TRUE(tracker.takeSustainedHostLoss());
|
||||
EXPECT_FALSE(tracker.takeSustainedHostLoss());
|
||||
}
|
||||
|
||||
TEST(UsbHwcdcSession, DebouncesAndReportsPhysicalLossEdgeOnce) {
|
||||
mesh::UsbHostPresenceDebouncer tracker;
|
||||
EXPECT_TRUE(tracker.observeHost(true, 100, 100, 2000));
|
||||
EXPECT_TRUE(tracker.observeHost(false, 110, 100, 2000));
|
||||
EXPECT_FALSE(tracker.takeHostLossEdge());
|
||||
EXPECT_TRUE(tracker.observeHost(false, 209, 100, 2000));
|
||||
EXPECT_FALSE(tracker.takeHostLossEdge());
|
||||
EXPECT_TRUE(tracker.observeHost(false, 210, 100, 2000));
|
||||
EXPECT_TRUE(tracker.takeHostLossEdge());
|
||||
EXPECT_FALSE(tracker.takeHostLossEdge());
|
||||
|
||||
// Repeated false samples belong to the same edge.
|
||||
EXPECT_TRUE(tracker.observeHost(false, 220, 100, 2000));
|
||||
EXPECT_FALSE(tracker.takeHostLossEdge());
|
||||
}
|
||||
|
||||
TEST(UsbHwcdcSession, IgnoresTransientPhysicalLossBeforeEdgeDebounce) {
|
||||
mesh::UsbHostPresenceDebouncer tracker;
|
||||
EXPECT_TRUE(tracker.observeHost(true, 100, 100, 2000));
|
||||
EXPECT_TRUE(tracker.observeHost(false, 110, 100, 2000));
|
||||
EXPECT_TRUE(tracker.observeHost(true, 150, 100, 2000));
|
||||
EXPECT_FALSE(tracker.takeHostLossEdge());
|
||||
}
|
||||
|
||||
TEST(UsbHwcdcSession, FrameProofAloneDoesNotInventPhysicalLossEdge) {
|
||||
mesh::UsbHostPresenceDebouncer tracker;
|
||||
EXPECT_TRUE(tracker.isClientConnected(
|
||||
false, 100, true, 100, 1, 2000, 600000, 100, 2000, false));
|
||||
EXPECT_FALSE(tracker.takeHostLossEdge());
|
||||
}
|
||||
|
||||
TEST(UsbHwcdcSession, DebounceHandlesMillisRollover) {
|
||||
mesh::UsbHostPresenceDebouncer tracker;
|
||||
const uint32_t started = 0xFFFFFFF0u;
|
||||
EXPECT_TRUE(tracker.observeHost(true, started, 100, 2000));
|
||||
EXPECT_TRUE(tracker.observeHost(false, started + 100u, 100, 2000));
|
||||
EXPECT_TRUE(tracker.observeHost(false, started + 2099u, 100, 2000));
|
||||
EXPECT_FALSE(tracker.observeHost(false, started + 2100u, 100, 2000));
|
||||
EXPECT_TRUE(tracker.takeSustainedHostLoss());
|
||||
}
|
||||
|
||||
TEST(UsbHwcdcSession, IdleLeaseIsAbsoluteWhileHostRemains) {
|
||||
mesh::UsbHostPresenceDebouncer tracker;
|
||||
EXPECT_FALSE(tracker.isClientConnected(
|
||||
true, 700100, true, 100, 1, 2000, 600000, 100, 2000, false));
|
||||
}
|
||||
|
||||
TEST(UsbHwcdcSession, ReplugDoesNotErasePendingOldSessionLoss) {
|
||||
mesh::UsbHostPresenceDebouncer tracker;
|
||||
EXPECT_TRUE(tracker.observeHost(true, 10, 100, 2000));
|
||||
EXPECT_TRUE(tracker.observeHost(false, 20, 100, 2000));
|
||||
EXPECT_FALSE(tracker.observeHost(false, 2020, 100, 2000));
|
||||
EXPECT_FALSE(tracker.observeHost(true, 2030, 100, 2000));
|
||||
EXPECT_TRUE(tracker.takeSustainedHostLoss());
|
||||
EXPECT_FALSE(tracker.takeSustainedHostLoss());
|
||||
}
|
||||
|
||||
TEST(UsbHwcdcSession, SustainedLossOverridesFreshFrameUntilReset) {
|
||||
mesh::UsbHostPresenceDebouncer tracker;
|
||||
EXPECT_TRUE(tracker.observeHost(true, 10, 100, 2000));
|
||||
EXPECT_TRUE(tracker.observeHost(false, 20, 100, 2000));
|
||||
EXPECT_FALSE(tracker.observeHost(false, 2020, 100, 2000));
|
||||
EXPECT_FALSE(tracker.isClientConnected(
|
||||
false, 2030, true, 2030, 1, 2000, 600000, 100, 2000, false));
|
||||
EXPECT_FALSE(tracker.isClientConnected(
|
||||
true, 2040, true, 2040, 2, 2000, 600000, 100, 2000, false));
|
||||
EXPECT_TRUE(tracker.takeSustainedHostLoss());
|
||||
tracker.reset();
|
||||
EXPECT_TRUE(tracker.isClientConnected(
|
||||
true, 2050, true, 2050, 3, 2000, 600000, 100, 2000, false));
|
||||
}
|
||||
|
||||
TEST(UsbHwcdcSession, FiniteReplyExtendsIdleAndGetsOneDrainWindow) {
|
||||
mesh::UsbHostPresenceDebouncer tracker;
|
||||
const uint32_t last_frame = 100;
|
||||
const uint32_t idle_limit = 600000;
|
||||
const uint32_t now = last_frame + idle_limit;
|
||||
|
||||
EXPECT_TRUE(tracker.isClientConnected(
|
||||
true, now, true, last_frame, 1, 2000, idle_limit, 100, 2000, true));
|
||||
EXPECT_TRUE(tracker.isClientConnected(
|
||||
true, now + 1, true, last_frame, 1, 2000, idle_limit, 100, 2000, false));
|
||||
EXPECT_TRUE(tracker.isClientConnected(
|
||||
true, now + 2000, true, last_frame, 1, 2000, idle_limit, 100, 2000, false));
|
||||
EXPECT_FALSE(tracker.isClientConnected(
|
||||
true, now + 2001, true, last_frame, 1, 2000, idle_limit, 100, 2000, false));
|
||||
EXPECT_FALSE(tracker.isClientConnected(
|
||||
true, now + 3000, true, last_frame, 1, 2000, idle_limit, 100, 2000, false));
|
||||
}
|
||||
|
||||
TEST(UsbHwcdcSession, NewFiniteReplyRearmsAfterDrain) {
|
||||
mesh::UsbHostPresenceDebouncer tracker;
|
||||
EXPECT_TRUE(tracker.isClientConnected(
|
||||
true, 700000, true, 1, 1, 2000, 600000, 100, 2000, true));
|
||||
EXPECT_TRUE(tracker.isClientConnected(
|
||||
true, 700001, true, 1, 1, 2000, 600000, 100, 2000, false));
|
||||
EXPECT_TRUE(tracker.isClientConnected(
|
||||
true, 701000, true, 1, 1, 2000, 600000, 100, 2000, true));
|
||||
EXPECT_TRUE(tracker.isClientConnected(
|
||||
true, 701001, true, 1, 1, 2000, 600000, 100, 2000, false));
|
||||
EXPECT_FALSE(tracker.isClientConnected(
|
||||
true, 703001, true, 1, 1, 2000, 600000, 100, 2000, false));
|
||||
}
|
||||
|
||||
TEST(UsbHwcdcSession, SustainedLossOverridesFiniteReplyAndResetClearsIt) {
|
||||
mesh::UsbHostPresenceDebouncer tracker;
|
||||
EXPECT_TRUE(tracker.observeHost(true, 10, 100, 2000));
|
||||
EXPECT_TRUE(tracker.isClientConnected(
|
||||
true, 700000, true, 1, 1, 2000, 600000, 100, 2000, true));
|
||||
EXPECT_TRUE(tracker.observeHost(false, 700010, 100, 2000));
|
||||
EXPECT_FALSE(tracker.observeHost(false, 702010, 100, 2000));
|
||||
EXPECT_FALSE(tracker.isClientConnected(
|
||||
false, 702011, true, 1, 1, 2000, 600000, 100, 2000, true));
|
||||
tracker.reset();
|
||||
EXPECT_FALSE(tracker.isClientConnected(
|
||||
true, 702012, true, 1, 1, 2000, 600000, 100, 2000, false));
|
||||
}
|
||||
|
||||
TEST(UsbHwcdcSession, FiniteReplyDrainHandlesMillisRollover) {
|
||||
mesh::UsbHostPresenceDebouncer tracker;
|
||||
const uint32_t started = 0xFFFFFFF0u;
|
||||
EXPECT_TRUE(tracker.isClientConnected(
|
||||
true, started, true, started - 600001u, 1,
|
||||
2000, 600000, 100, 2000, true));
|
||||
EXPECT_TRUE(tracker.isClientConnected(
|
||||
true, started + 1u, true, started - 600001u, 1,
|
||||
2000, 600000, 100, 2000, false));
|
||||
EXPECT_TRUE(tracker.isClientConnected(
|
||||
true, started + 2000u, true, started - 600001u, 1,
|
||||
2000, 600000, 100, 2000, false));
|
||||
EXPECT_FALSE(tracker.isClientConnected(
|
||||
true, started + 2001u, true, started - 600001u, 1,
|
||||
2000, 600000, 100, 2000, false));
|
||||
}
|
||||
|
||||
TEST(UsbHwcdcSession, SessionResetClearsFrameProofAndPartialIo) {
|
||||
BufferStream stream;
|
||||
ArduinoSerialInterface interface;
|
||||
interface.begin(stream, START_TOKEN);
|
||||
interface.enable();
|
||||
interface.enableFlowControl(true);
|
||||
|
||||
const uint8_t complete[] = {'<', 1, 0, 0x05};
|
||||
stream.push(complete, sizeof(complete));
|
||||
uint8_t frame[MAX_FRAME_SIZE] = {};
|
||||
ASSERT_EQ(interface.checkRecvFrame(frame), 1u);
|
||||
ASSERT_TRUE(interface.hasReceivedFrame());
|
||||
const uint32_t completed = interface.getCompletedFrameCount();
|
||||
|
||||
stream.write_capacity = 0;
|
||||
const uint8_t response[] = {0x06, 0x42};
|
||||
ASSERT_EQ(interface.writeFrame(response, sizeof(response)),
|
||||
sizeof(response));
|
||||
const uint8_t partial[] = {'<', 2, 0, 0x16};
|
||||
stream.push(partial, sizeof(partial));
|
||||
EXPECT_EQ(interface.checkRecvFrame(frame), 0u);
|
||||
ASSERT_TRUE(interface.hasPendingIO());
|
||||
|
||||
interface.resetSessionState();
|
||||
EXPECT_FALSE(interface.hasReceivedFrame());
|
||||
EXPECT_FALSE(interface.hasPendingIO());
|
||||
EXPECT_EQ(interface.getCompletedFrameCount(), completed);
|
||||
}
|
||||
|
||||
TEST(UsbMotaOwnerPolicy, MatchesEveryOtaCliAttachDetachSpelling) {
|
||||
EXPECT_TRUE(mesh::isUsbMotaOwnerTransitionCommand("ota folder on"));
|
||||
EXPECT_TRUE(mesh::isUsbMotaOwnerTransitionCommand("ota folder off"));
|
||||
|
||||
@@ -0,0 +1,159 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from pathlib import Path
|
||||
import re
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
class StorageLayoutCliTest(unittest.TestCase):
|
||||
def test_shared_matcher_formatter_and_host_fallback(self):
|
||||
source = r'''
|
||||
#include <helpers/StorageLayout.h>
|
||||
#include <string.h>
|
||||
|
||||
namespace mesh { class MainBoard {}; }
|
||||
|
||||
using mesh::cli::Esp32StorageLayoutWriter;
|
||||
using mesh::cli::StorageLayoutGetMatch;
|
||||
using mesh::cli::classifyStorageLayoutGet;
|
||||
|
||||
int main() {
|
||||
if (classifyStorageLayoutGet(nullptr) !=
|
||||
StorageLayoutGetMatch::NotMatched) return 1;
|
||||
if (classifyStorageLayoutGet("storage.layout") !=
|
||||
StorageLayoutGetMatch::Valid) return 2;
|
||||
if (classifyStorageLayoutGet("storage.layout extra") !=
|
||||
StorageLayoutGetMatch::InvalidArguments) return 3;
|
||||
if (classifyStorageLayoutGet("storage.layout\textra") !=
|
||||
StorageLayoutGetMatch::InvalidArguments) return 4;
|
||||
if (classifyStorageLayoutGet("storage.layoutx") !=
|
||||
StorageLayoutGetMatch::NotMatched) return 5;
|
||||
|
||||
char layout[160];
|
||||
Esp32StorageLayoutWriter writer(layout, sizeof(layout), 8UL * 1024UL * 1024UL);
|
||||
if (!writer.append("nvs", 0x9000, 20UL * 1024UL, false)) return 6;
|
||||
if (!writer.append("app0", 0x10000, 1984UL * 1024UL, true)) return 7;
|
||||
writer.finish();
|
||||
if (strcmp(layout,
|
||||
"> int:esp32=8192K ext:none; nvs@0x9000+20K,app0*@0x10000+1984K") != 0)
|
||||
return 8;
|
||||
|
||||
char empty[80];
|
||||
Esp32StorageLayoutWriter empty_writer(empty, sizeof(empty), 4UL * 1024UL * 1024UL);
|
||||
empty_writer.finish();
|
||||
if (strcmp(empty, "> int:esp32=4096K ext:none; partitions=none") != 0)
|
||||
return 9;
|
||||
|
||||
char bounded[64];
|
||||
Esp32StorageLayoutWriter bounded_writer(
|
||||
bounded, sizeof(bounded), 4UL * 1024UL * 1024UL);
|
||||
if (!bounded_writer.append("nvs", 0x9000, 20UL * 1024UL, false)) return 10;
|
||||
if (bounded_writer.append("large_app", 0x10000, 1984UL * 1024UL, true)) return 11;
|
||||
bounded_writer.finish();
|
||||
if (!bounded_writer.truncated() || strstr(bounded, ",...") == nullptr) return 12;
|
||||
if (strlen(bounded) >= sizeof(bounded)) return 13;
|
||||
|
||||
char bytes[24];
|
||||
mesh::cli::formatStorageBytes(bytes, sizeof(bytes), 1536);
|
||||
if (strcmp(bytes, "1.5 KiB") != 0) return 14;
|
||||
|
||||
mesh::MainBoard board;
|
||||
char reply[80];
|
||||
if (!mesh::cli::handleStorageLayoutGet(
|
||||
"storage.layout", board, reply, sizeof(reply))) return 15;
|
||||
if (strcmp(reply,
|
||||
"Error: storage layout unsupported on this platform") != 0)
|
||||
return 16;
|
||||
if (!mesh::cli::handleStorageLayoutGet(
|
||||
"storage.layout bad", board, reply, sizeof(reply))) return 17;
|
||||
if (strcmp(reply, "Error: use get storage.layout") != 0) return 18;
|
||||
if (mesh::cli::handleStorageLayoutGet(
|
||||
"storage.layoutx", board, reply, sizeof(reply))) return 19;
|
||||
return 0;
|
||||
}
|
||||
'''
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
executable = Path(temp_dir) / "storage_layout"
|
||||
compiled = subprocess.run(
|
||||
[
|
||||
"c++",
|
||||
"-std=c++11",
|
||||
f"-I{ROOT / 'src'}",
|
||||
"-x",
|
||||
"c++",
|
||||
"-",
|
||||
str(ROOT / "src/helpers/StorageLayout.cpp"),
|
||||
"-o",
|
||||
str(executable),
|
||||
],
|
||||
input=source,
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
self.assertEqual(compiled.returncode, 0, compiled.stderr)
|
||||
result = subprocess.run(
|
||||
[str(executable)], capture_output=True, text=True, check=False
|
||||
)
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
|
||||
def test_common_and_companion_use_one_platform_formatter(self):
|
||||
common = (ROOT / "src/helpers/CommonCLI.cpp").read_text()
|
||||
common_header = (ROOT / "src/helpers/CommonCLI.h").read_text()
|
||||
companion = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text()
|
||||
formatter = (ROOT / "src/helpers/StorageLayout.cpp").read_text()
|
||||
|
||||
self.assertIn('#include "StorageLayout.h"', common)
|
||||
self.assertIn(
|
||||
"mesh::cli::handleStorageLayoutGet(config, *_board, reply, 160)",
|
||||
common,
|
||||
)
|
||||
self.assertNotIn("handleStorageLayoutGetCmd", common)
|
||||
self.assertNotIn("handleStorageLayoutGetCmd", common_header)
|
||||
self.assertNotIn("esp_partition_find", common)
|
||||
|
||||
local_start = companion.index("bool MyMesh::handleLocalControlCommand(")
|
||||
local_end = companion.index(
|
||||
"\n#if COMPANION_FEATURE_TEMP_RADIO\nvoid MyMesh::serviceTempRadio()",
|
||||
local_start,
|
||||
)
|
||||
local = companion[local_start:local_end]
|
||||
self.assertRegex(
|
||||
local,
|
||||
re.compile(
|
||||
r'if \(strncmp\(command, "get ", 4\) == 0 &&\s+'
|
||||
r'mesh::cli::handleStorageLayoutGet\(command \+ 4, board, reply,\s+'
|
||||
r'reply_size\)\)',
|
||||
re.DOTALL,
|
||||
),
|
||||
)
|
||||
self.assertIn(
|
||||
'terminalOutput().print(" get storage.layout\\r\\n");',
|
||||
companion,
|
||||
)
|
||||
|
||||
# ESP-IDF discovery and every platform-specific result live only in the
|
||||
# shared formatter; neither role owns a second copy.
|
||||
self.assertEqual(formatter.count("esp_partition_find("), 1)
|
||||
self.assertEqual(formatter.count('> int:esp32=%luK ext:none; '), 0)
|
||||
writer = (ROOT / "src/helpers/StorageLayout.h").read_text()
|
||||
self.assertEqual(writer.count('> int:esp32=%luK ext:none; '), 1)
|
||||
|
||||
def test_companion_availability_is_documented(self):
|
||||
availability = (
|
||||
ROOT / "docs/cli_command_availability.md"
|
||||
).read_text()
|
||||
companion_start = availability.index("## Companion framed CLI")
|
||||
companion_end = availability.index("\n## nRF52", companion_start)
|
||||
companion_section = availability[companion_start:companion_end]
|
||||
self.assertIn("[`get storage.layout`]", companion_section)
|
||||
self.assertIn("Every Companion text terminal and command `0x42`", companion_section)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,121 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import gzip
|
||||
from pathlib import Path
|
||||
import re
|
||||
import unittest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SERVER = ROOT / "src" / "helpers" / "esp32" / "WebConfigServer.cpp"
|
||||
KEYS = ROOT / "src" / "helpers" / "WebConfigKeys.h"
|
||||
UI = ROOT / "webui" / "index.html"
|
||||
EMBEDDED_UI = ROOT / "src" / "helpers" / "esp32" / "WebConfigHtml.h"
|
||||
|
||||
|
||||
def function_body(source: str, start: str, end: str) -> str:
|
||||
return source[source.index(start) : source.index(end, source.index(start))]
|
||||
|
||||
|
||||
def embedded_ui() -> str:
|
||||
header = EMBEDDED_UI.read_text(encoding="utf-8")
|
||||
length = int(re.search(r"WEBCONFIG_HTML_GZ_LEN = (\d+);", header).group(1))
|
||||
array = header.split(
|
||||
"const uint8_t WEBCONFIG_HTML_GZ[] PROGMEM = {", 1
|
||||
)[1]
|
||||
blob = bytes(
|
||||
int(value, 16) for value in re.findall(r"0x([0-9a-f]{2})", array)
|
||||
)[:length]
|
||||
return gzip.decompress(blob).decode("utf-8")
|
||||
|
||||
|
||||
class WebConfigSecretMaskingTest(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.server = SERVER.read_text(encoding="utf-8")
|
||||
cls.keys = KEYS.read_text(encoding="utf-8")
|
||||
cls.ui = UI.read_text(encoding="utf-8")
|
||||
match = re.search(
|
||||
r'static const char SECRET_SENTINEL\[\]\s*=\s*"([^"]*)"\s*;',
|
||||
cls.server,
|
||||
)
|
||||
if match is None:
|
||||
raise AssertionError("WebConfigServer has no secret sentinel")
|
||||
cls.sentinel = match.group(1)
|
||||
|
||||
def test_server_and_captive_portal_use_exactly_eight_stars(self):
|
||||
self.assertEqual(self.sentinel, "********")
|
||||
self.assertEqual(len(self.sentinel), 8)
|
||||
|
||||
browser = re.search(r'var SENTINEL\s*=\s*"([^"]*)"\s*;', self.ui)
|
||||
self.assertIsNotNone(browser)
|
||||
self.assertEqual(browser.group(1), self.sentinel)
|
||||
|
||||
captive_portal = re.search(
|
||||
r'var SENTINEL\s*=\s*"([^"]*)"\s*;', embedded_ui()
|
||||
)
|
||||
self.assertIsNotNone(captive_portal)
|
||||
self.assertEqual(captive_portal.group(1), self.sentinel)
|
||||
|
||||
def test_config_get_masks_only_nonempty_wifi_password(self):
|
||||
get_handler = function_body(
|
||||
self.server,
|
||||
"void WebConfigServer::handleConfigGet(",
|
||||
"void WebConfigServer::handleConfigPost(",
|
||||
)
|
||||
self.assertRegex(
|
||||
get_handler,
|
||||
r'wifi\["pwd"\]\s*=\s*_wifi_password\[0\]\s*'
|
||||
r'\?\s*SECRET_SENTINEL\s*:\s*""\s*;',
|
||||
)
|
||||
self.assertEqual(get_handler.count('wifi["pwd"]'), 1)
|
||||
self.assertNotRegex(
|
||||
get_handler,
|
||||
r'wifi\["pwd"\]\s*=\s*(?:\(const char\s*\*\)\s*)?'
|
||||
r'_wifi_password\s*;',
|
||||
)
|
||||
|
||||
def test_config_post_skips_only_exact_sentinel_for_secret_keys(self):
|
||||
post_handler = function_body(
|
||||
self.server,
|
||||
"void WebConfigServer::handleConfigPost(",
|
||||
"void WebConfigServer::handleConfigResult(",
|
||||
)
|
||||
skip = re.search(
|
||||
r"if\s*\(\s*isSecretKey\(key\)\s*&&\s*"
|
||||
r"strcmp\(val,\s*SECRET_SENTINEL\)\s*==\s*0\s*\)\s*continue\s*;",
|
||||
post_handler,
|
||||
)
|
||||
self.assertIsNotNone(skip)
|
||||
|
||||
# The equality guard must run before a BatchEntry/CLI command is built;
|
||||
# otherwise an unchanged masked field could persist the placeholder.
|
||||
batch_entry = post_handler.index("BatchEntry& e")
|
||||
pre_persistence = post_handler[
|
||||
post_handler.index("int count = 0;") : batch_entry
|
||||
]
|
||||
self.assertEqual(pre_persistence.count("continue;"), 1)
|
||||
self.assertLess(skip.start(), batch_entry)
|
||||
self.assertLess(skip.start(), post_handler.index("count++;"))
|
||||
|
||||
# strcmp equality means close values remain real password updates, not
|
||||
# wildcard matches. Keep representative boundary cases explicit.
|
||||
def skipped(key: str, value: str) -> bool:
|
||||
return key == "wifi.pwd" and value == self.sentinel
|
||||
|
||||
self.assertTrue(skipped("wifi.pwd", "********"))
|
||||
for value in ("", "*******", "*********", "********x", "x********"):
|
||||
self.assertFalse(skipped("wifi.pwd", value), value)
|
||||
self.assertFalse(skipped("wifi.ssid", "********"))
|
||||
|
||||
def test_wifi_password_is_classified_as_a_secret_key(self):
|
||||
secret_classifier = function_body(
|
||||
self.keys,
|
||||
"static inline bool wcIsSecretKey(",
|
||||
"static inline bool wcIsSecretReadCommand(",
|
||||
)
|
||||
self.assertIn('strcmp(key, "wifi.pwd") == 0', secret_classifier)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -45,6 +45,163 @@ class WebConfigUiTest(unittest.TestCase):
|
||||
self.assertIn('if(!st.skipRadio&&$("#wz-rp").value==="")', save)
|
||||
self.assertIn("if(st.hasMqtt&&!st.skipMqtt&&", save)
|
||||
|
||||
def test_setup_wizard_opens_existing_wifi_scan_once(self):
|
||||
page = SOURCE.read_text(encoding="utf-8")
|
||||
self.assertIn("wizardScanOpened:false", page)
|
||||
self.assertIn('onclick="openScan(\'wz-ssid\')">Scan</button>', page)
|
||||
self.assertIn('onclick="startScan(true)">Rescan</button>', page)
|
||||
|
||||
open_scan = page[page.index("function openScan(target)"):
|
||||
page.index("function closeScan()")]
|
||||
self.assertIn('if(target==="wz-ssid")st.wizardScanOpened=true', open_scan)
|
||||
|
||||
auto_open = page[page.index("function openWizardScanOnce()"):
|
||||
page.index("function wzGo(n)")]
|
||||
self.assertIn("if(st.wizardScanOpened)return", auto_open)
|
||||
self.assertIn('openScan("wz-ssid")', auto_open)
|
||||
self.assertNotIn("startScan(true)", auto_open)
|
||||
|
||||
show_wizard = page[page.index("function showWizard()"):
|
||||
page.index("function wzGo(n)")]
|
||||
self.assertLess(show_wizard.index("loadConfig(capture)"),
|
||||
show_wizard.index("openWizardScanOnce()"))
|
||||
self.assertIn("wzRadioFill();openWizardScanOnce()", show_wizard)
|
||||
|
||||
# Scan completion may replace only the result list. Credentials change
|
||||
# only in the existing explicit network-click handler below it.
|
||||
poll_scan = page[page.index("function pollScan(url,epoch)"):
|
||||
page.index("function clearInheritedWiFiPassword")]
|
||||
self.assertNotIn(".value=", poll_scan)
|
||||
self.assertNotIn("dispatchEvent", poll_scan)
|
||||
|
||||
def test_scan_epoch_invalidates_old_timers_and_responses(self):
|
||||
page = SOURCE.read_text(encoding="utf-8")
|
||||
self.assertIn("scanEpoch:0", page)
|
||||
|
||||
close_scan = page[page.index("function closeScan()"):
|
||||
page.index("function startScan(force)")]
|
||||
self.assertIn("clearTimeout(st.scanTimer);st.scanTimer=0", close_scan)
|
||||
self.assertIn("st.scanEpoch++", close_scan)
|
||||
|
||||
start_scan = page[page.index("function startScan(force)"):
|
||||
page.index("function bars(rssi)")]
|
||||
self.assertIn("var epoch=++st.scanEpoch", start_scan)
|
||||
self.assertIn('pollScan(force?"/api/scan?rescan=1":"/api/scan",epoch)',
|
||||
start_scan)
|
||||
|
||||
poll_scan = page[page.index("function pollScan(url,epoch)"):
|
||||
page.index("function clearInheritedWiFiPassword")]
|
||||
# Check before issuing a request, after a response, and in the error
|
||||
# path so neither a stale timer nor a stale reply can alter the panel.
|
||||
self.assertEqual(poll_scan.count("if(epoch!==st.scanEpoch)return"), 3)
|
||||
self.assertIn('pollScan("/api/scan",epoch)', poll_scan)
|
||||
|
||||
def test_late_config_load_replays_visible_form_edits(self):
|
||||
page = SOURCE.read_text(encoding="utf-8")
|
||||
self.assertIn("configLoadCapture:null", page)
|
||||
self.assertIn("configEpoch:0", page)
|
||||
|
||||
listener = page[page.index('document.addEventListener("input"'):
|
||||
page.index("function updateSaveBar()")]
|
||||
self.assertIn("el.closest(capture.root)", listener)
|
||||
self.assertIn("capture.values[el.dataset.k]=elVal(el)", listener)
|
||||
self.assertIn("capture.confirms[el.dataset.cfm]=el.value", listener)
|
||||
self.assertIn("capture.wifiPasswordAutoCleared=!!ev.meshAutoWiFiPassword", listener)
|
||||
self.assertIn("capture.radio[el.dataset.rg]=elVal(el)", listener)
|
||||
self.assertIn("capture.rxps[el.dataset.rxps]=elVal(el)", listener)
|
||||
|
||||
load_config = page[page.index("function loadConfig(loadCapture)"):
|
||||
page.index("function markDirty")]
|
||||
self.assertIn("var epoch=++st.configEpoch", load_config)
|
||||
self.assertIn("loadCapture.loadEpoch=epoch", load_config)
|
||||
self.assertIn("if(epoch!==st.configEpoch)return false", load_config)
|
||||
self.assertIn("restoreConfigLoadEdits(loadCapture)", load_config)
|
||||
self.assertIn("return true", load_config)
|
||||
|
||||
restore = page[page.index("function restoreConfigLoadEdits"):
|
||||
page.index("function rxpsCfgValue")]
|
||||
self.assertIn('owns.call(capture.values,"wifi.ssid")', restore)
|
||||
self.assertIn('capture.values["wifi.ssid"]!==st.orig["wifi.ssid"]',
|
||||
restore)
|
||||
self.assertIn('!owns.call(capture.values,"wifi.pwd")', restore)
|
||||
self.assertIn('st.orig["wifi.pwd"]===SENTINEL', restore)
|
||||
self.assertIn('capture.values["wifi.pwd"]=""', restore)
|
||||
self.assertIn("capture.wifiPasswordAutoCleared=true", restore)
|
||||
self.assertIn("st.wifiPasswordAutoCleared=true", restore)
|
||||
self.assertIn("v!==st.orig[k]", restore)
|
||||
self.assertIn("if(dirty)st.dirty[k]=v;else delete st.dirty[k]", restore)
|
||||
self.assertIn("capture.confirms[k]", restore)
|
||||
self.assertIn("Object.keys(capture.radio)", restore)
|
||||
self.assertIn("var restoredRadio=radioCombo()", restore)
|
||||
self.assertIn("if(capture.rxps)", restore)
|
||||
|
||||
show_wizard = page[page.index("function showWizard()"):
|
||||
page.index("function wzGo(n)")]
|
||||
self.assertLess(show_wizard.index("st.configLoadCapture=capture"),
|
||||
show_wizard.index("loadConfig(capture)"))
|
||||
self.assertIn("if(st.configLoadCapture===capture)", show_wizard)
|
||||
self.assertIn("if(!applied)return", show_wizard)
|
||||
self.assertIn("st.configEpoch===capture.loadEpoch", show_wizard)
|
||||
|
||||
enter_app = page[page.index("function enterApp()"):
|
||||
page.index("function enterConsole()")]
|
||||
self.assertIn('root:"#v-app"', enter_app)
|
||||
self.assertIn("st.configLoadCapture=capture", enter_app)
|
||||
self.assertIn("loadConfig(capture)", enter_app)
|
||||
|
||||
def test_new_ssid_clears_only_inherited_masked_password(self):
|
||||
page = SOURCE.read_text(encoding="utf-8")
|
||||
clear_pwd = page[page.index("function clearInheritedWiFiPassword"):
|
||||
page.index('$("#scan-list").addEventListener')]
|
||||
self.assertIn('target==="wz-ssid"?"wz-wifi-pwd"', clear_pwd)
|
||||
self.assertIn('target==="app-ssid"?"app-wifi-pwd"', clear_pwd)
|
||||
self.assertIn('nextSsid===st.orig["wifi.ssid"]', clear_pwd)
|
||||
self.assertIn("pwd.value!==SENTINEL", clear_pwd)
|
||||
self.assertIn('pwd.value=""', clear_pwd)
|
||||
self.assertIn("st.wifiPasswordAutoCleared=true", clear_pwd)
|
||||
self.assertIn("st.wifiPasswordAutoCleared=false", clear_pwd)
|
||||
self.assertIn("pwd.value=SENTINEL", clear_pwd)
|
||||
self.assertIn("meshAutoWiFiPassword=true", clear_pwd)
|
||||
|
||||
choose = page[page.index('$("#scan-list").addEventListener'):
|
||||
page.index("/* ---------- wizard ---------- */")]
|
||||
self.assertLess(
|
||||
choose.index("clearInheritedWiFiPassword(st.scanTarget,selected)"),
|
||||
choose.index("el.value=selected"),
|
||||
)
|
||||
|
||||
listener = page[page.index('document.addEventListener("input"'):
|
||||
page.index("function updateSaveBar()")]
|
||||
self.assertIn(
|
||||
'if(el.dataset.k==="wifi.ssid")clearInheritedWiFiPassword(el.id,elVal(el))',
|
||||
listener,
|
||||
)
|
||||
|
||||
def test_password_review_distinguishes_explicit_clear(self):
|
||||
page = SOURCE.read_text(encoding="utf-8")
|
||||
review = page[page.index("function buildReview()"):
|
||||
page.index("function wizardSave()")]
|
||||
self.assertIn(
|
||||
'Object.prototype.hasOwnProperty.call(d,"wifi.pwd")', review
|
||||
)
|
||||
self.assertIn(
|
||||
'pwdChanged?(d["wifi.pwd"]?"******":"(open network)")', review
|
||||
)
|
||||
|
||||
def test_scan_panel_exposes_automatic_updates_accessibly(self):
|
||||
page = SOURCE.read_text(encoding="utf-8")
|
||||
self.assertIn('id="wz-scan-btn"', page)
|
||||
self.assertIn('id="app-scan-btn"', page)
|
||||
self.assertEqual(page.count('aria-controls="scan-panel"'), 2)
|
||||
self.assertIn('id="scan-panel" role="region"', page)
|
||||
self.assertIn('id="scan-list" aria-live="polite" aria-busy="true"', page)
|
||||
open_scan = page[page.index("function openScan(target)"):
|
||||
page.index("function closeScan()")]
|
||||
self.assertIn('setAttribute("aria-expanded"', open_scan)
|
||||
close_scan = page[page.index("function closeScan()"):
|
||||
page.index("function startScan(force)")]
|
||||
self.assertIn('setAttribute("aria-hidden","true")', close_scan)
|
||||
|
||||
def test_generated_asset_contains_show_controls(self):
|
||||
header = HEADER.read_text(encoding="utf-8")
|
||||
length = int(re.search(r"WEBCONFIG_HTML_GZ_LEN = (\d+);", header).group(1))
|
||||
@@ -59,6 +216,11 @@ class WebConfigUiTest(unittest.TestCase):
|
||||
self.assertIn("function toggleWiFiPassword(btn)", page)
|
||||
self.assertIn("function wzSkipRadio()", page)
|
||||
self.assertIn("function wzSkipMqtt()", page)
|
||||
self.assertIn("function openWizardScanOnce()", page)
|
||||
self.assertIn('openScan("wz-ssid")', page)
|
||||
self.assertIn("function restoreConfigLoadEdits(capture)", page)
|
||||
self.assertIn("function pollScan(url,epoch)", page)
|
||||
self.assertIn("function clearInheritedWiFiPassword(target,nextSsid)", page)
|
||||
|
||||
def test_skipped_mqtt_does_not_pin_display_in_setup_mode(self):
|
||||
main = (ROOT / "examples" / "companion_radio" / "main.cpp").read_text(
|
||||
|
||||
@@ -0,0 +1,573 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
"""Optional real-browser checks for WebConfig's asynchronous setup races."""
|
||||
|
||||
import gzip
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
HEADER = ROOT / "src" / "helpers" / "esp32" / "WebConfigHtml.h"
|
||||
|
||||
|
||||
def chromium_path():
|
||||
candidates = [
|
||||
shutil.which(name)
|
||||
for name in ("google-chrome", "chromium", "chromium-browser", "chrome", "msedge")
|
||||
]
|
||||
for base, rest in (
|
||||
(os.environ.get("ProgramFiles"), ("Google", "Chrome", "Application", "chrome.exe")),
|
||||
(os.environ.get("ProgramFiles(x86)"), ("Microsoft", "Edge", "Application", "msedge.exe")),
|
||||
(os.environ.get("LocalAppData"), ("Google", "Chrome", "Application", "chrome.exe")),
|
||||
):
|
||||
if base:
|
||||
candidates.append(str(Path(base).joinpath(*rest)))
|
||||
return next((path for path in candidates if path and Path(path).is_file()), None)
|
||||
|
||||
|
||||
BROWSER = chromium_path()
|
||||
|
||||
|
||||
def embedded_page():
|
||||
header = HEADER.read_text(encoding="utf-8")
|
||||
length = int(re.search(r"WEBCONFIG_HTML_GZ_LEN = (\d+);", header).group(1))
|
||||
array = header.split("const uint8_t WEBCONFIG_HTML_GZ[] PROGMEM = {", 1)[1]
|
||||
blob = bytes(
|
||||
int(value, 16) for value in re.findall(r"0x([0-9a-f]{2})", array)
|
||||
)[:length]
|
||||
return gzip.decompress(blob).decode("utf-8")
|
||||
|
||||
|
||||
class WebConfigUiRuntimeTest(unittest.TestCase):
|
||||
def setup_values(self):
|
||||
sys.path.insert(0, str(ROOT / "scripts"))
|
||||
try:
|
||||
import webconfig_mock_server as mock
|
||||
finally:
|
||||
sys.path.pop(0)
|
||||
|
||||
config = mock.default_config(True)
|
||||
config["wifi"]["ssid"] = "StoredNet"
|
||||
config["wifi"]["pwd"] = "********"
|
||||
status = {
|
||||
"mode": "setup",
|
||||
"auth": True,
|
||||
"mqtt": False,
|
||||
"cli": False,
|
||||
"wifi_psk64": True,
|
||||
"password_supported": True,
|
||||
"needs_password": False,
|
||||
"radio_optional": True,
|
||||
"needs_setup": False,
|
||||
"capabilities": 0,
|
||||
"max_cmds": 16,
|
||||
"name": "runtime-test",
|
||||
"role": "Companion",
|
||||
"board": "mock",
|
||||
"fw": "v1.17.1-test",
|
||||
"build_date": "runtime",
|
||||
}
|
||||
return status, config
|
||||
|
||||
def run_page(self, prelude, virtual_time=1500):
|
||||
if not BROWSER:
|
||||
self.skipTest("Chromium-family browser is unavailable")
|
||||
|
||||
source = embedded_page()
|
||||
markers = ('<script>\n"use strict";', '<script>"use strict";')
|
||||
marker = next((candidate for candidate in markers if candidate in source), None)
|
||||
self.assertIsNotNone(marker)
|
||||
page = source.replace(marker, prelude + marker, 1)
|
||||
|
||||
with tempfile.TemporaryDirectory(prefix="meshcore-webconfig-runtime-") as tmp:
|
||||
tmp_path = Path(tmp)
|
||||
html = tmp_path / "index.html"
|
||||
profile = tmp_path / "profile"
|
||||
profile.mkdir()
|
||||
html.write_text(page, encoding="utf-8")
|
||||
args = [
|
||||
BROWSER,
|
||||
"--headless=new",
|
||||
"--disable-gpu",
|
||||
"--no-first-run",
|
||||
"--no-default-browser-check",
|
||||
"--user-data-dir=" + str(profile),
|
||||
"--virtual-time-budget=" + str(virtual_time),
|
||||
"--dump-dom",
|
||||
html.as_uri(),
|
||||
]
|
||||
if os.name != "nt" and hasattr(os, "geteuid") and os.geteuid() == 0:
|
||||
args.insert(1, "--no-sandbox")
|
||||
result = subprocess.run(args, capture_output=True, timeout=30)
|
||||
|
||||
stderr = result.stderr.decode("utf-8", "replace")
|
||||
self.assertEqual(result.returncode, 0, stderr)
|
||||
dom = result.stdout.decode("utf-8", "replace")
|
||||
self.assertNotIn("data-test-error=", dom)
|
||||
return dom
|
||||
|
||||
def run_early_selection_case(self, explicit_password=None):
|
||||
status, config = self.setup_values()
|
||||
|
||||
prelude = """
|
||||
<script>
|
||||
(function(){
|
||||
var status=%s,config=%s;
|
||||
function response(value){
|
||||
return {ok:true,status:200,json:function(){return Promise.resolve(value)}};
|
||||
}
|
||||
window.fetch=function(path){
|
||||
if(path==="/api/status")return Promise.resolve(response(status));
|
||||
if(path==="/api/config")return new Promise(function(resolve){
|
||||
setTimeout(function(){resolve(response(config))},500);
|
||||
});
|
||||
if(path.indexOf("/api/scan")===0)return Promise.resolve(response({
|
||||
state:"done",networks:[{ssid:"EarlyNet",rssi:-40,enc:true,channel:6}]
|
||||
}));
|
||||
return Promise.resolve({ok:false,status:404,json:function(){return Promise.resolve({})}});
|
||||
};
|
||||
window.addEventListener("load",function(){
|
||||
setTimeout(function(){
|
||||
openScan("wz-ssid");
|
||||
setTimeout(function(){
|
||||
var network=document.querySelector("#scan-list .net");
|
||||
if(!network){document.body.setAttribute("data-test-error","missing-network");return}
|
||||
network.click();
|
||||
var explicit=%s;
|
||||
if(explicit!==null){
|
||||
var pwd=document.getElementById("wz-wifi-pwd");
|
||||
pwd.value=explicit;
|
||||
pwd.dispatchEvent(new Event("input",{bubbles:true}));
|
||||
}
|
||||
},50);
|
||||
},50);
|
||||
setTimeout(function(){
|
||||
var body=document.body,owns=Object.prototype.hasOwnProperty;
|
||||
body.setAttribute("data-test-ssid",document.getElementById("wz-ssid").value);
|
||||
body.setAttribute("data-test-wz-pwd",document.getElementById("wz-wifi-pwd").value);
|
||||
body.setAttribute("data-test-app-pwd",document.getElementById("app-wifi-pwd").value);
|
||||
body.setAttribute("data-test-orig-pwd",st.orig["wifi.pwd"]);
|
||||
body.setAttribute("data-test-dirty-ssid",st.dirty["wifi.ssid"]||"");
|
||||
body.setAttribute("data-test-dirty-pwd-present",
|
||||
owns.call(st.dirty,"wifi.pwd")?"yes":"no");
|
||||
body.setAttribute("data-test-dirty-pwd",st.dirty["wifi.pwd"]||"");
|
||||
},900);
|
||||
});
|
||||
})();
|
||||
</script>
|
||||
""" % (json.dumps(status), json.dumps(config), json.dumps(explicit_password))
|
||||
|
||||
return self.run_page(prelude)
|
||||
|
||||
def test_early_new_ssid_clears_password_loaded_later(self):
|
||||
dom = self.run_early_selection_case()
|
||||
self.assertIn('data-test-ssid="EarlyNet"', dom)
|
||||
self.assertIn('data-test-orig-pwd="********"', dom)
|
||||
self.assertIn('data-test-wz-pwd=""', dom)
|
||||
self.assertIn('data-test-app-pwd=""', dom)
|
||||
self.assertIn('data-test-dirty-ssid="EarlyNet"', dom)
|
||||
self.assertIn('data-test-dirty-pwd-present="yes"', dom)
|
||||
self.assertIn('data-test-dirty-pwd=""', dom)
|
||||
|
||||
def test_early_explicit_password_is_preserved(self):
|
||||
dom = self.run_early_selection_case("new-secret")
|
||||
self.assertIn('data-test-ssid="EarlyNet"', dom)
|
||||
self.assertIn('data-test-wz-pwd="new-secret"', dom)
|
||||
self.assertIn('data-test-app-pwd="new-secret"', dom)
|
||||
self.assertIn('data-test-dirty-pwd-present="yes"', dom)
|
||||
self.assertIn('data-test-dirty-pwd="new-secret"', dom)
|
||||
|
||||
def test_setup_automatically_opens_embedded_scan_picker(self):
|
||||
status, config = self.setup_values()
|
||||
prelude = """
|
||||
<script>
|
||||
(function(){
|
||||
var status=%s,config=%s;
|
||||
function response(value){
|
||||
return {ok:true,status:200,json:function(){return Promise.resolve(value)}};
|
||||
}
|
||||
window.fetch=function(path){
|
||||
if(path==="/api/status")return Promise.resolve(response(status));
|
||||
if(path==="/api/config")return new Promise(function(resolve){
|
||||
setTimeout(function(){resolve(response(config))},100);
|
||||
});
|
||||
if(path.indexOf("/api/scan")===0)return Promise.resolve(response({
|
||||
state:"done",networks:[{ssid:"NearbyNet",rssi:-35,enc:true,channel:11}]
|
||||
}));
|
||||
return Promise.resolve({ok:false,status:404,json:function(){return Promise.resolve({})}});
|
||||
};
|
||||
window.addEventListener("load",function(){
|
||||
setTimeout(function(){
|
||||
var body=document.body,panel=document.getElementById("scan-panel");
|
||||
body.setAttribute("data-test-scan-open",panel.classList.contains("hide")?"no":"yes");
|
||||
body.setAttribute("data-test-scan-hidden",panel.getAttribute("aria-hidden"));
|
||||
body.setAttribute("data-test-scan-expanded",document.getElementById("wz-scan-btn").getAttribute("aria-expanded"));
|
||||
body.setAttribute("data-test-scan-busy",document.getElementById("scan-list").getAttribute("aria-busy"));
|
||||
body.setAttribute("data-test-network-count",document.querySelectorAll("#scan-list .net").length);
|
||||
body.setAttribute("data-test-ssid",document.getElementById("wz-ssid").value);
|
||||
body.setAttribute("data-test-pwd",document.getElementById("wz-wifi-pwd").value);
|
||||
body.setAttribute("data-test-dirty-count",Object.keys(st.dirty).length);
|
||||
},700);
|
||||
});
|
||||
})();
|
||||
</script>
|
||||
""" % (json.dumps(status), json.dumps(config))
|
||||
|
||||
dom = self.run_page(prelude)
|
||||
self.assertIn('data-test-scan-open="yes"', dom)
|
||||
self.assertIn('data-test-scan-hidden="false"', dom)
|
||||
self.assertIn('data-test-scan-expanded="true"', dom)
|
||||
self.assertIn('data-test-scan-busy="false"', dom)
|
||||
self.assertIn('data-test-network-count="1"', dom)
|
||||
self.assertIn('data-test-ssid="StoredNet"', dom)
|
||||
self.assertIn('data-test-pwd="********"', dom)
|
||||
self.assertIn('data-test-dirty-count="0"', dom)
|
||||
|
||||
def test_typed_new_ssid_clears_inherited_password_and_review(self):
|
||||
status, config = self.setup_values()
|
||||
prelude = """
|
||||
<script>
|
||||
(function(){
|
||||
var status=%s,config=%s;
|
||||
function response(value){
|
||||
return {ok:true,status:200,json:function(){return Promise.resolve(value)}};
|
||||
}
|
||||
window.fetch=function(path){
|
||||
if(path==="/api/status")return Promise.resolve(response(status));
|
||||
if(path==="/api/config")return Promise.resolve(response(config));
|
||||
if(path.indexOf("/api/scan")===0)return Promise.resolve(response({state:"done",networks:[]}));
|
||||
return Promise.resolve({ok:false,status:404,json:function(){return Promise.resolve({})}});
|
||||
};
|
||||
window.addEventListener("load",function(){
|
||||
setTimeout(function(){
|
||||
var ssid=document.getElementById("wz-ssid");
|
||||
ssid.value="TypedNet";
|
||||
ssid.dispatchEvent(new Event("input",{bubbles:true}));
|
||||
buildReview();
|
||||
var body=document.body,owns=Object.prototype.hasOwnProperty;
|
||||
body.setAttribute("data-test-ssid",ssid.value);
|
||||
body.setAttribute("data-test-pwd",document.getElementById("wz-wifi-pwd").value);
|
||||
body.setAttribute("data-test-dirty-pwd",owns.call(st.dirty,"wifi.pwd")?"yes":"no");
|
||||
body.setAttribute("data-test-review-open",document.getElementById("wz-review").textContent.indexOf("(open network)")>=0?"yes":"no");
|
||||
},400);
|
||||
});
|
||||
})();
|
||||
</script>
|
||||
""" % (json.dumps(status), json.dumps(config))
|
||||
|
||||
dom = self.run_page(prelude)
|
||||
self.assertIn('data-test-ssid="TypedNet"', dom)
|
||||
self.assertIn('data-test-pwd=""', dom)
|
||||
self.assertIn('data-test-dirty-pwd="yes"', dom)
|
||||
self.assertIn('data-test-review-open="yes"', dom)
|
||||
|
||||
def test_returning_to_original_ssid_restores_masked_password(self):
|
||||
status, config = self.setup_values()
|
||||
prelude = """
|
||||
<script>
|
||||
(function(){
|
||||
var status=%s,config=%s;
|
||||
function response(value){
|
||||
return {ok:true,status:200,json:function(){return Promise.resolve(value)}};
|
||||
}
|
||||
window.fetch=function(path){
|
||||
if(path==="/api/status")return Promise.resolve(response(status));
|
||||
if(path==="/api/config")return Promise.resolve(response(config));
|
||||
if(path.indexOf("/api/scan")===0)return Promise.resolve(response({state:"done",networks:[]}));
|
||||
return Promise.resolve({ok:false,status:404,json:function(){return Promise.resolve({})}});
|
||||
};
|
||||
window.addEventListener("load",function(){
|
||||
setTimeout(function(){
|
||||
var ssid=document.getElementById("wz-ssid");
|
||||
ssid.value="OtherNet";
|
||||
ssid.dispatchEvent(new Event("input",{bubbles:true}));
|
||||
ssid.value="StoredNet";
|
||||
ssid.dispatchEvent(new Event("input",{bubbles:true}));
|
||||
var body=document.body,owns=Object.prototype.hasOwnProperty;
|
||||
body.setAttribute("data-test-pwd",document.getElementById("wz-wifi-pwd").value);
|
||||
body.setAttribute("data-test-dirty-ssid",owns.call(st.dirty,"wifi.ssid")?"yes":"no");
|
||||
body.setAttribute("data-test-dirty-pwd",owns.call(st.dirty,"wifi.pwd")?"yes":"no");
|
||||
body.setAttribute("data-test-auto-cleared",st.wifiPasswordAutoCleared?"yes":"no");
|
||||
},400);
|
||||
});
|
||||
})();
|
||||
</script>
|
||||
""" % (json.dumps(status), json.dumps(config))
|
||||
|
||||
dom = self.run_page(prelude)
|
||||
self.assertIn('data-test-pwd="********"', dom)
|
||||
self.assertIn('data-test-dirty-ssid="no"', dom)
|
||||
self.assertIn('data-test-dirty-pwd="no"', dom)
|
||||
self.assertIn('data-test-auto-cleared="no"', dom)
|
||||
|
||||
def test_advanced_editor_edit_during_config_load_is_preserved(self):
|
||||
status, config = self.setup_values()
|
||||
status["mode"] = "lan"
|
||||
prelude = """
|
||||
<script>
|
||||
(function(){
|
||||
var status=%s,config=%s;
|
||||
function response(value){
|
||||
return {ok:true,status:200,json:function(){return Promise.resolve(value)}};
|
||||
}
|
||||
window.fetch=function(path){
|
||||
if(path==="/api/status")return Promise.resolve(response(status));
|
||||
if(path==="/api/config")return new Promise(function(resolve){
|
||||
setTimeout(function(){resolve(response(config))},500);
|
||||
});
|
||||
return Promise.resolve({ok:false,status:404,json:function(){return Promise.resolve({})}});
|
||||
};
|
||||
window.addEventListener("load",function(){
|
||||
setTimeout(function(){
|
||||
var name=document.querySelector('#v-app [data-k="name"]');
|
||||
name.value="Early App Edit";
|
||||
name.dispatchEvent(new Event("input",{bubbles:true}));
|
||||
},75);
|
||||
setTimeout(function(){
|
||||
var body=document.body,name=document.querySelector('#v-app [data-k="name"]');
|
||||
body.setAttribute("data-test-name",name.value);
|
||||
body.setAttribute("data-test-dirty-name",st.dirty.name||"");
|
||||
body.setAttribute("data-test-capture",st.configLoadCapture===null?"clear":"set");
|
||||
},850);
|
||||
});
|
||||
})();
|
||||
</script>
|
||||
""" % (json.dumps(status), json.dumps(config))
|
||||
|
||||
dom = self.run_page(prelude)
|
||||
self.assertIn('data-test-name="Early App Edit"', dom)
|
||||
self.assertIn('data-test-dirty-name="Early App Edit"', dom)
|
||||
self.assertIn('data-test-capture="clear"', dom)
|
||||
|
||||
def test_second_load_auto_password_clear_remains_restorable(self):
|
||||
status, config = self.setup_values()
|
||||
prelude = """
|
||||
<script>
|
||||
(function(){
|
||||
var status=%s,config=%s,configCalls=0;
|
||||
function response(value){
|
||||
return {ok:true,status:200,json:function(){return Promise.resolve(value)}};
|
||||
}
|
||||
window.fetch=function(path){
|
||||
if(path==="/api/status")return Promise.resolve(response(status));
|
||||
if(path==="/api/config"){
|
||||
configCalls++;
|
||||
if(configCalls===1)return Promise.resolve(response(config));
|
||||
return new Promise(function(resolve){
|
||||
setTimeout(function(){resolve(response(config))},400);
|
||||
});
|
||||
}
|
||||
if(path.indexOf("/api/scan")===0)return Promise.resolve(response({state:"done",networks:[]}));
|
||||
return Promise.resolve({ok:false,status:404,json:function(){return Promise.resolve({})}});
|
||||
};
|
||||
window.addEventListener("load",function(){
|
||||
// The setup wizard's first load has populated both password fields with
|
||||
// the stored-secret sentinel. Entering the advanced editor starts a second
|
||||
// captured load; change its SSID while that request is still outstanding.
|
||||
setTimeout(function(){enterApp()},75);
|
||||
setTimeout(function(){
|
||||
var ssid=document.getElementById("app-ssid");
|
||||
ssid.value="OtherNet";
|
||||
ssid.dispatchEvent(new Event("input",{bubbles:true}));
|
||||
},125);
|
||||
setTimeout(function(){
|
||||
var body=document.body,owns=Object.prototype.hasOwnProperty;
|
||||
var ssid=document.getElementById("app-ssid");
|
||||
var pwd=document.getElementById("app-wifi-pwd");
|
||||
|
||||
ssid.value="StoredNet";
|
||||
ssid.dispatchEvent(new Event("input",{bubbles:true}));
|
||||
body.setAttribute("data-test-restored-pwd",pwd.value);
|
||||
body.setAttribute("data-test-restored-dirty-ssid",
|
||||
owns.call(st.dirty,"wifi.ssid")?"yes":"no");
|
||||
body.setAttribute("data-test-restored-dirty-pwd",
|
||||
owns.call(st.dirty,"wifi.pwd")?"yes":"no");
|
||||
body.setAttribute("data-test-restored-auto",
|
||||
st.wifiPasswordAutoCleared?"yes":"no");
|
||||
|
||||
// A real password edit after an automatic clear is authoritative. It
|
||||
// must not be replaced with the sentinel when the SSID returns to the
|
||||
// stored network.
|
||||
ssid.value="OtherNet";
|
||||
ssid.dispatchEvent(new Event("input",{bubbles:true}));
|
||||
pwd.value="manual-secret";
|
||||
pwd.dispatchEvent(new Event("input",{bubbles:true}));
|
||||
ssid.value="StoredNet";
|
||||
ssid.dispatchEvent(new Event("input",{bubbles:true}));
|
||||
body.setAttribute("data-test-manual-pwd",pwd.value);
|
||||
body.setAttribute("data-test-manual-dirty-ssid",
|
||||
owns.call(st.dirty,"wifi.ssid")?"yes":"no");
|
||||
body.setAttribute("data-test-manual-dirty-pwd",
|
||||
owns.call(st.dirty,"wifi.pwd")?"yes":"no");
|
||||
body.setAttribute("data-test-manual-auto",
|
||||
st.wifiPasswordAutoCleared?"yes":"no");
|
||||
body.setAttribute("data-test-config-calls",String(configCalls));
|
||||
body.setAttribute("data-test-capture",
|
||||
st.configLoadCapture===null?"clear":"set");
|
||||
},700);
|
||||
});
|
||||
})();
|
||||
</script>
|
||||
""" % (json.dumps(status), json.dumps(config))
|
||||
|
||||
dom = self.run_page(prelude)
|
||||
self.assertIn('data-test-restored-pwd="********"', dom)
|
||||
self.assertIn('data-test-restored-dirty-ssid="no"', dom)
|
||||
self.assertIn('data-test-restored-dirty-pwd="no"', dom)
|
||||
self.assertIn('data-test-restored-auto="no"', dom)
|
||||
self.assertIn('data-test-manual-pwd="manual-secret"', dom)
|
||||
self.assertIn('data-test-manual-dirty-ssid="no"', dom)
|
||||
self.assertIn('data-test-manual-dirty-pwd="yes"', dom)
|
||||
self.assertIn('data-test-manual-auto="no"', dom)
|
||||
self.assertIn('data-test-config-calls="2"', dom)
|
||||
self.assertIn('data-test-capture="clear"', dom)
|
||||
|
||||
def test_delayed_app_load_preserves_only_edited_radio_field(self):
|
||||
status, config = self.setup_values()
|
||||
status["mode"] = "lan"
|
||||
prelude = """
|
||||
<script>
|
||||
(function(){
|
||||
var status=%s,config=%s;
|
||||
function response(value){
|
||||
return {ok:true,status:200,json:function(){return Promise.resolve(value)}};
|
||||
}
|
||||
window.fetch=function(path){
|
||||
if(path==="/api/status")return Promise.resolve(response(status));
|
||||
if(path==="/api/config")return new Promise(function(resolve){
|
||||
setTimeout(function(){resolve(response(config))},500);
|
||||
});
|
||||
return Promise.resolve({ok:false,status:404,json:function(){return Promise.resolve({})}});
|
||||
};
|
||||
window.addEventListener("load",function(){
|
||||
setTimeout(function(){
|
||||
var sf=document.querySelector('#v-app [data-rg="sf"]');
|
||||
sf.value="9";
|
||||
sf.dispatchEvent(new Event("input",{bubbles:true}));
|
||||
},75);
|
||||
setTimeout(function(){
|
||||
var body=document.body;
|
||||
function radioValue(key){
|
||||
return document.querySelector('#v-app [data-rg="'+key+'"]').value;
|
||||
}
|
||||
body.setAttribute("data-test-radio-freq",radioValue("freq"));
|
||||
body.setAttribute("data-test-radio-bw",radioValue("bw"));
|
||||
body.setAttribute("data-test-radio-sf",radioValue("sf"));
|
||||
body.setAttribute("data-test-radio-cr",radioValue("cr"));
|
||||
body.setAttribute("data-test-radio-orig",st.orig.radio);
|
||||
body.setAttribute("data-test-radio-dirty",st.dirty.radio||"");
|
||||
body.setAttribute("data-test-capture",
|
||||
st.configLoadCapture===null?"clear":"set");
|
||||
},850);
|
||||
});
|
||||
})();
|
||||
</script>
|
||||
""" % (json.dumps(status), json.dumps(config))
|
||||
|
||||
dom = self.run_page(prelude)
|
||||
self.assertIn('data-test-radio-freq="910.525"', dom)
|
||||
self.assertIn('data-test-radio-bw="62.5"', dom)
|
||||
self.assertIn('data-test-radio-sf="9"', dom)
|
||||
self.assertIn('data-test-radio-cr="5"', dom)
|
||||
self.assertIn('data-test-radio-orig="910.525,62.5,7,5"', dom)
|
||||
self.assertIn('data-test-radio-dirty="910.525,62.5,9,5"', dom)
|
||||
self.assertIn('data-test-capture="clear"', dom)
|
||||
|
||||
def test_closed_picker_ignores_late_scan_response(self):
|
||||
status, config = self.setup_values()
|
||||
prelude = """
|
||||
<script>
|
||||
(function(){
|
||||
var status=%s,config=%s;
|
||||
function response(value){
|
||||
return {ok:true,status:200,json:function(){return Promise.resolve(value)}};
|
||||
}
|
||||
window.fetch=function(path){
|
||||
if(path==="/api/status")return Promise.resolve(response(status));
|
||||
if(path==="/api/config")return Promise.resolve(response(config));
|
||||
if(path.indexOf("/api/scan")===0)return new Promise(function(resolve){
|
||||
setTimeout(function(){resolve(response({
|
||||
state:"done",networks:[{ssid:"LateNet",rssi:-40,enc:true,channel:6}]
|
||||
}))},450);
|
||||
});
|
||||
return Promise.resolve({ok:false,status:404,json:function(){return Promise.resolve({})}});
|
||||
};
|
||||
window.addEventListener("load",function(){
|
||||
setTimeout(function(){
|
||||
closeScan();
|
||||
document.getElementById("scan-list").textContent="closed";
|
||||
},100);
|
||||
setTimeout(function(){
|
||||
var body=document.body,panel=document.getElementById("scan-panel");
|
||||
body.setAttribute("data-test-scan-hidden",panel.getAttribute("aria-hidden"));
|
||||
body.setAttribute("data-test-scan-expanded",document.getElementById("wz-scan-btn").getAttribute("aria-expanded"));
|
||||
body.setAttribute("data-test-network-count",document.querySelectorAll("#scan-list .net").length);
|
||||
body.setAttribute("data-test-scan-text",document.getElementById("scan-list").textContent);
|
||||
},800);
|
||||
});
|
||||
})();
|
||||
</script>
|
||||
""" % (json.dumps(status), json.dumps(config))
|
||||
|
||||
dom = self.run_page(prelude)
|
||||
self.assertIn('data-test-scan-hidden="true"', dom)
|
||||
self.assertIn('data-test-scan-expanded="false"', dom)
|
||||
self.assertIn('data-test-network-count="0"', dom)
|
||||
self.assertIn('data-test-scan-text="closed"', dom)
|
||||
|
||||
def test_newer_config_load_wins_when_older_reply_finishes_last(self):
|
||||
status, old_config = self.setup_values()
|
||||
new_config = json.loads(json.dumps(old_config))
|
||||
new_config["wifi"]["ssid"] = "AppNet"
|
||||
prelude = """
|
||||
<script>
|
||||
(function(){
|
||||
var status=%s,oldConfig=%s,newConfig=%s,configCalls=0;
|
||||
function response(value){
|
||||
return {ok:true,status:200,json:function(){return Promise.resolve(value)}};
|
||||
}
|
||||
window.fetch=function(path){
|
||||
if(path==="/api/status")return Promise.resolve(response(status));
|
||||
if(path==="/api/config"){
|
||||
configCalls++;
|
||||
if(configCalls===1)return new Promise(function(resolve){
|
||||
setTimeout(function(){resolve(response(oldConfig))},500);
|
||||
});
|
||||
return Promise.resolve(response(newConfig));
|
||||
}
|
||||
if(path.indexOf("/api/scan")===0)return Promise.resolve(response({state:"done",networks:[]}));
|
||||
return Promise.resolve({ok:false,status:404,json:function(){return Promise.resolve({})}});
|
||||
};
|
||||
window.addEventListener("load",function(){
|
||||
setTimeout(function(){enterApp()},75);
|
||||
setTimeout(function(){
|
||||
var body=document.body;
|
||||
body.setAttribute("data-test-config-calls",configCalls);
|
||||
body.setAttribute("data-test-orig-ssid",st.orig["wifi.ssid"]);
|
||||
body.setAttribute("data-test-field-ssid",document.getElementById("app-ssid").value);
|
||||
body.setAttribute("data-test-scan-hidden",document.getElementById("scan-panel").getAttribute("aria-hidden"));
|
||||
body.setAttribute("data-test-capture",st.configLoadCapture===null?"clear":"set");
|
||||
},800);
|
||||
});
|
||||
})();
|
||||
</script>
|
||||
""" % (json.dumps(status), json.dumps(old_config), json.dumps(new_config))
|
||||
|
||||
dom = self.run_page(prelude)
|
||||
self.assertIn('data-test-config-calls="2"', dom)
|
||||
self.assertIn('data-test-orig-ssid="AppNet"', dom)
|
||||
self.assertIn('data-test-field-ssid="AppNet"', dom)
|
||||
self.assertIn('data-test-scan-hidden="true"', dom)
|
||||
self.assertIn('data-test-capture="clear"', dom)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user