overland-controller/tests/test_http_api_contract.py

import json
import re
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
FIXTURES = ROOT / "tests" / "fixtures" / "http_api"
FIRMWARE = ROOT / "firmware" / "esp32" / "overland-controller" / "overland-controller.ino"


def load_fixture(name):
    return json.loads((FIXTURES / name).read_text())


def firmware_source():
    return FIRMWARE.read_text()


def assert_keys(payload, keys):
    missing = set(keys) - set(payload)
    assert not missing, f"missing keys: {sorted(missing)}"


def registered_routes():
    source = firmware_source()
    literal_pattern = re.compile(r'server\.on\("([^"]+)"(?:,\s*(HTTP_[A-Z]+))?')
    macro_pattern = re.compile(r'server\.on\(API_V1\("([^"]+)"\)(?:,\s*(HTTP_[A-Z]+))?')

    routes = {(match.group(1), match.group(2) or "ANY") for match in literal_pattern.finditer(source)}
    routes.update({
        (f"/api/v1{match.group(1)}", match.group(2) or "ANY")
        for match in macro_pattern.finditer(source)
    })
    return routes


def test_firmware_registers_current_http_contract_routes():
    routes = registered_routes()

    expected_routes = {
        ("/api/v1/status", "ANY"),
        ("/api/v1/health", "HTTP_GET"),
        ("/api/v1/capabilities", "HTTP_GET"),
        ("/api/v1/config", "HTTP_GET"),
        ("/api/v1/config/export", "HTTP_GET"),
        ("/api/v1/config/import", "HTTP_POST"),
        ("/api/v1/relay/set", "HTTP_POST"),
        ("/api/v1/config/wifi", "HTTP_GET"),
        ("/api/v1/config/wifi", "HTTP_POST"),
        ("/api/v1/wifi/connect", "HTTP_POST"),
        ("/api/v1/wifi/clear", "HTTP_POST"),
        ("/api/v1/temps/scan", "HTTP_POST"),
        ("/api/v1/temps/assign", "HTTP_POST"),
        ("/api/v1/temps/clear", "HTTP_POST"),
    }

    assert expected_routes <= routes


def test_firmware_keeps_root_compatibility_aliases():
    routes = registered_routes()

    compatibility_routes = {
        ("/status", "ANY"),
        ("/config", "HTTP_GET"),
        ("/relay/set", "HTTP_POST"),
        ("/config/wifi", "HTTP_GET"),
        ("/config/wifi", "HTTP_POST"),
        ("/wifi/connect", "HTTP_POST"),
        ("/wifi/clear", "HTTP_POST"),
        ("/temps/scan", "HTTP_POST"),
        ("/temps/assign", "HTTP_POST"),
        ("/temps/clear", "HTTP_POST"),
    }

    assert compatibility_routes <= routes


def test_embedded_webui_uses_versioned_api_routes():
    source = firmware_source()

    assert 'const API_BASE="/api/v1";' in source
    assert 'fetch(api("' in source
    assert 'fetch("/status"' not in source
    assert 'fetch("/config' not in source
    assert 'fetch("/relay' not in source
    assert 'fetch("/temps' not in source
    assert 'fetch("/wifi' not in source
    assert "/bms/reconnect" not in source
    assert "/system/restart" not in source


def test_status_fixture_matches_dashboard_contract_shape():
    payload = load_fixture("status_response.json")

    assert_keys(payload, [
        "type",
        "timestamp",
        "battery",
        "temps",
        "relays",
        "vehicle",
        "network",
        "alarms",
        "system",
        "config",
    ])
    assert payload["type"] == "status_response"

    assert_keys(payload["battery"], [
        "source",
        "connected",
        "soc",
        "voltage",
        "current",
        "remaining_ah",
        "capacity_ah",
        "runtime_hours",
        "temperature_f",
        "cycle_count",
        "cell_count",
        "ntc_count",
        "cell_voltages",
        "cell_min_voltage",
        "cell_max_voltage",
        "cell_delta_mv",
        "cells_valid",
    ])
    assert_keys(payload["temps"][0], [
        "id",
        "name",
        "enabled",
        "weather",
        "online",
        "temperature_f",
    ])
    assert payload["hardware_profile"] == "generic_esp32_2ch_relay"
    assert payload["output_count"] == 2
    assert_keys(payload["relays"][0], [
        "id",
        "name",
        "role",
        "hardware_channel",
        "hardware_profile",
        "pin",
        "enabled",
        "available",
        "state",
    ])
    assert_keys(payload["network"], [
        "wifi_enabled",
        "uart_connected",
        "ap_enabled",
        "ap_ip",
        "sta_enabled",
        "sta_connected",
        "sta_ssid",
        "sta_ip",
        "saved_network_count",
        "saved_networks",
    ])
    assert_keys(payload["system"], [
        "firmware_name",
        "firmware_version",
        "build_date",
        "build_time",
        "uptime_seconds",
    ])


def test_config_fixture_matches_current_config_response_shape():
    payload = load_fixture("config_response.json")

    assert_keys(payload, [
        "device_name",
        "hardware_profile",
        "output_count",
        "relays",
        "bms",
        "temperature_sensor_count",
        "temperature_sensors",
    ])
    assert payload["hardware_profile"] == "generic_esp32_2ch_relay"
    assert payload["output_count"] == 2
    assert_keys(payload["relays"][0], [
        "id",
        "name",
        "role",
        "hardware_channel",
        "hardware_profile",
        "pin",
        "enabled",
        "available",
    ])
    assert_keys(payload["bms"], ["enabled", "name", "address", "address_type"])
    assert_keys(payload["temperature_sensors"][0], [
        "id",
        "name",
        "address",
        "enabled",
        "weather",
    ])


def test_health_and_capabilities_contracts_are_registered():
    source = firmware_source()

    assert '"health_response"' in source
    assert '"capabilities_response"' in source
    assert '"GET /api/v1/health"' in source
    assert '"GET /api/v1/capabilities"' in source
    assert '"config_backup_restore"' in source


def test_config_backup_restore_contract_is_registered():
    source = firmware_source()

    assert 'API_V1("/config/export")' in source
    assert 'API_V1("/config/import")' in source
    assert '"config_export_response"' in source
    assert '"password"' in source
    assert '"invalid_config"' in source


def test_relay_set_contract_fixtures_and_firmware_errors():
    request = load_fixture("relay_set_request.json")
    response = load_fixture("relay_set_response.json")

    assert_keys(request, ["id", "state"])
    assert_keys(response, ["type", "ok", "id", "state"])
    assert response["type"] == "relay_response"
    assert response["ok"] is True

    source = firmware_source()
    assert '"invalid_json"' in source
    assert '"unknown_relay"' in source


def test_relay_api_is_output_count_agnostic():
    source = firmware_source()

    assert 'doc["hardware_profile"] = HARDWARE_PROFILE;' in source
    assert 'doc["output_count"] = MAX_RELAYS;' in source
    assert 'for (int i = 0; i < MAX_RELAYS; i++)' in source
    assert 'relay["hardware_channel"] = appConfig.relays[i].hardwareChannel;' in source
    assert 'relay["available"] = appConfig.relays[i].available;' in source
    assert 'relays.state[index] = enabled;' in source
    assert 'relays.relay1' not in source
    assert 'relays.relay2' not in source


def test_temp_scan_assign_clear_contract_fixtures():
    scan = load_fixture("temp_scan_response.json")
    assign_request = load_fixture("temp_assign_request.json")
    assign_response = load_fixture("temp_assign_response.json")
    clear_request = load_fixture("temp_clear_request.json")
    clear_response = load_fixture("temp_clear_response.json")

    assert_keys(scan, ["type", "ok", "devices"])
    assert scan["type"] == "temp_scan_response"
    assert_keys(scan["devices"][0], ["index", "address"])

    assert_keys(assign_request, ["id", "index"])
    assert_keys(assign_response, ["device_name", "relays", "bms", "temperature_sensors"])
    assert assign_response["temperature_sensors"][0]["address"]
    assert assign_response["temperature_sensors"][0]["enabled"] is True

    assert_keys(clear_request, ["id"])
    assert_keys(clear_response, ["device_name", "relays", "bms", "temperature_sensors"])
    assert clear_response["temperature_sensors"][0]["address"] == ""
    assert clear_response["temperature_sensors"][0]["enabled"] is False

    source = firmware_source()
    assert '"unknown_temp_sensor"' in source
    assert '"invalid_temp_selection"' in source


def test_wifi_config_contract_fixtures():
    request = load_fixture("wifi_config_request.json")
    response = load_fixture("wifi_config_response.json")

    assert_keys(request, ["networks"])
    assert_keys(request["networks"][0], ["ssid", "password", "priority"])
    assert_keys(response, ["type", "ok", "wifi"])
    assert response["type"] == "wifi_config_response"
    assert response["ok"] is True
    assert_keys(response["wifi"], [
        "ap_enabled",
        "sta_enabled",
        "network_count",
        "active_ssid",
        "sta_connected",
        "ap_ip",
        "sta_ip",
        "networks",
    ])
    assert_keys(response["wifi"]["networks"][0], [
        "index",
        "ssid",
        "priority",
        "password_set",
    ])


def test_http_error_fixture_and_known_error_codes_are_stable():
    payload = load_fixture("error_response.json")
    source = firmware_source()

    assert payload == {"ok": False, "error": "invalid_json"}

    for error_code in [
        "invalid_json",
        "invalid_config",
        "unknown_relay",
        "unknown_temp_sensor",
        "invalid_temp_selection",
        "invalid_bms_selection",
        "invalid_relay_route",
        "missing_relay_action",
        "invalid_relay_action",
    ]:
        assert error_code in source


def test_status_endpoint_supports_field_selection():
    source = firmware_source()

    assert 'server.hasArg("fields")' in source
    assert 'server.arg("fields")' in source
    assert 'findStatusFieldIndex' in source
    assert '"invalid_field"' in source
    assert 'Unknown field' in source
    assert 'server.send(400, "application/json", errorOutput)' in source


def test_embedded_webui_uses_partial_status_for_overview_polling():
    source = firmware_source()

    assert 'const STATUS_FIELDS_OVERVIEW="battery,temps,relays,vehicle,network,alarms,system";' in source
    assert 'function statusUrl()' in source
    assert 'api("/status?fields="+STATUS_FIELDS_OVERVIEW)' in source
    assert 'fetch(statusUrl(),{cache:"no-store"})' in source
    assert 'function mergeStatus(previous,next)' in source
    assert 'api("/status?fields=config")' in source


def test_cargo_ap_uses_wpa2_password():
    source = firmware_source()

    assert "Preferences apPrefs;" in source
    assert "generateDefaultApPassword" in source
    assert "validApPassword" in source
    assert "loadApConfig();" in source
    assert "startAccessPoint()" in source
    assert "WiFi.softAP(apSsid.c_str(), apPassword.c_str())" in source
    assert 'network["ap_auth"] = "wpa2";' in source
    assert 'server.on(API_V1("/config/ap"), HTTP_GET, handleGetApConfig);' in source
    assert 'server.on(API_V1("/config/ap"), HTTP_POST, handleUpdateApConfig);' in source


def test_cargo_ap_password_is_not_returned_by_get_config_ap():
    source = firmware_source()

    get_start = source.index("void handleGetApConfig()")
    get_end = source.index("void handleUpdateApConfig()")
    get_body = source[get_start:get_end]

    assert 'response["password_set"]' in get_body
    assert 'response["password"]' not in get_body


def test_embedded_webui_has_ap_settings_controls():
    source = firmware_source()

    assert "Access Point" in source
    assert 'id="apSsidInput"' in source
    assert 'id="apPasswordInput"' in source
    assert "function saveApConfig()" in source
    assert 'api("/config/ap")' in source
    assert "toggleApPassword" in source
    assert "renderApConfig(data)" in source
    assert "Dashboard auto-migration is planned but not implemented yet" in source


def test_embedded_webui_config_page_is_grouped():
    source = firmware_source()

    assert "configGroupTitle" in source
    assert "Home / Starlink WiFi" in source
    assert "Relay Outputs" in source
    assert "Temperature Assignment" in source
    assert "Maintenance" in source


def test_embedded_webui_config_subtabs():
    source = firmware_source()

    assert "function showConfigSubtab(name)" in source
    assert "config-general" in source
    assert "config-wifi" in source
    assert "config-relays" in source
    assert "config-temps" in source
    assert "config-bms" in source
    assert "config-maintenance" in source
    assert "WiFi / AP" in source
    assert "Maintenance" in source


def test_embedded_webui_api_badge_labels_are_clear():
    source = firmware_source()

    assert "API Connecting..." in source
    assert "API Live" in source
    assert "API Error" in source
    assert 'textContent="Offline"' not in source


def test_embedded_webui_config_page_is_grouped():
    source = firmware_source()

    assert "configGroupTitle" in source
    assert "General" in source
    assert "Home / Starlink WiFi" in source
    assert "Relay Outputs" in source
    assert "Temperature Assignment" in source
    assert "BMS Config" in source
    assert "Maintenance" in source


def test_embedded_webui_config_subtabs():
    source = firmware_source()

    assert "function showConfigSubtab(name)" in source
    assert "config-general" in source
    assert "config-wifi" in source
    assert "config-relays" in source
    assert "config-temps" in source
    assert "config-bms" in source
    assert "config-maintenance" in source
    assert "WiFi / AP" in source
    assert "subtabbtn active" in source


def test_cargo_oled_support_is_optional():
    source = firmware_source()

    assert "#define OLED_ENABLED 0" in source
    assert "#if OLED_ENABLED" in source
    assert "void oledBegin()" in source
    assert "void oledLoop()" in source
    assert "void oledShowSetupPage()" in source
    assert "OLED_I2C_ADDR = 0x3C" in source
    assert "Adafruit_SSD1306" in source


def test_cargo_oled_shows_setup_and_service_pages():
    source = firmware_source()

    assert 'oledClear("Setup / AP")' in source
    assert 'oledClear("Battery")' in source
    assert 'oledClear("Relays")' in source
    assert 'oledClear("Network")' in source
    assert 'oledClear("Alarms")' in source
    assert "oledShowSetupPage();" in source
    assert "oledLoop();" in source


def test_gpio_pin_plan_assigns_oled_to_21_22():
    config = (ROOT / "firmware" / "esp32" / "overland-controller" / "config.h").read_text()
    source = firmware_source()

    assert "#define OLED_I2C_SDA_PIN 21" in config
    assert "#define OLED_I2C_SCL_PIN 22" in config
    assert "#define OLED_SETUP_BUTTON_PIN 25" in config
    assert "const int OLED_I2C_SDA = OLED_I2C_SDA_PIN;" in source
    assert "const int OLED_I2C_SCL = OLED_I2C_SCL_PIN;" in source
    assert "const int OLED_BUTTON_PIN = OLED_SETUP_BUTTON_PIN;" in source


def test_legacy_dashboard_uart_is_disabled_by_default():
    config = (ROOT / "firmware" / "esp32" / "overland-controller" / "config.h").read_text()
    source = firmware_source()

    assert "#define DASHBOARD_UART_ENABLED 0" in config
    assert "DASHBOARD_UART_TX_PIN 21" not in config
    assert "DASHBOARD_UART_RX_PIN 22" not in config
    assert "#if DASHBOARD_UART_ENABLED" in source
    assert "pollDashboardUart();" in source


def test_temp_config_lookup_is_available_to_http_handlers():
    source = firmware_source()

    assert "int findTempConfigIndexById(" in source
    assert "configIndex = findTempConfigIndexById(id);" in source
    assert "findTempConfigIndexForUart" not in source

    helper_pos = source.index("int findTempConfigIndexById(")
    first_uart_guard_pos = source.index("#if DASHBOARD_UART_ENABLED")
    assert helper_pos < first_uart_guard_pos


def test_embedded_webui_preserves_config_input_while_polling():
    source = firmware_source()

    assert "const configTouched={};" in source
    assert "function markConfigFieldTouched(id)" in source
    assert "function setConfigValue(id,value)" in source
    assert "document.activeElement===el" in source
    assert 'document.addEventListener("input"' in source
    assert 'setConfigValue("apSsidInput"' in source
    assert "deviceNameInput" in source
    assert "bmsNameInput" in source
    assert "tempEnabledCount" in source


def test_ap_config_can_be_reset_to_defaults():
    source = firmware_source()

    assert "void resetApConfigToDefaults()" in source
    assert 'saveApConfig("OverlandController", newPassword);' in source
    assert "void handleResetApConfig()" in source
    assert 'server.on(API_V1("/config/ap/reset"), HTTP_POST, handleResetApConfig);' in source
    assert 'api("/config/ap/reset")' in source
    assert "Reset AP to Default" in source
    assert "regenerated password" in source


def test_embedded_webui_has_consistent_button_intent_styles():
    source = firmware_source()

    assert "/* UI polish v2 */" in source
    assert "button.save" in source
    assert "button.danger" in source
    assert "button.secondary" in source
    assert "configGroupTitle" in source
    assert "linear-gradient" in source
    assert 'class="save" onclick="saveApConfig()"' in source
    assert 'class="danger" onclick="resetApConfig()"' in source
    assert 'class="save" onclick="saveWifi()"' in source


def test_embedded_webui_theme_v3():
    source = firmware_source()

    assert "/* UI theme v3 */" in source
    assert "--accent:#5f8fcf" in source
    assert "@media(min-width:980px)" in source
    assert "#configPage" in source
    assert "grid-template-columns:190px 1fr" in source
    assert "button.save" in source
    assert "button.danger" in source
    assert "button.secondary" in source


def test_embedded_webui_main_pages_are_mutually_exclusive():
    source = firmware_source()

    assert "/* Desktop page layout fix */" in source
    assert ".page:not(.active){display:none!important}" in source
    assert "#configPage.active" in source
    assert "Main page navigation override" in source
    assert 'overview:"overviewPage"' in source
    assert 'battery:"batteryPage"' in source
    assert 'config:"configPage"' in source
    assert 'showConfigSubtab("general")' in source


def test_embedded_webui_mobile_temperature_config_cards():
    source = firmware_source()

    assert "/* Mobile temperature config layout */" in source
    assert "tempConfigCard" in source
    assert "tempConfigControls" in source
    assert "Temp Slot " in source
    assert "Clear Assignment" in source
    assert "Weather Badge" in source
    assert "@media(max-width:700px)" in source


def test_embedded_webui_temp_config_uses_mobile_cards():
    source = firmware_source()
    start = source.index('const tempBox=$("tempConfigList");')
    end = source.index("async function saveRelayConfig", start)
    body = source[start:end]

    assert "tempConfigCard" in body
    assert "Temp Slot " in body
    assert "Clear Assignment" in body
    assert "Weather Badge" in body
    assert "configrow tempConfigRow" not in body
    assert "Address:" not in body


def test_embedded_webui_formats_uptime():
    source = firmware_source()

    assert "function formatUptime(totalSeconds)" in source
    assert 'parts.push(days+"d")' in source
    assert 'parts.push(hours+"h")' in source
    assert 'parts.push(minutes+"m")' in source
    assert 'parts.push(seconds+"s")' in source
    assert 'formatUptime(s.uptime_seconds||0)' in source
    assert '+" sec"' not in source