diff --git a/src/display_controller.py b/src/display_controller.py index dada2974..a535f477 100644 --- a/src/display_controller.py +++ b/src/display_controller.py @@ -44,6 +44,20 @@ from src.common.sync_manager import DisplaySyncManager, SyncRole # Get logger with consistent configuration logger = get_logger(__name__) +# How long startup will wait for plugins to fetch their first data before +# showing anything. Each plugin's update blocks for up to the executor's 30s +# timeout and they run one after another, so the uncapped total is the sum of +# every slow plugin: 82 seconds on the worst boot measured, with a blank panel +# throughout. Whatever does not finish in time is picked up by the scheduled +# update tick moments later, with the display already running. +_INITIAL_UPDATE_BUDGET_SECONDS = 20.0 + +# The least budget worth starting a plugin with. Below this the plugin is +# deferred instead: granting it a floor would let the pass run past its +# deadline, and granting it the true remainder would record a timeout for a +# slot it never had a chance to use. +_MIN_INITIAL_UPDATE_TIMEOUT_SECONDS = 2.0 + # Vegas mode import (lazy loaded to avoid circular imports) _vegas_mode_imported = False VegasModeCoordinator = None @@ -463,7 +477,7 @@ class DisplayController: # Initial data update for plugins (ensures data available on first display) logger.info("Performing initial plugin data update...") update_start = time.time() - self._update_modules() + self._update_modules(deadline=update_start + _INITIAL_UPDATE_BUDGET_SECONDS) logger.info("Initial plugin update completed in %.3f seconds", time.time() - update_start) # Initialize Vegas mode coordinator @@ -819,14 +833,42 @@ class DisplayController: self._cached_target_brightness = normal_brightness # persist for minute-gate return normal_brightness - def _update_modules(self): - """Update all plugin modules.""" + def _update_modules(self, deadline: Optional[float] = None): + """Update all plugin modules. + + Args: + deadline: Wall-clock time after which remaining plugins are left + for the scheduled update tick instead of being waited on. Each + update blocks this thread for up to the executor's timeout, and + they run one after another, so without a bound the total is the + sum of every slow plugin on the system. Measured at startup on + a live rig: 82 seconds, 55 and 26 on the two boots before -- all + of it with nothing on the panel. + """ if not self.plugin_manager: return - + # Update all loaded plugins plugins_dict = getattr(self.plugin_manager, 'loaded_plugins', None) or getattr(self.plugin_manager, 'plugins', {}) + deferred = [] for plugin_id, plugin_instance in plugins_dict.items(): + update_timeout = None + if deadline is not None: + update_timeout = deadline - time.time() + if update_timeout < _MIN_INITIAL_UPDATE_TIMEOUT_SECONDS: + # Too little left to be worth starting. Deferring rather + # than granting a floor keeps the budget a real ceiling -- + # clamping up to a minimum let a plugin that began with a + # sliver left run on past the deadline -- and a plugin + # handed a slot it cannot use would just be recorded as + # having timed out. + # + # Nothing is lost either way: a plugin that has never + # updated is immediately due, so run_scheduled_updates() + # picks it up within seconds, with the display already + # running. + deferred.append(plugin_id) + continue # Check circuit breaker before attempting update if hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker: if self.plugin_manager.health_tracker.should_skip_plugin(plugin_id): @@ -835,7 +877,13 @@ class DisplayController: # Use PluginExecutor if available for safe execution if hasattr(self.plugin_manager, 'plugin_executor'): - success = self.plugin_manager.plugin_executor.execute_update(plugin_instance, plugin_id) + # The remaining budget is the timeout, so the pass cannot + # run past its deadline. Bounding the loop alone did not do + # it: the last plugin to start could still block for the + # executor's full 30s, which turned a 20s budget into a 31.8s + # pass on the rig. + success = self.plugin_manager.plugin_executor.execute_update( + plugin_instance, plugin_id, timeout=update_timeout) if success and hasattr(self.plugin_manager, 'plugin_last_update'): self.plugin_manager.plugin_last_update[plugin_id] = time.time() else: @@ -854,6 +902,12 @@ class DisplayController: if hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker: self.plugin_manager.health_tracker.record_failure(plugin_id, exc) + if deferred: + logger.info( + "Initial update budget spent; %d plugin(s) left to the update " + "tick so the display can start: %s", + len(deferred), ", ".join(deferred)) + def _tick_plugin_updates_for_vegas(self) -> None: """Run scheduled plugin updates and tell Vegas mode which plugins actually got fresh data, so it can hot-swap them into the scroll diff --git a/src/display_manager.py b/src/display_manager.py index c33e0861..9cc7f622 100644 --- a/src/display_manager.py +++ b/src/display_manager.py @@ -25,6 +25,7 @@ the same object. import json import os +import socket import tempfile if os.getenv("EMULATOR", "false") == "true": from RGBMatrixEmulator import RGBMatrix, RGBMatrixOptions @@ -517,6 +518,91 @@ class DisplayManager: logger.warning(f"[BRIGHTNESS] Matrix does not support brightness property: {e}", exc_info=True) return -1 + @staticmethod + def _local_ip() -> Optional[str]: + """This device's address on the network it routes through, or None. + + Deliberately not `hostname -I` or a systemctl probe for AP mode, which + is how the web launcher does it: both spawn processes with multi-second + timeouts, and this runs on the startup path the rest of this change + exists to shorten. Connecting a UDP socket sends no packets -- it only + asks the kernel which source address it would use -- so it costs + microseconds and works with the network down, as long as a route + exists. + """ + sock = None + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.settimeout(0.2) + sock.connect(("8.8.8.8", 80)) # nosec B104 - no traffic; selects a route + ip = sock.getsockname()[0] + return ip if ip and not ip.startswith("127.") else None + except OSError: + return None + finally: + if sock is not None: + try: + sock.close() + except OSError: + pass + + def _fitting_font(self, lines, width): + """The largest font from the usual ladder that fits every line.""" + candidates = [self.font, + ("assets/fonts/4x6-font.ttf", 6)] + for candidate in candidates: + try: + font = candidate + if isinstance(candidate, tuple): + font = ImageFont.truetype(candidate[0], candidate[1]) + if all(self.draw.textlength(t, font=font) <= width for t in lines): + return font + except (OSError, ValueError, AttributeError): + continue + return self.font + + def _draw_startup_banner(self, lines, width: int, height: int) -> None: + """Centre `lines` over whatever the test pattern already drew. + + This screen stays on the panel for the whole initial plugin update, and + on a headless Pi it is the only place the device's address appears + without going looking for it -- so it has to be readable off a wall, + not merely present. + + The font is chosen to fit rather than fixed at 8px: "Initializing" is + 96px in PressStart2P, which ran off the side of a 64px panel even + before an address was added. And the pattern is punched out behind the + text, because the diagonal runs through the middle of the panel, which + is exactly where this sits. + + The text stays blue. It is not decoration: the pattern draws one pure + channel per element -- red border, green diagonal, blue text -- so that + a glance at the panel says whether led_rgb_sequence is right. Swap the + wiring to BGR and the border comes up blue and this text red. Drawing + it white would light all three channels and destroy the only blue + reference on the screen, which is why it is worth a comment rather + than a quiet preference. + """ + if not lines: + return + font = self._fitting_font(lines, width - 2) + line_height = self.draw.textbbox((0, 0), "Ag", font=font)[3] + 1 + block_height = line_height * len(lines) + block_top = max(1, (height - block_height) // 2) + block_width = max(self.draw.textlength(t, font=font) for t in lines) + block_left = max(0, (width - block_width) // 2) + + self.draw.rectangle( + [block_left - 2, block_top - 1, + block_left + block_width + 1, block_top + block_height], + fill=(0, 0, 0)) + + for row, line in enumerate(lines): + line_width = self.draw.textlength(line, font=font) + self.draw.text( + (max(0, (width - line_width) // 2), block_top + row * line_height), + line, font=font, fill=(0, 0, 255)) + def _draw_test_pattern(self): """Draw a test pattern to verify the display is working.""" try: @@ -536,8 +622,11 @@ class DisplayManager: # Draw a diagonal line self.draw.line([0, 0, self.matrix.width-1, self.matrix.height-1], fill=(0, 255, 0)) - # Draw some text - changed from "TEST" to "Initializing" with smaller font - self.draw.text((10, 10), "Initializing", font=self.font, fill=(0, 0, 255)) + lines = ["Initializing"] + ip = self._local_ip() + if ip: + lines.append(ip) + self._draw_startup_banner(lines, self.matrix.width, self.matrix.height) # Update the display once after everything is drawn self.update_display() diff --git a/test/test_initial_update_budget.py b/test/test_initial_update_budget.py new file mode 100644 index 00000000..4b74cc8e --- /dev/null +++ b/test/test_initial_update_budget.py @@ -0,0 +1,224 @@ +"""Tests that startup does not wait indefinitely for plugins to fetch data. + +DisplayController.__init__ calls _update_modules() once, to populate plugin +data before the first frame. It walks every loaded plugin in turn, and each +update blocks the calling thread for up to the executor's 30s timeout, so the +uncapped total is the sum of every slow plugin on the system. + +Profiled on a live rig with py-spy, the main thread sat 9.34s in + + display_controller._update_modules + -> plugin_executor.execute_update + -> execute_with_timeout -> threading.join + +and the controller's own log put the full pass at 82 seconds on the worst +boot measured (55 and 26 on the two before). The panel shows nothing for all +of it. + +Nothing is lost by stopping early: a plugin that has never updated is +immediately due, so run_scheduled_updates() collects it seconds later with the +display already running. +""" + +import os +import time +from unittest.mock import Mock + +import pytest + +# display_controller imports display_manager, which binds the hardware +# rgbmatrix module unless EMULATOR=true is set before import (same convention +# as test_display_controller_vegas_tick.py). +os.environ.setdefault("EMULATOR", "true") + +from src.display_controller import ( # noqa: E402 + DisplayController, _INITIAL_UPDATE_BUDGET_SECONDS, + _MIN_INITIAL_UPDATE_TIMEOUT_SECONDS) + + +class FakeExecutor: + """Records which plugins were updated, and can make some of them slow.""" + + def __init__(self, cost=0.0, slow=()): + self.updated = [] + self.cost = cost + self.slow = set(slow) + + def execute_update(self, plugin, plugin_id, timeout=None): + self.updated.append(plugin_id) + if plugin_id in self.slow: + time.sleep(self.cost) + return True + + +@pytest.fixture +def tiny_floor(monkeypatch): + """Shrink the "worth starting" floor so timing tests stay quick.""" + import src.display_controller as mod + monkeypatch.setattr(mod, "_MIN_INITIAL_UPDATE_TIMEOUT_SECONDS", 0.01) + + +def _controller(plugin_ids, executor): + c = DisplayController.__new__(DisplayController) + c.plugin_manager = Mock() + # Both attributes, because _update_modules reads + # `loaded_plugins or plugins` and an empty dict is falsy. + c.plugin_manager.loaded_plugins = {pid: Mock() for pid in plugin_ids} + c.plugin_manager.plugins = dict(c.plugin_manager.loaded_plugins) + c.plugin_manager.plugin_executor = executor + c.plugin_manager.plugin_last_update = {} + c.plugin_manager.health_tracker = None + return c + + +class TestTheBudgetIsRespected: + def test_without_a_deadline_every_plugin_is_updated(self): + ex = FakeExecutor() + _controller(['a', 'b', 'c'], ex)._update_modules() + assert ex.updated == ['a', 'b', 'c'] + + def test_a_passed_deadline_stops_the_pass(self): + ex = FakeExecutor() + _controller(['a', 'b', 'c'], ex)._update_modules(deadline=time.time() - 1) + assert ex.updated == [], "updated %r after the deadline" % ex.updated + + def test_slow_plugins_do_not_drag_in_the_rest(self, tiny_floor): + # One plugin burns the whole budget; the remainder must be left alone + # rather than each adding its own wait. + ex = FakeExecutor(cost=0.3, slow={'slow'}) + c = _controller(['slow'] + ['p%d' % i for i in range(20)], ex) + started = time.time() + c._update_modules(deadline=started + 0.2) + elapsed = time.time() - started + + assert ex.updated == ['slow'], "updated %r" % ex.updated + # Bounded by the one in-flight update, not by twenty more. + assert elapsed < 1.0, "%.2fs" % elapsed + + def test_a_generous_deadline_still_gets_everything(self): + ex = FakeExecutor() + c = _controller(['a', 'b', 'c'], ex) + c._update_modules(deadline=time.time() + 30) + assert ex.updated == ['a', 'b', 'c'] + + def test_the_deadline_is_checked_before_each_plugin(self, tiny_floor): + # Not just once up front: the budget can be spent partway through. + ex = FakeExecutor(cost=0.15, slow={'a', 'b', 'c', 'd'}) + c = _controller(['a', 'b', 'c', 'd'], ex) + c._update_modules(deadline=time.time() + 0.2) + assert 0 < len(ex.updated) < 4, "updated %r" % ex.updated + + +class TestThePassIsBoundedInPractice: + def test_the_last_plugin_cannot_overrun_the_budget(self): + # Checking the deadline before each plugin is not enough on its own: + # one that starts with a moment left could still block for the + # executor's full timeout. On the rig that turned a 20s budget into a + # 31.8s pass, so the remaining budget is passed down as the timeout. + seen = [] + + class Executor: + def execute_update(self, plugin, plugin_id, timeout=None): + seen.append(timeout) + return True + + c = _controller(['a', 'b', 'c'], Executor()) + deadline = time.time() + 5 + c._update_modules(deadline=deadline) + + assert seen and all(t is not None for t in seen), seen + assert all(t <= 5.01 for t in seen), seen + # The exact remainder, never clamped up: clamping would let the pass + # run past its deadline. Anything below the floor is deferred instead, + # so what does start always has a usable slot. + assert all(t >= _MIN_INITIAL_UPDATE_TIMEOUT_SECONDS for t in seen), seen + + def test_without_a_deadline_the_executor_default_is_left_alone(self): + seen = [] + + class Executor: + def execute_update(self, plugin, plugin_id, timeout=None): + seen.append(timeout) + return True + + _controller(['a'], Executor())._update_modules() + assert seen == [None], seen + + +class TestTheBudgetItself: + def test_it_is_short_enough_to_be_worth_having(self): + # The measured uncapped worst case was 82s; a budget near that would + # not bound anything. + assert _INITIAL_UPDATE_BUDGET_SECONDS <= 30 + + def test_it_is_long_enough_for_a_quick_plugin_or_two(self): + assert _INITIAL_UPDATE_BUDGET_SECONDS >= 5 + + +class TestNothingIsSilentlyDropped: + def test_deferred_plugins_are_named_in_the_log(self, caplog): + ex = FakeExecutor() + c = _controller(['a', 'b'], ex) + with caplog.at_level('INFO'): + c._update_modules(deadline=time.time() - 1) + text = "\n".join(r.getMessage() for r in caplog.records) + assert 'a' in text and 'b' in text, text + assert 'budget' in text.lower(), text + + def test_nothing_is_logged_when_all_of_them_ran(self, caplog): + ex = FakeExecutor() + c = _controller(['a'], ex) + with caplog.at_level('INFO'): + c._update_modules(deadline=time.time() + 30) + assert not any('budget' in r.getMessage().lower() for r in caplog.records) + + +class TestItDoesNotBreakTheOrdinaryPaths: + def test_no_plugin_manager_is_harmless(self): + c = DisplayController.__new__(DisplayController) + c.plugin_manager = None + c._update_modules(deadline=time.time() - 1) # must not raise + + def test_an_empty_plugin_set_is_harmless(self): + ex = FakeExecutor() + _controller([], ex)._update_modules(deadline=time.time() + 5) + assert ex.updated == [] + + +class TestTooLittleBudgetDefersRatherThanClamps: + def test_a_plugin_starting_below_the_floor_is_deferred(self): + ex = FakeExecutor() + c = _controller(['a'], ex) + # Just under the floor: previously this was clamped up to the floor and + # run anyway, which pushed the pass past its deadline. + c._update_modules( + deadline=time.time() + _MIN_INITIAL_UPDATE_TIMEOUT_SECONDS - 0.05) + assert ex.updated == [], "started a plugin it could not give a slot to" + + def test_a_plugin_starting_above_the_floor_still_runs(self): + ex = FakeExecutor() + c = _controller(['a'], ex) + c._update_modules( + deadline=time.time() + _MIN_INITIAL_UPDATE_TIMEOUT_SECONDS + 1) + assert ex.updated == ['a'] + + def test_the_timeout_is_the_remainder_not_the_floor(self): + seen = [] + + class Executor: + def execute_update(self, plugin, plugin_id, timeout=None): + seen.append(timeout) + return True + + c = _controller(['a'], Executor()) + c._update_modules(deadline=time.time() + 9) + assert seen and 8.5 <= seen[0] <= 9.01, seen + + def test_the_pass_cannot_outlast_its_deadline(self, tiny_floor): + # Every plugin sleeps well past the budget; the deferral keeps the + # whole pass inside it rather than overrunning by a floor's worth. + ex = FakeExecutor(cost=0.4, slow={'a', 'b', 'c', 'd', 'e'}) + c = _controller(['a', 'b', 'c', 'd', 'e'], ex) + started = time.time() + c._update_modules(deadline=started + 0.5) + assert time.time() - started < 1.2, "%.2fs" % (time.time() - started) diff --git a/test/test_initializing_screen.py b/test/test_initializing_screen.py new file mode 100644 index 00000000..9f6f3f3c --- /dev/null +++ b/test/test_initializing_screen.py @@ -0,0 +1,190 @@ +"""Tests the startup screen that shows while plugins fetch their first data. + +That screen is on the panel for the whole initial-update window, and on a +headless Pi it is the only place the device's address appears without going +looking for it -- so it now carries the address as well as "Initializing". + +Two things have to hold. It must fit every supported panel: the old fixed +8px PressStart2P drew "Initializing" 96px wide at x=10, which ran off the +side of a 64px panel before an address was ever added. And the lookup must be +cheap, because this runs on the startup path that the rest of this change +exists to shorten. +""" + +import os +import time + +from PIL import Image, ImageDraw, ImageFont +import pytest + +os.environ.setdefault("EMULATOR", "true") + +from src.display_manager import DisplayManager # noqa: E402 + +SIZES = [(64, 32), (128, 32), (128, 64), (256, 32), (512, 64)] + + +class FakeMatrix: + def __init__(self, width, height): + self.width, self.height = width, height + + +def _manager(width, height): + dm = DisplayManager.__new__(DisplayManager) + dm.image = Image.new('RGB', (width, height)) + dm.draw = ImageDraw.Draw(dm.image) + dm.matrix = FakeMatrix(width, height) + dm.font = ImageFont.truetype('assets/fonts/PressStart2P-Regular.ttf', 8) + return dm + + +def _layout(dm, lines): + """The geometry _draw_startup_banner uses.""" + font = dm._fitting_font(lines, dm.matrix.width - 2) + line_height = dm.draw.textbbox((0, 0), "Ag", font=font)[3] + 1 + top = max(1, (dm.matrix.height - line_height * len(lines)) // 2) + widths = [dm.draw.textlength(t, font=font) for t in lines] + return font, widths, top, top + line_height * len(lines) + + +def _render_over_pattern(width, height, lines): + """Draw the test pattern, then the banner over it, as startup does.""" + dm = _manager(width, height) + dm.draw.rectangle([0, 0, width - 1, height - 1], outline=(255, 0, 0)) + dm.draw.line([0, 0, width - 1, height - 1], fill=(0, 255, 0)) + dm._draw_startup_banner(lines, width, height) + return dm + + +class TestTheAddressLookup: + def test_it_never_reports_loopback(self): + # A loopback address on the panel would be actively misleading -- it is + # not something anyone can browse to. + ip = DisplayManager._local_ip() + assert ip is None or not ip.startswith("127."), ip + + def test_it_looks_like_an_address_when_there_is_one(self): + ip = DisplayManager._local_ip() + if ip is None: + pytest.skip("host has no routable address") + parts = ip.split(".") + assert len(parts) == 4 and all(p.isdigit() for p in parts), ip + + def test_it_is_cheap_enough_for_the_startup_path(self): + DisplayManager._local_ip() # warm anything cacheable + started = time.perf_counter() + for _ in range(20): + DisplayManager._local_ip() + per_call = (time.perf_counter() - started) / 20 + # `hostname -I` with its 2s timeout, which the web launcher uses, would + # be thousands of times this. + assert per_call < 0.05, "%.1f ms per call" % (per_call * 1000) + + def test_it_returns_none_rather_than_raising(self, monkeypatch): + import src.display_manager as mod + + def no_network(*a, **k): + raise OSError("network is unreachable") + + monkeypatch.setattr(mod.socket, "socket", no_network) + assert DisplayManager._local_ip() is None + + +class TestItFitsEveryPanel: + @pytest.mark.parametrize("width,height", SIZES) + def test_both_lines_fit_with_an_address(self, width, height): + dm = _manager(width, height) + _font, widths, top, bottom = _layout(dm, ["Initializing", "255.255.255.255"]) + assert all(w <= width - 2 for w in widths), (width, widths) + assert bottom <= height and top >= 0, (top, bottom, height) + + @pytest.mark.parametrize("width,height", SIZES) + def test_it_still_fits_with_no_address(self, width, height): + dm = _manager(width, height) + _font, widths, _top, bottom = _layout(dm, ["Initializing"]) + assert all(w <= width - 2 for w in widths), (width, widths) + assert bottom <= height, (bottom, height) + + def test_the_smallest_panel_drops_to_a_narrower_font(self): + # The regression this guards: PressStart2P at 8px is 96px wide for + # "Initializing", which does not fit 64px however it is positioned. + dm = _manager(64, 32) + font, widths, _t, _b = _layout(dm, ["Initializing", "10.0.20.104"]) + assert font is not dm.font, "kept a font that cannot fit" + assert max(widths) <= 62, widths + + def test_a_roomy_panel_keeps_the_larger_font(self): + dm = _manager(256, 32) + font, _w, _t, _b = _layout(dm, ["Initializing", "10.0.20.104"]) + assert font is dm.font, "needlessly shrank on a panel with room" + + +class TestPlacement: + @pytest.mark.parametrize("width,height", SIZES) + def test_the_lines_are_centred(self, width, height): + dm = _manager(width, height) + lines = ["Initializing", "10.0.20.104"] + _font, widths, _t, _b = _layout(dm, lines) + for w in widths: + left = max(0, (width - w) // 2) + assert abs((left + (left + w)) - width) <= 2, (left, w, width) + + def test_the_address_sits_under_the_word(self): + dm = _manager(128, 64) + font, _w, top, bottom = _layout(dm, ["Initializing", "10.0.20.104"]) + line_height = dm.draw.textbbox((0, 0), "Ag", font=font)[3] + 1 + assert bottom - top == line_height * 2 + + +class TestItIsActuallyReadable: + """The point of the address is that someone can read it off the wall.""" + + @pytest.mark.parametrize("width,height", SIZES) + def test_the_diagonal_does_not_cross_the_text(self, width, height): + lines = ["Initializing", "10.0.20.104"] + dm = _render_over_pattern(width, height, lines) + _font, widths, top, bottom = _layout(dm, lines) + # textlength returns a float, so these must be floored before they + # can index pixels. + block_width = int(max(widths)) + left = int(max(0, (width - block_width) // 2)) + + px = dm.image.load() + green = 0 + for y in range(int(top), min(int(bottom), height)): + for x in range(left, min(left + block_width, width)): + r, g, b = px[x, y] + if g > 128 and r < 128 and b < 128: + green += 1 + assert green == 0, "%d green pixels behind the text at %dx%d" % ( + green, width, height) + + @pytest.mark.parametrize("width,height", SIZES) + def test_the_text_stays_pure_blue(self, width, height): + # Not a style choice. The pattern lights one channel per element -- + # red border, green diagonal, blue text -- so a glance says whether + # led_rgb_sequence is right: wire it BGR and the border comes up blue + # and this text red. White text would light all three and destroy the + # only blue reference on the screen. + dm = _render_over_pattern(width, height, ["Initializing", "10.0.20.104"]) + px = dm.image.load() + blue = sum(1 for y in range(height) for x in range(width) + if px[x, y] == (0, 0, 255)) + assert blue > 20, "only %d blue pixels at %dx%d" % (blue, width, height) + white = sum(1 for y in range(height) for x in range(width) + if px[x, y] == (255, 255, 255)) + assert white == 0, "%d white pixels would muddy the channel check" % white + + def test_each_element_lights_one_channel(self): + # The whole point of the pattern: three pure primaries on screen. + dm = _render_over_pattern(128, 64, ["Initializing", "10.0.20.104"]) + seen = set(dm.image.getdata()) + assert (255, 0, 0) in seen, "no pure red border" + assert (0, 255, 0) in seen, "no pure green diagonal" + assert (0, 0, 255) in seen, "no pure blue text" + + def test_nothing_is_drawn_for_no_lines(self): + dm = _manager(128, 64) + before = dm.image.tobytes() + dm._draw_startup_banner([], 128, 64) + assert dm.image.tobytes() == before