test_neuralos.py

"""Verification suite for NeuralOS Hypervisor v3.

Runs entirely on the deterministic MockNeuralEngine: no torch needed, so the
compositor / kernel / protocol logic is testable in CI in <1s.
"""
import os
import struct
import json

os.environ["NEURALOS_ENGINE"] = "mock"
os.environ["NEURALOS_CACHE"] = "/tmp/neuralos_test_cache"

import numpy as np
import pytest
from PIL import Image

import app as N


@pytest.fixture(autouse=True)
def fresh():
    """Rebuild module state per test."""
    N.STORE = N.AssetStore()
    N.SPRITES = N.SpriteLibrary(N.STORE)
    N.KERNEL = N.OSKernel(N.SPRITES)
    N.COMPOSITOR = N.Compositor(N.KERNEL, N.SPRITES, N.STORE)
    N.ENGINE = N.MockNeuralEngine()
    yield


# --------------------------------------------------------------------------- #
# compositor
# --------------------------------------------------------------------------- #

def test_merge_rects_tile_aligned():
    r = N.merge_rects([(5, 5, 10, 10)], (N.FB, N.FB))
    assert r == [(0, 0, N.TILE, N.TILE)]


def test_merge_rects_clamps_to_fb():
    r = N.merge_rects([(N.FB - 5, N.FB - 5, 100, 100)], (N.FB, N.FB))
    assert r == [(N.FB - N.TILE, N.FB - N.TILE, N.TILE, N.TILE)]


def test_layerstack_order_and_alpha():
    st = N.LayerStack((32, 32))
    st.put("a", Image.new("RGBA", (32, 32), (255, 0, 0, 255)), 0, 0, 1)
    st.put("b", Image.new("RGBA", (16, 16), (0, 0, 255, 255)), 0, 0, 2)
    out = st.composite()
    assert out.getpixel((2, 2)) == (0, 0, 255)
    assert out.getpixel((24, 24)) == (255, 0, 0)


def test_put_is_identity_stable():
    """Unchanged layer must not dirty the grid (else patches never fire)."""
    st = N.LayerStack((32, 32))
    img = Image.new("RGBA", (8, 8), (1, 2, 3, 255))
    st.put("x", img, 0, 0, 1)
    st.take_dirty_rects()
    st.put("x", img, 0, 0, 1)
    assert st.take_dirty_rects() == []


def test_composite_clip_only_region():
    st = N.LayerStack((64, 64))
    st.put("bg", Image.new("RGBA", (64, 64), (10, 20, 30, 255)), 0, 0, 1)
    tile = st.composite(clip=(32, 32, 16, 16))
    assert tile.size == (16, 16)


# --------------------------------------------------------------------------- #
# sprite library
# --------------------------------------------------------------------------- #

def test_wallpaper_and_icons_cached():
    w1 = N.SPRITES.wallpaper()
    w2 = N.SPRITES.wallpaper()
    assert w1 is w2
    assert w1.size == (N.FB, N.FB)
    assert N.SPRITES.icon("notepad") is N.SPRITES.icon("notepad")


def test_window_frame_dimensions():
    f = N.SPRITES.window_frame(200, 150, "Test")
    assert f.size == (200, 150)


# --------------------------------------------------------------------------- #
# kernel / hit testing
# --------------------------------------------------------------------------- #

def test_icon_dblclick_opens_window():
    k = N.KERNEL
    x, y, w, h = k.icon_rect(0)
    assert k.hit(x + 5, y + 5)[0] == "icon"
    win = k.open(k.desk_icons[0][0])
    assert win is not None and len(k.windows) == 1
    assert k.focused == win.pid


def test_start_button_toggles_and_menu_hit():
    k = N.KERNEL
    sx, sy, sw, sh = k.start_button_rect()
    assert k.hit(sx + 5, sy + 5)[0] == "start"
    k.menu_open = True
    mx, my, mw, mh = k.menu_rect()
    assert k.hit(mx + 100, my + 30)[0] in ("menu", "menu_item")


def test_close_button_hit_region():
    k = N.KERNEL
    win = k.open("notepad")
    cx, cy, cw, ch = win.ctrl_rects()["close"]
    kind, obj, extra = k.hit(cx + 2, cy + 2)
    assert kind == "ctrl" and extra == "close" and obj.pid == win.pid


def test_window_move_clamped():
    k = N.KERNEL
    win = k.open("cmd")
    k.move(win.pid, -9999, 9999)
    assert win.x <= N.FB and win.y <= N.FB
    assert win.x + win.w >= 0


def test_taskbar_button_hit():
    k = N.KERNEL
    win = k.open("paint")
    bx, by, bw, bh = k.taskbar_button_rect(0)
    kind, obj, _ = k.hit(bx + 5, by + 5)
    assert kind == "taskbar_btn" and obj.pid == win.pid


# --------------------------------------------------------------------------- #
# Text2OS
# --------------------------------------------------------------------------- #

def test_keyword_intents():
    ops = N.keyword_intents("please open notepad and the browser")
    apps = [o.get("app") for o in ops if o["op"] == "open"]
    assert "notepad" in apps and "browser" in apps


def test_parse_intents_falls_back_without_llm():
    # mock engine returns "" for text2text -> deterministic keyword path
    ops = N.parse_intents("launch cmd")
    assert any(o["op"] == "open" and o.get("app") == "cmd" for o in ops)


# --------------------------------------------------------------------------- #
# LASM
# --------------------------------------------------------------------------- #

def test_lasm_ops():
    N.COMPOSITOR.build()
    vm = N.LatentVM(N.COMPOSITOR.stack, N.SPRITES)
    logs = vm.run([
        ["LFILL", "bg", [0, 0, 0, 255]],
        ["LSET", "panel", 10, 10, "a test panel", 128, 128],
        ["LCOPY", "panel", "copy", 200, 200],
        ["LNOISE", "panel", 8],
    ])
    assert any(l.startswith("LFILL") for l in logs)
    assert any(l.startswith("LSET") for l in logs)
    assert any(l.startswith("LCOPY") for l in logs)
    assert not any(l.startswith("ERR") for l in logs)


# --------------------------------------------------------------------------- #
# protocol
# --------------------------------------------------------------------------- #

def test_pack_roundtrip():
    blob = struct.pack(">BB I", N.MAGIC, N.T_FULL, 0) + b""
    payload = b"\x00\x01\x02"
    msg = N.pack(N.T_PATCH, {"x": 10, "y": 20}, payload)
    assert msg[0] == N.MAGIC
    assert msg[1] == N.T_PATCH
    hlen = struct.unpack(">I", msg[2:6])[0]
    header = json.loads(msg[6:6 + hlen])
    assert header == {"x": 10, "y": 20}
    assert msg[6 + hlen:] == payload


# --------------------------------------------------------------------------- #
# render pipeline: full -> patch -> idle
# --------------------------------------------------------------------------- #

def test_render_full_then_patch_then_idle():
    c = N.COMPOSITOR
    kind, _, payload = c.render()
    assert kind == "full" and len(payload) > 0
    assert c.last is not None

    # open a window -> some tiles change -> patch or full, never "none"
    N.KERNEL.open("notepad")
    c.stack.mark_all_dirty()
    kind2, patches, _ = c.render()
    assert kind2 in ("full", "patch")

    # nothing changed -> idle, nothing sent
    kind3, _, _ = c.render()
    assert kind3 == "none"


def test_patch_smaller_than_full():
    """A cursor move must ship far less data than a full 1024x1024 frame."""
    c = N.COMPOSITOR
    kind, _, full = c.render()
    assert kind == "full"
    nbytes_full = len(full)

    N.KERNEL.mouse = (500, 500)
    c.render()                       # consume the cursor-move dirty tiles
    N.KERNEL.mouse = (520, 520)
    kind2, patches, _ = c.render()
    if kind2 == "patch":
        nbytes_patch = sum(len(p[4]) for p in patches)
        assert nbytes_patch < nbytes_full


def test_metrics_are_populated():
    N.COMPOSITOR.render()
    m = N.COMPOSITOR.metrics
    assert m["frames"] >= 1
    assert m["build_ms"] >= 0 and m["frame_ms"] >= 0