Rewritten in python, functionality updated. Weather service improved, now closer to I38's method.

This commit is contained in:
Storm Dragon committed 2026-10-23 15:13:12 -04:00
1 parent f2d1fafc94
commit 38889ded77
26 files changed
+2334 -747

No files matched your search

+253
View File
@@ -0,0 +1,253 @@
"""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_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()
+360
View File
@@ -0,0 +1,360 @@
"""Opt-in desktop checks: TALKING_CLOCK_GUI_TESTS=1 python3 -m unittest discover -s tests."""
from dataclasses import replace
from datetime import datetime
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"))
@unittest.skipUnless(os.environ.get("TALKING_CLOCK_GUI_TESTS") == "1", "requires an opted-in GTK display")
class GuiTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
from talking_clock.gui import Gtk
if not Gtk.init_check()[0]:
raise RuntimeError("GUI tests require an accessible display")
def setUp(self):
from talking_clock.config import Config, Settings
from talking_clock.runtime import Clock
from talking_clock.gui import ClockWindow
from talking_clock.speech import Catalog
self.discovery = patch("talking_clock.gui.discover",
return_value=Catalog(("espeak-ng", "rhvoice"), "espeak-ng", ("English (America)",), "en-us"))
self.mock_discover = self.discovery.start()
self.addCleanup(self.discovery.stop)
self.directory = tempfile.TemporaryDirectory()
self.clock = Clock(Config(self.directory.name), Settings(chime=False, speak=False), Mock())
self.clock.audio.weather.detect_location = Mock(return_value=None)
self.window = ClockWindow(self.clock)
def tearDown(self):
self.window.destroy()
self.directory.cleanup()
def wait_until(self, predicate):
from talking_clock.gui import GLib
loop = GLib.MainLoop()
deadline = time.monotonic() + 2
def check():
if predicate() or time.monotonic() >= deadline:
loop.quit()
return False
return True
GLib.timeout_add(10, check)
loop.run()
self.assertTrue(predicate())
def test_initial_focus_tab_and_shift_tab(self):
from talking_clock.gui import Gtk
self.assertIs(self.window.get_focus(), self.window.time)
self.assertFalse(self.window.time.get_accessible().get_name())
self.assertFalse(self.window.time.get_accessible().get_description())
self.window.child_focus(Gtk.DirectionType.TAB_FORWARD)
self.assertIsInstance(self.window.get_focus(), Gtk.Button)
self.assertEqual(self.window.get_focus().get_label(), "_Settings")
self.window.time.grab_focus()
self.window.child_focus(Gtk.DirectionType.TAB_BACKWARD)
# Toplevel Tab traversal wraps after leaving its container.
if self.window.get_focus() is None:
self.window.child_focus(Gtk.DirectionType.TAB_BACKWARD)
self.assertIsInstance(self.window.get_focus(), Gtk.Button)
def test_display_updates_and_supports_24_hour_and_hidden_seconds(self):
with patch("talking_clock.gui.datetime") as clock_time:
clock_time.now.return_value = datetime(2026, 10, 23, 14, 5, 6).astimezone()
self.window.update_time()
self.assertEqual(self.window.time.get_text(), "2:05:06 PM")
clock_time.now.return_value = datetime(2026, 10, 23, 14, 5, 7).astimezone()
self.window.update_time()
self.assertEqual(self.window.time.get_text(), "2:05:07 PM")
self.clock.settings = replace(self.clock.settings, time_format="24", seconds=False)
self.window.update_time()
self.assertEqual(self.window.time.get_text(), "14:05")
def test_space_speaks_only_time_even_if_scheduled_speech_is_disabled(self):
from talking_clock.gui import Gdk
self.clock.settings = replace(self.clock.settings, speak=False, chime=True, weather=True)
original = replace(self.clock.settings)
self.clock.audio.speak = Mock()
self.clock.audio.chime = Mock()
self.clock.audio.weather.report = Mock()
event = Gdk.Event.new(Gdk.EventType.KEY_PRESS)
event.keyval = Gdk.KEY_space
event.state = Gdk.ModifierType(0)
with patch("talking_clock.gui.datetime") as clock_time:
clock_time.now.return_value = datetime(2026, 10, 23, 14, 5, 6).astimezone()
self.assertTrue(self.window.time.emit("key-press-event", event))
self.clock.audio.worker.join(timeout=2)
self.assertFalse(self.clock.audio.worker.is_alive())
self.assertEqual(self.clock.audio.speak.call_args.args[0], "2 oh 5 P M")
self.clock.audio.chime.assert_not_called()
self.clock.audio.weather.report.assert_not_called()
self.assertEqual(self.clock.settings, original)
def test_modified_space_and_other_keys_do_not_trigger_speech(self):
from talking_clock.gui import Gdk
self.clock.audio.announce = Mock()
event = Gdk.Event.new(Gdk.EventType.KEY_PRESS)
event.keyval = Gdk.KEY_space
event.state = Gdk.ModifierType.CONTROL_MASK
self.assertFalse(self.window.time_key_pressed(self.window.time, event))
event.keyval = Gdk.KEY_Tab
event.state = Gdk.ModifierType(0)
self.assertFalse(self.window.time_key_pressed(self.window.time, event))
self.clock.audio.announce.assert_not_called()
def test_space_weather_is_independent_of_scheduled_weather(self):
from talking_clock.gui import Gdk
event = Gdk.Event.new(Gdk.EventType.KEY_PRESS)
event.keyval = Gdk.KEY_space
event.state = Gdk.ModifierType(0)
for scheduled in (False, True):
for on_space in (False, True):
with self.subTest(scheduled=scheduled, on_space=on_space):
self.clock.settings = replace(self.clock.settings, weather=scheduled,
weather_on_space=on_space)
original = replace(self.clock.settings)
self.clock.audio.speak = Mock()
self.clock.audio.weather.report = Mock(return_value="77 degrees Fahrenheit and cloudy")
self.assertTrue(self.window.time_key_pressed(self.window.time, event))
self.clock.audio.worker.join(timeout=2)
self.assertFalse(self.clock.audio.worker.is_alive())
text = self.clock.audio.speak.call_args.args[0]
self.assertEqual("77 degrees Fahrenheit and cloudy" in text, on_space)
self.assertEqual(self.clock.audio.weather.report.call_count, int(on_space))
self.assertEqual(self.clock.settings, original)
def test_settings_tabs_save_and_apply_immediately(self):
from talking_clock.gui import SettingsDialog
dialog = SettingsDialog(self.window, self.clock)
try:
self.assertEqual(dialog.notebook.get_n_pages(), 5)
self.assertEqual(set(dialog.controls), set(vars(self.clock.settings)))
dialog.controls["font_size"].set_value(100)
dialog.controls["high_contrast"].set_active(True)
dialog.controls["half_hour"].set_active(True)
dialog.controls["time_format"].set_active_id("24")
dialog.controls["weather_on_space"].set_active(True)
dialog.save()
self.assertEqual(self.clock.settings.font_size, 100)
self.assertTrue(self.clock.settings.half_hour)
self.assertTrue(self.clock.settings.high_contrast)
self.assertTrue(self.clock.settings.weather_on_space)
self.assertEqual(self.clock.config.load(), self.clock.settings)
finally:
dialog.destroy()
def test_combo_accessible_names_and_label_relations(self):
from talking_clock.gui import Atk, SettingsDialog
names = {"soundpack": "Soundpack", "module": "Module", "voice": "Voice",
"rate": "Rate", "pitch": "Pitch", "volume": "Volume",
"time_format": "Time format", "temperature_unit": "Temperature unit"}
dialog = SettingsDialog(self.window, self.clock)
try:
self.wait_until(lambda: dialog.controls["voice"].get_sensitive())
for key, name in names.items():
control = dialog.controls[key]
targets = [control, control.get_child()] if control.get_has_entry() else [control]
for target in targets:
with self.subTest(control=key, widget=type(target).__name__):
accessible = target.get_accessible()
self.assertEqual(accessible.get_name(), name)
relation = accessible.ref_relation_set().get_relation_by_type(Atk.RelationType.LABELLED_BY)
self.assertIsNotNone(relation)
self.assertEqual(relation.get_target()[0].get_name(), name)
finally:
dialog.destroy()
def test_speech_test_uses_unsaved_controls_without_saving(self):
from talking_clock.gui import SettingsDialog
dialog = SettingsDialog(self.window, self.clock)
try:
self.wait_until(lambda: dialog.controls["voice"].get_sensitive())
original = replace(self.clock.settings)
dialog.controls["module"].set_active_id("espeak-ng")
self.wait_until(lambda: dialog.controls["voice"].get_sensitive())
dialog.controls["voice"].set_active_id("English (America)")
for key, value in (("rate", "30"), ("pitch", "-20"), ("volume", "50")):
dialog.controls[key].set_active_id(value)
self.clock.audio.speak = Mock()
self.clock.audio.chime = Mock()
self.clock.audio.weather.report = Mock()
dialog.test_speech_button.clicked()
self.clock.audio.worker.join(timeout=2)
self.assertFalse(self.clock.audio.worker.is_alive())
preview = self.clock.audio.speak.call_args.args[2]
self.assertEqual((preview.module, preview.voice, preview.rate, preview.pitch, preview.volume),
("espeak-ng", "English (America)", 30, -20, 50))
self.assertTrue(preview.speak)
self.clock.audio.chime.assert_not_called()
self.clock.audio.weather.report.assert_not_called()
self.assertEqual(self.clock.settings, original)
self.assertFalse(self.clock.config.path.exists())
finally:
dialog.destroy()
def test_speech_controls_are_combos_and_reload_voices_for_selected_module(self):
from talking_clock.gui import Gtk, SettingsDialog
from talking_clock.speech import Catalog
self.clock.settings = replace(self.clock.settings, rate=37)
dialog = SettingsDialog(self.window, self.clock)
try:
self.wait_until(lambda: dialog.controls["voice"].get_sensitive())
for key in ("module", "voice", "rate", "pitch", "volume"):
self.assertIsInstance(dialog.controls[key], Gtk.ComboBoxText)
self.assertEqual(dialog.controls["rate"].get_active_id(), "37")
self.mock_discover.return_value = Catalog(("espeak-ng", "rhvoice"), "rhvoice", ("Alan",), "en-us")
dialog.controls["module"].set_active_id("rhvoice")
self.wait_until(lambda: dialog.controls["voice"].get_sensitive())
self.mock_discover.assert_called_with("rhvoice")
self.assertTrue(dialog.controls["voice"].set_active_id("Alan"))
self.assertFalse(dialog.controls["voice"].set_active_id("English (America)"))
dialog.controls["pitch"].set_active_id("-20")
dialog.controls["volume"].set_active_id("50")
dialog.save()
self.assertEqual(self.clock.settings.module, "rhvoice")
self.assertEqual(self.clock.settings.voice, "Alan")
self.assertEqual(self.clock.settings.rate, 37)
self.assertEqual(self.clock.settings.pitch, -20)
self.assertEqual(self.clock.settings.volume, 50)
finally:
dialog.destroy()
def test_dispatcher_failure_preserves_saved_speech_controls(self):
from talking_clock.gui import SettingsDialog
self.clock.settings = replace(self.clock.settings, module="offline", voice="Saved voice")
self.mock_discover.side_effect = RuntimeError("Dispatcher unavailable")
dialog = SettingsDialog(self.window, self.clock)
try:
self.wait_until(lambda: dialog.controls["voice"].get_sensitive())
self.assertIn("Saved selections are preserved", dialog.speech_status.get_text())
dialog.save()
self.assertEqual(self.clock.settings.module, "offline")
self.assertEqual(self.clock.settings.voice, "Saved voice")
finally:
dialog.destroy()
def test_window_close_stops_audio_and_scheduler(self):
self.clock.audio.stop = Mock()
self.window.destroy()
self.assertTrue(self.clock.stopped.is_set())
self.clock.audio.stop.assert_called_once()
def test_control_q_closes_clock_from_settings_button(self):
from talking_clock.gui import Gdk, Gtk
self.clock.audio.stop = Mock()
self.window.child_focus(Gtk.DirectionType.TAB_FORWARD)
self.assertIsInstance(self.window.get_focus(), Gtk.Button)
self.assertTrue(Gtk.accel_groups_activate(self.window, Gdk.KEY_q, Gdk.ModifierType.CONTROL_MASK))
self.assertTrue(self.clock.stopped.is_set())
self.clock.audio.stop.assert_called_once()
def test_control_q_from_settings_stops_program_without_saving(self):
from talking_clock.gui import Gdk, Gtk
self.window.open_settings(None)
self.window.dialog.controls["font_size"].set_value(100)
self.assertTrue(Gtk.accel_groups_activate(self.window.dialog, Gdk.KEY_q,
Gdk.ModifierType.CONTROL_MASK))
self.assertTrue(self.clock.stopped.is_set())
self.assertIsNone(self.window.dialog)
self.assertEqual(self.clock.settings.font_size, 72)
self.assertFalse(self.clock.config.path.exists())
def test_shutdown_with_settings_open_exits_main_loop(self):
from talking_clock.gui import GLib, Gtk
self.window.open_settings(None)
self.assertIsNotNone(self.window.dialog)
GLib.idle_add(lambda: self.window.destroy() or False)
Gtk.main()
self.assertIsNone(self.window.dialog)
self.assertTrue(self.clock.stopped.is_set())
def test_detected_city_prefills_weather_fields(self):
from talking_clock.gui import GLib, SettingsDialog
self.clock.audio.weather.detect_location.return_value = {
"location": "Raleigh, North Carolina, US", "latitude": "35.78", "longitude": "-78.64"}
dialog = SettingsDialog(self.window, self.clock)
try:
deadline = time.monotonic() + 2
loop = GLib.MainLoop()
def ready():
if dialog.controls["location"].get_text() or time.monotonic() >= deadline:
loop.quit()
return False
return True
GLib.timeout_add(10, ready)
loop.run()
self.assertEqual(dialog.controls["location"].get_text(), "Raleigh, North Carolina, US")
self.assertEqual(dialog.controls["latitude"].get_text(), "35.78")
self.assertEqual(dialog.controls["longitude"].get_text(), "-78.64")
dialog.save()
self.assertEqual(self.clock.config.load().latitude, "35.78")
finally:
dialog.destroy()
def test_editing_city_does_not_reuse_old_coordinates(self):
from talking_clock.gui import SettingsDialog
self.clock.settings = replace(self.clock.settings, location="Raleigh", latitude="35.78", longitude="-78.64")
dialog = SettingsDialog(self.window, self.clock)
try:
dialog.controls["location"].set_text("Berlin, Germany")
dialog.save()
self.assertEqual(self.clock.settings.location, "Berlin, Germany")
self.assertEqual(self.clock.settings.latitude, "")
self.assertEqual(self.clock.settings.longitude, "")
finally:
dialog.destroy()
def test_gui_process_replaces_headless_process_and_remains_single_instance(self):
from talking_clock.config import Config, Settings
from talking_clock.runtime import send_command
root = Path(__file__).resolve().parents[1]
with tempfile.TemporaryDirectory() as directory:
base = Path(directory)
Config(base / "config/talking-clock").save(Settings(chime=False, speak=False))
environment = dict(os.environ, XDG_RUNTIME_DIR=directory, XDG_CONFIG_HOME=str(base / "config"))
command = [sys.executable, str(root / "talking-clock")]
processes = []
try:
headless = subprocess.Popen(command + ["-H"], env=environment,
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
processes.append(headless)
deadline = time.monotonic() + 5
while not (base / "talking-clock/control.sock").exists() and time.monotonic() < deadline:
time.sleep(0.02)
self.assertIsNone(headless.poll())
gui = subprocess.Popen(command, env=environment, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
processes.append(gui)
headless.communicate(timeout=5)
self.assertEqual(headless.returncode, 0)
deadline = time.monotonic() + 5
while time.monotonic() < deadline:
if send_command(base / "talking-clock", "status") == "gui":
break
time.sleep(0.02)
else:
self.fail("Replacement GUI never became ready")
self.assertIsNone(gui.poll())
duplicate = subprocess.run(command, env=environment, capture_output=True, timeout=5)
self.assertEqual(duplicate.returncode, 0, duplicate.stderr)
self.assertIsNone(gui.poll())
subprocess.run(command + ["-k"], env=environment, check=True, capture_output=True, timeout=5)
gui.communicate(timeout=5)
self.assertEqual(gui.returncode, 0)
finally:
for process in processes:
if process.poll() is None:
process.terminate()
process.communicate(timeout=5)
if __name__ == "__main__":
unittest.main()
+97
View File
@@ -0,0 +1,97 @@
"""Speech discovery is read-only; all dispatcher connections here are mocked."""
import json
from pathlib import Path
import sys
import tempfile
from types import SimpleNamespace
import unittest
from unittest.mock import Mock, patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from talking_clock.config import Config, Settings
from talking_clock.speech import connect, discover, matching_voices, system_language
class SpeechTests(unittest.TestCase):
def test_locale_precedence_encoding_and_c_fallback(self):
self.assertEqual(system_language({"LC_ALL": "de_DE.UTF-8", "LANG": "en_US.UTF-8"}), "de-de")
self.assertEqual(system_language({"LC_ALL": "C.UTF-8", "LANGUAGE": "es_MX:en"}), "es-mx")
self.assertEqual(system_language({"LC_MESSAGES": "fr_CA.UTF-8@modifier", "LANG": "en_US.UTF-8"}), "fr-ca")
self.assertEqual(system_language({"LANG": "pt_BR.UTF-8"}), "pt-br")
self.assertEqual(system_language({"LC_ALL": "POSIX", "LANG": "C"}), "en")
def test_voice_filter_prefers_region_and_falls_back_to_language(self):
voices = (("British", "en-gb", ""), ("American", "en_US", ""),
("French", "fr", ""), ("Unknown", None, None),
("British", "en-gb", "other"), ("English", "EN", ""))
self.assertEqual(matching_voices(voices, "en-us"), ("American",))
self.assertEqual(matching_voices(voices, "en-au"), ("American", "British", "English"))
self.assertEqual(matching_voices(voices, "en"), ("English",))
self.assertEqual(matching_voices(voices, "fr-fr"), ("French",))
self.assertEqual(matching_voices(voices, "es"), ())
def test_discovery_queries_selected_module_and_closes_without_speaking(self):
client = Mock()
client.list_output_modules.return_value = ("rhvoice", "espeak-ng")
client.list_synthesis_voices.return_value = (("American", "en-us", ""), ("French", "fr", ""))
with patch("talking_clock.speech.connect", return_value=client), patch(
"talking_clock.speech.system_language", return_value="en-us"):
catalog = discover("espeak-ng")
self.assertEqual(catalog.modules, ("espeak-ng", "rhvoice"))
self.assertEqual(catalog.voices, ("American",))
client.set_output_module.assert_called_once_with("espeak-ng")
client.close.assert_called_once()
client.speak.assert_not_called()
def test_system_default_module_is_queried_without_overriding_it(self):
client = Mock()
client.list_output_modules.return_value = ("espeak-ng",)
client.get_output_module.return_value = "espeak-ng"
client.list_synthesis_voices.return_value = ()
with patch("talking_clock.speech.connect", return_value=client):
self.assertEqual(discover().module, "espeak-ng")
client.set_output_module.assert_not_called()
client.speak.assert_not_called()
def test_connection_does_not_autostart_dispatcher(self):
module = SimpleNamespace(SSIPClient=Mock())
with patch.dict(sys.modules, {"speechd": module}):
connect()
module.SSIPClient.assert_called_once_with("talking-clock", component="settings", autospawn=False)
def test_unavailable_dispatcher_and_module_are_reported(self):
with patch("talking_clock.speech.connect", side_effect=OSError("offline")):
with self.assertRaisesRegex(RuntimeError, "offline"):
discover()
client = Mock()
client.list_output_modules.return_value = ()
with patch("talking_clock.speech.connect", return_value=client):
with self.assertRaisesRegex(RuntimeError, "unavailable"):
discover("missing")
client.close.assert_called_once()
def test_numeric_limits_and_saved_intermediate_values(self):
for key in ("rate", "pitch", "volume"):
for value in (-100, -37, 0, 37, 100):
settings = Settings(**{key: value})
self.assertIs(settings.validate(), settings)
for value in (-101, 101, "10", True):
with self.assertRaises(ValueError):
Settings(**{key: value}).validate()
def test_old_python_speech_preferences_migrate_to_dispatcher_defaults(self):
with tempfile.TemporaryDirectory() as directory:
config = Config(directory)
config.path.write_text(json.dumps({"voice": "custom", "voice_command": "some old command",
"rhvoice": "alan", "gama_path": "/old", "font_size": 100}))
settings = config.load()
self.assertEqual(settings.module, "")
self.assertEqual(settings.voice, "")
self.assertEqual(settings.font_size, 100)
config.save(settings)
data = json.loads(config.path.read_text())
self.assertFalse(set(data) & {"voice_command", "rhvoice", "gama_path"})
if __name__ == "__main__":
unittest.main()
+151
View File
@@ -0,0 +1,151 @@
"""Weather checks use mocked JSON and an isolated cache, without network access."""
from dataclasses import replace
from datetime import datetime, timezone
from email.message import Message
from email.utils import format_datetime
from pathlib import Path
import sys
import tempfile
import threading
import unittest
from unittest.mock import Mock, patch
from urllib.error import HTTPError
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from talking_clock.audio import Audio
from talking_clock.config import Settings
from talking_clock.weather import WeatherClient
CURRENT = {"current": {"temperature_2m": 25, "weather_code": 3},
"current_units": {"temperature_2m": "°C"}}
PLACE = {"city": "Raleigh", "region": "North Carolina", "country": "US", "loc": "35.78,-78.64"}
class WeatherTests(unittest.TestCase):
def setUp(self):
self.temporary = tempfile.TemporaryDirectory()
self.addCleanup(self.temporary.cleanup)
self.client = WeatherClient(self.temporary.name)
self.client._request = Mock(return_value=CURRENT)
self.settings = Settings(latitude="35.78", longitude="-78.64")
def test_units_share_cache_and_conditions_are_plain_text(self):
with patch("talking_clock.weather.time.time", return_value=1000):
self.assertEqual(self.client.report(self.settings), "77 degrees Fahrenheit and cloudy")
self.assertEqual(self.client.report(replace(self.settings, temperature_unit="C")),
"25 degrees Celsius and cloudy")
self.client._request.assert_called_once()
self.assertIn("current=temperature_2m%2Cweather_code", self.client._request.call_args.args[0])
def test_negative_temperature_and_clear_condition(self):
self.client._request.return_value = {"current": {"temperature_2m": -10, "weather_code": 0},
"current_units": {"temperature_2m": "°C"}}
self.assertEqual(self.client.report(self.settings), "14 degrees Fahrenheit and clear")
def test_every_second_checks_and_restart_only_fetch_at_next_quarter(self):
for now in range(1000, 1800):
with patch("talking_clock.weather.time.time", return_value=now):
self.assertIsNotNone(self.client.report(self.settings))
self.assertEqual(self.client._request.call_count, 1)
restarted = WeatherClient(self.temporary.name)
restarted._request = self.client._request
with patch("talking_clock.weather.time.time", return_value=1799):
self.assertIsNotNone(restarted.report(self.settings))
self.assertEqual(self.client._request.call_count, 1)
with patch("talking_clock.weather.time.time", return_value=1800):
self.assertIsNotNone(restarted.report(self.settings))
self.assertEqual(self.client._request.call_count, 2)
def test_changed_coordinates_do_not_bypass_request_budget(self):
with patch("talking_clock.weather.time.time", return_value=1000):
self.assertIsNotNone(self.client.report(self.settings))
self.assertIsNone(self.client.report(replace(self.settings, latitude="40")))
self.client._request.assert_called_once()
def test_parallel_requests_share_disk_lock(self):
other = WeatherClient(self.temporary.name)
other._request = self.client._request
with patch("talking_clock.weather.time.time", return_value=1000):
threads = [threading.Thread(target=client.report, args=(self.settings,))
for client in (self.client, other, self.client, other)]
for thread in threads:
thread.start()
for thread in threads:
thread.join(5)
self.assertFalse(thread.is_alive())
self.client._request.assert_called_once()
def test_ip_location_is_cached_for_a_day_and_used_without_geocoding(self):
self.client._request.side_effect = [PLACE, CURRENT, PLACE]
with patch("talking_clock.weather.time.time", return_value=1000):
place = self.client.detect_location()
self.assertEqual(place["location"], "Raleigh, North Carolina, US")
self.assertIsNotNone(self.client.report(Settings()))
self.assertEqual(self.client._request.call_count, 2)
with patch("talking_clock.weather.time.time", return_value=87399):
self.assertEqual(self.client.detect_location(), place)
self.assertEqual(self.client._request.call_count, 2)
with patch("talking_clock.weather.time.time", return_value=87400):
self.assertEqual(self.client.detect_location(), place)
self.assertEqual(self.client._request.call_count, 3)
def test_manual_city_uses_geocoding(self):
self.client._request.side_effect = [
{"results": [{"name": "Berlin", "country": "Germany", "latitude": 52.52, "longitude": 13.41}]},
CURRENT]
self.assertIsNotNone(self.client.report(Settings(location="Berlin, Germany")))
self.assertIn("geocoding-api.open-meteo.com", self.client._request.call_args_list[0].args[0])
self.assertIn("latitude=52.52", self.client._request.call_args_list[1].args[0])
def test_invalid_responses_are_not_spoken_or_retried_each_second(self):
for invalid in ({}, {"current": {"temperature_2m": None, "weather_code": 3}},
{"current": {"temperature_2m": 25, "weather_code": None}},
{"results": []}):
with self.subTest(invalid=invalid), tempfile.TemporaryDirectory() as directory:
client = WeatherClient(directory)
client._request = Mock(return_value=invalid)
with patch("talking_clock.weather.time.time", return_value=1000), self.assertLogs(level="WARNING"):
self.assertIsNone(client.report(self.settings))
with patch("talking_clock.weather.time.time", return_value=1001):
self.assertIsNone(client.report(self.settings))
client._request.assert_called_once()
def test_http_429_honors_retry_after_and_persists_failure_backoff(self):
for retry in ("7200", format_datetime(datetime.fromtimestamp(8200, timezone.utc))):
with self.subTest(retry=retry), tempfile.TemporaryDirectory() as directory:
client = WeatherClient(directory)
headers = Message()
headers["Retry-After"] = retry
error = HTTPError("https://example.test", 429, "limit", headers, None)
self.addCleanup(error.close)
client._request = Mock(side_effect=error)
with patch("talking_clock.weather.time.time", return_value=1000), self.assertLogs(level="WARNING"):
self.assertIsNone(client.report(self.settings))
restarted = WeatherClient(directory)
restarted._request = client._request
with patch("talking_clock.weather.time.time", return_value=8199):
self.assertIsNone(restarted.report(self.settings))
client._request.assert_called_once()
with patch("talking_clock.weather.time.time", return_value=8200), self.assertLogs(level="WARNING"):
self.assertIsNone(restarted.report(self.settings))
self.assertEqual(client._request.call_count, 2)
def test_cache_failure_does_not_contact_server(self):
blocker = Path(self.temporary.name) / "not-a-directory"
blocker.touch()
client = WeatherClient(blocker)
client._request = Mock()
with self.assertLogs(level="WARNING"):
self.assertIsNone(client.report(self.settings))
client._request.assert_not_called()
def test_weather_failure_still_speaks_the_time(self):
audio = Audio()
audio.weather.report = Mock(return_value=None)
audio.speak = Mock()
audio._announce(datetime(2026, 10, 23, 12), Settings(weather=True, chime=False))
self.assertEqual(audio.speak.call_args.args[0], "12 o'clock P M")
if __name__ == "__main__":
unittest.main()