mikecarper
6b3f7df4d0
Share one bounded WiFi outage and recovery schedule across MQTT, automatic NTP, Companion station ownership and WebConfig. Preserve enabled settings and foreground access while allowing normal sleep after confirmed physical shutdown. Honor failed SDK client stops before handing off WiFi, protect clean teardown and reconfiguration, and retry transient ESP-NOW channel restoration without repeated disconnects. Add production-method, SDK failure and real sleep regressions to CI. Validation: native MQTT/NTP/clock/reconnect suites; production power and wrapper failure tests; browser, room HTTP, OTA and combined bridge regressions; Full Heltec V4 repeater, room and companion builds with capability, flash and linked-memory checks.
74 lines
3.8 KiB
Python
74 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Execute actual recovery/login/logout gates with real JSON and fake sockets."""
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
from test_replay_reset_integration import extract_braced
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
class WebConfigRecoveryAuthTest(unittest.TestCase):
|
|
def compile_run(self, source, expect_rejection=False):
|
|
compiler = shutil.which("g++") or shutil.which("clang++")
|
|
self.assertIsNotNone(compiler)
|
|
json_header = next((ROOT / ".pio/libdeps").glob("*/ArduinoJson/src/ArduinoJson.h"), None)
|
|
self.assertIsNotNone(json_header, "native dependency headers are required")
|
|
with tempfile.TemporaryDirectory(prefix="meshcore-webconfig-auth-") as temporary:
|
|
work = Path(temporary)
|
|
generated = work / "auth.cpp"
|
|
generated.write_text(source)
|
|
binary = work / "auth"
|
|
sanitizers = (["-fsanitize=address,undefined", "-fno-sanitize-recover=all",
|
|
"-fno-pie", "-no-pie"] if sys.platform.startswith("linux") else [])
|
|
compiled = subprocess.run([compiler, "-std=c++17", "-Wall", "-Wextra",
|
|
*sanitizers, "-I" + str(json_header.parent),
|
|
str(generated), "-o", str(binary)],
|
|
capture_output=True, text=True, timeout=60)
|
|
self.assertEqual(compiled.returncode, 0, compiled.stderr)
|
|
result = subprocess.run([str(binary)], capture_output=True, text=True, timeout=10)
|
|
if expect_rejection:
|
|
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertIn("Assertion", result.stderr)
|
|
else:
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
|
|
def fixture(self):
|
|
source = (ROOT / "src/helpers/esp32/WebConfigServer.cpp").read_text()
|
|
methods = "\n".join(extract_braced(source, signature) for signature in (
|
|
"static bool fixedTimeEquals(", "bool WebConfigServer::checkAuth(",
|
|
"void WebConfigServer::handleLogin(", "void WebConfigServer::handleLogout("))
|
|
# The actual config POST entry gate must reject a replacement password
|
|
# before parsing or queueing it; exercise its exact production prefix.
|
|
post = extract_braced(source, "void WebConfigServer::handleConfigPost(")
|
|
gate = post[post.index("{") + 1:post.index(" if (req->contentLength()")]
|
|
methods += "\nvoid WebConfigServer::checkConfigPost(AsyncWebServerRequest* req) {" + gate + "\nconfig_accepted=true;\n}"
|
|
return (ROOT / "test/fixtures/webconfig_recovery_auth.cpp").read_text().replace("@METHODS@", methods)
|
|
|
|
def test_saved_network_fallback_requires_current_admin_session(self):
|
|
self.compile_run(self.fixture())
|
|
|
|
def test_old_interface_only_fallback_fails_executable_regression(self):
|
|
fixture = self.fixture()
|
|
old = fixture.replace("if (_initial_setup || _wifi_ssid[0] == 0) { noteNetworkActivity(); return true; }", "return true;", 1)
|
|
self.assertNotEqual(old, fixture)
|
|
self.compile_run(old, expect_rejection=True)
|
|
|
|
def test_old_unauthenticated_logout_fails_executable_regression(self):
|
|
fixture = self.fixture()
|
|
signature = "void WebConfigServer::handleLogout("
|
|
start = fixture.index(signature, fixture.index("@METHODS@") if "@METHODS@" in fixture else 0)
|
|
prefix, logout = fixture[:start], fixture[start:]
|
|
gate = ' if (!checkAuth(req)) { req->send(401, "application/json", "{\\"error\\":\\"auth\\"}"); return; }\n'
|
|
old = prefix + logout.replace(gate, "", 1)
|
|
self.assertNotEqual(old, fixture)
|
|
self.compile_run(old, expect_rejection=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|