Rename project from lel to aman
This commit is contained in:
parent
2c570c7a87
commit
09090102a2
12 changed files with 77 additions and 77 deletions
|
|
@ -1,246 +0,0 @@
|
|||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SRC = ROOT / "src"
|
||||
if str(SRC) not in sys.path:
|
||||
sys.path.insert(0, str(SRC))
|
||||
|
||||
import leld
|
||||
from config import Config, VocabularyReplacement
|
||||
|
||||
|
||||
class FakeDesktop:
|
||||
def __init__(self):
|
||||
self.inject_calls = []
|
||||
self.quit_calls = 0
|
||||
|
||||
def inject_text(
|
||||
self,
|
||||
text: str,
|
||||
backend: str,
|
||||
*,
|
||||
remove_transcription_from_clipboard: bool = False,
|
||||
) -> None:
|
||||
self.inject_calls.append((text, backend, remove_transcription_from_clipboard))
|
||||
|
||||
def request_quit(self) -> None:
|
||||
self.quit_calls += 1
|
||||
|
||||
|
||||
class FakeSegment:
|
||||
def __init__(self, text: str):
|
||||
self.text = text
|
||||
|
||||
|
||||
class FakeModel:
|
||||
def __init__(self, text: str = "hello world"):
|
||||
self.text = text
|
||||
self.last_kwargs = {}
|
||||
|
||||
def transcribe(self, _audio, language=None, vad_filter=None):
|
||||
self.last_kwargs = {
|
||||
"language": language,
|
||||
"vad_filter": vad_filter,
|
||||
}
|
||||
return [FakeSegment(self.text)], self.last_kwargs
|
||||
|
||||
|
||||
class FakeHintModel:
|
||||
def __init__(self, text: str = "hello world"):
|
||||
self.text = text
|
||||
self.last_kwargs = {}
|
||||
|
||||
def transcribe(
|
||||
self,
|
||||
_audio,
|
||||
language=None,
|
||||
vad_filter=None,
|
||||
hotwords=None,
|
||||
initial_prompt=None,
|
||||
):
|
||||
self.last_kwargs = {
|
||||
"language": language,
|
||||
"vad_filter": vad_filter,
|
||||
"hotwords": hotwords,
|
||||
"initial_prompt": initial_prompt,
|
||||
}
|
||||
return [FakeSegment(self.text)], self.last_kwargs
|
||||
|
||||
|
||||
class FakeAIProcessor:
|
||||
def process(self, text, lang="en", **_kwargs):
|
||||
return text
|
||||
|
||||
|
||||
class FakeAudio:
|
||||
def __init__(self, size: int):
|
||||
self.size = size
|
||||
|
||||
|
||||
class DaemonTests(unittest.TestCase):
|
||||
def _config(self) -> Config:
|
||||
cfg = Config()
|
||||
return cfg
|
||||
|
||||
@patch("leld.stop_audio_recording", return_value=FakeAudio(8))
|
||||
@patch("leld.start_audio_recording", return_value=(object(), object()))
|
||||
def test_toggle_start_stop_injects_text(self, _start_mock, _stop_mock):
|
||||
desktop = FakeDesktop()
|
||||
with patch("leld._build_whisper_model", return_value=FakeModel()):
|
||||
daemon = leld.Daemon(self._config(), desktop, verbose=False)
|
||||
daemon.ai_processor = FakeAIProcessor()
|
||||
daemon._start_stop_worker = (
|
||||
lambda stream, record, trigger, process_audio: daemon._stop_and_process(
|
||||
stream, record, trigger, process_audio
|
||||
)
|
||||
)
|
||||
|
||||
daemon.toggle()
|
||||
self.assertEqual(daemon.get_state(), leld.State.RECORDING)
|
||||
|
||||
daemon.toggle()
|
||||
|
||||
self.assertEqual(daemon.get_state(), leld.State.IDLE)
|
||||
self.assertEqual(desktop.inject_calls, [("hello world", "clipboard", False)])
|
||||
|
||||
@patch("leld.stop_audio_recording", return_value=FakeAudio(8))
|
||||
@patch("leld.start_audio_recording", return_value=(object(), object()))
|
||||
def test_shutdown_stops_recording_without_injection(self, _start_mock, _stop_mock):
|
||||
desktop = FakeDesktop()
|
||||
with patch("leld._build_whisper_model", return_value=FakeModel()):
|
||||
daemon = leld.Daemon(self._config(), desktop, verbose=False)
|
||||
daemon.ai_processor = FakeAIProcessor()
|
||||
daemon._start_stop_worker = (
|
||||
lambda stream, record, trigger, process_audio: daemon._stop_and_process(
|
||||
stream, record, trigger, process_audio
|
||||
)
|
||||
)
|
||||
|
||||
daemon.toggle()
|
||||
self.assertEqual(daemon.get_state(), leld.State.RECORDING)
|
||||
|
||||
self.assertTrue(daemon.shutdown(timeout=0.2))
|
||||
self.assertEqual(daemon.get_state(), leld.State.IDLE)
|
||||
self.assertEqual(desktop.inject_calls, [])
|
||||
|
||||
@patch("leld.stop_audio_recording", return_value=FakeAudio(8))
|
||||
@patch("leld.start_audio_recording", return_value=(object(), object()))
|
||||
def test_dictionary_replacement_applies_after_ai(self, _start_mock, _stop_mock):
|
||||
desktop = FakeDesktop()
|
||||
model = FakeModel(text="good morning martha")
|
||||
cfg = self._config()
|
||||
cfg.vocabulary.replacements = [VocabularyReplacement(source="Martha", target="Marta")]
|
||||
|
||||
with patch("leld._build_whisper_model", return_value=model):
|
||||
daemon = leld.Daemon(cfg, desktop, verbose=False)
|
||||
daemon.ai_processor = FakeAIProcessor()
|
||||
daemon._start_stop_worker = (
|
||||
lambda stream, record, trigger, process_audio: daemon._stop_and_process(
|
||||
stream, record, trigger, process_audio
|
||||
)
|
||||
)
|
||||
|
||||
daemon.toggle()
|
||||
daemon.toggle()
|
||||
|
||||
self.assertEqual(desktop.inject_calls, [("good morning Marta", "clipboard", False)])
|
||||
|
||||
def test_transcribe_skips_hints_when_model_does_not_support_them(self):
|
||||
desktop = FakeDesktop()
|
||||
model = FakeModel(text="hello")
|
||||
cfg = self._config()
|
||||
cfg.vocabulary.terms = ["Docker", "Systemd"]
|
||||
|
||||
with patch("leld._build_whisper_model", return_value=model):
|
||||
daemon = leld.Daemon(cfg, desktop, verbose=False)
|
||||
|
||||
result = daemon._transcribe(object())
|
||||
|
||||
self.assertEqual(result, "hello")
|
||||
self.assertNotIn("hotwords", model.last_kwargs)
|
||||
self.assertNotIn("initial_prompt", model.last_kwargs)
|
||||
|
||||
def test_transcribe_applies_hints_when_model_supports_them(self):
|
||||
desktop = FakeDesktop()
|
||||
model = FakeHintModel(text="hello")
|
||||
cfg = self._config()
|
||||
cfg.vocabulary.terms = ["Systemd"]
|
||||
cfg.vocabulary.replacements = [VocabularyReplacement(source="docker", target="Docker")]
|
||||
|
||||
with patch("leld._build_whisper_model", return_value=model):
|
||||
daemon = leld.Daemon(cfg, desktop, verbose=False)
|
||||
|
||||
result = daemon._transcribe(object())
|
||||
|
||||
self.assertEqual(result, "hello")
|
||||
self.assertIn("Docker", model.last_kwargs["hotwords"])
|
||||
self.assertIn("Systemd", model.last_kwargs["hotwords"])
|
||||
self.assertIn("Preferred vocabulary", model.last_kwargs["initial_prompt"])
|
||||
|
||||
def test_verbose_flag_controls_transcript_logging(self):
|
||||
desktop = FakeDesktop()
|
||||
cfg = self._config()
|
||||
|
||||
with patch("leld._build_whisper_model", return_value=FakeModel()):
|
||||
daemon = leld.Daemon(cfg, desktop, verbose=False)
|
||||
self.assertFalse(daemon.log_transcript)
|
||||
|
||||
with patch("leld._build_whisper_model", return_value=FakeModel()):
|
||||
daemon_verbose = leld.Daemon(cfg, desktop, verbose=True)
|
||||
self.assertTrue(daemon_verbose.log_transcript)
|
||||
|
||||
@patch("leld.stop_audio_recording", return_value=FakeAudio(8))
|
||||
@patch("leld.start_audio_recording", return_value=(object(), object()))
|
||||
def test_passes_clipboard_remove_option_to_desktop(self, _start_mock, _stop_mock):
|
||||
desktop = FakeDesktop()
|
||||
model = FakeModel(text="hello world")
|
||||
cfg = self._config()
|
||||
cfg.injection.remove_transcription_from_clipboard = True
|
||||
|
||||
with patch("leld._build_whisper_model", return_value=model):
|
||||
daemon = leld.Daemon(cfg, desktop, verbose=False)
|
||||
daemon.ai_processor = FakeAIProcessor()
|
||||
daemon._start_stop_worker = (
|
||||
lambda stream, record, trigger, process_audio: daemon._stop_and_process(
|
||||
stream, record, trigger, process_audio
|
||||
)
|
||||
)
|
||||
|
||||
daemon.toggle()
|
||||
daemon.toggle()
|
||||
|
||||
self.assertEqual(desktop.inject_calls, [("hello world", "clipboard", True)])
|
||||
|
||||
def test_state_changes_are_debug_level(self):
|
||||
desktop = FakeDesktop()
|
||||
with patch("leld._build_whisper_model", return_value=FakeModel()):
|
||||
daemon = leld.Daemon(self._config(), desktop, verbose=False)
|
||||
|
||||
with self.assertLogs(level="DEBUG") as logs:
|
||||
daemon.set_state(leld.State.RECORDING)
|
||||
|
||||
self.assertTrue(
|
||||
any("DEBUG:root:state: idle -> recording" in line for line in logs.output)
|
||||
)
|
||||
|
||||
|
||||
class LockTests(unittest.TestCase):
|
||||
def test_lock_rejects_second_instance(self):
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
with patch.dict(os.environ, {"XDG_RUNTIME_DIR": td}, clear=False):
|
||||
first = leld._lock_single_instance()
|
||||
try:
|
||||
with self.assertRaises(SystemExit) as ctx:
|
||||
leld._lock_single_instance()
|
||||
self.assertIn("already running", str(ctx.exception))
|
||||
finally:
|
||||
first.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
Add table
Add a link
Reference in a new issue