bfe-core1106-sdk/qemu/tests/test_qmp_drive.py

#!/usr/bin/env python3
"""Offline tests for qmp.py's drive(): the QMP socket and the control channel
are faked, so this runs in well under two seconds with no VM.

What is worth pinning is the contract the docstring makes: every step gets a
results.jsonl row of ok / fail / fatal and the run CONTINUES, so one run
reports every broken expectation. The region verbs are where that was once
false (flare-edge issue #147): a reference whose box did not fit the
screendump raised out of drive() as a traceback, and a tolerance on the line
was never compared with the one the reference was captured under.

    python3 test_qmp_drive.py
"""
import json
import os
import sys
import tempfile
import time
import unittest

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import qmp  # noqa: E402

SIZE = 64


def ppm_bytes(fill=0):
    return b"P6\n%d %d\n255\n" % (SIZE, SIZE) + bytes([fill]) * (SIZE * SIZE * 3)


class FakeCtl:
    """Canned replies in the shapes warden_debug.c actually produces."""

    def __init__(self, path, timeout=15.0):
        self.path = path

    def send(self, cmd):
        if cmd == "page":
            return "Demo/Rows"
        if cmd == "stats":
            return "page: Demo/Rows\ncpu: 12%\nfps: 10\nrender: 3.20 ms/frame\nrga: 0%\nidle: 0"
        if cmd == "wake":
            return "wake: ok"
        if cmd.startswith("@cat "):
            return json.dumps({"a": {"b": 1}, "list": [1, 2], "name": "warden"})
        if cmd.startswith("hit "):
            return 'hit 47,676: obj text="" box=12,640,72x72'
        if cmd.startswith("nav "):
            return cmd + ": ok"
        return ""


def fake_rpc(sock, sock_file, obj):
    if obj.get("execute") == "screendump":
        with open(obj["arguments"]["filename"], "wb") as fh:
            fh.write(ppm_bytes())
    return {}


def run_script(text, refs=None):
    """-> (exit code or None, {cmd: row}, rows) for one drive() over TEXT."""
    outdir = tempfile.mkdtemp(prefix="qmpdrive.")
    script = os.path.join(outdir, "s.txt")
    with open(script, "w") as fh:
        fh.write(text)
    refs_path = os.path.join(outdir, "refs.json")
    if refs is not None:
        with open(refs_path, "w") as fh:
            json.dump(refs, fh)
    saved = qmp.rpc, qmp.Ctl
    qmp.rpc, qmp.Ctl = fake_rpc, FakeCtl
    rc = None
    try:
        try:
            qmp.drive(None, None, script, outdir, SIZE, ctl_path="fake",
                      console_path=None, refs_path=refs_path)
        except SystemExit as e:
            rc = e.code
    finally:
        qmp.rpc, qmp.Ctl = saved
    with open(os.path.join(outdir, "results.jsonl")) as fh:
        rows = [json.loads(line) for line in fh if line.strip()]
    return rc, {r["cmd"]: r for r in rows}, rows


class PureHelpers(unittest.TestCase):
    def test_every_pixel_round_trips_through_lvgl_calibration(self):
        # lv_evdev.c _evdev_calibrate: px = axis * (width - 1) / AXIS_MAX,
        # integer division. A tap requested at px must land at px, for every
        # px, or a hit-confirmed target can be missed by one pixel.
        for size in (720, 480, 1024):
            for px in range(size):
                axis = qmp.to_axis(px, size)
                self.assertTrue(0 <= axis <= qmp.AXIS_MAX)
                back = axis * (size - 1) // qmp.AXIS_MAX
                self.assertEqual(back, px, f"size {size}: px {px} -> axis {axis} -> {back}")

    def test_resolve_path_and_ops(self):
        doc = {"a": {"b": [5, 6]}, "s": "connected"}
        self.assertEqual(qmp.resolve_path(doc, "a.b[1]"), (6, None))
        self.assertIsNotNone(qmp.resolve_path(doc, "a.c")[1])
        self.assertTrue(qmp.apply_op("eq", 1, "1"))
        self.assertTrue(qmp.apply_op("ne", 1, "2"))
        self.assertTrue(qmp.apply_op("contains", "connected", "nect"))
        self.assertTrue(qmp.apply_op("len_ge", [1, 2], "2"))
        self.assertTrue(qmp.apply_op("len_eq", [1, 2], "2"))
        self.assertTrue(qmp.apply_op("gt", 3.0, "2"))
        self.assertTrue(qmp.apply_op("lt", 1, "2"))
        self.assertTrue(qmp.apply_op("eq", "connected", "connected"))

    def test_parse_stats(self):
        got = qmp.parse_stats(FakeCtl("x").send("stats"))
        self.assertEqual(got, {"cpu": 12.0, "fps": 10.0, "render": 3.2, "idle": 0.0})


class DriveVerbs(unittest.TestCase):
    def test_every_channel_passes_on_a_healthy_ui(self):
        rc, by, rows = run_script(
            "wake\n"
            "assert_page Demo/Rows\n"
            "assert_hit 47 676 obj box=12,640,72x72\n"
            "assert_stat idle eq 0\n"
            "wait_json a.b eq 1 2\n"
            "wait_json list len_ge 2 2\n"
            "assert_json name eq warden\n"
            "assert_stat fps gt 0\n"
            "nav Demo/Rows\n"
            "capture_region r1 0 0 8 8 exact\n"
            "assert_region r1 exact\n"
            "assert_region r1\n"
        )
        self.assertIsNone(rc, [r for r in rows if r["status"] != "ok"])
        self.assertEqual(len(rows), 12)
        self.assertTrue(all(r["status"] == "ok" for r in rows))

    def test_mismatches_are_fails_not_stops(self):
        t0 = time.monotonic()
        rc, by, rows = run_script(
            "assert_page Demo/Other\n"
            "assert_hit 47 676 obj box=0,0,1x1\n"
            "wait_json a.b eq 2 1\n"
            "assert_json a.zz eq 1\n"
            "assert_stat fps lt 0\n"
            "assert_page Demo/Rows\n"
        )
        self.assertEqual(rc, 1)
        self.assertEqual([r["status"] for r in rows],
                         ["fail", "fail", "fail", "fail", "fail", "ok"])
        self.assertIn("moved", by["assert_hit 47 676 obj box=0,0,1x1"]["detail"])
        self.assertGreaterEqual(time.monotonic() - t0, 1.0, "wait_json must honour its timeout")

    def test_region_faults_are_per_step_fatal(self):
        # A pre-seeded reference whose box does not fit a 64x64 screendump,
        # a name with no reference, a tolerance other than the captured one,
        # and a capture box off the screen: each is FATAL for its own step
        # and the assert_page after them still runs.
        refs = {"big": {"x": 10, "y": 10, "w": 1000, "h": 1000, "tolerance": "exact",
                        "phash": "0" * 16, "structural": "00" * 32}}
        rc, by, rows = run_script(
            "assert_region big exact\n"
            "assert_region nope exact\n"
            "capture_region r1 0 0 8 8 exact\n"
            "assert_region r1 loose\n"
            "capture_region huge 0 0 999 999 exact\n"
            "capture_region r2 0 0 8 8 fuzzy\n"
            "assert_page Demo/Rows\n",
            refs=refs,
        )
        self.assertEqual(rc, 1)
        self.assertEqual([r["status"] for r in rows],
                         ["fatal", "fatal", "ok", "fatal", "fatal", "fail", "ok"])
        self.assertIn("unusable", by["assert_region big exact"]["detail"])
        self.assertIn("no reference", by["assert_region nope exact"]["detail"])
        self.assertIn("captured as exact, script expects loose",
                      by["assert_region r1 loose"]["detail"])
        self.assertIn("cannot capture", by["capture_region huge 0 0 999 999 exact"]["detail"])
        self.assertIn("unknown tolerance", by["capture_region r2 0 0 8 8 fuzzy"]["detail"])

    def test_unknown_verb_is_fatal_for_the_run(self):
        # A silently-ignored line is a test that proves nothing, so this one
        # is the documented exception to "the run continues": drive() exits
        # with the message rather than recording a row.
        rc, by, rows = run_script("assert_page Demo/Rows\nfrobnicate 1 2\n")
        self.assertIsInstance(rc, str)
        self.assertIn("unknown command 'frobnicate'", rc)
        self.assertEqual([r["status"] for r in rows], ["ok"])


if __name__ == "__main__":
    unittest.main(verbosity=1)