Add conformance suite; client: raw PPM screenshot + PPM parser 8ac19619e24b

# Conformance suite

pytest tests that assert an emulator correctly implements the control contract

(`../SPEC.md`, 0.1.x). Point them at any emulator — the reference x16emu, FAB,

the Neo6502 — and they check the same surface. Platform-agnostic: framebuffer

dimensions come from the PPM header, addresses use `$0000`, and `bank` is

exercised without requiring the platform to have banking.

## Run

Attach to an already-running emulator:

```

RRDC_PORT=8080 pytest -v

```

Or let the suite launch one (`{port}` is substituted into the command):

```

RRDC_EMU_CMD="SDL_VIDEODRIVER=dummy SDL_AUDIODRIVER=dummy \

    /path/to/x16emu -rom rom.bin -warp -controlport {port}" \

RRDC_PORT=8391 pytest -v

```

If nothing is listening and no `RRDC_EMU_CMD` is set, the tests **skip** (a bare

`pytest` in CI without an emulator is a skip, not a failure).

## What it checks

- `/status` — required keys present, contract MAJOR matches, sane types.

- `/regs` — a non-empty JSON object of integer registers.

- `/mem` — returns exactly `len` bytes; honours `bank`.

- `/screenshot` — a well-formed PPM whose body size matches its dimensions.

- `/step` — deterministic: equal steps advance the frame counter by equal,

  positive amounts, and `/step` returns the halted counter.

- `/pause` + `/resume` — toggle the paused state.

- Errors — unknown path → 404; `GET /step` → 405.

## Environment

| Var | Default | Meaning |

|---|---|---|

| `RRDC_PORT` | `8080` | control port to attach to / launch on |

| `RRDC_EMU_CMD` | — | shell command to launch an emulator (`{port}` substituted) |

| `RRDC_BOOT_TIMEOUT` | `20` | seconds to wait for `/status` |

Requires `pytest`. Nothing else — the client is stdlib-only (Pillow is optional

and unused here).

"""Conformance harness fixtures.

Provides an `emu` fixture (an EmuControl client). Two modes:

- Attach: point RRDC_PORT at an already-running emulator's control port.

- Launch: set RRDC_EMU_CMD to a shell command that starts one (the literal

  {port} is substituted). The suite spawns it, waits for /status, and tears it

  down afterward.

If nothing is listening and no RRDC_EMU_CMD is set, the tests SKIP (so a bare

`pytest` in CI without an emulator is a skip, not a failure).

"""

import os

import signal

import subprocess

import sys

import pytest

HERE = os.path.dirname(os.path.abspath(__file__))

sys.path.insert(0, os.path.join(HERE, "..", "client"))

from emu_control import EmuControl  # noqa: E402

def _kill(proc):

    for sig in (signal.SIGTERM, signal.SIGKILL):

        try:

            os.killpg(os.getpgid(proc.pid), sig)

            proc.wait(timeout=5)

            return

        except Exception:

            continue

@pytest.fixture(scope="session")

def emu():

    port = int(os.environ.get("RRDC_PORT", "8080"))

    cmd = os.environ.get("RRDC_EMU_CMD")

    proc = None

    if cmd:

        proc = subprocess.Popen(

            cmd.replace("{port}", str(port)), shell=True, start_new_session=True,

            stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)

    client = EmuControl(port)

    try:

        timeout = float(os.environ.get("RRDC_BOOT_TIMEOUT", "20"))

        client.wait_ready(tries=max(1, int(timeout / 0.1)), delay=0.1)

    except Exception as ex:

        if proc is not None:

            _kill(proc)

        pytest.skip(f"no emulator on control port {port}: {ex} "

                    f"(set RRDC_EMU_CMD to launch one, or RRDC_PORT to attach)")

    yield client

    if proc is not None:

        _kill(proc)

"""Contract conformance tests (SPEC.md 0.1.x). Platform-agnostic: framebuffer

dimensions come from the PPM header, addresses use $0000, and `bank` is

exercised but banking is not required."""

import urllib.error

import urllib.request

from emu_control import parse_ppm_header

def _raw(emu, method, path):

    """Raw request returning (status_code, body) without the client's wrapping,

    so we can assert on 4xx status codes."""

    req = urllib.request.Request(emu.base + path, method=method)

    try:

        with urllib.request.urlopen(req, timeout=emu.timeout) as r:

            return r.status, r.read()

    except urllib.error.HTTPError as e:

        return e.code, e.read()

def test_status_contract(emu):

    st = emu.status()

    for k in ("contract", "emulator", "platform", "frame", "paused", "running"):

        assert k in st, f"missing /status.{k}"

    assert str(st["contract"]).split(".")[0] == "0", "unexpected contract MAJOR"

    assert isinstance(st["platform"], str) and st["platform"]

    assert isinstance(st["frame"], int)

    assert isinstance(st["paused"], bool)

def test_regs_object(emu):

    r = emu.regs()

    assert isinstance(r, dict) and r, "regs must be a non-empty JSON object"

    assert all(isinstance(v, int) for v in r.values()), "register values must be ints"

def test_mem_exact_length(emu):

    assert len(emu.mem(0x0000, 32)) == 32

    assert len(emu.mem(0x0000, 256)) == 256

    assert len(emu.mem(0x0000, 0)) == 0

def test_mem_bank_param_returns_length(emu):

    # bank is optional (ignored on flat platforms) but must still honour len.

    assert len(emu.mem(0x0000, 8, bank=0)) == 8

def test_screenshot_ppm(emu):

    data = emu.screenshot_ppm()

    w, h, maxval, off = parse_ppm_header(data)

    assert w > 0 and h > 0, f"bad dimensions {w}x{h}"

    assert maxval == 255

    assert len(data) == off + 3 * w * h, "PPM body size mismatches dimensions"

def test_step_is_deterministic(emu):

    emu.pause()

    try:

        a = emu.frame()

        b = emu.step(2); c = emu.frame()

        assert b == c, "/step must return the halted frame counter"

        d = emu.step(2); e = emu.frame()

        assert d == e

        assert c - a > 0, "/step must advance the frame counter"

        assert (c - a) == (e - c), "equal steps must advance by equal amounts"

    finally:

        emu.resume()

def test_pause_resume_toggle(emu):

    emu.pause()

    assert emu.status()["paused"] is True

    emu.resume()

    assert emu.status()["paused"] is False

def test_unknown_path_404(emu):

    code, _ = _raw(emu, "GET", "/no-such-endpoint")

    assert code == 404

def test_step_requires_post(emu):

    code, _ = _raw(emu, "GET", "/step?frames=1")

    assert code == 405

def parse_ppm_header(data: bytes):

    """Parse a binary PPM (P6) header. Returns (width, height, maxval, offset)

    where offset is the index of the first pixel byte. Raises on malformed

    input. Stdlib-only — no Pillow needed."""

    if data[:2] != b"P6":

        raise EmuControlError(f"not a P6 PPM (magic {data[:2]!r})")

    fields, i, n = [], 2, len(data)

    while len(fields) < 3 and i < n:

        while i < n and data[i:i + 1].isspace():

            i += 1

        if i < n and data[i:i + 1] == b"#":            # comment to end of line

            while i < n and data[i:i + 1] not in (b"\n", b"\r"):

                i += 1

            continue

        start = i

        while i < n and not data[i:i + 1].isspace():

            i += 1

        fields.append(int(data[start:i]))

    if len(fields) < 3:

        raise EmuControlError("truncated PPM header")

    return fields[0], fields[1], fields[2], i + 1        # +1: single whitespace

    def screenshot_ppm(self) -> bytes:

        """Return the live screen as raw PPM (P6) bytes. Stdlib-only."""

        body, _ = self._get("/screenshot")

        return body

        body = self.screenshot_ppm()