mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-29 18:46:59 +00:00
f106e0ebc2
'hey hor' triggered the wake word: the default sensitivity was 0.5, which for openWakeWord IS the raw per-frame score threshold — openWakeWord's own permissive baseline that near-misses clear. Raised the default to 0.6 so phonetic near-misses fall short while real 'hey hermes' (typically 0.9+) still fires easily. Also fixed a real cross-engine inconsistency found while checking: the sensitivity knob is documented 'higher = stricter' and behaves that way for openWakeWord (threshold = sensitivity) and sherpa (0.05 + 0.4*s), but Porcupine's own 'sensitivities' param runs the opposite way (higher = MORE false alarms, per Picovoice). Turning sensitivity up made Porcupine looser — backwards. Now inverted (1 - sensitivity) so 'higher = stricter' holds for every engine. - tools/wake_word.py: default 0.6; _sensitivity fallback uses _DEFAULTS; Porcupine sensitivity inverted with rationale - hermes_cli/config.py + docs: default + consistent-direction note - tests: Porcupine inversion, default>=0.6 regression, fallback-to-default
1049 lines
40 KiB
Python
1049 lines
40 KiB
Python
"""Tests for tools.wake_word — the "Hey Hermes" hotword detector.
|
|
|
|
No live audio or network: the sounddevice import is faked, engines are stubbed,
|
|
and lazy-dep availability is monkeypatched. Covers config resolution, engine
|
|
dispatch, the requirements probe, the detector fire/cooldown loop, and the
|
|
process-wide singleton lifecycle.
|
|
"""
|
|
|
|
import multiprocessing
|
|
import os
|
|
import sys
|
|
import threading
|
|
import time
|
|
import types
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
import tools.wake_word as ww
|
|
|
|
|
|
# ── Config helpers ───────────────────────────────────────────────────────
|
|
|
|
|
|
def test_config_defaults_and_clamping():
|
|
assert ww._provider({}) == "openwakeword"
|
|
assert ww._provider({"provider": "Porcupine"}) == "porcupine"
|
|
assert ww._sensitivity({"sensitivity": 5}) == 1.0
|
|
assert ww._sensitivity({"sensitivity": -1}) == 0.0
|
|
# Invalid input falls back to the configured default, not a hardcoded 0.5.
|
|
assert ww._sensitivity({"sensitivity": "nope"}) == ww._DEFAULTS["sensitivity"]
|
|
assert ww._sensitivity({}) == ww._DEFAULTS["sensitivity"]
|
|
assert ww.wake_phrase({"phrase": "hey hermes"}) == "hey hermes"
|
|
assert ww.wake_phrase({}) == "hey hermes"
|
|
|
|
|
|
def test_wake_surface_enabled_gate():
|
|
# Disabled → never, regardless of surface.
|
|
assert ww.wake_surface_enabled("cli", {"enabled": False, "surface": "cli"}) is False
|
|
# auto → every surface is eligible; ownership still admits only one.
|
|
for s in ("cli", "tui", "gui"):
|
|
assert ww.wake_surface_enabled(s, {"enabled": True, "surface": "auto"}) is True
|
|
# Pinned surface → only that one.
|
|
cfg = {"enabled": True, "surface": "tui"}
|
|
assert ww.wake_surface_enabled("tui", cfg) is True
|
|
assert ww.wake_surface_enabled("cli", cfg) is False
|
|
assert ww.wake_surface_enabled("gui", cfg) is False
|
|
# Missing/blank surface defaults to auto.
|
|
assert ww.wake_surface_enabled("gui", {"enabled": True}) is True
|
|
|
|
|
|
def test_looks_like_path():
|
|
assert ww._looks_like_path("models/hey_hermes.onnx")
|
|
assert ww._looks_like_path("custom.ppn")
|
|
assert not ww._looks_like_path("hey_jarvis")
|
|
|
|
|
|
def test_load_wake_word_config_is_a_dict_with_defaults():
|
|
# Wired into DEFAULT_CONFIG, so a real load returns the section shape.
|
|
cfg = ww.load_wake_word_config()
|
|
assert isinstance(cfg, dict)
|
|
assert cfg.get("enabled") is False
|
|
assert cfg.get("provider") == "openwakeword"
|
|
|
|
|
|
def test_load_wake_word_config_guards_non_dict(monkeypatch):
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config", lambda: {"wake_word": "oops"}
|
|
)
|
|
assert ww.load_wake_word_config() == {}
|
|
|
|
|
|
# ── Engine dispatch ──────────────────────────────────────────────────────
|
|
|
|
|
|
def test_build_engine_dispatch(monkeypatch):
|
|
monkeypatch.setattr(ww, "_OpenWakeWordEngine", lambda cfg: "oww")
|
|
monkeypatch.setattr(ww, "_PorcupineEngine", lambda cfg: "pv")
|
|
assert ww._build_engine({"provider": "openwakeword"}) == "oww"
|
|
assert ww._build_engine({"provider": "porcupine"}) == "pv"
|
|
with pytest.raises(ValueError):
|
|
ww._build_engine({"provider": "bogus"})
|
|
|
|
|
|
# ── Requirements probe ───────────────────────────────────────────────────
|
|
|
|
|
|
def _voice_loop_ready(monkeypatch, stt=True, tts=True):
|
|
"""Pin the STT/TTS probes so requirements tests don't depend on the
|
|
test venv's installed voice stack."""
|
|
monkeypatch.setattr(ww, "_stt_ready", lambda: stt)
|
|
monkeypatch.setattr(ww, "_tts_ready", lambda: tts)
|
|
|
|
|
|
def test_requirements_openwakeword_available(monkeypatch):
|
|
_voice_loop_ready(monkeypatch)
|
|
monkeypatch.setattr(ww, "_audio_available", lambda: True)
|
|
monkeypatch.setattr("tools.lazy_deps.is_available", lambda f: True)
|
|
r = ww.check_wake_word_requirements(
|
|
{"provider": "openwakeword", "phrase": "hey hermes"}
|
|
)
|
|
assert r["available"] is True
|
|
assert r["provider"] == "openwakeword"
|
|
assert r["phrase"] == "hey hermes"
|
|
|
|
|
|
def test_requirements_need_stt_and_tts(monkeypatch):
|
|
"""No STT/TTS → wake refuses to arm (mic would hear you, then nothing)."""
|
|
monkeypatch.setattr(ww, "_audio_available", lambda: True)
|
|
monkeypatch.setattr("tools.lazy_deps.is_available", lambda f: True)
|
|
|
|
_voice_loop_ready(monkeypatch, stt=False, tts=True)
|
|
r = ww.check_wake_word_requirements({"provider": "openwakeword"})
|
|
assert r["available"] is False
|
|
assert r["stt_available"] is False
|
|
assert "speech-to-text" in r["hint"]
|
|
assert "text-to-speech" not in r["hint"]
|
|
|
|
_voice_loop_ready(monkeypatch, stt=True, tts=False)
|
|
r = ww.check_wake_word_requirements({"provider": "openwakeword"})
|
|
assert r["available"] is False
|
|
assert r["tts_available"] is False
|
|
assert "text-to-speech" in r["hint"]
|
|
|
|
_voice_loop_ready(monkeypatch, stt=False, tts=False)
|
|
r = ww.check_wake_word_requirements({"provider": "openwakeword"})
|
|
assert r["available"] is False
|
|
assert "speech-to-text and text-to-speech" in r["hint"]
|
|
|
|
|
|
def test_tts_ready_is_a_probe_never_an_installer(monkeypatch):
|
|
"""_tts_ready must NOT trigger lazy pip installs from a status poll.
|
|
|
|
Regression: check_tts_requirements → _import_edge_tts → lazy_deps.ensure
|
|
ran pip inside wake.status; a slow/failed install froze the poll and
|
|
unmounted the desktop ear. Uninstalled-but-lazy-installable counts as
|
|
ready WITHOUT calling ensure/check.
|
|
"""
|
|
import types as _types
|
|
|
|
monkeypatch.setattr(
|
|
ww, "_tts_ready", ww.__dict__["_tts_ready"]
|
|
) # use the real implementation
|
|
fake_tts = _types.SimpleNamespace(
|
|
_get_provider=lambda cfg: "edge",
|
|
_load_tts_config=lambda: {},
|
|
check_tts_requirements=lambda: (_ for _ in ()).throw(
|
|
AssertionError("check_tts_requirements must not run when deps are missing")
|
|
),
|
|
)
|
|
monkeypatch.setitem(sys.modules, "tools.tts_tool", fake_tts)
|
|
|
|
# Deps missing + lazy installs allowed → ready (installs at first speak).
|
|
monkeypatch.setattr("tools.lazy_deps.is_available", lambda f: False)
|
|
monkeypatch.setattr("tools.lazy_deps._allow_lazy_installs", lambda: True)
|
|
assert ww._tts_ready() is True
|
|
|
|
# Deps missing + lazy installs disabled → not ready.
|
|
monkeypatch.setattr("tools.lazy_deps._allow_lazy_installs", lambda: False)
|
|
assert ww._tts_ready() is False
|
|
|
|
# Deps present → falls through to the real requirements check.
|
|
fake_tts.check_tts_requirements = lambda: True
|
|
monkeypatch.setattr("tools.lazy_deps.is_available", lambda f: True)
|
|
assert ww._tts_ready() is True
|
|
|
|
|
|
def test_requirements_porcupine_needs_access_key(monkeypatch):
|
|
monkeypatch.delenv("PORCUPINE_ACCESS_KEY", raising=False)
|
|
monkeypatch.setattr(ww, "_audio_available", lambda: True)
|
|
monkeypatch.setattr("tools.lazy_deps.is_available", lambda f: True)
|
|
r = ww.check_wake_word_requirements({"provider": "porcupine"})
|
|
assert r["available"] is False
|
|
assert r["access_key_set"] is False
|
|
assert "PORCUPINE_ACCESS_KEY" in r["hint"]
|
|
|
|
|
|
def test_requirements_unavailable_without_audio(monkeypatch):
|
|
monkeypatch.setattr(ww, "_audio_available", lambda: False)
|
|
monkeypatch.setattr("tools.lazy_deps.is_available", lambda f: True)
|
|
r = ww.check_wake_word_requirements({"provider": "openwakeword"})
|
|
assert r["available"] is False
|
|
assert r["audio_available"] is False
|
|
|
|
|
|
def test_requirements_fresh_install_lazy_allowed(monkeypatch):
|
|
"""Deps missing + lazy installs allowed → available, so /wake on can
|
|
reach the engine constructor's ``lazy_deps.ensure()`` call.
|
|
|
|
Regression: the audio probe imports sounddevice/numpy — packages the
|
|
lazy installer would fetch — so gating ``available`` on it made the
|
|
lazy-install path unreachable on a fresh machine (the /wake on handler
|
|
printed the pip hint and bailed before ensure() ever ran).
|
|
"""
|
|
def _boom():
|
|
raise AssertionError("audio probe must not run while deps are missing")
|
|
|
|
_voice_loop_ready(monkeypatch)
|
|
monkeypatch.setattr(ww, "_audio_available", _boom)
|
|
monkeypatch.setattr("tools.lazy_deps.is_available", lambda f: False)
|
|
monkeypatch.setattr("tools.lazy_deps._allow_lazy_installs", lambda: True)
|
|
r = ww.check_wake_word_requirements({"provider": "openwakeword"})
|
|
assert r["available"] is True
|
|
assert r["deps_available"] is False
|
|
assert r["hint"] == ""
|
|
|
|
|
|
def test_requirements_fresh_install_lazy_disabled(monkeypatch):
|
|
"""Deps missing + lazy installs disabled → unavailable, with the manual
|
|
pip command as the remediation hint."""
|
|
monkeypatch.setattr(ww, "_audio_available", lambda: True)
|
|
monkeypatch.setattr("tools.lazy_deps.is_available", lambda f: False)
|
|
monkeypatch.setattr("tools.lazy_deps._allow_lazy_installs", lambda: False)
|
|
monkeypatch.setattr(
|
|
"tools.lazy_deps.feature_install_command", lambda f: f"uv pip install {f}"
|
|
)
|
|
r = ww.check_wake_word_requirements({"provider": "openwakeword"})
|
|
assert r["available"] is False
|
|
assert r["deps_available"] is False
|
|
assert "install" in r["hint"]
|
|
|
|
|
|
def test_requirements_deps_present_but_no_audio_hint(monkeypatch):
|
|
"""Once deps ARE installed, a failing audio probe blocks with a mic hint
|
|
(lazy installs can't fix a missing audio device)."""
|
|
monkeypatch.setattr(ww, "_audio_available", lambda: False)
|
|
monkeypatch.setattr("tools.lazy_deps.is_available", lambda f: True)
|
|
monkeypatch.setattr("tools.lazy_deps._allow_lazy_installs", lambda: True)
|
|
r = ww.check_wake_word_requirements({"provider": "openwakeword"})
|
|
assert r["available"] is False
|
|
assert "audio device" in r["hint"]
|
|
|
|
|
|
# ── openWakeWord engine (bundled model + base-model fetch) ───────────────
|
|
|
|
|
|
def _install_fake_openwakeword(monkeypatch):
|
|
"""Swap in a fake ``openwakeword`` so the engine builds with no network.
|
|
|
|
Returns a ``calls`` dict recording every ``download_models`` invocation.
|
|
"""
|
|
calls = {"download": []}
|
|
|
|
class _FakeModel:
|
|
def __init__(self, wakeword_models, inference_framework="onnx"):
|
|
self.wakeword_models = list(wakeword_models)
|
|
self.models = {"hey_hermes": object()}
|
|
|
|
def predict(self, frame):
|
|
return {"hey_hermes": 0.0}
|
|
|
|
def reset(self):
|
|
pass
|
|
|
|
oww = types.ModuleType("openwakeword")
|
|
oww.utils = types.SimpleNamespace(
|
|
download_models=lambda names=[]: calls["download"].append(list(names))
|
|
)
|
|
model_mod = types.ModuleType("openwakeword.model")
|
|
model_mod.Model = _FakeModel
|
|
|
|
monkeypatch.setitem(sys.modules, "openwakeword", oww)
|
|
monkeypatch.setitem(sys.modules, "openwakeword.model", model_mod)
|
|
monkeypatch.setattr("tools.lazy_deps.ensure", lambda *a, **k: None)
|
|
return calls
|
|
|
|
|
|
def test_openwakeword_ensures_base_models_for_custom_path(monkeypatch):
|
|
# Regression: a custom ``.onnx`` path used to skip download_models entirely,
|
|
# so a fresh install crashed at load time on a missing melspectrogram.onnx.
|
|
# The base feature models must be ensured for a custom path too.
|
|
calls = _install_fake_openwakeword(monkeypatch)
|
|
eng = ww._OpenWakeWordEngine(
|
|
{"provider": "openwakeword", "openwakeword": {"model": "/models/hey_hermes.onnx"}}
|
|
)
|
|
assert calls["download"] == [["/models/hey_hermes.onnx"]]
|
|
assert eng._labels == ["hey_hermes"]
|
|
|
|
|
|
def test_openwakeword_fetches_builtin_by_name(monkeypatch):
|
|
calls = _install_fake_openwakeword(monkeypatch)
|
|
ww._OpenWakeWordEngine({"provider": "openwakeword", "openwakeword": {"model": "hey_jarvis"}})
|
|
assert calls["download"] == [["hey_jarvis"]]
|
|
|
|
|
|
def test_bundled_hey_hermes_model_ships_on_disk():
|
|
# The "hey hermes" wake word works out of the box only if the model is
|
|
# actually bundled. Both framework artifacts must exist and be non-trivial.
|
|
for framework in ("onnx", "tflite"):
|
|
path = ww._bundled_wakeword_path(framework)
|
|
assert os.path.exists(path), path
|
|
assert os.path.getsize(path) > 1024, path
|
|
|
|
|
|
@pytest.mark.parametrize("model_value", [None, "", "hey_hermes", "hey hermes", "HEY_HERMES"])
|
|
def test_openwakeword_default_resolves_to_bundled_model(monkeypatch, model_value):
|
|
# The default (and any "hey_hermes" alias) must load the bundled file, not be
|
|
# passed through as a bogus built-in name that openWakeWord can't resolve.
|
|
# Which artifact is bundled follows the platform's default backend (tflite on
|
|
# macOS ARM64, where openWakeWord's onnx path scores near-zero).
|
|
calls = _install_fake_openwakeword(monkeypatch)
|
|
sub = {} if model_value is None else {"model": model_value}
|
|
ww._OpenWakeWordEngine({"provider": "openwakeword", "openwakeword": sub})
|
|
(downloaded,) = calls["download"]
|
|
assert downloaded == [ww._bundled_wakeword_path(ww.default_inference_framework())]
|
|
|
|
|
|
def test_openwakeword_bundled_model_matches_framework(monkeypatch):
|
|
calls = _install_fake_openwakeword(monkeypatch)
|
|
# Pin the tflite runtime as present so this exercises artifact selection,
|
|
# not runtime availability — off-Darwin the bridge legitimately returns
|
|
# False and the engine falls back to onnx (covered separately below).
|
|
monkeypatch.setattr(ww, "ensure_tflite_runtime", lambda: True)
|
|
ww._OpenWakeWordEngine(
|
|
{"provider": "openwakeword", "openwakeword": {"inference_framework": "tflite"}}
|
|
)
|
|
(downloaded,) = calls["download"]
|
|
assert downloaded == [ww._bundled_wakeword_path("tflite")]
|
|
|
|
|
|
# ── platform-aware backend selection (openWakeWord onnx is broken on macOS ARM64,
|
|
# upstream dscripka/openWakeWord#336) ────────────────────────────────────────
|
|
|
|
def test_default_framework_is_tflite_on_macos_arm64(monkeypatch):
|
|
monkeypatch.setattr(ww.sys, "platform", "darwin")
|
|
monkeypatch.setattr("platform.machine", lambda: "arm64")
|
|
assert ww.default_inference_framework() == "tflite"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"plat,machine",
|
|
[("linux", "x86_64"), ("linux", "aarch64"), ("win32", "AMD64"), ("darwin", "x86_64")],
|
|
)
|
|
def test_default_framework_is_onnx_elsewhere(monkeypatch, plat, machine):
|
|
# Only the broken platform changes behaviour; everyone else keeps onnx.
|
|
monkeypatch.setattr(ww.sys, "platform", plat)
|
|
monkeypatch.setattr("platform.machine", lambda: machine)
|
|
assert ww.default_inference_framework() == "onnx"
|
|
|
|
|
|
def test_explicit_framework_overrides_platform_default(monkeypatch):
|
|
# An operator who pins a backend keeps it, even on macOS ARM64.
|
|
calls = _install_fake_openwakeword(monkeypatch)
|
|
monkeypatch.setattr(ww.sys, "platform", "darwin")
|
|
monkeypatch.setattr("platform.machine", lambda: "arm64")
|
|
ww._OpenWakeWordEngine(
|
|
{"provider": "openwakeword", "openwakeword": {"inference_framework": "onnx"}}
|
|
)
|
|
(downloaded,) = calls["download"]
|
|
assert downloaded == [ww._bundled_wakeword_path("onnx")]
|
|
|
|
|
|
# ── ambient-speech rejection: consecutive-frame confirmation ──────────────────
|
|
|
|
def _openwakeword_engine_with_scores(monkeypatch, cfg_wake, scores):
|
|
"""Build a real _OpenWakeWordEngine whose predict() replays ``scores``."""
|
|
seq = iter(scores)
|
|
|
|
class _ScriptedModel:
|
|
def __init__(self, wakeword_models, inference_framework="onnx"):
|
|
self.models = {"hey_hermes": object()}
|
|
|
|
def predict(self, frame):
|
|
return {"hey_hermes": next(seq)}
|
|
|
|
def reset(self):
|
|
pass
|
|
|
|
oww = types.ModuleType("openwakeword")
|
|
oww.utils = types.SimpleNamespace(download_models=lambda names=[]: None)
|
|
model_mod = types.ModuleType("openwakeword.model")
|
|
model_mod.Model = _ScriptedModel
|
|
monkeypatch.setitem(sys.modules, "openwakeword", oww)
|
|
monkeypatch.setitem(sys.modules, "openwakeword.model", model_mod)
|
|
monkeypatch.setattr("tools.lazy_deps.ensure", lambda *a, **k: None)
|
|
monkeypatch.setattr(ww, "ensure_tflite_runtime", lambda: True)
|
|
return ww._OpenWakeWordEngine({"provider": "openwakeword", **cfg_wake})
|
|
|
|
|
|
def test_confirmation_frames_reject_single_frame_spike(monkeypatch):
|
|
# A lone over-threshold frame (ambient phoneme) must NOT fire with the
|
|
# default 3-frame confirmation; the streak resets on the next quiet frame.
|
|
eng = _openwakeword_engine_with_scores(
|
|
monkeypatch,
|
|
{"sensitivity": 0.5, "confirmation_frames": 3},
|
|
[0.9, 0.0, 0.0, 0.9, 0.0],
|
|
)
|
|
assert [eng.process(None) for _ in range(5)] == [False, False, False, False, False]
|
|
|
|
|
|
def test_confirmation_frames_fire_on_sustained_phrase(monkeypatch):
|
|
# Three consecutive over-threshold frames (a real utterance) fire exactly
|
|
# once, on the third frame.
|
|
eng = _openwakeword_engine_with_scores(
|
|
monkeypatch,
|
|
{"sensitivity": 0.5, "confirmation_frames": 3},
|
|
[0.9, 0.9, 0.9, 0.0],
|
|
)
|
|
assert [eng.process(None) for _ in range(4)] == [False, False, True, False]
|
|
|
|
|
|
def test_confirmation_frames_one_restores_single_frame_behavior(monkeypatch):
|
|
# confirmation_frames=1 is the old behavior: fire on the first frame.
|
|
eng = _openwakeword_engine_with_scores(
|
|
monkeypatch,
|
|
{"sensitivity": 0.5, "confirmation_frames": 1},
|
|
[0.9, 0.0],
|
|
)
|
|
assert eng.process(None) is True
|
|
|
|
|
|
def test_confirmation_streak_resets_on_engine_reset(monkeypatch):
|
|
# A pause (reset) between two over-threshold frames must not let a
|
|
# pre-pause frame count toward the post-resume streak.
|
|
eng = _openwakeword_engine_with_scores(
|
|
monkeypatch,
|
|
{"sensitivity": 0.5, "confirmation_frames": 2},
|
|
[0.9, 0.9, 0.9],
|
|
)
|
|
assert eng.process(None) is False # streak = 1
|
|
eng.reset() # streak -> 0
|
|
assert eng.process(None) is False # streak = 1 again, not 2
|
|
assert eng.process(None) is True # streak = 2 -> fire
|
|
|
|
|
|
def test_confirmation_frames_config_clamped(monkeypatch):
|
|
assert ww._confirmation_frames({"confirmation_frames": 0}) == 1
|
|
assert ww._confirmation_frames({"confirmation_frames": 99}) == 10
|
|
assert ww._confirmation_frames({"confirmation_frames": "x"}) == ww._DEFAULT_CONFIRMATION_FRAMES
|
|
assert ww._confirmation_frames({}) == ww._DEFAULT_CONFIRMATION_FRAMES
|
|
|
|
|
|
def test_porcupine_sensitivity_is_inverted_to_match_shared_contract(monkeypatch):
|
|
# Our config contract is "higher sensitivity = stricter" for every engine.
|
|
# Porcupine's own `sensitivities` param means the OPPOSITE (higher = looser,
|
|
# more false alarms), so the engine must pass 1 - sensitivity.
|
|
captured = {}
|
|
|
|
class _FakePorcupine:
|
|
frame_length = 512
|
|
|
|
def process(self, frame):
|
|
return -1
|
|
|
|
def _create(**kwargs):
|
|
captured.update(kwargs)
|
|
return _FakePorcupine()
|
|
|
|
pv = types.ModuleType("pvporcupine")
|
|
pv.create = _create
|
|
monkeypatch.setitem(sys.modules, "pvporcupine", pv)
|
|
monkeypatch.setattr("tools.lazy_deps.ensure", lambda *a, **k: None)
|
|
monkeypatch.setenv("PORCUPINE_ACCESS_KEY", "test-key")
|
|
|
|
# Strict (0.9) → Porcupine gets a low 0.1 (few false alarms).
|
|
ww._PorcupineEngine({"provider": "porcupine", "sensitivity": 0.9})
|
|
assert captured["sensitivities"] == [pytest.approx(0.1)]
|
|
|
|
# Loose (0.2) → Porcupine gets a high 0.8.
|
|
ww._PorcupineEngine({"provider": "porcupine", "sensitivity": 0.2})
|
|
assert captured["sensitivities"] == [pytest.approx(0.8)]
|
|
|
|
|
|
def test_default_sensitivity_is_stricter_than_openwakeword_baseline():
|
|
# Regression: the 0.5 default let near-misses ("hey hor") through. The
|
|
# default must sit above openWakeWord's permissive 0.5 baseline.
|
|
assert ww._DEFAULTS["sensitivity"] >= 0.6
|
|
assert ww._sensitivity({}) >= 0.6
|
|
|
|
|
|
def test_macos_tflite_refuses_silent_onnx_downgrade(monkeypatch):
|
|
# openWakeWord silently falls back to onnx when no tflite runtime imports.
|
|
# On macOS ARM64 that lands on the broken backend, so we must raise instead
|
|
# of arming a listener that can never fire.
|
|
_install_fake_openwakeword(monkeypatch)
|
|
monkeypatch.setattr(ww.sys, "platform", "darwin")
|
|
monkeypatch.setattr("platform.machine", lambda: "arm64")
|
|
monkeypatch.setattr(ww, "ensure_tflite_runtime", lambda: False)
|
|
with pytest.raises(RuntimeError, match="ai-edge-litert"):
|
|
ww._OpenWakeWordEngine({"provider": "openwakeword", "openwakeword": {}})
|
|
|
|
|
|
def test_non_macos_tflite_falls_back_to_onnx(monkeypatch):
|
|
# Off macOS the onnx backend works, so a missing tflite runtime is a
|
|
# downgrade, not a hard failure.
|
|
calls = _install_fake_openwakeword(monkeypatch)
|
|
monkeypatch.setattr(ww.sys, "platform", "linux")
|
|
monkeypatch.setattr("platform.machine", lambda: "x86_64")
|
|
monkeypatch.setattr(ww, "ensure_tflite_runtime", lambda: False)
|
|
ww._OpenWakeWordEngine(
|
|
{"provider": "openwakeword", "openwakeword": {"inference_framework": "tflite"}}
|
|
)
|
|
(downloaded,) = calls["download"]
|
|
assert downloaded == [ww._bundled_wakeword_path("onnx")]
|
|
|
|
|
|
def test_requirements_report_missing_tflite_runtime(monkeypatch):
|
|
# A missing runtime must surface as unavailable + an actionable hint rather
|
|
# than an armed-but-deaf detector.
|
|
monkeypatch.setattr(ww.sys, "platform", "darwin")
|
|
monkeypatch.setattr("platform.machine", lambda: "arm64")
|
|
monkeypatch.setattr(ww, "ensure_tflite_runtime", lambda: False)
|
|
monkeypatch.setattr("tools.lazy_deps.is_available", lambda feature: feature != "wake.openwakeword.tflite")
|
|
monkeypatch.setattr("tools.lazy_deps._allow_lazy_installs", lambda: False)
|
|
monkeypatch.setattr(ww, "_audio_available", lambda: True)
|
|
reqs = ww.check_wake_word_requirements({"provider": "openwakeword", "openwakeword": {}})
|
|
assert reqs["available"] is False
|
|
assert "ai-edge-litert" in reqs["hint"]
|
|
|
|
|
|
# ── sherpa-onnx open-vocabulary engine ───────────────────────────────────
|
|
|
|
|
|
def _install_fake_sherpa(monkeypatch, tmp_path):
|
|
"""Fake sherpa_onnx + a fake model dir so the engine builds offline."""
|
|
calls = {"text2token": [], "spotter": [], "results": []}
|
|
|
|
model_dir = tmp_path / "kws-model"
|
|
model_dir.mkdir()
|
|
for name in (
|
|
"tokens.txt",
|
|
"bpe.model",
|
|
"encoder-epoch-12-avg-2-chunk-16-left-64.onnx",
|
|
"decoder-epoch-12-avg-2-chunk-16-left-64.onnx",
|
|
"joiner-epoch-12-avg-2-chunk-16-left-64.onnx",
|
|
):
|
|
(model_dir / name).write_bytes(b"x")
|
|
|
|
class _FakeStream:
|
|
def accept_waveform(self, sample_rate, samples):
|
|
pass
|
|
|
|
class _FakeSpotter:
|
|
def __init__(self, **kwargs):
|
|
calls["spotter"].append(kwargs)
|
|
|
|
def create_stream(self):
|
|
return _FakeStream()
|
|
|
|
def is_ready(self, stream):
|
|
return bool(calls["results"])
|
|
|
|
def decode_stream(self, stream):
|
|
pass
|
|
|
|
def get_result(self, stream):
|
|
return calls["results"].pop(0) if calls["results"] else ""
|
|
|
|
def reset_stream(self, stream):
|
|
pass
|
|
|
|
def _fake_text2token(phrases, tokens, tokens_type, bpe_model):
|
|
calls["text2token"].append(list(phrases))
|
|
return [p.split() for p in phrases]
|
|
|
|
sherpa = types.ModuleType("sherpa_onnx")
|
|
sherpa.KeywordSpotter = _FakeSpotter
|
|
sherpa.text2token = _fake_text2token
|
|
monkeypatch.setitem(sys.modules, "sherpa_onnx", sherpa)
|
|
monkeypatch.setattr("tools.lazy_deps.ensure", lambda *a, **k: None)
|
|
|
|
# numpy is an optional voice-extra dep, lazy-installed at runtime — CI's
|
|
# hermetic slices don't have it. process() only calls asarray(...)/32768,
|
|
# so a minimal stub keeps these tests runnable without the real package.
|
|
if "numpy" not in sys.modules:
|
|
class _FakeArr(list):
|
|
def __truediv__(self, other):
|
|
return self
|
|
|
|
np_stub = types.ModuleType("numpy")
|
|
np_stub.float32 = "float32"
|
|
np_stub.asarray = lambda x, dtype=None: _FakeArr(x)
|
|
monkeypatch.setitem(sys.modules, "numpy", np_stub)
|
|
return calls, model_dir
|
|
|
|
|
|
def test_sherpa_engine_tokenizes_configured_phrase_at_runtime(monkeypatch, tmp_path):
|
|
# The open-vocab core: the phrase from config is tokenized at runtime —
|
|
# no per-phrase model, no training artifact.
|
|
calls, model_dir = _install_fake_sherpa(monkeypatch, tmp_path)
|
|
eng = ww._SherpaKwsEngine({
|
|
"provider": "sherpa",
|
|
"phrase": "purple monkey dishwasher",
|
|
"sherpa": {"model_dir": str(model_dir)},
|
|
})
|
|
assert calls["text2token"] == [["PURPLE MONKEY DISHWASHER"]]
|
|
# keywords file was materialized with an underscored display name
|
|
with open(eng._keywords_file) as f:
|
|
line = f.read().strip()
|
|
assert line.endswith("@PURPLE_MONKEY_DISHWASHER")
|
|
eng.close()
|
|
assert not os.path.exists(eng._keywords_file)
|
|
|
|
|
|
def test_sherpa_engine_process_fires_and_resets(monkeypatch, tmp_path):
|
|
calls, model_dir = _install_fake_sherpa(monkeypatch, tmp_path)
|
|
eng = ww._SherpaKwsEngine({
|
|
"provider": "sherpa", "phrase": "hey hermes",
|
|
"sherpa": {"model_dir": str(model_dir)},
|
|
})
|
|
frame = [0] * eng.frame_length
|
|
assert eng.process(frame) is False # no result queued
|
|
calls["results"].append("HEY_HERMES")
|
|
assert eng.process(frame) is True # queued result → fire
|
|
old_stream = eng._stream
|
|
eng.reset()
|
|
assert eng._stream is not old_stream # fresh decoder state
|
|
|
|
|
|
def test_sherpa_provider_routing(monkeypatch, tmp_path):
|
|
calls, model_dir = _install_fake_sherpa(monkeypatch, tmp_path)
|
|
for alias in ("sherpa", "sherpa-onnx", "kws", "open"):
|
|
eng = ww._build_engine({
|
|
"provider": alias, "phrase": "x",
|
|
"sherpa": {"model_dir": str(model_dir)},
|
|
})
|
|
assert isinstance(eng, ww._SherpaKwsEngine)
|
|
|
|
|
|
def test_sherpa_requirements_probe_uses_sherpa_feature(monkeypatch):
|
|
seen = {}
|
|
monkeypatch.setattr(ww, "_audio_available", lambda: True)
|
|
monkeypatch.setattr(
|
|
"tools.lazy_deps.is_available", lambda f: seen.setdefault("feature", f) or True
|
|
)
|
|
r = ww.check_wake_word_requirements({"provider": "sherpa", "phrase": "anything at all"})
|
|
assert seen["feature"] == "wake.sherpa"
|
|
assert r["provider"] == "sherpa"
|
|
assert r["phrase"] == "anything at all"
|
|
|
|
|
|
# ── Multi-profile phrase routing ─────────────────────────────────────────
|
|
|
|
|
|
def test_sherpa_engine_enrolls_all_profile_phrases(monkeypatch, tmp_path):
|
|
calls, model_dir = _install_fake_sherpa(monkeypatch, tmp_path)
|
|
monkeypatch.setattr(ww, "_active_profile_name", lambda: "default")
|
|
monkeypatch.setattr(
|
|
ww, "enrolled_profile_phrases",
|
|
lambda: {"coder": "hey coder", "trader": "hey trader"},
|
|
)
|
|
eng = ww._SherpaKwsEngine({
|
|
"provider": "sherpa", "phrase": "hey hermes",
|
|
"sherpa": {"model_dir": str(model_dir)},
|
|
})
|
|
with open(eng._keywords_file, encoding="utf-8") as f:
|
|
lines = f.read().strip().splitlines()
|
|
assert len(lines) == 3
|
|
assert eng._display_to_profile == {
|
|
"HEY_HERMES": "default",
|
|
"HEY_CODER": "coder",
|
|
"HEY_TRADER": "trader",
|
|
}
|
|
eng.close()
|
|
|
|
|
|
def test_sherpa_engine_profile_routing_can_be_disabled(monkeypatch, tmp_path):
|
|
calls, model_dir = _install_fake_sherpa(monkeypatch, tmp_path)
|
|
monkeypatch.setattr(ww, "_active_profile_name", lambda: "default")
|
|
monkeypatch.setattr(
|
|
ww, "enrolled_profile_phrases", lambda: {"coder": "hey coder"}
|
|
)
|
|
eng = ww._SherpaKwsEngine({
|
|
"provider": "sherpa", "phrase": "hey hermes", "profile_routing": False,
|
|
"sherpa": {"model_dir": str(model_dir)},
|
|
})
|
|
assert eng._display_to_profile == {"HEY_HERMES": "default"}
|
|
eng.close()
|
|
|
|
|
|
def test_sherpa_engine_match_maps_back_to_profile(monkeypatch, tmp_path):
|
|
calls, model_dir = _install_fake_sherpa(monkeypatch, tmp_path)
|
|
monkeypatch.setattr(ww, "_active_profile_name", lambda: "default")
|
|
monkeypatch.setattr(
|
|
ww, "enrolled_profile_phrases", lambda: {"coder": "hey coder"}
|
|
)
|
|
eng = ww._SherpaKwsEngine({
|
|
"provider": "sherpa", "phrase": "hey hermes",
|
|
"sherpa": {"model_dir": str(model_dir)},
|
|
})
|
|
frame = [0] * eng.frame_length
|
|
calls["results"].append("HEY_CODER")
|
|
assert eng.process(frame) is True
|
|
assert eng.last_match == ("hey coder", "coder")
|
|
calls["results"].append("HEY_HERMES")
|
|
assert eng.process(frame) is True
|
|
assert eng.last_match == ("hey hermes", "default")
|
|
|
|
|
|
def test_enrolled_profile_phrases_reads_profile_configs(monkeypatch, tmp_path):
|
|
profiles_root = tmp_path / "profiles"
|
|
for name, body in (
|
|
("coder", "wake_word:\n enabled: true\n phrase: hey coder\n"),
|
|
("trader", "wake_word:\n enabled: true\n"), # phrase defaults
|
|
("quiet", "wake_word:\n enabled: false\n"), # not enrolled
|
|
("empty", ""), # no wake_word at all
|
|
):
|
|
d = profiles_root / name
|
|
d.mkdir(parents=True)
|
|
(d / "config.yaml").write_text(body, encoding="utf-8")
|
|
|
|
class _Info:
|
|
def __init__(self, name):
|
|
self.name = name
|
|
|
|
import types as _types
|
|
fake_profiles = _types.ModuleType("hermes_cli.profiles")
|
|
fake_profiles.list_profiles = lambda: [
|
|
_Info(p.name) for p in sorted(profiles_root.iterdir())
|
|
]
|
|
fake_profiles.get_profile_dir = lambda name: str(profiles_root / name)
|
|
fake_profiles.get_active_profile_name = lambda: "default"
|
|
monkeypatch.setitem(sys.modules, "hermes_cli.profiles", fake_profiles)
|
|
|
|
phrases = ww.enrolled_profile_phrases()
|
|
assert phrases == {"coder": "hey coder", "trader": "hey trader"}
|
|
|
|
|
|
def test_get_last_match_reads_detector_engine(monkeypatch):
|
|
class _Eng:
|
|
last_match = ("hey coder", "coder")
|
|
|
|
class _Det:
|
|
engine = _Eng()
|
|
|
|
monkeypatch.setattr(ww, "_detector", _Det())
|
|
assert ww.get_last_match() == ("hey coder", "coder")
|
|
monkeypatch.setattr(ww, "_detector", None)
|
|
assert ww.get_last_match() is None
|
|
|
|
|
|
# ── Detector loop ────────────────────────────────────────────────────────
|
|
|
|
|
|
class _FakeStream:
|
|
"""Always-readable input stream that yields trivial frames."""
|
|
|
|
def __init__(self, **_kw):
|
|
self.closed = False
|
|
|
|
def start(self):
|
|
pass
|
|
|
|
def read(self, n):
|
|
time.sleep(0.01)
|
|
return [0] * n, False
|
|
|
|
def stop(self):
|
|
pass
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
class _FakeEngine:
|
|
frame_length = 4
|
|
|
|
def __init__(self, fire=True):
|
|
self._fire = fire
|
|
self.closed = False
|
|
self.resets = 0
|
|
|
|
def process(self, frame):
|
|
return self._fire
|
|
|
|
def reset(self):
|
|
self.resets += 1
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
def _fake_audio(monkeypatch):
|
|
fake_sd = types.SimpleNamespace(InputStream=lambda **kw: _FakeStream(**kw))
|
|
monkeypatch.setattr(ww, "_import_audio", lambda: (fake_sd, None))
|
|
|
|
|
|
class _Frame(list):
|
|
"""List with numpy-ish abs()/max() so the silence probe sees real peaks."""
|
|
|
|
def __abs__(self):
|
|
return _Frame(abs(x) for x in self)
|
|
|
|
def max(self):
|
|
return max(self) if self else 0
|
|
|
|
|
|
class _SilentStream(_FakeStream):
|
|
"""Stream that always yields near-zero frames (dead macOS mic)."""
|
|
|
|
def read(self, n):
|
|
time.sleep(0.005)
|
|
return _Frame([0] * n), False
|
|
|
|
|
|
class _LoudStream(_FakeStream):
|
|
"""Stream that yields audible frames."""
|
|
|
|
def read(self, n):
|
|
time.sleep(0.005)
|
|
return _Frame([500] * n), False
|
|
|
|
|
|
def test_detector_flags_silent_stream_and_recovers(monkeypatch):
|
|
"""A stream of zeros sets audio_silent (macOS no-permission mode); audio clears it."""
|
|
monkeypatch.setattr(ww, "_SILENCE_ALERT_SECONDS", 0.001) # trip on the first frame
|
|
stream_cls = {"cls": _SilentStream}
|
|
fake_sd = types.SimpleNamespace(InputStream=lambda **kw: stream_cls["cls"](**kw))
|
|
monkeypatch.setattr(ww, "_import_audio", lambda: (fake_sd, None))
|
|
|
|
det = ww.WakeWordDetector(_FakeEngine(fire=False), lambda: None)
|
|
det.start()
|
|
try:
|
|
deadline = time.monotonic() + 2.0
|
|
while not det.audio_silent and time.monotonic() < deadline:
|
|
time.sleep(0.01)
|
|
assert det.audio_silent is True
|
|
assert ww.audio_is_silent() is False # module accessor needs the singleton
|
|
|
|
monkeypatch.setattr(ww, "_detector", det)
|
|
assert ww.audio_is_silent() is True
|
|
|
|
# Audio returns (permission granted / real mic) — flag clears.
|
|
det.pause()
|
|
stream_cls["cls"] = _LoudStream
|
|
det.resume()
|
|
deadline = time.monotonic() + 2.0
|
|
while det.audio_silent and time.monotonic() < deadline:
|
|
time.sleep(0.01)
|
|
assert det.audio_silent is False
|
|
finally:
|
|
monkeypatch.setattr(ww, "_detector", None)
|
|
det.stop()
|
|
|
|
|
|
def test_detector_fires_once_under_cooldown(monkeypatch):
|
|
_fake_audio(monkeypatch)
|
|
calls = []
|
|
eng = _FakeEngine(fire=True)
|
|
det = ww.WakeWordDetector(eng, lambda: calls.append(1), cooldown=10.0)
|
|
det.start()
|
|
time.sleep(0.25)
|
|
det.stop()
|
|
assert len(calls) == 1 # high cooldown suppresses repeats
|
|
assert eng.closed is True
|
|
assert det.running is False
|
|
|
|
|
|
def test_detector_refires_after_cooldown(monkeypatch):
|
|
_fake_audio(monkeypatch)
|
|
calls = []
|
|
det = ww.WakeWordDetector(_FakeEngine(fire=True), lambda: calls.append(1), cooldown=0.05)
|
|
det.start()
|
|
time.sleep(0.3)
|
|
det.stop()
|
|
assert len(calls) >= 2
|
|
|
|
|
|
def test_detector_no_fire_when_engine_quiet(monkeypatch):
|
|
_fake_audio(monkeypatch)
|
|
calls = []
|
|
det = ww.WakeWordDetector(_FakeEngine(fire=False), lambda: calls.append(1))
|
|
det.start()
|
|
time.sleep(0.15)
|
|
det.stop()
|
|
assert calls == []
|
|
|
|
|
|
def test_detector_resets_engine_on_each_start(monkeypatch):
|
|
# Clearing the engine buffer on (re)start is what stops a resume right after
|
|
# a voice turn from re-firing on stale audio (the runaway wake loop).
|
|
_fake_audio(monkeypatch)
|
|
eng = _FakeEngine(fire=False)
|
|
det = ww.WakeWordDetector(eng, lambda: None)
|
|
det.start()
|
|
time.sleep(0.05)
|
|
det.pause()
|
|
det.resume()
|
|
time.sleep(0.05)
|
|
det.stop()
|
|
assert eng.resets >= 2 # initial start + resume
|
|
|
|
|
|
def test_detector_pause_resume(monkeypatch):
|
|
_fake_audio(monkeypatch)
|
|
det = ww.WakeWordDetector(_FakeEngine(fire=False), lambda: None)
|
|
det.start()
|
|
time.sleep(0.05)
|
|
assert det.running is True
|
|
det.pause()
|
|
assert det.running is False
|
|
det.resume()
|
|
time.sleep(0.05)
|
|
assert det.running is True
|
|
det.stop()
|
|
assert det.running is False
|
|
|
|
|
|
# ── Singleton lifecycle ──────────────────────────────────────────────────
|
|
|
|
|
|
def test_singleton_lifecycle(monkeypatch, tmp_path):
|
|
_fake_audio(monkeypatch)
|
|
monkeypatch.setattr(ww, "_build_engine", lambda cfg: _FakeEngine(fire=False))
|
|
monkeypatch.setattr(ww, "_lock_path", lambda: tmp_path / "wake.lock")
|
|
owner = object()
|
|
|
|
assert ww.is_listening() is False
|
|
det = ww.start_listening(lambda: None, owner=owner, config={})
|
|
time.sleep(0.05)
|
|
assert ww.is_listening() is True
|
|
assert ww.owns_listener(owner) is True
|
|
|
|
# Re-entrant start returns the same detector and re-arms it.
|
|
det2 = ww.start_listening(lambda: None, owner=owner, config={})
|
|
assert det2 is det
|
|
|
|
assert ww.pause_listening(owner=owner) is True
|
|
assert ww.is_listening() is False
|
|
assert ww.resume_listening(owner=owner) is True
|
|
time.sleep(0.05)
|
|
assert ww.is_listening() is True
|
|
|
|
assert ww.stop_listening(owner=owner) is True
|
|
assert ww.is_listening() is False
|
|
|
|
|
|
def test_second_owner_cannot_mutate_listener(monkeypatch, tmp_path):
|
|
_fake_audio(monkeypatch)
|
|
monkeypatch.setattr(ww, "_build_engine", lambda cfg: _FakeEngine(fire=False))
|
|
monkeypatch.setattr(ww, "_lock_path", lambda: tmp_path / "wake.lock")
|
|
owner, intruder = object(), object()
|
|
first_callback = lambda: None
|
|
|
|
detector = ww.start_listening(first_callback, owner=owner, config={})
|
|
with pytest.raises(ww.WakeWordInUse):
|
|
ww.start_listening(lambda: None, owner=intruder, config={})
|
|
|
|
assert detector.on_wake is first_callback
|
|
assert ww.pause_listening(owner=intruder) is False
|
|
assert ww.resume_listening(owner=intruder) is False
|
|
assert ww.stop_listening(owner=intruder) is False
|
|
assert ww.owns_listener(owner) is True
|
|
assert ww.stop_listening(owner=owner) is True
|
|
|
|
|
|
def test_detection_callback_can_pause_and_close_stream(monkeypatch, tmp_path):
|
|
streams = []
|
|
|
|
def _stream(**kw):
|
|
stream = _FakeStream(**kw)
|
|
streams.append(stream)
|
|
return stream
|
|
|
|
fake_sd = types.SimpleNamespace(InputStream=_stream)
|
|
monkeypatch.setattr(ww, "_import_audio", lambda: (fake_sd, None))
|
|
monkeypatch.setattr(ww, "_build_engine", lambda cfg: _FakeEngine(fire=True))
|
|
monkeypatch.setattr(ww, "_lock_path", lambda: tmp_path / "wake.lock")
|
|
owner = object()
|
|
paused = threading.Event()
|
|
|
|
def _on_wake():
|
|
if ww.pause_listening(owner=owner):
|
|
paused.set()
|
|
|
|
ww.start_listening(_on_wake, owner=owner, config={})
|
|
assert paused.wait(2)
|
|
assert ww.is_listening() is False
|
|
assert streams[0].closed is True
|
|
assert ww.stop_listening(owner=owner) is True
|
|
|
|
|
|
def test_startup_failure_releases_owner_and_machine_lock(monkeypatch, tmp_path):
|
|
class _BrokenSoundDevice:
|
|
@staticmethod
|
|
def InputStream(**_kw):
|
|
raise OSError("no microphone")
|
|
|
|
lock_path = tmp_path / "wake.lock"
|
|
monkeypatch.setattr(ww, "_import_audio", lambda: (_BrokenSoundDevice, None))
|
|
monkeypatch.setattr(ww, "_build_engine", lambda cfg: _FakeEngine(fire=False))
|
|
monkeypatch.setattr(ww, "_lock_path", lambda: lock_path)
|
|
owner = object()
|
|
|
|
with pytest.raises(RuntimeError, match="Failed to open"):
|
|
ww.start_listening(lambda: None, owner=owner, config={})
|
|
|
|
assert ww.owns_listener(owner) is False
|
|
handle = ww._acquire_machine_lock(lock_path)
|
|
ww._release_machine_lock(handle)
|
|
|
|
|
|
def test_stream_failure_releases_owner_and_machine_lock(monkeypatch, tmp_path):
|
|
class _FailingStream(_FakeStream):
|
|
def read(self, _n):
|
|
raise OSError("device disconnected")
|
|
|
|
fake_sd = types.SimpleNamespace(InputStream=lambda **kw: _FailingStream(**kw))
|
|
engine = _FakeEngine(fire=False)
|
|
lock_path = tmp_path / "wake.lock"
|
|
monkeypatch.setattr(ww, "_import_audio", lambda: (fake_sd, None))
|
|
monkeypatch.setattr(ww, "_build_engine", lambda cfg: engine)
|
|
monkeypatch.setattr(ww, "_lock_path", lambda: lock_path)
|
|
owner = object()
|
|
|
|
ww.start_listening(lambda: None, owner=owner, config={})
|
|
deadline = time.time() + 2
|
|
while ww.owns_listener(owner) and time.time() < deadline:
|
|
time.sleep(0.01)
|
|
|
|
assert ww.owns_listener(owner) is False
|
|
assert engine.closed is True
|
|
handle = ww._acquire_machine_lock(lock_path)
|
|
ww._release_machine_lock(handle)
|
|
|
|
|
|
def _hold_machine_lock(path: str, ready, release) -> None:
|
|
from tools import wake_word
|
|
|
|
handle = wake_word._acquire_machine_lock(Path(path))
|
|
ready.set()
|
|
release.wait(10)
|
|
assert handle is not None
|
|
|
|
|
|
def test_machine_lock_is_released_when_owner_process_exits(tmp_path):
|
|
lock_path = tmp_path / "wake.lock"
|
|
ctx = multiprocessing.get_context("spawn")
|
|
ready = ctx.Event()
|
|
release = ctx.Event()
|
|
process = ctx.Process(
|
|
target=_hold_machine_lock,
|
|
args=(str(lock_path), ready, release),
|
|
)
|
|
process.start()
|
|
try:
|
|
assert ready.wait(10)
|
|
with pytest.raises(ww.WakeWordInUse):
|
|
ww._acquire_machine_lock(lock_path)
|
|
release.set()
|
|
process.join(10)
|
|
assert process.exitcode == 0
|
|
handle = ww._acquire_machine_lock(lock_path)
|
|
ww._release_machine_lock(handle)
|
|
finally:
|
|
release.set()
|
|
if process.is_alive():
|
|
process.terminate()
|
|
process.join(10)
|