320 lines
17 KiB
Python
320 lines
17 KiB
Python
"""Focused behavior checks; audio commands are mocked so tests stay silent."""
|
|
from dataclasses import replace
|
|
from datetime import datetime
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
|
from talking_clock.audio import Audio, spoken_time
|
|
from talking_clock.config import Config, Settings, read_legacy, resolve_soundpack, soundpacks
|
|
from talking_clock.runtime import Clock, Instance, Scheduler, due, runtime_directory
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
class SettingsTests(unittest.TestCase):
|
|
def test_round_trip_and_atomic_replacement(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
config = Config(directory)
|
|
settings = Settings(module="espeak-ng", voice="English (America)", rate=37,
|
|
font_size=100, weather_on_space=True)
|
|
config.save(settings)
|
|
self.assertEqual(config.load(), settings)
|
|
config.save(replace(settings, font_size=120))
|
|
self.assertEqual(config.load().font_size, 120)
|
|
self.assertEqual(list(Path(directory).iterdir()), [config.path])
|
|
self.assertEqual(config.path.stat().st_mode & 0o777, 0o600)
|
|
|
|
def test_invalid_settings_preserve_existing_file(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
config = Config(directory)
|
|
config.save(Settings())
|
|
with self.assertRaises(ValueError):
|
|
config.save(Settings(font_size=0))
|
|
self.assertEqual(config.load().font_size, 72)
|
|
for data in ({"hourly": "false"}, {"font_size": True}, {"rate": 101}, {"volume": "10"}, []):
|
|
config.path.write_text(json.dumps(data))
|
|
with self.assertRaises(ValueError):
|
|
config.load()
|
|
|
|
def test_legacy_import_is_data_and_preserves_location(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "talking-clockrc"
|
|
path.write_text('format=24\nspeak=false\nvoice="rhvoice-alan"\n'
|
|
'zipcode="Raleigh North Carolina"\nsound=play -q\n'
|
|
'ignored=$(touch /tmp/this-must-never-run)\n')
|
|
settings = read_legacy(path)
|
|
self.assertEqual(settings.voice, "")
|
|
self.assertEqual(settings.module, "")
|
|
self.assertFalse(settings.speak)
|
|
self.assertTrue(settings.weather)
|
|
self.assertEqual(settings.location, "Raleigh North Carolina")
|
|
self.assertEqual(Config(directory).load(), settings)
|
|
|
|
def test_user_soundpack_wins_and_xdg_is_respected(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
user, system = Path(directory) / "user", Path(directory) / "system"
|
|
for root in (user, system):
|
|
(root / "talking-clock/sounds/bells").mkdir(parents=True)
|
|
with patch.dict(os.environ, {"XDG_DATA_HOME": str(user), "XDG_DATA_DIRS": str(system)}):
|
|
self.assertEqual(soundpacks()["bells"], user / "talking-clock/sounds/bells")
|
|
self.assertEqual(resolve_soundpack("bells"), user / "talking-clock/sounds/bells")
|
|
self.assertTrue(resolve_soundpack("default").is_dir())
|
|
self.assertEqual(resolve_soundpack(str(system)), system)
|
|
|
|
def test_live_reload_keeps_previous_settings_on_error(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
config = Config(directory)
|
|
config.save(Settings(speak=False, chime=False))
|
|
instance = Mock()
|
|
clock = Clock(config, config.load(), instance)
|
|
config.save(Settings(speak=False, chime=False, font_size=100))
|
|
clock.tick()
|
|
self.assertEqual(clock.settings.font_size, 100)
|
|
config.path.write_text('{"font_size": "bad"}')
|
|
clock.last_check = 0
|
|
with self.assertLogs("talking_clock.runtime", level="ERROR"):
|
|
clock.tick()
|
|
self.assertEqual(clock.settings.font_size, 100)
|
|
clock.stop()
|
|
|
|
|
|
class SchedulingTests(unittest.TestCase):
|
|
def test_independent_intervals(self):
|
|
for hourly in (False, True):
|
|
for half in (False, True):
|
|
for quarter in (False, True):
|
|
settings = Settings(hourly=hourly, half_hour=half, quarter_hour=quarter)
|
|
for minute, expected in ((0, hourly), (15, quarter), (30, half), (45, quarter), (17, False)):
|
|
self.assertEqual(due(datetime(2026, 10, 23, 12, minute), settings), expected)
|
|
|
|
def test_no_repeat_startup_burst_or_catch_up(self):
|
|
audio = Mock()
|
|
scheduler = Scheduler(audio)
|
|
settings = Settings(hourly=True, half_hour=True, quarter_hour=True)
|
|
scheduler.tick(datetime(2026, 10, 23, 12, 0, 40), settings)
|
|
audio.announce.assert_not_called()
|
|
scheduler.tick(datetime(2026, 10, 23, 12, 1), settings)
|
|
scheduler.tick(datetime(2026, 10, 23, 12, 15), settings)
|
|
scheduler.tick(datetime(2026, 10, 23, 12, 15, 1), settings)
|
|
self.assertEqual(audio.announce.call_count, 1)
|
|
scheduler.tick(datetime(2026, 10, 23, 12, 5), settings) # wall clock moves back
|
|
scheduler.tick(datetime(2026, 10, 23, 12, 15), settings)
|
|
self.assertEqual(audio.announce.call_count, 1)
|
|
scheduler.tick(datetime(2026, 10, 23, 16, 2), settings) # resume after sleep
|
|
self.assertEqual(audio.announce.call_count, 1)
|
|
|
|
def test_midnight_and_time_formats(self):
|
|
self.assertEqual(spoken_time(datetime(2026, 10, 23, 0, 0), "12"), "12 o'clock A M")
|
|
self.assertEqual(spoken_time(datetime(2026, 10, 23, 0, 0), "24"), "zero hundred hours")
|
|
self.assertEqual(spoken_time(datetime(2026, 10, 23, 13, 5), "24"), "13 oh 5")
|
|
|
|
|
|
class AudioTests(unittest.TestCase):
|
|
def test_default_hour_and_quarter_chimes(self):
|
|
audio = Audio()
|
|
audio.run = Mock(return_value=True)
|
|
audio.chime(datetime(2026, 10, 23, 3, 0), Settings())
|
|
self.assertEqual([Path(call.args[0][-1]).name for call in audio.run.call_args_list],
|
|
["prepend.ogg", "bell.ogg", "bell.ogg", "bell.ogg"])
|
|
audio.run.reset_mock()
|
|
audio.chime(datetime(2026, 10, 23, 3, 30), Settings())
|
|
self.assertEqual(Path(audio.run.call_args.args[0][-1]).name, "30.ogg")
|
|
|
|
def test_recorded_hour_replaces_repeated_bell(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
(Path(directory) / "12.ogg").touch()
|
|
audio = Audio()
|
|
audio.run = Mock(return_value=True)
|
|
audio.chime(datetime(2026, 10, 23, 0, 0), Settings(soundpack=directory))
|
|
self.assertEqual(audio.run.call_count, 1)
|
|
self.assertEqual(audio.run.call_args.args[0][-1], str(Path(directory) / "12.ogg"))
|
|
|
|
def test_speech_dispatcher_receives_all_controls_and_text_as_one_argument(self):
|
|
audio = Audio()
|
|
audio.run = Mock(return_value=True)
|
|
with patch("talking_clock.audio.system_language", return_value="en-us"):
|
|
audio.speak("12 o'clock A M", datetime(2026, 10, 23),
|
|
Settings(module="espeak-ng", voice="English (America)", rate=30, pitch=-20, volume=40))
|
|
audio.run.assert_called_once_with([
|
|
"spd-say", "-w", "-P", "important", "-N", "talking-clock", "-l", "en-us",
|
|
"-r", "30", "-p", "-20", "-i", "40", "-o", "espeak-ng",
|
|
"-y", "English (America)", "--", "12 o'clock A M"])
|
|
|
|
def test_stop_terminates_active_process_and_blocks_new_audio(self):
|
|
audio = Audio()
|
|
process = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"],
|
|
start_new_session=True)
|
|
try:
|
|
audio.process = process
|
|
audio.stop()
|
|
self.assertIsNotNone(process.poll())
|
|
with patch("talking_clock.audio.subprocess.Popen") as launch:
|
|
self.assertFalse(audio.run(["must-not-run"]))
|
|
launch.assert_not_called()
|
|
finally:
|
|
if process.poll() is None:
|
|
process.kill()
|
|
process.wait()
|
|
|
|
|
|
class LifecycleTests(unittest.TestCase):
|
|
def test_automatic_install_locations(self):
|
|
cases = (
|
|
("local-bin", "1000", (".local/bin",), False, ".local/bin", ".local/share"),
|
|
("home-bin", "1000", ("bin",), False, "bin", ".local/share"),
|
|
("prefer-local", "1000", ("bin", ".local/bin"), False, ".local/bin", ".local/share"),
|
|
("xdg-data", "1000", ("bin",), True, "bin", "custom data"),
|
|
("root", "0", (), False, "/usr/local/bin", "/usr/share"),
|
|
)
|
|
for name, uid, path_dirs, custom_data, bin_suffix, data_suffix in cases:
|
|
with self.subTest(name=name), tempfile.TemporaryDirectory() as directory:
|
|
base = Path(directory)
|
|
home = base / "user home"
|
|
home.mkdir()
|
|
tools = base / "tools"
|
|
tools.mkdir()
|
|
fake_id = tools / "id"
|
|
fake_id.write_text(f"#!/bin/sh\nprintf '%s\\n' {uid}\n")
|
|
fake_id.chmod(0o755)
|
|
stage = base / "stage"
|
|
environment = dict(os.environ, HOME=str(home), DESTDIR=str(stage),
|
|
PATH=":".join([str(tools)] +
|
|
[str(home / p) + "/" for p in path_dirs] +
|
|
[os.defpath]))
|
|
for key in ("PREFIX", "XDG_DATA_HOME", "PYTHONPATH"):
|
|
environment.pop(key, None)
|
|
if custom_data:
|
|
environment["XDG_DATA_HOME"] = str(home / data_suffix)
|
|
subprocess.run(["bash", str(ROOT / "INSTALL")], env=environment,
|
|
check=True, capture_output=True, timeout=10)
|
|
bindir = Path(bin_suffix) if uid == "0" else home / bin_suffix
|
|
datadir = Path(data_suffix) if uid == "0" else home / data_suffix
|
|
launcher = stage / bindir.relative_to("/") / "talking-clock"
|
|
data = stage / datadir.relative_to("/")
|
|
self.assertTrue((data / "applications/talking-clock.desktop").is_file())
|
|
result = subprocess.run([str(launcher), "-h"], env=environment,
|
|
capture_output=True, text=True, timeout=5)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("-H", result.stdout)
|
|
environment.update(PYTHONPATH=str(data / "talking-clock/python"),
|
|
XDG_DATA_HOME=str(base / "empty-user"),
|
|
XDG_DATA_DIRS=str(base / "empty-system"))
|
|
result = subprocess.run([sys.executable, "-c",
|
|
"from talking_clock.config import resolve_soundpack; "
|
|
"print(resolve_soundpack('default'))"],
|
|
env=environment, capture_output=True, text=True, timeout=5)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(result.stdout.strip(), str(data / "talking-clock/sounds/default"))
|
|
|
|
def test_user_install_requires_home_bin_on_path_before_writing(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
base = Path(directory)
|
|
tools = base / "tools"
|
|
tools.mkdir()
|
|
fake_id = tools / "id"
|
|
fake_id.write_text("#!/bin/sh\nprintf '1000\\n'\n")
|
|
fake_id.chmod(0o755)
|
|
stage = base / "stage"
|
|
environment = dict(os.environ, HOME=str(base / "home"), DESTDIR=str(stage),
|
|
PATH=f"{tools}:{base}/home/bin-other:{os.defpath}")
|
|
environment.pop("PREFIX", None)
|
|
result = subprocess.run(["bash", str(ROOT / "INSTALL")], env=environment,
|
|
capture_output=True, text=True, timeout=10)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("Add ~/.local/bin or ~/bin to PATH", result.stderr)
|
|
self.assertFalse(stage.exists())
|
|
|
|
def test_staged_installer_launcher_and_soundpack(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
environment = dict(os.environ, DESTDIR=directory, PREFIX="/usr/local")
|
|
subprocess.run(["bash", str(ROOT / "INSTALL")], env=environment,
|
|
check=True, capture_output=True, timeout=10)
|
|
prefix = Path(directory) / "usr/local"
|
|
result = subprocess.run([str(prefix / "bin/talking-clock"), "-h"],
|
|
capture_output=True, text=True, timeout=5)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("-H", result.stdout)
|
|
environment.update(PYTHONPATH=str(prefix / "share/talking-clock/python"),
|
|
XDG_DATA_HOME=str(Path(directory) / "empty-user"),
|
|
XDG_DATA_DIRS=str(Path(directory) / "empty-system"))
|
|
result = subprocess.run([sys.executable, "-c",
|
|
"from talking_clock.config import resolve_soundpack; "
|
|
"print(resolve_soundpack('default'))"],
|
|
env=environment, capture_output=True, text=True, timeout=5)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(result.stdout.strip(), str(prefix / "share/talking-clock/sounds/default"))
|
|
|
|
def test_single_instance_lock_and_stale_socket_recovery(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
first, second = Instance(Path(directory)), Instance(Path(directory))
|
|
try:
|
|
self.assertTrue(first.acquire())
|
|
self.assertFalse(second.acquire())
|
|
first.close()
|
|
self.assertTrue(second.acquire())
|
|
finally:
|
|
first.close()
|
|
second.close()
|
|
|
|
def test_reject_non_private_runtime_directory(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
target = Path(directory) / "talking-clock"
|
|
target.mkdir(mode=0o755)
|
|
target.chmod(0o755)
|
|
with patch.dict(os.environ, {"XDG_RUNTIME_DIR": directory}):
|
|
with self.assertRaises(OSError):
|
|
runtime_directory()
|
|
|
|
def test_headless_duplicate_kill_and_restart_without_gtk(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
base = Path(directory)
|
|
config = Config(base / "config/talking-clock")
|
|
config.save(Settings(chime=False, speak=False))
|
|
environment = dict(os.environ, XDG_CONFIG_HOME=str(base / "config"),
|
|
XDG_RUNTIME_DIR=directory, DISPLAY="", WAYLAND_DISPLAY="")
|
|
command = [sys.executable, str(ROOT / "talking-clock")]
|
|
for _ in range(2):
|
|
process = subprocess.Popen(command + ["-H"], env=environment,
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
try:
|
|
socket_path = base / "talking-clock/control.sock"
|
|
for _ in range(100):
|
|
if socket_path.exists() or process.poll() is not None:
|
|
break
|
|
time.sleep(0.02)
|
|
self.assertIsNone(process.poll())
|
|
result = subprocess.run(command + ["-H"], env=environment,
|
|
capture_output=True, text=True, timeout=5)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("already running", result.stdout)
|
|
# A failed attempt to open a display must leave headless mode alive.
|
|
result = subprocess.run(command, env=environment,
|
|
capture_output=True, text=True, timeout=5)
|
|
self.assertEqual(result.returncode, 1, result.stderr)
|
|
self.assertIsNone(process.poll())
|
|
result = subprocess.run(command + ["-k"], env=environment,
|
|
capture_output=True, text=True, timeout=5)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
_, errors = process.communicate(timeout=5)
|
|
self.assertEqual(process.returncode, 0, errors.decode())
|
|
self.assertFalse(socket_path.exists())
|
|
finally:
|
|
if process.poll() is None:
|
|
process.terminate()
|
|
process.communicate(timeout=5)
|
|
result = subprocess.run(command + ["-k"], env=environment,
|
|
capture_output=True, timeout=5)
|
|
self.assertEqual(result.returncode, 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|