mirror of
https://github.com/esphome/esphome.git
synced 2026-10-02 01:10:22 +00:00
[snapshot][SDL] Display headless mode and snapshots (#17917)
Co-authored-by: Claude Opus 5 <noreply@anthropic.com> Co-authored-by: Jesse Hills <3060199+jesserockz@users.noreply.github.com>
This commit is contained in:
co-authored by
Claude Opus 5
Jesse Hills
parent
567f7f9196
commit
f0e2eb96bd
@@ -0,0 +1,101 @@
|
||||
"""Tests for the sdl display schema, in particular the headless option."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from esphome import config_validation as cv
|
||||
from esphome.components.sdl.display import (
|
||||
CONF_SDL_ID,
|
||||
CONFIG_SCHEMA,
|
||||
headless_final_validate,
|
||||
)
|
||||
from esphome.config import Config
|
||||
from esphome.const import PlatformFramework
|
||||
from esphome.core import ID
|
||||
from esphome.final_validate import full_config
|
||||
from esphome.types import ConfigType
|
||||
from tests.component_tests.types import SetCoreConfigCallable
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _host_platform(set_core_config: SetCoreConfigCallable) -> None:
|
||||
set_core_config(PlatformFramework.HOST_NATIVE)
|
||||
|
||||
|
||||
def _config(**extra: object) -> ConfigType:
|
||||
config: ConfigType = {
|
||||
"dimensions": {"width": 320, "height": 240},
|
||||
# sdl2-config is not necessarily installed in the test environment
|
||||
"sdl_options": "-lSDL2",
|
||||
}
|
||||
config.update(extra)
|
||||
return config
|
||||
|
||||
|
||||
def test_defaults_to_windowed() -> None:
|
||||
"""A display without the option is not headless."""
|
||||
assert CONFIG_SCHEMA(_config())["headless"] is False
|
||||
|
||||
|
||||
def test_headless_accepted() -> None:
|
||||
"""A headless display needs nothing beyond the dimensions."""
|
||||
assert CONFIG_SCHEMA(_config(headless=True))["headless"] is True
|
||||
|
||||
|
||||
def test_headless_rejects_window_options() -> None:
|
||||
"""Window options are meaningless without a window."""
|
||||
with pytest.raises(cv.Invalid, match="has no effect"):
|
||||
CONFIG_SCHEMA(
|
||||
_config(headless=True, window_options={"position": {"x": 0, "y": 0}})
|
||||
)
|
||||
|
||||
|
||||
def test_headless_rejects_snapshot_key() -> None:
|
||||
"""A headless display has no keyboard, so the action is the only way in."""
|
||||
with pytest.raises(cv.Invalid, match="snapshot.take"):
|
||||
CONFIG_SCHEMA(_config(headless=True, snapshot_key="SDLK_F12"))
|
||||
|
||||
|
||||
def test_snapshot_key_accepted_when_windowed() -> None:
|
||||
"""The key is only valid alongside a window."""
|
||||
config = CONFIG_SCHEMA(_config(snapshot_key="SDLK_F12"))
|
||||
assert str(config["snapshot_key"]) == "SDLK_F12"
|
||||
|
||||
|
||||
def _declare_sdl_display(headless: bool) -> ID:
|
||||
"""Register a full_config with a single sdl display declaration and return a reference to it.
|
||||
|
||||
Mirrors what the real config pipeline leaves behind: a "display" domain entry plus a
|
||||
declare_ids record id_declaration_match_schema uses to find it again.
|
||||
"""
|
||||
declared_id = ID("my_sdl", is_declaration=True)
|
||||
fc = Config()
|
||||
fc["display"] = [
|
||||
{
|
||||
"platform": "sdl",
|
||||
"id": declared_id,
|
||||
"headless": headless,
|
||||
"dimensions": {"width": 320, "height": 240},
|
||||
}
|
||||
]
|
||||
fc.declare_ids.append((declared_id, ["display", 0, "id"]))
|
||||
full_config.set(fc)
|
||||
return ID("my_sdl")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("platform", ["binary_sensor", "touchscreen"])
|
||||
def test_headless_final_validate_rejects_headless_display(platform: str) -> None:
|
||||
"""binary_sensor and touchscreen both need a window, so a headless display is rejected."""
|
||||
sdl_ref = _declare_sdl_display(headless=True)
|
||||
schema = headless_final_validate(platform)
|
||||
with pytest.raises(cv.Invalid, match="needs a window"):
|
||||
schema({CONF_SDL_ID: sdl_ref})
|
||||
|
||||
|
||||
@pytest.mark.parametrize("platform", ["binary_sensor", "touchscreen"])
|
||||
def test_headless_final_validate_accepts_windowed_display(platform: str) -> None:
|
||||
"""The same platforms are accepted once the display has a window."""
|
||||
sdl_ref = _declare_sdl_display(headless=False)
|
||||
schema = headless_final_validate(platform)
|
||||
schema({CONF_SDL_ID: sdl_ref}) # Should not raise.
|
||||
@@ -14,6 +14,15 @@ display:
|
||||
position:
|
||||
x: 100
|
||||
y: 100
|
||||
snapshot_key: SDLK_F12
|
||||
|
||||
- platform: sdl
|
||||
id: headless_display
|
||||
headless: true
|
||||
show_test_card: true
|
||||
dimensions:
|
||||
width: 320
|
||||
height: 240
|
||||
|
||||
- platform: sdl
|
||||
id: second_display
|
||||
@@ -46,3 +55,21 @@ binary_sensor:
|
||||
sdl_id: sdl_sdl_display
|
||||
id: key_enter
|
||||
key: SDLK_RETURN
|
||||
|
||||
esphome:
|
||||
# A name of your own is only good for one snapshot - a second one under the same name fails
|
||||
# rather than writing over the first - so these run once rather than on a repeating interval.
|
||||
on_boot:
|
||||
- delay: 2s
|
||||
- snapshot.take:
|
||||
id: headless_display
|
||||
filename: test_card.bmp
|
||||
- snapshot.take:
|
||||
id: headless_display
|
||||
filename: !lambda 'return "shot.bmp";'
|
||||
|
||||
interval:
|
||||
# A generated name has the time in it, so this one can repeat.
|
||||
- interval: 10s
|
||||
then:
|
||||
- snapshot.take: sdl_sdl_display
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
# Config-only test for the headless and screenshot options. The combinations that must be
|
||||
# rejected are covered by tests/component_tests/sdl/test_sdl.py; this file checks that the
|
||||
# accepted forms validate together.
|
||||
host:
|
||||
mac_address: "62:23:45:AF:B3:DD"
|
||||
|
||||
display:
|
||||
- platform: sdl
|
||||
id: headless_display
|
||||
headless: true
|
||||
dimensions: 320x240
|
||||
|
||||
- platform: sdl
|
||||
id: windowed_display
|
||||
dimensions: 320x240
|
||||
snapshot_key: SDLK_F12
|
||||
|
||||
binary_sensor:
|
||||
- platform: sdl
|
||||
sdl_id: windowed_display
|
||||
id: key_up
|
||||
key: SDLK_UP
|
||||
|
||||
interval:
|
||||
- interval: 10s
|
||||
then:
|
||||
- snapshot.take:
|
||||
id: headless_display
|
||||
filename: periodic.bmp
|
||||
@@ -0,0 +1,34 @@
|
||||
display:
|
||||
- platform: snapshot
|
||||
id: snapshot_display
|
||||
update_interval: 1s
|
||||
show_test_card: true
|
||||
# An odd width exercises the row padding in the BMP writer
|
||||
dimensions:
|
||||
width: 101
|
||||
height: 64
|
||||
|
||||
- platform: snapshot
|
||||
id: snapshot_rotated
|
||||
rotation: 90
|
||||
dimensions: 320x240
|
||||
lambda: |-
|
||||
it.filled_rectangle(0, 0, 40, 20, Color(0xFF, 0x80, 0x00));
|
||||
|
||||
esphome:
|
||||
# A name of your own is only good for one snapshot - a second one under the same name fails
|
||||
# rather than writing over the first - so these run once rather than on a repeating interval.
|
||||
on_boot:
|
||||
- delay: 2s
|
||||
- snapshot.take:
|
||||
id: snapshot_display
|
||||
filename: test_card.bmp
|
||||
- snapshot.take:
|
||||
id: snapshot_rotated
|
||||
filename: !lambda 'return "rotated.bmp";'
|
||||
|
||||
interval:
|
||||
# A generated name has the time in it, so this one can repeat.
|
||||
- interval: 10s
|
||||
then:
|
||||
- snapshot.take: snapshot_display
|
||||
@@ -0,0 +1,5 @@
|
||||
host:
|
||||
mac_address: "62:23:45:AF:B3:DE"
|
||||
|
||||
packages:
|
||||
snapshot: !include common.yaml
|
||||
@@ -0,0 +1,26 @@
|
||||
"""Shared utilities for ESPHome integration tests - keeping output from failing tests."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
#: Where a failing test leaves output for someone to look at afterwards. pytest's own
|
||||
#: temporary folder is no use on a CI runner, which throws the whole workspace away when
|
||||
#: the job ends; the workflow uploads this folder instead when a job fails.
|
||||
ARTIFACT_DIR = Path(__file__).resolve().parents[2] / "test_artifacts"
|
||||
|
||||
|
||||
def keep_artifact(name: str, data: bytes) -> Path:
|
||||
"""Write ``data`` where it can still be read after the run, and return the path.
|
||||
|
||||
Args:
|
||||
name: File name to write under the artifact folder.
|
||||
data: Contents to write.
|
||||
|
||||
Returns:
|
||||
The full path written.
|
||||
"""
|
||||
ARTIFACT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
path = ARTIFACT_DIR / name
|
||||
path.write_bytes(data)
|
||||
return path
|
||||
@@ -0,0 +1,161 @@
|
||||
"""Shared utilities for ESPHome integration tests - reading BMP snapshots."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from collections.abc import Awaitable, Callable
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
import struct
|
||||
|
||||
# Size of the smallest BMP header pair (file header plus BITMAPINFOHEADER).
|
||||
_MIN_HEADER_SIZE = 54
|
||||
|
||||
# How long capture_when_drawn() keeps asking for a picture with something on it.
|
||||
DRAW_TIMEOUT = 15.0
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Bmp:
|
||||
"""A decoded BMP image."""
|
||||
|
||||
width: int
|
||||
height: int
|
||||
bits: int
|
||||
#: Pixel data with the per row padding stripped, so it depends only on the image itself.
|
||||
pixels: bytes
|
||||
|
||||
|
||||
class NotABmpError(Exception):
|
||||
"""The data is not a BMP at all, as opposed to a BMP that is still being written."""
|
||||
|
||||
|
||||
def parse_bmp(data: bytes) -> Bmp | None:
|
||||
"""Decode a BMP, or return None if the data is not a complete image yet.
|
||||
|
||||
Raises:
|
||||
NotABmpError: If the data cannot become a valid BMP however much more is appended.
|
||||
"""
|
||||
# Writes go to the file in order, so a short read is always a prefix of what will be there.
|
||||
# Anything wrong in a prefix we have already read is wrong for good, and worth saying now
|
||||
# rather than reporting as a timeout later.
|
||||
if len(data) >= 2 and data[:2] != b"BM":
|
||||
raise NotABmpError(f"expected a BMP, got {data[:2]!r}")
|
||||
if len(data) < _MIN_HEADER_SIZE:
|
||||
return None
|
||||
file_size = struct.unpack_from("<I", data, 2)[0]
|
||||
offset = struct.unpack_from("<I", data, 10)[0]
|
||||
width, height = struct.unpack_from("<ii", data, 18)
|
||||
bits = struct.unpack_from("<H", data, 28)[0]
|
||||
rows = abs(height)
|
||||
row_size = ((width * bits + 31) // 32) * 4
|
||||
if width <= 0 or rows == 0 or bits == 0 or offset < _MIN_HEADER_SIZE:
|
||||
raise NotABmpError(
|
||||
f"BMP header makes no sense: {width}x{height}, {bits} bits, "
|
||||
f"pixels at offset {offset}"
|
||||
)
|
||||
if file_size < offset + row_size * rows:
|
||||
raise NotABmpError(
|
||||
f"BMP header claims {file_size} bytes, too few for {width}x{rows} "
|
||||
f"at {bits} bits"
|
||||
)
|
||||
if len(data) < file_size:
|
||||
return None
|
||||
used = width * bits // 8
|
||||
pixels = b"".join(
|
||||
data[offset + row * row_size : offset + row * row_size + used]
|
||||
for row in range(rows)
|
||||
)
|
||||
return Bmp(width=width, height=rows, bits=bits, pixels=pixels)
|
||||
|
||||
|
||||
async def wait_for_bmp(path: Path, timeout: float = 5.0) -> Bmp:
|
||||
"""Wait for a complete BMP file to appear at ``path`` and return it.
|
||||
|
||||
The file is created before any of its contents are written, so waiting for it to exist is
|
||||
not enough - a read that wins the race sees a truncated image. Keep reading until the
|
||||
headers say the whole image is there.
|
||||
|
||||
Args:
|
||||
path: The file to wait for.
|
||||
timeout: Maximum time to wait in seconds.
|
||||
|
||||
Returns:
|
||||
The decoded image.
|
||||
|
||||
Raises:
|
||||
AssertionError: If no complete image is readable within ``timeout``.
|
||||
NotABmpError: If what was written is not a BMP. This is reported as soon as it is
|
||||
seen, so a device that writes the wrong thing is named for what it did rather
|
||||
than waiting out the timeout.
|
||||
"""
|
||||
loop = asyncio.get_running_loop()
|
||||
deadline = loop.time() + timeout
|
||||
while True:
|
||||
try:
|
||||
data = path.read_bytes()
|
||||
except FileNotFoundError:
|
||||
data = b""
|
||||
if (image := parse_bmp(data)) is not None:
|
||||
return image
|
||||
if loop.time() >= deadline:
|
||||
break
|
||||
await asyncio.sleep(0.05)
|
||||
if not data:
|
||||
raise AssertionError(f"no snapshot appeared at {path} within {timeout}s")
|
||||
raise AssertionError(
|
||||
f"{path} was still incomplete after {timeout}s ({len(data)} bytes)"
|
||||
)
|
||||
|
||||
|
||||
def is_blank(image: Bmp) -> bool:
|
||||
"""True if every pixel of the image is the same colour.
|
||||
|
||||
Whole pixels are counted rather than byte values: a plain background is usually made of more
|
||||
than one distinct byte, so counting bytes would find several of them in a blank screen.
|
||||
"""
|
||||
return len({image.pixels[i : i + 3] for i in range(0, len(image.pixels), 3)}) <= 1
|
||||
|
||||
|
||||
async def capture_when_drawn(
|
||||
take: Callable[[str], Awaitable[None]],
|
||||
directory: Path,
|
||||
prefix: str = "drawn",
|
||||
timeout: float = DRAW_TIMEOUT,
|
||||
) -> tuple[Bmp, Path]:
|
||||
"""Ask for snapshots until one has something drawn on it, and return it and where it went.
|
||||
|
||||
A display holds one flat colour until it first draws, which is one update interval after it
|
||||
starts - long enough that a test connecting over the API can easily get in first. Capturing
|
||||
once and hoping would compare a blank screen against whatever the test expects, reporting a
|
||||
drawing fault where the real trouble was timing.
|
||||
|
||||
Args:
|
||||
take: Asks the device for a snapshot under the name it is given.
|
||||
directory: Where the device writes them.
|
||||
prefix: Start of the names asked for. Each attempt needs its own, because a snapshot never
|
||||
writes over a file that is already there.
|
||||
timeout: How long to keep asking.
|
||||
|
||||
Returns:
|
||||
The first image that is not one flat colour, and the path it was read from.
|
||||
|
||||
Raises:
|
||||
AssertionError: If nothing had been drawn within ``timeout``.
|
||||
"""
|
||||
loop = asyncio.get_running_loop()
|
||||
deadline = loop.time() + timeout
|
||||
attempt = 0
|
||||
while True:
|
||||
attempt += 1
|
||||
path = directory / f"{prefix}-{attempt}.bmp"
|
||||
await take(path.name)
|
||||
image = await wait_for_bmp(path)
|
||||
if not is_blank(image):
|
||||
return image, path
|
||||
if loop.time() >= deadline:
|
||||
raise AssertionError(
|
||||
f"the screen was still a single flat colour after {timeout}s and "
|
||||
f"{attempt} captures - nothing was drawn"
|
||||
)
|
||||
await asyncio.sleep(0.5)
|
||||
@@ -0,0 +1,53 @@
|
||||
esphome:
|
||||
name: lvgl-headless-render-test
|
||||
host:
|
||||
|
||||
api:
|
||||
actions:
|
||||
# The name comes from the test so it can capture more than once: a snapshot never writes over
|
||||
# a file that is already there, so a fixed name could only ever be captured once.
|
||||
- action: take_screenshot
|
||||
variables:
|
||||
name: string
|
||||
then:
|
||||
- snapshot.take:
|
||||
id: lvgl_display
|
||||
filename: !lambda return name;
|
||||
|
||||
logger:
|
||||
level: DEBUG
|
||||
|
||||
display:
|
||||
# A display with no screen, so what LVGL draws depends on LVGL alone - nothing about the machine
|
||||
# running the test, and no graphics library outside this repository, can move the result.
|
||||
- platform: snapshot
|
||||
id: lvgl_display
|
||||
auto_clear_enabled: false
|
||||
dimensions:
|
||||
width: 300
|
||||
height: 300
|
||||
|
||||
# The widgets are spelled out here rather than left to the built in "Hello World" screen, which
|
||||
# LVGL builds when nothing is configured: that screen contains a spinner, and an animation cannot
|
||||
# produce the same picture twice.
|
||||
#
|
||||
# Everything that affects the rendered pixels is set explicitly, so the expected hash in the test
|
||||
# depends only on the drawing code and the built in font. In particular the background comes from a
|
||||
# full screen object rather than from the theme, so adjusting a theme default does not break this.
|
||||
lvgl:
|
||||
displays: lvgl_display
|
||||
default_font: montserrat_14
|
||||
widgets:
|
||||
- obj:
|
||||
width: 100%
|
||||
height: 100%
|
||||
bg_color: 0x000080
|
||||
bg_opa: cover
|
||||
border_width: 0
|
||||
radius: 0
|
||||
pad_all: 0
|
||||
widgets:
|
||||
- label:
|
||||
align: center
|
||||
text: "Hello World!"
|
||||
text_color: 0xFFFFFF
|
||||
@@ -0,0 +1,29 @@
|
||||
esphome:
|
||||
name: sdl-headless-screenshot-test
|
||||
host:
|
||||
|
||||
api:
|
||||
actions:
|
||||
# The name comes from the test so it can capture more than once while it waits for the first
|
||||
# frame: a snapshot never writes over a file that is already there.
|
||||
- action: take_screenshot
|
||||
variables:
|
||||
name: string
|
||||
then:
|
||||
- snapshot.take:
|
||||
id: sdl_display
|
||||
filename: !lambda return name;
|
||||
|
||||
logger:
|
||||
level: DEBUG
|
||||
|
||||
display:
|
||||
- platform: sdl
|
||||
id: sdl_display
|
||||
headless: true
|
||||
show_test_card: true
|
||||
update_interval: 100ms
|
||||
# An odd width exercises the row padding in the BMP writer
|
||||
dimensions:
|
||||
width: 101
|
||||
height: 64
|
||||
@@ -0,0 +1,28 @@
|
||||
esphome:
|
||||
name: snapshot-display-test
|
||||
host:
|
||||
|
||||
api:
|
||||
actions:
|
||||
# The name comes from the test so it can ask for several in a row and check what each one
|
||||
# does with it.
|
||||
- action: take_snapshot
|
||||
variables:
|
||||
name: string
|
||||
then:
|
||||
- snapshot.take:
|
||||
id: snapshot_display
|
||||
filename: !lambda return name;
|
||||
|
||||
logger:
|
||||
level: DEBUG
|
||||
|
||||
display:
|
||||
- platform: snapshot
|
||||
id: snapshot_display
|
||||
show_test_card: true
|
||||
update_interval: 100ms
|
||||
# An odd width exercises the row padding in the BMP writer
|
||||
dimensions:
|
||||
width: 101
|
||||
height: 64
|
||||
@@ -0,0 +1,83 @@
|
||||
"""Integration test that checks what LVGL actually draws, using a display with no screen.
|
||||
|
||||
The rendered screen is compared against a hash rather than a checked in reference image, so the
|
||||
repository does not have to carry a binary file. If a change to the drawing code or to the bundled
|
||||
LVGL alters the output, this test fails and prints the hash it saw; update EXPECTED_SHA256 once the
|
||||
new image has been looked at and found to be correct.
|
||||
|
||||
The picture is drawn and encoded entirely by code in this repository, so nothing installed on the
|
||||
machine running the test takes part in the result.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from .artifact_utils import keep_artifact
|
||||
from .bmp_utils import capture_when_drawn
|
||||
from .types import APIClientConnectedFactory, RunCompiledFunction
|
||||
|
||||
WIDTH = 300
|
||||
HEIGHT = 300
|
||||
|
||||
# sha256 of the pixel data of a 300x300 screen showing "Hello World!" centred in white on a dark
|
||||
# blue background, drawn with the built in montserrat_14 font. To regenerate, run this test and
|
||||
# take the hash it reports.
|
||||
EXPECTED_SHA256 = "a995b002dd1d183c47514da15ab9a60a3e7d788c2e24386a02fddd48655092ed"
|
||||
# Bundled LVGL version (esphome/components/lvgl/__init__.py, LVGL_VERSION) the hash above was
|
||||
# generated against. A version bump can shift anti-aliasing enough to change the hash even though
|
||||
# nothing is actually wrong -- if this test fails, check that first before regenerating the hash.
|
||||
EXPECTED_LVGL_VERSION = "9.5.0"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_lvgl_headless_render(
|
||||
yaml_config: str,
|
||||
run_compiled: RunCompiledFunction,
|
||||
api_client_connected: APIClientConnectedFactory,
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""LVGL draws the expected screen on a 300x300 display with no screen behind it."""
|
||||
snapshot_dir = tmp_path / "snapshots"
|
||||
monkeypatch.setenv("ESPHOME_SNAPSHOT_DIR", str(snapshot_dir))
|
||||
|
||||
async with run_compiled(yaml_config), api_client_connected() as client:
|
||||
_, services = await client.list_entities_services()
|
||||
service = next(s for s in services if s.name == "take_screenshot")
|
||||
|
||||
async def take(name: str) -> None:
|
||||
await client.execute_service(service, {"name": name})
|
||||
|
||||
# The background is not the whole picture: LVGL must have drawn on it. Waiting for that
|
||||
# rather than for a fixed time keeps a slow first frame from being reported as a hash
|
||||
# mismatch, which would look like a drawing regression.
|
||||
image, capture = await capture_when_drawn(take, snapshot_dir, prefix="render")
|
||||
assert (image.width, image.height, image.bits) == (WIDTH, HEIGHT, 24)
|
||||
|
||||
digest = hashlib.sha256(image.pixels).hexdigest()
|
||||
if digest != EXPECTED_SHA256:
|
||||
# Kept outside the temporary folder so CI can upload it; see artifact_utils.
|
||||
kept = keep_artifact(
|
||||
"lvgl_headless_render_actual.bmp", capture.read_bytes()
|
||||
)
|
||||
|
||||
from esphome.components.lvgl import LVGL_VERSION
|
||||
|
||||
version_hint = ""
|
||||
if LVGL_VERSION != EXPECTED_LVGL_VERSION:
|
||||
version_hint = (
|
||||
f"the bundled LVGL version changed ({EXPECTED_LVGL_VERSION} -> "
|
||||
f"{LVGL_VERSION}), which is the likely cause\n"
|
||||
)
|
||||
pytest.fail(
|
||||
f"rendered screen does not match the expected hash\n"
|
||||
f"{version_hint}"
|
||||
f" expected: {EXPECTED_SHA256}\n"
|
||||
f" actual: {digest}\n"
|
||||
f"the image that was rendered has been kept at {kept}\n"
|
||||
f"on CI it is in the integration-test-artifacts upload for this job"
|
||||
)
|
||||
@@ -0,0 +1,49 @@
|
||||
"""Integration test for headless SDL rendering and snapshot capture.
|
||||
|
||||
How a file is named and written is the same for every display that can take a snapshot and is
|
||||
covered by test_snapshot_display; what is tested here is that SDL renders and can be read back
|
||||
with no display server present.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from .bmp_utils import capture_when_drawn
|
||||
from .types import APIClientConnectedFactory, RunCompiledFunction
|
||||
|
||||
WIDTH = 101
|
||||
HEIGHT = 64
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sdl_headless_screenshot(
|
||||
yaml_config: str,
|
||||
run_compiled: RunCompiledFunction,
|
||||
api_client_connected: APIClientConnectedFactory,
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""A headless SDL display renders with no display server and can be captured."""
|
||||
snapshot_dir = tmp_path / "snapshots"
|
||||
# The device reads this when it writes a file; the subprocess inherits our environment, so it
|
||||
# must be set before the binary is launched.
|
||||
monkeypatch.setenv("ESPHOME_SNAPSHOT_DIR", str(snapshot_dir))
|
||||
# Make sure the run really is headless even when the test machine has a display.
|
||||
monkeypatch.delenv("DISPLAY", raising=False)
|
||||
monkeypatch.delenv("WAYLAND_DISPLAY", raising=False)
|
||||
|
||||
async with run_compiled(yaml_config), api_client_connected() as client:
|
||||
_, services = await client.list_entities_services()
|
||||
service = next(s for s in services if s.name == "take_screenshot")
|
||||
|
||||
async def take(name: str) -> None:
|
||||
await client.execute_service(service, {"name": name})
|
||||
|
||||
# The test card is drawn in several colours, so once it is on the screen the picture is
|
||||
# not one flat shade. Capturing until that is true waits out the first update rather than
|
||||
# racing it.
|
||||
image, _ = await capture_when_drawn(take, snapshot_dir)
|
||||
assert (image.width, image.height, image.bits) == (WIDTH, HEIGHT, 24)
|
||||
@@ -0,0 +1,78 @@
|
||||
"""Integration test for the snapshot display and the file writing shared with other displays."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from pathlib import Path
|
||||
|
||||
from aioesphomeapi import LogLevel
|
||||
import pytest
|
||||
|
||||
from .bmp_utils import capture_when_drawn, wait_for_bmp
|
||||
from .types import APIClientConnectedFactory, RunCompiledFunction
|
||||
|
||||
WIDTH = 101
|
||||
HEIGHT = 64
|
||||
|
||||
# Part of the message the writer logs when it will not write over a file that is already there.
|
||||
REFUSAL_MESSAGE = b"not overwriting"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_snapshot_display(
|
||||
yaml_config: str,
|
||||
run_compiled: RunCompiledFunction,
|
||||
api_client_connected: APIClientConnectedFactory,
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""A display with no screen draws into memory and writes what it drew to a file."""
|
||||
snapshot_dir = tmp_path / "snapshots"
|
||||
# The device reads this when it writes a file; the subprocess inherits our environment, so it
|
||||
# must be set before the binary is launched.
|
||||
monkeypatch.setenv("ESPHOME_SNAPSHOT_DIR", str(snapshot_dir))
|
||||
|
||||
async with run_compiled(yaml_config), api_client_connected() as client:
|
||||
_, services = await client.list_entities_services()
|
||||
service = next(s for s in services if s.name == "take_snapshot")
|
||||
|
||||
async def take(name: str) -> None:
|
||||
await client.execute_service(service, {"name": name})
|
||||
|
||||
# The test card is drawn in several colours, so once it is on the screen the picture is
|
||||
# not one flat shade. Capturing until that is true waits out the first update rather than
|
||||
# racing it.
|
||||
image, capture = await capture_when_drawn(take, snapshot_dir)
|
||||
assert (image.width, image.height, image.bits) == (WIDTH, HEIGHT, 24)
|
||||
|
||||
# An extension is only added when there is not one already, whatever its case.
|
||||
await take("UPPER.BMP")
|
||||
await wait_for_bmp(snapshot_dir / "UPPER.BMP")
|
||||
|
||||
# A name that tries to lead somewhere else is cut back to one harmless name in the
|
||||
# snapshot directory.
|
||||
await take("../escape")
|
||||
await wait_for_bmp(snapshot_dir / ".._escape.bmp")
|
||||
|
||||
# A second capture under a name already used must fail rather than write over the first.
|
||||
# Wait for the device to report the refusal: on its own, an unchanged file cannot tell a
|
||||
# refusal apart from a request the device has not got to yet, so a regression that wrote
|
||||
# over the file could still pass on a busy machine.
|
||||
refused = asyncio.Event()
|
||||
|
||||
def on_log(msg) -> None:
|
||||
if REFUSAL_MESSAGE in msg.message:
|
||||
refused.set()
|
||||
|
||||
client.subscribe_logs(on_log, log_level=LogLevel.LOG_LEVEL_DEBUG)
|
||||
|
||||
before = capture.read_bytes()
|
||||
await take(capture.name)
|
||||
await asyncio.wait_for(refused.wait(), timeout=10.0)
|
||||
assert capture.read_bytes() == before
|
||||
# Nothing beyond what was asked for, leaving out however many captures it took to wait
|
||||
# for the first frame.
|
||||
written = sorted(
|
||||
p.name for p in snapshot_dir.iterdir() if not p.name.startswith("drawn-")
|
||||
)
|
||||
assert written == [".._escape.bmp", "UPPER.BMP"]
|
||||
Reference in New Issue
Block a user