diff --git a/micro/app_main.py b/micro/app_main.py index 0db1efc..6d6a822 100644 --- a/micro/app_main.py +++ b/micro/app_main.py @@ -76,7 +76,14 @@ def IRQ_button_press(pin): status_led=None, timeout_seconds=config["wifi"]["timeout_seconds"], ) -network.connect_with_fallback(watchdog_feed=watchdog.feed) +printer_api_ready = config_loader.is_runtime_config_ready(config) +if printer_api_ready: + network.connect_with_fallback(watchdog_feed=watchdog.feed) +else: + # Keep the setup UI reachable until a complete, explicit printer + # configuration has been saved. In particular, do not poll or clear the + # baked-in/default printer ID while settings are incomplete. + network.start_access_point() if network.is_ap_mode(): print("Wi-Fi unavailable; connect to the setup access point to update settings") flasher.on() @@ -117,6 +124,10 @@ def handle_pending_button_press(): global PENDING_BUTTON_PRESS global PRINTER_STATUS_UPDATE_REQUIRED + if not printer_api_ready: + PENDING_BUTTON_PRESS = False + return + try: with_network_connection( lambda: api.clear_plate(config["printer"]["id"]) @@ -133,6 +144,10 @@ def handle_printer_status_update(): global PRINTER_AWAITING_PLATE_CLEAR, PRINTER_STATUS_UPDATE_REQUIRED global CHAMBER_LIGHT_IS_ON + if not printer_api_ready: + PRINTER_STATUS_UPDATE_REQUIRED = False + return + try: response = with_network_connection( lambda: api.get_printer_status(config["printer"]["id"]) diff --git a/micro/config_loader.py b/micro/config_loader.py index 6b45c0c..b57ec98 100644 --- a/micro/config_loader.py +++ b/micro/config_loader.py @@ -3,6 +3,7 @@ except ImportError: import json +import os import random @@ -36,7 +37,9 @@ def generate_hostname(): "request_timeout_seconds": 3, }, "printer": { - "id": 3, + # A printer must be selected in setup. Never guess a device ID: a + # default ID can send a button press to an unrelated printer. + "id": None, "poll_interval_seconds": 5, }, "led": { @@ -55,20 +58,175 @@ def generate_hostname(): } +_STRING_FIELDS = { + "wifi": ("ssid", "password", "hostname", "ap_password"), + "api": ("base_url", "key"), + "button": ("pull", "trigger"), + "web": ("password",), +} +_NUMBER_FIELDS = { + "wifi": ("timeout_seconds",), + "api": ("request_timeout_seconds",), + "printer": ("poll_interval_seconds",), + "led": ("pin", "flash_interval_ms"), + "button": ("pin", "debounce_ms"), +} +_KNOWN_SECTIONS = tuple(DEFAULT_CONFIG.keys()) + + def load_config(path="config.json"): config = _copy_dict(DEFAULT_CONFIG) - try: - with open(path) as config_file: - loaded_config = json.load(config_file) - except OSError: - print("Config file not found, using defaults:", path) + path = str(path) + loaded_config = _load_config_file(path) + source_path = path + if loaded_config is None: + backup_path = path + ".bak" + loaded_config = _load_config_file(backup_path) + if loaded_config is not None: + source_path = backup_path + + if loaded_config is None: + print("Config file missing or invalid, using defaults:", path) return config _deep_update(config, loaded_config) + if source_path != path: + print("Recovered configuration from backup:", source_path) return config +def is_runtime_config_ready(config): + """Return whether the settings can safely address a printer API.""" + if not isinstance(config, dict): + return False + + wifi = config.get("wifi") + api = config.get("api") + printer = config.get("printer") + if not isinstance(wifi, dict) or not isinstance(api, dict): + return False + if not isinstance(printer, dict): + return False + + printer_id = printer.get("id") + return bool( + wifi.get("ssid") + and api.get("base_url") + and api.get("key") + and isinstance(printer_id, int) + and not isinstance(printer_id, bool) + and printer_id >= 0 + ) + + +def save_config(path, config): + """Replace a config atomically, retaining the previous copy until verified.""" + path = str(path) + if not isinstance(config, dict): + raise ValueError("Configuration must be an object") + + temporary_path = path + ".tmp" + backup_path = path + ".bak" + serialized = json.dumps(config) + with open(temporary_path, "w") as config_file: + config_file.write(serialized) + config_file.write("\n") + sync = getattr(os, "sync", None) + if sync is not None: + sync() + + staged_config = _load_config_file(temporary_path) + if staged_config != config: + _remove_file(temporary_path) + raise ValueError("Could not verify saved configuration") + + if _load_config_file(path) is not None: + _remove_file(backup_path) + os.rename(path, backup_path) + else: + # Keep any valid prior backup until the new config is installed. + _remove_file(path) + + try: + os.rename(temporary_path, path) + except OSError: + # Restore a known-good config when the replacement could not be made. + if ( + _load_config_file(path) is None + and _load_config_file(backup_path) is not None + ): + try: + os.rename(backup_path, path) + except OSError: + pass + _remove_file(temporary_path) + raise + + if _load_config_file(path) != config: + _remove_file(path) + if _load_config_file(backup_path) is not None: + try: + os.rename(backup_path, path) + except OSError: + pass + raise ValueError("Could not verify active configuration") + _remove_file(backup_path) + + +def _load_config_file(path): + try: + with open(path) as config_file: + loaded_config = json.load(config_file) + except (OSError, ValueError, TypeError): + return None + if not _is_valid_config_shape(loaded_config): + return None + return loaded_config + + +def _is_valid_config_shape(config): + if not isinstance(config, dict): + return False + + for section in _KNOWN_SECTIONS: + if section in config and not isinstance(config[section], dict): + return False + + for section, fields in _STRING_FIELDS.items(): + values = config.get(section, {}) + for field in fields: + if field in values and not isinstance(values[field], str): + return False + + for section, fields in _NUMBER_FIELDS.items(): + values = config.get(section, {}) + for field in fields: + value = values.get(field) + if field in values and ( + not isinstance(value, (int, float)) or isinstance(value, bool) + ): + return False + + if "id" in config.get("printer", {}): + printer_id = config["printer"]["id"] + if printer_id is not None and ( + not isinstance(printer_id, int) + or isinstance(printer_id, bool) + or printer_id < 0 + ): + return False + + return True + + +def _remove_file(path): + try: + os.remove(path) + except OSError: + pass + + def _copy_dict(source): result = {} diff --git a/micro/web_config.py b/micro/web_config.py index 76b91c6..2038cbb 100644 --- a/micro/web_config.py +++ b/micro/web_config.py @@ -12,8 +12,11 @@ try: import ota_manager + import config_loader except ImportError: - from micro import ota_manager + from micro import ota_manager, config_loader + +save_config = config_loader.save_config CONFIG_PATH = "config.json" @@ -292,13 +295,6 @@ def build_config(form, current_config): return config -def save_config(path, config): - serialized = _json_dumps(config) - with open(path, "w") as config_file: - config_file.write(serialized) - config_file.write("\n") - - def render_config_page(config, message="", update_status=None): wifi = config.get("wifi", {}) api = config.get("api", {}) diff --git a/tests/test_config_loader.py b/tests/test_config_loader.py index fb1259e..657c535 100644 --- a/tests/test_config_loader.py +++ b/tests/test_config_loader.py @@ -8,10 +8,83 @@ def test_web_setup_default_hostname_has_four_character_suffix(): assert re.fullmatch(r"bambutton-[A-Z0-9]{4}", config_loader.DEFAULT_HOSTNAME) -def test_explicit_hostname_is_preserved_when_loading_config(tmp_path): +def test_partial_config_keeps_explicit_settings_and_fills_defaults(tmp_path): config_path = tmp_path / "config.json" config_path.write_text(json.dumps({"wifi": {"hostname": "shop-button"}})) config = config_loader.load_config(config_path) assert config["wifi"]["hostname"] == "shop-button" + assert config["wifi"]["ssid"] == "" + assert config["api"]["base_url"] == "" + assert config["printer"]["id"] is None + assert not config_loader.is_runtime_config_ready(config) + + +def test_malformed_config_recovers_from_valid_backup(tmp_path): + config_path = tmp_path / "config.json" + backup_path = tmp_path / "config.json.bak" + config_path.write_text('{"wifi":') + backup_path.write_text(json.dumps({ + "wifi": {"ssid": "shop"}, + "api": {"base_url": "http://bambuddy/api/v1"}, + "printer": {"id": 7}, + })) + + config = config_loader.load_config(config_path) + + assert config["wifi"]["ssid"] == "shop" + assert config["printer"]["id"] == 7 + + +def test_malformed_config_without_backup_uses_bootable_defaults(tmp_path): + config_path = tmp_path / "config.json" + config_path.write_text('{"wifi":') + + config = config_loader.load_config(config_path) + + assert config["wifi"]["ssid"] == "" + assert config["api"]["base_url"] == "" + assert config["printer"]["id"] is None + assert config["web"]["password"] + assert not config_loader.is_runtime_config_ready(config) + + +def test_nested_invalid_active_config_recovers_from_valid_backup(tmp_path): + config_path = tmp_path / "config.json" + backup_path = tmp_path / "config.json.bak" + config_path.write_text(json.dumps({"wifi": None})) + backup_path.write_text(json.dumps({ + "wifi": {"ssid": "shop"}, + "api": {"base_url": "http://bambuddy/api/v1", "key": "secret"}, + "printer": {"id": 7}, + })) + + config = config_loader.load_config(config_path) + + assert config["wifi"]["ssid"] == "shop" + assert config["printer"]["id"] == 7 + assert config_loader.is_runtime_config_ready(config) + + +def test_incomplete_config_never_inherits_a_printer_id(tmp_path): + config_path = tmp_path / "config.json" + config_path.write_text(json.dumps({ + "wifi": {"ssid": "shop"}, + "api": {"base_url": "http://bambuddy/api/v1", "key": "secret"}, + })) + + config = config_loader.load_config(config_path) + + assert config["printer"]["id"] is None + assert not config_loader.is_runtime_config_ready(config) + + +def test_non_numeric_printer_id_is_rejected_and_uses_safe_defaults(tmp_path): + config_path = tmp_path / "config.json" + config_path.write_text(json.dumps({"printer": {"id": "3"}})) + + config = config_loader.load_config(config_path) + + assert config["printer"]["id"] is None + assert not config_loader.is_runtime_config_ready(config) diff --git a/tests/test_config_startup.py b/tests/test_config_startup.py new file mode 100644 index 0000000..a1e35d8 --- /dev/null +++ b/tests/test_config_startup.py @@ -0,0 +1,131 @@ +import copy +import sys +from pathlib import Path +from types import SimpleNamespace + +from micro import config_loader + + +APP_MAIN = Path(__file__).parents[1] / "micro" / "app_main.py" + + +class StopMainLoop(BaseException): + pass + + +def test_incomplete_config_starts_setup_ap_without_printer_api_requests(monkeypatch): + config = copy.deepcopy(config_loader.DEFAULT_CONFIG) + api_calls = [] + + class FakeNetwork: + def __init__(self): + self.ap_starts = 0 + self.ap_mode = False + + def connect_with_fallback(self, watchdog_feed=None): + raise AssertionError("incomplete settings must not connect to Wi-Fi") + + def start_access_point(self): + self.ap_starts += 1 + self.ap_mode = True + + def is_ap_mode(self): + return self.ap_mode + + def is_connected(self): + return False + + def ifconfig(self): + return ("192.168.4.1", "255.255.255.0", "192.168.4.1", "192.168.4.1") + + def mode(self): + return "access point" + + class FakeAPI: + def clear_plate(self, printer_id): + api_calls.append(("clear", printer_id)) + + def get_printer_status(self, printer_id): + api_calls.append(("status", printer_id)) + + class FakeButton: + def __init__(self, **kwargs): + pass + + def start(self): + pass + + class FakeFlasher: + def __init__(self, **kwargs): + pass + + def start(self): + pass + + def on(self): + pass + + class FakeWDT: + def __init__(self, timeout): + pass + + def feed(self): + pass + + class FakeOTA: + def __init__(self, reset=None): + pass + + def status(self): + return {"current_version": "test"} + + def confirm_boot(self): + return True + + class FakeWebServer: + def __init__(self, **kwargs): + pass + + def poll(self): + return False + + network = FakeNetwork() + modules = { + "machine": SimpleNamespace(WDT=FakeWDT, reset=lambda: None), + "time": SimpleNamespace( + ticks_ms=lambda: 0, + ticks_diff=lambda left, right: left - right, + ticks_add=lambda value, delta: value + delta, + sleep_ms=lambda delay: (_ for _ in ()).throw(StopMainLoop()), + ), + "bambuddy_api": SimpleNamespace(BambuddyAPI=lambda *args: FakeAPI()), + "config_loader": SimpleNamespace( + load_config=lambda: config, + is_runtime_config_ready=config_loader.is_runtime_config_ready, + ), + "gpio_button": SimpleNamespace(GPIOButton=FakeButton), + "led_flasher": SimpleNamespace(LedFlasher=FakeFlasher), + "ota_manager": SimpleNamespace(OTAUpdateManager=FakeOTA), + "wifi": SimpleNamespace( + WiFi=lambda **kwargs: network, + DEFAULT_HOSTNAME="bambutton", + DEFAULT_AP_PASSWORD="bambutton", + AP_RETRY_INTERVAL_SECONDS=30, + ), + "periodic_timer": SimpleNamespace( + PeriodicTimer=lambda **kwargs: SimpleNamespace(start=lambda: None) + ), + "web_config": SimpleNamespace(WebConfigServer=FakeWebServer), + } + for name, module in modules.items(): + monkeypatch.setitem(sys.modules, name, module) + + namespace = {"__name__": "app_main_config_gate_test"} + try: + exec(compile(APP_MAIN.read_text(), str(APP_MAIN), "exec"), namespace) + except StopMainLoop: + pass + + assert network.ap_starts == 1 + assert network.is_ap_mode() is True + assert api_calls == [] diff --git a/tests/test_web_config.py b/tests/test_web_config.py index f9bc4eb..0b15189 100644 --- a/tests/test_web_config.py +++ b/tests/test_web_config.py @@ -127,6 +127,35 @@ def test_save_config_writes_json(tmp_path): web_config.save_config(str(path), config) assert json.loads(path.read_text()) == config + assert not (tmp_path / "config.json.bak").exists() + assert not (tmp_path / "config.json.tmp").exists() + + +def test_save_config_restores_previous_file_when_replace_fails(tmp_path, monkeypatch): + path = tmp_path / "config.json" + old_config = base_config() + new_config = web_config.build_config(valid_form(), old_config) + path.write_text(json.dumps(old_config)) + real_rename = web_config.config_loader.os.rename + + def fail_staged_replace(source, destination): + if str(source).endswith(".tmp"): + raise OSError("simulated interrupted replacement") + return real_rename(source, destination) + + monkeypatch.setattr(web_config.config_loader.os, "rename", fail_staged_replace) + + with pytest.raises(OSError, match="simulated interrupted"): + web_config.save_config(path, new_config) + + assert json.loads(path.read_text()) == old_config + assert not (tmp_path / "config.json.bak").exists() + assert not (tmp_path / "config.json.tmp").exists() + + +def test_save_config_rejects_non_object_config(tmp_path): + with pytest.raises(ValueError, match="object"): + web_config.save_config(tmp_path / "config.json", []) def test_config_page_contains_form_and_does_not_require_formatting_js():