Cognitive_Agent/tests/unit/test_energy_monitor_frontend_contract.py

from __future__ import annotations

import json
import shutil
import subprocess
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
WEB = ROOT / "apps" / "web-console"


def source(name: str) -> str:
    return (WEB / name).read_text(encoding="utf-8")


def node_runtime() -> str:
    candidates = [
        shutil.which("node"),
        Path.home() / ".cache" / "codex-runtimes" / "codex-primary-runtime" / "dependencies" / "node" / "bin" / "node.exe",
        Path.home() / ".cache" / "codex-runtimes" / "codex-primary-runtime" / "dependencies" / "node" / "bin" / "node",
    ]
    for candidate in candidates:
        if candidate and Path(candidate).is_file():
            return str(candidate)
    raise AssertionError("Node.js is required to execute the energy monitor runtime contracts")


def test_energy_monitor_assets_are_versioned_and_cached() -> None:
    html = source("index.html")
    service_worker = source("service-worker.js")

    for asset in (
        "/energy-monitor.css?v=energy-monitor-4",
        "/energy-monitor.js?v=energy-monitor-4",
    ):
        assert asset in html
        assert f'"{asset}"' in service_worker
    assert html.index("/assistant-chat.js?v=codex-assistant-26") < html.index(
        "/energy-monitor.js?v=energy-monitor-4"
    )
    assert 'const CACHE_VERSION = "cognitive-agent-pad-v276"' in service_worker


def test_energy_monitor_uses_the_fifth_shared_dock_icon_and_starts_closed() -> None:
    script = source("energy-monitor.js")

    assert 'trigger.id = "energyMonitorToggle"' in script
    assert 'trigger.className = "energy-monitor-toggle v2-panel-control-button"' in script
    assert 'trigger.setAttribute("aria-controls", "energyMonitorLayer")' in script
    assert 'trigger.setAttribute("aria-pressed", "false")' in script
    assert 'trigger.setAttribute("aria-expanded", "false")' in script
    assert 'data-lucide="activity"' in script
    assert '$("#v2PanelControlDock").append(trigger)' in script
    assert 'layer.hidden = true' in script
    assert 'layer.inert = true' in script
    assert 'setOpen(true)' in script


def test_energy_monitor_is_an_independent_fixed_modal() -> None:
    script = source("energy-monitor.js")
    styles = source("energy-monitor.css")

    assert 'document.body.append(layer)' in script
    assert 'layer.setAttribute("aria-modal", "true")' in script
    assert "setBackgroundInert(true)" in script
    assert "setBackgroundInert(false)" in script
    assert "position: fixed;" in styles
    assert "z-index: 5000;" in styles
    assert ".energy-monitor-layer[hidden]" in styles
    assert "overscroll-behavior: contain;" in styles
    for forbidden in (
        "setSidePanelVisibility",
        "setSidebarCollapsed",
        "setDrawerOpen",
        "device-power-drawer-open",
        "assistant-drawer-open",
        "data-open-panels",
    ):
        assert forbidden not in script
    assert ".workspace {" not in styles
    assert ".joint-twin-shell {" not in styles
    assert ".v2-unified-brand h1" not in styles


def test_energy_monitor_covers_all_current_controllable_devices() -> None:
    script = source("energy-monitor.js")
    expected = {
        "passenger": "副驾座椅",
        "driver": "主驾座椅",
        "climate": "空调",
        "shade": "遮阳帘",
        "windows": "车窗",
        "reading": "阅读灯",
        "ambient": "氛围灯",
    }

    for device_id, name in expected.items():
        assert f'id: "{device_id}"' in script
        assert f'name: "{name}"' in script
    assert 'id: "headlight"' not in script
    assert "ENERGY_MONITORED_DEVICES.length" in script
    assert "data-energy-device-power" in script
    assert "data-energy-device-energy" in script
    assert "data-energy-device-sparkline" in script
    assert "data-energy-device-share" in script


def test_energy_monitor_has_measured_simulated_and_estimated_data_sources() -> None:
    script = source("energy-monitor.js")

    assert 'window.addEventListener("cognitive-agent:power-telemetry"' in script
    assert "window.updateDevicePowerTelemetry = linkedUpdater" in script
    assert "telemetry.power_w ?? telemetry.total_power_w ?? telemetry.current_power_w" in script
    assert 'source: "estimated"' in script
    assert "function telemetrySource(telemetry, payload)" in script
    assert 'telemetry?.simulated === true || payload?.simulated === true' in script
    assert 'telemetry?.estimated === true || payload?.estimated === true' in script
    assert 'return "estimated";' in script
    assert 'measured: "车端实测"' in script
    assert 'simulated: "模拟数据"' in script
    assert 'estimated: "状态估算"' in script
    assert "未接入独立电表的设备显示明确标注的状态估算值" in script
    assert "MEASURED_STALE_MS = 15000" in script
    assert "MAX_DEVICE_POWER_W = 100000" in script


def test_energy_monitor_samples_integrates_and_bounds_its_curves() -> None:
    script = source("energy-monitor.js")

    assert "SAMPLE_INTERVAL_MS = 1000" in script
    assert "HISTORY_LIMIT = 60" in script
    assert "((previous + values[id]) / 2) * elapsedMs / 3600000" in script
    assert "Math.min(5000" not in script
    assert "monitor.frames.splice(0, monitor.frames.length - HISTORY_LIMIT)" in script
    assert 'id="energyMonitorChartLine"' in script
    assert 'id="energyMonitorChartArea"' in script
    assert "polylinePoints(values, maximum" in script
    assert 'window.setInterval(() => sampleNow(), SAMPLE_INTERVAL_MS)' in script
    assert 'ingestPowerTelemetry(event.detail || {}, { sample: false })' in script
    assert 'ingestPowerTelemetry(payload, { sample: false })' in script
    assert 'new CustomEvent("cognitive-agent:energy-snapshot"' in script


def test_energy_monitor_tracks_ai_and_direct_control_activity() -> None:
    script = source("energy-monitor.js")
    assistant = source("assistant-chat.js")

    assert "function markControlActivity(payload)" in script
    assert 'window.addEventListener("cognitive-agent:vehicle-can-control-request"' in script
    assert 'window.addEventListener("cognitive-agent:validated-can-commands"' in script
    assert 'window.addEventListener("cognitive-agent:vehicle-control-activity"' in script
    assert 'window.addEventListener("cognitive-agent:request-complete"' in script
    assert "eventPayload?.actions" in script
    assert "eventPayload?.validated_actions" in script
    assert "eventPayload?.intent_ir?.spec?.actions" in script
    assert "eventPayload?.function_config" in script
    assert "eventPayload?.function_configs" in script
    assert "eventPayload?.command_plans" in script
    assert "notifyControlActivity: markControlActivity" in script
    assert "completeControlActivity" in script
    assert 'event.type === "action.progress" && event.payload' in assistant
    assert 'new CustomEvent("cognitive-agent:vehicle-control-activity"' in assistant
    assert 'new CustomEvent("cognitive-agent:validated-can-commands"' not in assistant


def test_energy_monitor_reads_persistent_seat_state_for_both_seats() -> None:
    app = source("app.js")
    script = source("energy-monitor.js")

    assert 'card.dataset.ventilationState = "unknown"' in app
    assert 'seatCard.dataset[`${feature}State`]' in app
    assert 'seatCard.dataset[`${feature}Level`]' in app
    assert "seatCard.dataset.massageState" in app
    assert 'const card = $(\'.seat-card[data-seat="passenger"]\')' in script
    assert "card?.dataset.ventilationState" in script


def test_energy_monitor_escape_is_consumed_and_focus_is_restored() -> None:
    script = source("energy-monitor.js")

    assert 'document.addEventListener("keydown", captureModalKeys, true)' in script
    assert 'if (event.key === "Escape")' in script
    assert "event.stopImmediatePropagation()" in script
    assert "monitor.returnFocus = document.activeElement" in script
    assert "target.focus()" in script
    assert "focusableElements()" in script
    assert 'if (event.key !== "Tab") return' in script
    assert "!layer?.contains(document.activeElement)" in script
    assert "monitor.returnFocus.isConnected" in script


def test_energy_monitor_is_responsive_and_reduced_motion_safe() -> None:
    styles = source("energy-monitor.css")

    for media_query in (
        "@media (max-width: 980px)",
        "@media (max-width: 760px)",
        "@media (max-width: 460px)",
        "@media (max-width: 340px)",
        "@media (prefers-reduced-motion: reduce)",
    ):
        assert media_query in styles
    assert "overflow: auto;" in styles
    assert "grid-template-columns: minmax(0, 1fr);" in styles
    assert "font-variant-numeric: tabular-nums;" in styles
    assert "animation: none;" in styles


def test_energy_monitor_runtime_aggregation_and_invalid_input_are_deterministic() -> None:
    node = node_runtime()
    harness = r"""
const path = require("path");
let clock = 1000000;
Date.now = () => clock;
global.document = {
  activeElement: null,
  addEventListener() {},
  querySelector() { return null; },
  querySelectorAll() { return []; },
};
global.window = {
  addEventListener() {},
  dispatchEvent() {},
  setInterval() { return 1; },
  setTimeout() {},
};
global.CustomEvent = class CustomEvent { constructor(type, init) { this.type = type; this.detail = init?.detail; } };
global.HTMLElement = class HTMLElement {};
require(path.resolve(process.argv[1]));
const api = window.CognitiveAgentEnergyMonitor;
const measuredPayload = { source: "vehicle_power_sensor", devices: {
  passenger_seat: { power_w: 120 },
  driver_seat: { power_w: 80 },
  climate: { power_w: 700, trip_energy_kwh: 0.42 },
  roof: { power_w: 20 },
  window: { power_w: 10 },
  reading_light: { power_w: 30 },
  ambient_light: { power_w: 40 },
}};
const first = api.ingestTelemetry(measuredPayload);
clock += 1000;
api.ingestTelemetry({ source: "physical_can", devices: { climate: { power_w: 760 } } }, { sample: false });
const second = api.sampleNow({ now: clock, render: false });
const simulated = api.ingestTelemetry({ device: "passenger_seat", power_w: 121, simulated: true });
const unlabelled = api.ingestTelemetry({ device: "driver_seat", power_w: 81 });
const derived = api.ingestTelemetry({ device: "roof", power_w: 21, source: "derived" });
const invalid = api.ingestTelemetry({ devices: {
  passenger_seat: { power_w: -3 },
  climate: { power_w: 1e308 },
  unknown_device: { power_w: 9999 },
}});
api.ingestTelemetry(measuredPayload, { sample: false });
api.sampleNow({ now: clock, render: false });
const energyBeforeGap = api.getSnapshot().total_energy_wh;
clock += 60000;
api.ingestTelemetry(measuredPayload, { sample: false });
const longGap = api.sampleNow({ now: clock, render: false });
const longGapEnergyDelta = longGap.total_energy_wh - energyBeforeGap;
clock += 16000;
const staleEnergy = api.sampleNow({ now: clock, render: false });
api.ingestTelemetry({ source: "vehicle_power_sensor", devices: {
  ambient_light: { today_energy_kwh: 0.125 },
}}, { sample: false });
const energyOnly = api.sampleNow({ now: clock, render: false });
const base = clock;
for (let index = 1; index <= 70; index += 1) {
  api.sampleNow({ now: base + index * 1000, render: false });
}
const bounded = api.getSnapshot();
process.stdout.write(JSON.stringify({
  first, second, simulated, unlabelled, derived, invalid, longGapEnergyDelta, staleEnergy, energyOnly, bounded,
}));
"""
    completed = subprocess.run(
        [node, "-e", harness, str(WEB / "energy-monitor.js")],
        check=True,
        capture_output=True,
        text=True,
        encoding="utf-8",
    )
    output = json.loads(completed.stdout)

    assert round(output["first"]["total_power_w"]) == 1000
    assert output["first"]["source"] == "measured"
    assert output["first"]["devices"]["climate"]["reported_energy_wh"] == 420
    assert output["first"]["devices"]["climate"]["reported_energy_label"] == "本次行程"
    assert round(output["second"]["total_power_w"]) == 1060
    assert output["simulated"]["devices"]["passenger"]["source"] == "simulated"
    assert output["unlabelled"]["devices"]["driver"]["source"] == "estimated"
    assert output["derived"]["devices"]["shade"]["source"] == "estimated"
    assert output["derived"]["source"] == "measured_simulated_estimated"
    assert output["derived"]["measured_count"] == 4
    assert output["derived"]["simulated_count"] == 1
    assert output["derived"]["estimated_count"] == 2
    assert round(output["invalid"]["devices"]["passenger"]["power_w"]) == 121
    assert round(output["invalid"]["devices"]["climate"]["power_w"]) == 760
    assert 16.6 < output["longGapEnergyDelta"] < 16.8
    assert output["staleEnergy"]["devices"]["climate"]["source"] == "estimated"
    assert output["staleEnergy"]["devices"]["climate"]["reported_energy_wh"] == 420
    assert output["staleEnergy"]["devices"]["climate"]["reported_energy_source"] == "measured"
    assert output["energyOnly"]["devices"]["ambient"]["reported_energy_wh"] == 125
    assert output["energyOnly"]["devices"]["ambient"]["reported_energy_label"] == "今日累计"
    assert output["bounded"]["sample_count"] == 60
    assert output["bounded"]["source"] == "estimated"
    assert output["bounded"]["estimated_count"] == 7
    assert output["bounded"]["total_power_w"] >= 0
    assert output["bounded"]["total_energy_wh"] >= 0


def test_energy_monitor_runtime_maps_ai_operations_to_active_devices() -> None:
    node = node_runtime()
    harness = r"""
const path = require("path");
let clock = 2000000;
Date.now = () => clock;
global.document = {
  activeElement: null,
  addEventListener() {},
  querySelector() { return null; },
  querySelectorAll() { return []; },
};
global.window = {
  addEventListener() {},
  dispatchEvent() {},
  setInterval() { return 1; },
  setTimeout() {},
};
global.CustomEvent = class CustomEvent { constructor(type, init) { this.type = type; this.detail = init?.detail; } };
global.HTMLElement = class HTMLElement {};
require(path.resolve(process.argv[1]));
const api = window.CognitiveAgentEnergyMonitor;
const idle = api.sampleNow({ now: clock, render: false });
const devices = api.notifyControlActivity({ result: { status: "succeeded", actions: [
  { domain: "seat", capability: "seat.set_backrest_recline", target: { instance: "passenger" } },
  { domain: "seat", capability: "seat.set_driver_axis_native", target: { zone: "front_left" } },
  { domain: "window", capability: "window.set_motion", target: { zone: "all" } },
  { domain: "roof", capability: "roof.set_shade_motion", target: { zone: "vehicle" } },
  { domain: "climate", capability: "climate.set_temperature", target: { zone: "driver" } },
  { domain: "interior_light", capability: "interior_light.set_reading_power", target: { zone: "all" } },
  { domain: "ambient_light", capability: "ambient_light.set_power", target: { zone: "vehicle" } },
] } });
const directMappings = [
  "passenger_seat", "driver_seat", "climate", "roof", "window", "reading_light", "ambient_light",
].map(device_id => [device_id, api.notifyControlActivity({ device_id })]);
const intentShape = api.notifyControlActivity({ status: "processing", intent_ir: { spec: { actions: [
  { domain: "climate", capability: "climate.set_power", target: { zone: "vehicle" } },
] } } });
const singleConfigShape = api.notifyControlActivity({ status: "processing", function_config: { spec: {
  domain: "interior_light",
  operations: [{ capability: "interior_light.set_indoor_power", target: { zone: "vehicle" } }],
} } });
const previewDevices = api.notifyControlActivity({ result: {
  status: "succeeded",
  orchestration_route: "mode_profile_create_preview",
  actions: [{ domain: "ambient_light", capability: "ambient_light.set_power" }],
} });
const failedDevices = api.notifyControlActivity({ result: {
  status: "compile_failed",
  actions: [{ domain: "climate", capability: "climate.set_power" }],
} });
clock += 1000;
const active = api.sampleNow({ now: clock, render: false });
clock += 5000;
const settled = api.sampleNow({ now: clock, render: false });
const failureDevices = api.notifyControlActivity({
  status: "processing",
  request_id: "req-energy-failure",
  actions: [{ domain: "seat", capability: "seat.set_backrest_recline", target: { instance: "passenger" } }],
});
const failureIncrement = api.notifyControlActivity({
  status: "processing",
  request_id: "req-energy-failure",
  actions: [{ domain: "climate", capability: "climate.set_temperature", target: { zone: "vehicle" } }],
});
clock += 1000;
const failureActive = api.sampleNow({ now: clock, render: false });
api.completeControlActivity({
  result: { status: "failed", request_id: "req-energy-failure" },
  error: { message: "vehicle rejected command" },
});
clock += 1;
const failureCleared = api.sampleNow({ now: clock, render: false });
process.stdout.write(JSON.stringify({
  idle, devices, directMappings, intentShape, singleConfigShape, previewDevices, failedDevices, active,
  settled, failureDevices, failureIncrement, failureActive, failureCleared,
}));
"""
    completed = subprocess.run(
        [node, "-e", harness, str(WEB / "energy-monitor.js")],
        check=True,
        capture_output=True,
        text=True,
        encoding="utf-8",
    )
    output = json.loads(completed.stdout)

    assert set(output["devices"]) == {
        "passenger", "driver", "windows", "shade", "climate", "reading", "ambient"
    }
    assert output["directMappings"] == [
        ["passenger_seat", ["passenger"]],
        ["driver_seat", ["driver"]],
        ["climate", ["climate"]],
        ["roof", ["shade"]],
        ["window", ["windows"]],
        ["reading_light", ["reading"]],
        ["ambient_light", ["ambient"]],
    ]
    assert output["intentShape"] == ["climate"]
    assert output["singleConfigShape"] == ["reading"]
    assert output["previewDevices"] == []
    assert output["failedDevices"] == []
    assert output["active"]["devices"]["passenger"]["power_w"] > output["idle"]["devices"]["passenger"]["power_w"] + 100
    assert output["active"]["devices"]["driver"]["power_w"] > output["idle"]["devices"]["driver"]["power_w"] + 70
    assert output["active"]["devices"]["windows"]["power_w"] > output["idle"]["devices"]["windows"]["power_w"] + 100
    assert output["active"]["devices"]["shade"]["power_w"] > output["idle"]["devices"]["shade"]["power_w"] + 70
    assert output["active"]["devices"]["climate"]["power_w"] > output["idle"]["devices"]["climate"]["power_w"] + 70
    assert output["active"]["devices"]["reading"]["power_w"] > output["idle"]["devices"]["reading"]["power_w"] + 4
    assert output["active"]["devices"]["ambient"]["power_w"] > output["idle"]["devices"]["ambient"]["power_w"] + 3
    assert output["failureDevices"] == ["passenger"]
    assert output["failureIncrement"] == ["climate"]
    assert output["failureActive"]["devices"]["passenger"]["power_w"] > output["settled"]["devices"]["passenger"]["power_w"] + 100
    assert output["failureCleared"]["devices"]["passenger"]["power_w"] < output["failureActive"]["devices"]["passenger"]["power_w"] - 100
    assert output["failureCleared"]["devices"]["climate"]["power_w"] < output["failureActive"]["devices"]["climate"]["power_w"] - 70