113 lines
3.2 KiB
Python
113 lines
3.2 KiB
Python
import sys
|
|
import types
|
|
import unittest
|
|
import importlib
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
|
|
|
soundGeneratorStub = types.ModuleType("cthulhu.sound_generator")
|
|
|
|
|
|
class _Icon:
|
|
def __init__(self, path="", name=""):
|
|
self.path = path
|
|
self.name = name
|
|
|
|
def isValid(self):
|
|
return True
|
|
|
|
|
|
class _Tone:
|
|
def __init__(self, duration=0.1, frequency=440, volume=1.0, wave=0):
|
|
self.duration = duration
|
|
self.frequency = frequency
|
|
self.volume = volume
|
|
self.wave = wave
|
|
|
|
|
|
soundGeneratorStub.Icon = _Icon
|
|
soundGeneratorStub.Tone = _Tone
|
|
sys.modules.setdefault("cthulhu.sound_generator", soundGeneratorStub)
|
|
|
|
from cthulhu import settings
|
|
from cthulhu import piper_audio_player
|
|
from cthulhu import sound
|
|
from cthulhu import sound_sink
|
|
|
|
|
|
class _FakeProcess:
|
|
def poll(self):
|
|
return None
|
|
|
|
|
|
class SoundSinkTests(unittest.TestCase):
|
|
def test_auto_sink_prefers_autoaudiosink(self):
|
|
candidates = sound_sink._get_sink_element_candidates(settings.SOUND_SINK_AUTO)
|
|
self.assertGreaterEqual(len(candidates), 1)
|
|
self.assertEqual(candidates[0], "autoaudiosink")
|
|
|
|
def test_gstreamer_init_check_bool_result_does_not_break_imports(self):
|
|
modules = [
|
|
sound_sink,
|
|
sound,
|
|
piper_audio_player,
|
|
]
|
|
|
|
with mock.patch.object(sound_sink.Gst, "init_check", return_value=True):
|
|
reloadedModules = []
|
|
for module in modules:
|
|
reloaded = importlib.reload(module)
|
|
reloadedModules.append(reloaded)
|
|
self.assertTrue(reloaded._gstreamerAvailable)
|
|
|
|
for module in reloadedModules:
|
|
importlib.reload(module)
|
|
|
|
|
|
class PlayerRecoveryTests(unittest.TestCase):
|
|
def test_worker_diagnostic_marks_restart_required(self):
|
|
player = sound.Player()
|
|
|
|
player._handleWorkerDiagnostic("RECOVERY REQUIRED: lost audio sink")
|
|
|
|
self.assertTrue(player._workerRestartRequired)
|
|
self.assertEqual(player._workerRestartReason, "lost audio sink")
|
|
|
|
def test_ensure_worker_restarts_when_recovery_is_required(self):
|
|
player = sound.Player()
|
|
player._workerProcess = _FakeProcess()
|
|
player._workerSink = settings.SOUND_SINK_AUTO
|
|
player._workerRestartRequired = True
|
|
player._workerRestartReason = "lost audio sink"
|
|
|
|
stopReasons = []
|
|
startedSinks = []
|
|
|
|
with (
|
|
mock.patch.object(
|
|
sound_sink,
|
|
"get_configured_sound_sink",
|
|
return_value=settings.SOUND_SINK_AUTO,
|
|
),
|
|
mock.patch.object(
|
|
player,
|
|
"_stopWorkerLocked",
|
|
side_effect=lambda reason: stopReasons.append(reason),
|
|
),
|
|
mock.patch.object(
|
|
player,
|
|
"_startWorkerLocked",
|
|
side_effect=lambda configuredSink: startedSinks.append(configuredSink) or True,
|
|
),
|
|
):
|
|
self.assertTrue(player._ensureWorkerLocked())
|
|
|
|
self.assertEqual(stopReasons, ["lost audio sink"])
|
|
self.assertEqual(startedSinks, [settings.SOUND_SINK_AUTO])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|