Compare commits

..
Author SHA1 Message Date
ChuckandClaude Opus 5.5 686b00d224 fix(fonts): call the font manager's forget methods without a None-able local
Pylint E1102 (Codacy) read getattr(..., None) as possibly not callable.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-04 17:11:09 -04:00
ChuckandClaude Opus 5.5 10ac608b9d fix(fonts): unloading a plugin forgets its manifest fonts
PluginManager.unload_plugin() and the failed-load cleanup only called
FontManager.forget_manager_fonts(), which drops usage data. The plugin's
manifest registrations stayed: plugin_fonts / plugin_font_catalogs, its
plugin_id::family entries in font_catalog, and cached font objects for them
-- so its fonts kept resolving after unload and a family a reinstalled
manifest dropped stayed registered. The deprecated unregister_plugin_fonts
did this cleanup but nothing called it; it was removed in #708.

Add FontManager.forget_plugin_fonts(plugin_id) and call it from both
paths alongside forget_manager_fonts (each guarded on its own, so a font
manager stub with only one still works). FontManager takes no locks, so
like forget_manager_fonts it uses atomic pops over snapshots. A reload
(unload + load) registers the manifest fonts again and they resolve.

Raised by CodeRabbit on #709.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-04 17:00:28 -04:00
9 changed files with 210 additions and 250 deletions
+8 -12
View File
@@ -19,20 +19,16 @@ accepts both, but the store flags the old spelling as deprecated
## Unreleased
### A scrolling screen held by its plugin's update() is reported
- While a plugin's `update()` runs it holds the plugin's lock, and that
plugin's frames are skipped: on a scroller, a frozen strip, with nothing
logged (and a freeze of 5 s or more is a gap, not a freeze, to the frame
stats). The high-FPS loop now times each run of skipped frames; one of
250 ms or more logs `Display of <plugin> held N ms by its update()`
(rate-limited per plugin) when it ends, and is recorded on the plugin's
health as a `display hold` busy skip, which never counts toward the
circuit breaker. The 1 Hz loop is left out: its frames are a second apart,
so one skipped frame there measures nothing and freezes nothing visible.
### Fixed
- Unloading a plugin now forgets the fonts its manifest registered, not only
the fonts it reported using. Its `plugin_id::family` entries kept resolving
and their cached font objects stayed alive until a restart, and a family a
reinstalled plugin's manifest dropped stayed registered. The new
`FontManager.forget_plugin_fonts(plugin_id)` does the cleanup;
`PluginManager.unload_plugin()` and a failed load call it alongside
`forget_manager_fonts()`, and a reload registers the manifest's fonts again.
- The web preview and `/api/v3/display/current` no longer stay black for a
whole screen that draws its card once and then holds it. The snapshot is
written from `update_display()` at most once per write interval, so a frame
+3 -1
View File
@@ -203,6 +203,7 @@ Current methods:
| `measure_text(text, font)` | `(width, height, baseline)` |
| `get_font_height(font)` | Line height |
| `register_plugin_fonts(plugin_id, font_manifest)` | Register a plugin's fonts (core calls it at load) |
| `forget_plugin_fonts(plugin_id)` | Drop a plugin's manifest fonts and their cached objects (core calls it when a plugin unloads) |
| `clear_cache()` | Drop cached fonts and metrics |
| `font_catalog` (attribute) | Family name → file path |
@@ -218,5 +219,6 @@ Removed in 3.8.0, after logging a deprecation warning on first call since
| `get_performance_stats()` | — |
| `set_override()`, `remove_override()`, `get_overrides()` | a font field in your plugin's config schema |
| `get_manager_fonts()`, `get_detected_fonts()` | — |
| `get_plugin_fonts()`, `unregister_plugin_fonts()` | — |
| `get_plugin_fonts()` | — |
| `unregister_plugin_fonts()` | `forget_plugin_fonts()` (core calls it on unload) |
| `add_font()`, `remove_font()`, `validate_font()` | the web UI's Fonts tab |
+2 -64
View File
@@ -1164,55 +1164,6 @@ class DisplayController:
except Exception: # pylint: disable=broad-except
logger.exception("Error running scheduled plugin updates")
#: A run of frames skipped because a plugin's update() held its lock is
#: reported once it has lasted this long.
DISPLAY_HOLD_REPORT_SECONDS = 0.25
#: (plugin_id, monotonic start) of the current run of skipped frames.
_display_hold: Optional[Tuple[str, float]] = None
def _note_display_hold(self, plugin_id: str, held: bool) -> None:
"""Report how long a plugin's update() kept its display() from drawing.
While update() runs on the worker it holds the plugin's lock, and every
frame of that plugin's screen is skipped: the panel keeps showing the
last frame, which on a scroller is a frozen strip. Nothing said so --
the frames are not failures, and a scroll freeze of 5 s or more is a
gap to the frame stats, not a freeze. This times each such run and,
when it ends after DISPLAY_HOLD_REPORT_SECONDS or more, logs it
(rate-limited per plugin) and records it on the plugin's health as a
busy skip, which never touches the circuit breaker. Only frames of the
high-FPS loop are timed (see _display_once's ``report_hold``).
"""
# The clock is read only when a run starts or ends: on a frame that
# draws with no run open, this is one attribute check.
current = self._display_hold
if held:
if current is None or current[0] != plugin_id:
self._display_hold = (plugin_id, time.monotonic())
return
if current is None:
return
self._display_hold = None
if current[0] != plugin_id:
return
seconds = time.monotonic() - current[1]
if seconds < self.DISPLAY_HOLD_REPORT_SECONDS:
return
pm = self.plugin_manager
warn = getattr(pm, '_warn_rate_limited', None)
if warn is not None:
warn(f"display-hold:{plugin_id}",
"Display of %s held %.0f ms by its update()",
plugin_id, seconds * 1000.0)
tracker = getattr(pm, 'health_tracker', None)
record = getattr(tracker, 'record_busy_skip', None)
if record is not None:
try:
record(plugin_id, "display hold", seconds)
except Exception: # pylint: disable=broad-except
logger.debug("Could not record a display hold", exc_info=True)
@contextmanager
def _display_lock_or_skip(self, plugin_id):
"""Try-lock guard keeping a plugin's display() off its in-flight update().
@@ -1238,7 +1189,7 @@ class DisplayController:
lock.release()
def _display_once(self, plugin, mode: str, accepts_display_mode: bool,
force_clear: bool = False, report_hold: bool = False):
force_clear: bool = False):
"""Call ``plugin.display()`` directly for one frame of a render loop.
Frames after a screen's first dispatch come through here rather than
@@ -1254,12 +1205,6 @@ class DisplayController:
``display_mode`` so plugins with several modes stay on it.
accepts_display_mode: Whether display() takes ``display_mode``.
force_clear: Passed through to display().
report_hold: Time runs of frames skipped because update() holds
the plugin's lock (see _note_display_hold). Only the high-FPS
loop asks: its frames are ~8 ms apart, so a run measures the
hold, and a held scroller is a frozen strip. The 1 Hz loop's
frames are a second apart, so one skipped frame there would
read as a 1 s hold of a screen that did not visibly change.
Each call is timed (two monotonic reads) and handed to
PluginManager.note_display_duration, which logs and records slow
@@ -1274,12 +1219,6 @@ class DisplayController:
display_watchdog.watchdog.beat()
plugin_id = getattr(plugin, 'plugin_id', None)
with self._display_lock_or_skip(plugin_id) as can_display:
if report_hold and plugin_id:
self._note_display_hold(plugin_id, held=not can_display)
elif can_display and self._display_hold is not None:
# A drawn frame outside the high-FPS loop: whatever run was
# open is over, unreported.
self._display_hold = None
if not can_display:
return True
started = time.monotonic()
@@ -4048,8 +3987,7 @@ class DisplayController:
_frame_start = time.perf_counter()
try:
result = self._display_once(
manager_to_display, active_mode, _accepts_display_mode,
report_hold=True)
manager_to_display, active_mode, _accepts_display_mode)
if isinstance(result, bool) and not result:
logger.debug("Display returned False, breaking early")
break
+33
View File
@@ -222,6 +222,39 @@ class FontManager:
logger.error(f"Error registering fonts for plugin {plugin_id}: {e}", exc_info=True)
return False
def forget_plugin_fonts(self, plugin_id: str) -> bool:
"""Drop the fonts ``plugin_id``'s manifest registered: its manifest
and catalog, its ``plugin_id::family`` entries in font_catalog, and
cached font objects for those families. Called by core when a plugin
is unloaded, so a reload registers from its current manifest and a
removed plugin's fonts stop resolving.
FontManager takes no locks; like forget_manager_fonts this relies on
single dict operations being atomic and iterates snapshots, so a
render thread calling get_font() meanwhile cannot break it. Returns
True if the plugin had registered fonts.
"""
prefix = f"{plugin_id}::"
manifest = self.plugin_fonts.pop(plugin_id, None)
catalog = self.plugin_font_catalogs.pop(plugin_id, None)
# Every namespaced entry, not just the families in the catalog: one
# whose file failed to load never made it into the catalog, and a
# caller may have added one directly.
for family in list(self.font_catalog):
if family.startswith(prefix):
self.font_catalog.pop(family, None)
# get_font() keys the cache f"{family}_{size_px}".
dropped = [key for key in list(self.font_cache) if key.startswith(prefix)]
for key in dropped:
self.font_cache.pop(key, None)
if dropped:
# Font objects someone may hold were dropped; see cache_generation.
self.cache_generation += 1
if manifest is None and catalog is None:
return False
logger.info("Forgot fonts of plugin %s", plugin_id)
return True
def _validate_font_manifest(self, font_manifest: Dict[str, Any]) -> bool:
"""Validate the structure of a plugin's font manifest."""
required_fields = ["fonts"]
+18 -11
View File
@@ -597,11 +597,21 @@ class PluginManager:
self.plugin_loader.unregister_plugin_modules(plugin_id)
except Exception as e: # pragma: no cover - defensive
self.logger.debug("Could not drop modules of %s: %s", plugin_id, e)
try:
if self.font_manager is not None and hasattr(self.font_manager, 'forget_manager_fonts'):
self.font_manager.forget_manager_fonts(plugin_id)
except Exception as e:
self.logger.debug("Could not forget fonts of %s: %s", plugin_id, e)
self._forget_plugin_fonts(plugin_id)
def _forget_plugin_fonts(self, plugin_id: str) -> None:
"""Drop what the FontManager holds for a plugin: the fonts its
instance reported using (the Fonts tab's "Used by") and the fonts its
manifest registered. Never raises."""
if self.font_manager is None:
return
for name in ('forget_manager_fonts', 'forget_plugin_fonts'):
if not hasattr(self.font_manager, name):
continue
try:
getattr(self.font_manager, name)(plugin_id)
except Exception as e:
self.logger.debug("Could not forget fonts of %s (%s): %s", plugin_id, name, e)
#: Config keys the **core** reads out of a plugin's own config block. The
#: plugin never declares them, so a schema with
@@ -846,12 +856,9 @@ class PluginManager:
# Delegate sub-module and cached-module cleanup to the loader
self.plugin_loader.unregister_plugin_modules(plugin_id)
# Its font registrations go with it (the Fonts tab's "Used by").
try:
if self.font_manager is not None and hasattr(self.font_manager, 'forget_manager_fonts'):
self.font_manager.forget_manager_fonts(plugin_id)
except Exception as e:
self.logger.debug("Could not forget fonts of %s: %s", plugin_id, e)
# Its font registrations go with it: the fonts it reported using
# and the ones its manifest registered.
self._forget_plugin_fonts(plugin_id)
# Update state
self.state_manager.set_state(plugin_id, PluginState.UNLOADED)
-142
View File
@@ -1,142 +0,0 @@
"""The report of a scrolling screen held by its plugin's update().
While a plugin's update() runs it holds the plugin's lock, and its screen's
frames are skipped -- on a scroller, a frozen strip -- with nothing logged.
_note_display_hold times each such run and reports one of
DISPLAY_HOLD_REPORT_SECONDS or more.
"""
import threading
import types
from unittest.mock import MagicMock
import pytest
class _Clock:
"""display_controller's clock: moves only when run() sleeps or a test says."""
def __init__(self, start=10_000.0):
self.t = start
def now(self):
return self.t
def sleep(self, seconds):
self.t += max(seconds, 0.0005)
def module(self):
return types.SimpleNamespace(time=self.now, monotonic=self.now,
perf_counter=self.now, sleep=self.sleep)
@pytest.fixture
def clock(monkeypatch):
c = _Clock()
monkeypatch.setattr("src.display_controller.time", c.module())
return c
class _Locks:
"""get_plugin_lock for one plugin, whose lock the test can hold."""
def __init__(self):
self.lock = threading.Lock()
def __call__(self, plugin_id):
return self.lock
@pytest.fixture
def held(test_display_controller):
c = test_display_controller
locks = _Locks()
c.plugin_manager.get_plugin_lock = locks
c.plugin_manager._warn_rate_limited = MagicMock()
c.plugin_manager.health_tracker = MagicMock()
c._display_hold = None
return c, locks.lock
def _plugin(plugin_id):
p = MagicMock()
p.plugin_id = plugin_id
p.display.return_value = True
return p
class TestTheDisplayHoldReport:
def test_a_long_hold_is_reported_when_it_ends(self, held, clock):
c, lock = held
ticker = _plugin("ticker")
lock.acquire() # update() running
assert c._display_once(ticker, "ticker", False, report_hold=True) is True
clock.t += 0.2
assert c._display_once(ticker, "ticker", False, report_hold=True) is True
assert ticker.display.call_count == 0
c.plugin_manager._warn_rate_limited.assert_not_called()
clock.t += 0.2
lock.release() # update() done
c._display_once(ticker, "ticker", False, report_hold=True)
assert ticker.display.call_count == 1
key, message, plugin_id, ms = c.plugin_manager._warn_rate_limited.call_args[0]
assert key == "display-hold:ticker" and plugin_id == "ticker"
assert "held" in message and ms == pytest.approx(400.0)
c.plugin_manager.health_tracker.record_busy_skip.assert_called_once_with(
"ticker", "display hold", pytest.approx(0.4))
def test_a_short_hold_is_not(self, held, clock):
c, lock = held
ticker = _plugin("ticker")
lock.acquire()
c._display_once(ticker, "ticker", False, report_hold=True)
clock.t += 0.1
lock.release()
c._display_once(ticker, "ticker", False, report_hold=True)
c.plugin_manager._warn_rate_limited.assert_not_called()
c.plugin_manager.health_tracker.record_busy_skip.assert_not_called()
def test_a_hold_that_ends_on_another_plugins_screen_is_not_blamed_on_it(
self, held, clock):
c, lock = held
lock.acquire()
c._display_once(_plugin("ticker"), "ticker", False, report_hold=True)
clock.t += 1.0
lock.release()
c._display_once(_plugin("clock"), "clock", False, report_hold=True)
c.plugin_manager._warn_rate_limited.assert_not_called()
assert c._display_hold is None
def test_frames_that_draw_report_nothing(self, held, clock):
c, _lock = held
ticker = _plugin("ticker")
for _ in range(5):
c._display_once(ticker, "ticker", False, report_hold=True)
clock.t += 0.5
c.plugin_manager._warn_rate_limited.assert_not_called()
def test_the_1hz_loop_reports_no_holds(self, held, clock):
# A static screen's frames are a second apart: one skipped frame is
# not a measured hold, and nothing on the panel froze. (On ledpi the
# first version reported every such skip as "held 1000 ms".)
c, lock = held
board = _plugin("board")
lock.acquire()
c._display_once(board, "board", False)
clock.t += 1.0
lock.release()
c._display_once(board, "board", False)
c.plugin_manager._warn_rate_limited.assert_not_called()
c.plugin_manager.health_tracker.record_busy_skip.assert_not_called()
def test_a_run_left_open_is_dropped_by_a_1hz_frame(self, held, clock):
c, lock = held
ticker = _plugin("ticker")
lock.acquire()
c._display_once(ticker, "ticker", False, report_hold=True)
lock.release()
c._display_once(ticker, "ticker", False) # the 1 Hz loop draws
assert c._display_hold is None
clock.t += 5.0
c._display_once(ticker, "ticker", False, report_hold=True)
c.plugin_manager._warn_rate_limited.assert_not_called()
+143
View File
@@ -185,6 +185,149 @@ class TestPluginFonts:
assert fm.font_catalog["my-plugin::bundled"] == str(plugin_dir / "fonts" / "Bundled.ttf")
class TestForgetPluginFonts:
"""forget_plugin_fonts drops what a plugin's manifest registered. Before
it, unloading a plugin left its fonts resolvable and its cached font
objects alive until a restart."""
@staticmethod
def _register(fm, root, plugin_id, family="bundled"):
plugin_dir = root / plugin_id
(plugin_dir / "fonts").mkdir(parents=True, exist_ok=True)
font_file = plugin_dir / "fonts" / f"{family}.ttf"
if not font_file.exists(): # a loaded font may hold it open (Windows)
shutil.copy(resolve_asset_path("assets/fonts/PressStart2P-Regular.ttf"), font_file)
manifest = {"fonts": [{"family": family, "source": f"plugin://fonts/{family}.ttf"}]}
assert fm.register_plugin_fonts(plugin_id, manifest, plugin_dir=plugin_dir)
return plugin_dir
@staticmethod
def _entries_of(fm, plugin_id):
prefix = f"{plugin_id}::"
return {
"plugin_fonts": plugin_id in fm.plugin_fonts,
"plugin_font_catalogs": plugin_id in fm.plugin_font_catalogs,
"font_catalog": [k for k in fm.font_catalog if k.startswith(prefix)],
"font_cache": [k for k in fm.font_cache if k.startswith(prefix)],
}
NONE = {"plugin_fonts": False, "plugin_font_catalogs": False,
"font_catalog": [], "font_cache": []}
def test_unload_leaves_no_plugin_entries(self, fm, tmp_path):
self._register(fm, tmp_path, "alpha")
fm.resolve_font("alpha.title", "bundled", 8, plugin_id="alpha")
fm.get_font("alpha::bundled", 10)
assert self._entries_of(fm, "alpha")["font_cache"] # cached before
gen = fm.cache_generation
assert fm.forget_plugin_fonts("alpha") is True
assert self._entries_of(fm, "alpha") == self.NONE
assert fm.cache_generation == gen + 1
# The family no longer resolves to the plugin's file.
assert fm.font_catalog.get("alpha::bundled") is None
def test_other_plugins_and_core_fonts_are_untouched(self, fm, tmp_path):
self._register(fm, tmp_path, "alpha")
self._register(fm, tmp_path, "beta")
# A plugin whose id is a prefix of another's must not take it along.
self._register(fm, tmp_path, "alpha-two")
for pid in ("alpha", "beta", "alpha-two"):
fm.get_font(f"{pid}::bundled", 8)
core_font = fm.get_font("press_start", 8)
beta_before = self._entries_of(fm, "beta")
alpha_two_before = self._entries_of(fm, "alpha-two")
fm.forget_plugin_fonts("alpha")
assert self._entries_of(fm, "beta") == beta_before
assert self._entries_of(fm, "alpha-two") == alpha_two_before
assert fm.get_font("press_start", 8) is core_font
def test_reload_re_registers_cleanly(self, fm, tmp_path):
plugin_dir = self._register(fm, tmp_path, "alpha")
old = fm.get_font("alpha::bundled", 8)
fm.forget_plugin_fonts("alpha")
self._register(fm, tmp_path, "alpha")
assert fm.font_catalog["alpha::bundled"] == str(plugin_dir / "fonts" / "bundled.ttf")
font = fm.resolve_font("alpha.title", "bundled", 8, plugin_id="alpha")
assert isinstance(font, ImageFont.FreeTypeFont)
assert font is not old # loaded fresh, not the dropped cache entry
def test_a_family_the_new_manifest_drops_stops_resolving(self, fm, tmp_path):
self._register(fm, tmp_path, "alpha", family="old_face")
fm.forget_plugin_fonts("alpha")
self._register(fm, tmp_path, "alpha", family="new_face")
assert "alpha::old_face" not in fm.font_catalog
assert "alpha::new_face" in fm.font_catalog
def test_unknown_plugin_is_a_no_op(self, fm):
catalog = dict(fm.font_catalog)
gen = fm.cache_generation
assert fm.forget_plugin_fonts("never-registered") is False
assert fm.font_catalog == catalog
assert fm.cache_generation == gen
class TestPluginManagerReloadFonts:
"""Through PluginManager: unloading a plugin forgets its manifest fonts,
and reload_plugin (unload + load) registers them again so they resolve."""
PLUGIN_ID = "font-reload-demo"
MODULE = "plugin_font_reload_demo"
def test_unload_forgets_and_reload_resolves(self, tmp_path):
import sys
from src.plugin_system.plugin_manager import PluginManager
plugins_dir = tmp_path / "plugins"
plugin_dir = plugins_dir / self.PLUGIN_ID
(plugin_dir / "fonts").mkdir(parents=True)
shutil.copy(resolve_asset_path("assets/fonts/PressStart2P-Regular.ttf"),
plugin_dir / "fonts" / "Bundled.ttf")
manifest = {"id": self.PLUGIN_ID, "name": "Demo", "class_name": "Demo",
"entry_point": "manager.py",
"fonts": {"fonts": [{"family": "bundled",
"source": "plugin://fonts/Bundled.ttf"}]}}
(plugin_dir / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
(plugin_dir / "manager.py").write_text(
"class Demo:\n"
" def __init__(self, plugin_id, config, display_manager, cache_manager, plugin_manager):\n"
" self.enabled = True\n", encoding="utf-8")
pm = PluginManager(plugins_dir=str(plugins_dir))
fm = FontManager({})
pm.font_manager = fm
pm.plugin_manifests[self.PLUGIN_ID] = manifest
key = f"{self.PLUGIN_ID}::bundled"
try:
assert pm.load_plugin(self.PLUGIN_ID) is True
assert key in fm.font_catalog
fm.register_manager_font(self.PLUGIN_ID, "demo.title", "bundled", 8)
old = fm.resolve_font("demo.title", "bundled", 8, plugin_id=self.PLUGIN_ID)
assert pm.unload_plugin(self.PLUGIN_ID) is True
assert self.PLUGIN_ID not in fm.plugin_fonts
assert self.PLUGIN_ID not in fm.plugin_font_catalogs
assert key not in fm.font_catalog
assert not [k for k in fm.font_cache if k.startswith(f"{self.PLUGIN_ID}::")]
assert self.PLUGIN_ID not in fm.manager_fonts
assert pm.reload_plugin(self.PLUGIN_ID) is True
assert fm.font_catalog[key] == str(plugin_dir / "fonts" / "Bundled.ttf")
font = fm.resolve_font("demo.title", "bundled", 8, plugin_id=self.PLUGIN_ID)
assert isinstance(font, ImageFont.FreeTypeFont)
assert font is not old
finally:
sys.modules.pop(self.MODULE, None)
class TestDownloadFont:
"""_download_font: plugin fonts declared by URL, cached in temp_font_dir."""
@@ -69,6 +69,7 @@ def test_fixed_plugin_loads_new_code_after_failed_load(plugin_env, first_source)
assert MODULE_NAME not in sys.modules
assert PLUGIN_ID not in pm.plugin_loader._loaded_modules
pm.font_manager.forget_manager_fonts.assert_called_with(PLUGIN_ID)
pm.font_manager.forget_plugin_fonts.assert_called_with(PLUGIN_ID)
(plugin_dir / "manager.py").write_text(_FIXED, encoding="utf-8")
assert pm.load_plugin(PLUGIN_ID) is True
@@ -26,15 +26,8 @@ import pytest
@pytest.fixture
def client(monkeypatch):
from web_interface import app as web_app
def client():
from web_interface.app import app
# The captive-portal before_request hook shells out to systemctl/nmcli
# whenever its 30s cache is cold, so on a Linux host whether a request
# here runs subprocess depended on how long ago the previous one was --
# and several tests below stub subprocess. Pin it: no test in this file
# is about AP mode.
monkeypatch.setattr(web_app, 'is_ap_mode_active', lambda: False)
app.config['TESTING'] = True
with app.test_client() as c:
yield c
@@ -1143,23 +1136,12 @@ class TestPixletEditorHostDefaultsButDoesNotOverride:
captured['env'] = env
return FakeProcess()
# Swap the route module's own ``subprocess`` binding, not the shared
# ``subprocess.Popen``: patching the attribute on the real module is
# process-wide, and the app's before_request hook (the captive-portal
# check) runs ``subprocess.run`` -- ``with Popen(...)`` -- whenever its
# 30s AP-mode cache is cold on a host with systemctl. On the Linux CI
# runner that handed it this FakeProcess and 500'd the request, but
# only when the previous request was more than 30s earlier.
fake_subprocess = types.ModuleType('subprocess')
fake_subprocess.__dict__.update(mod.subprocess.__dict__)
fake_subprocess.Popen = fake_popen
with patch.object(mod, '_validate_starlark_app_path',
return_value=(app_dir, None)), \
patch.object(mod, '_PIXLET_EDITOR_SCRIPT', script), \
patch.object(mod, '_PIXLET_EDITOR_STATE', state_file), \
patch.object(mod, '_find_pixlet_binary', return_value='/usr/bin/pixlet'), \
patch.object(mod, 'subprocess', fake_subprocess), \
patch.object(mod.subprocess, 'Popen', side_effect=fake_popen), \
patch.dict(os.environ):
if operator_host is None:
os.environ.pop('PIXLET_EDITOR_HOST', None)