7 Commits
30 changed files with 1321 additions and 190 deletions
+4 -2
View File
@@ -9,7 +9,8 @@ driver=gstreamerDriver
#driver=genericDriver
# Sound themes. These are the pack of sounds used for sound alerts.
# Sound packs may be located at /usr/share/sounds
# Sound packs may be located at ~/.local/stormux/fenrir/sounds,
# /usr/share/sounds/fenrir, or /usr/share/sounds/fenrirscreenreader.
theme=default
# Sound volume controls how loud the sounds for your selected soundpack are.
@@ -201,7 +202,8 @@ date_format=%%A, %%B %%d, %%Y
auto_spell_check=True
# Language for spell checking (format: language_COUNTRY, e.g., en_US, en_GB, es_ES)
spell_check_language=en_US
# path for your scripts "script_keys" functionality
# path for your scripts "script_keys" functionality.
# User-local scripts in ~/.local/stormux/fenrir are loaded first.
script_path=/usr/share/fenrirscreenreader/scripts
# Override default commands or add custom commands without modifying Fenrir installation
# Leave empty to use default commands only
+1 -1
View File
@@ -41,7 +41,7 @@ def create_argument_parser():
argumentParser.add_argument(
'-s', '--setting',
metavar='SETTING-FILE',
default='/etc/fenrir/settings/settings.conf',
default=None,
help='Path to custom settings file'
)
argumentParser.add_argument(
@@ -22,7 +22,10 @@ class command:
return _("sends the following keypress to the terminal or application")
def run(self):
self.env["input"]["key_forward"] = 3
if self.env["runtime"]["InputManager"].no_key_pressed():
self.env["input"]["key_forward"] = 1
else:
self.env["input"]["key_forward"] = -1
self.env["runtime"]["OutputManager"].present_text(
_("Forward next keypress"), interrupt=True
)
@@ -22,10 +22,7 @@ class command:
return _("Saves your current Fenrir settings so they are the default.")
def run(self):
settings_file = self.env["runtime"][
"SettingsManager"
].get_settings_file()
self.env["runtime"]["SettingsManager"].save_settings(settings_file)
self.env["runtime"]["SettingsManager"].save_settings()
self.env["runtime"]["OutputManager"].present_text(
_("Settings saved."), interrupt=True
)
@@ -6,7 +6,6 @@
import _thread
import os
import subprocess
from subprocess import PIPE
from subprocess import Popen
@@ -53,10 +52,11 @@ class command:
def _thread_run(self):
try:
callstring = (
self.script_path + " " + self.env["general"]["curr_user"]
p = Popen(
[self.script_path, self.env["general"]["curr_user"]],
stdout=PIPE,
stderr=PIPE,
)
p = Popen(callstring, stdout=PIPE, stderr=PIPE, shell=True)
stdout, stderr = p.communicate()
stdout = stdout.decode("utf-8")
stderr = stderr.decode("utf-8")
@@ -48,7 +48,7 @@ class command:
for curr_key in self.env["input"]["curr_input"]:
if curr_key not in filter_list:
return
self.env["runtime"]["OutputManager"].interrupt_output()
self.env["runtime"]["OutputManager"].interrupt_output_async()
def set_callback(self, callback):
pass
@@ -28,15 +28,21 @@ class command:
# Only announce numlock changes if an actual numlock key was pressed
# AND the LED state actually changed (some numpads send spurious NUMLOCK events)
current_input = self.env["input"]["curr_input"]
previous_input = self.env["input"]["prev_input"]
relevant_input = current_input or previous_input
# Check if this is a genuine numlock key press by verifying:
# 1. KEY_NUMLOCK is in the current input sequence
# 2. The LED state has actually changed
# 3. This isn't just a side effect from a KP_ key (which some buggy numpads do)
is_genuine_numlock = (
current_input and
"KEY_NUMLOCK" in current_input and
not any(key.startswith("KEY_KP") for key in current_input if isinstance(key, str))
relevant_input and
"KEY_NUMLOCK" in relevant_input and
not any(
key.startswith("KEY_KP")
for key in relevant_input
if isinstance(key, str)
)
)
if is_genuine_numlock:
@@ -22,23 +22,24 @@ class command(config_command):
def run(self):
current_theme = self.get_setting("sound", "theme", "default")
current_theme_name = os.path.basename(
os.path.normpath(current_theme)
)
# Present current theme
self.present_text(f"Current sound theme: {current_theme}")
# Look for available sound themes
sound_paths = [
"/usr/share/sounds",
"/usr/share/fenrirscreenreader/sounds",
os.path.expanduser("~/.local/share/fenrirscreenreader/sounds"),
]
sound_paths = self.env[
"runtime"
]["SettingsManager"].get_sound_theme_roots()
available_themes = self.get_available_themes(sound_paths)
if len(available_themes) > 1:
# For this implementation, cycle through available themes
try:
current_index = available_themes.index(current_theme)
current_index = available_themes.index(current_theme_name)
next_index = (current_index + 1) % len(available_themes)
new_theme = available_themes[next_index]
except ValueError:
+115 -55
View File
@@ -174,90 +174,121 @@ class CommandManager:
)
continue
def get_script_paths(self, script_path=""):
if script_path:
candidate_paths = [script_path]
else:
settings_manager = self.env["runtime"]["SettingsManager"]
candidate_paths = [
settings_manager.get_user_script_path(),
settings_manager.get_setting("general", "script_path"),
os.path.join(fenrir_path, "../../config/scripts/"),
]
script_paths = []
seen_paths = set()
for path in candidate_paths:
if not path:
continue
normalized_path = os.path.abspath(os.path.expanduser(path))
if normalized_path in seen_paths:
continue
script_paths.append(normalized_path)
seen_paths.add(normalized_path)
return script_paths
def load_script_commands(self, section="commands", script_path=""):
if script_path == "":
script_path = self.env["runtime"]["SettingsManager"].get_setting(
"general", "script_path"
)
if not script_path.endswith("/"):
script_path += "/"
if not os.path.exists(script_path):
if os.path.exists(fenrir_path + "/../../config/scripts/"):
script_path = fenrir_path + "/../../config/scripts/"
else:
self.env["runtime"]["DebugManager"].write_debug_out(
"scriptpath not exists:" + script_path,
debug.DebugLevel.WARNING,
script_paths = self.get_script_paths(script_path)
loaded_any_path = False
loaded_script_names = set()
for path in script_paths:
loaded_any_path = (
self.load_script_commands_from_path(
section, path, loaded_script_names
)
return
or loaded_any_path
)
if not loaded_any_path:
self.env["runtime"]["DebugManager"].write_debug_out(
"No script paths available:" + str(script_paths),
debug.DebugLevel.WARNING,
)
def load_script_commands_from_path(
self, section, script_path, loaded_script_names=None
):
if loaded_script_names is None:
loaded_script_names = set()
if not os.path.exists(script_path):
return False
if not os.path.isdir(script_path):
self.env["runtime"]["DebugManager"].write_debug_out(
"scriptpath not a directory:" + script_path,
debug.DebugLevel.ERROR,
)
return
return False
if not os.access(script_path, os.R_OK):
self.env["runtime"]["DebugManager"].write_debug_out(
"scriptpath not readable:" + script_path,
debug.DebugLevel.ERROR,
)
return
command_list = glob.glob(script_path + "*")
return False
command_list = sorted(glob.glob(os.path.join(script_path, "*")))
sub_command = fenrir_path + "/commands/commands/subprocess.py"
for command in command_list:
invalid = False
try:
if not os.path.isfile(command):
continue
file_name, file_extension = os.path.splitext(command)
file_name = file_name.split("/")[-1]
if file_name.startswith("__"):
continue
if file_name.upper() in self.env["commands"][section]:
command_name = file_name.upper()
script_name = self.get_script_name(file_name)
if script_name in loaded_script_names:
self.env["runtime"]["DebugManager"].write_debug_out(
"Skip script with duplicate script name:"
+ command_name,
debug.DebugLevel.INFO,
)
continue
if command_name in self.env["commands"][section]:
self.env["runtime"]["DebugManager"].write_debug_out(
"Skip script with duplicate command name:"
+ command_name,
debug.DebugLevel.INFO,
)
continue
shortcut = self.get_script_shortcut(file_name)
if not shortcut:
continue
shortcut_key = str(shortcut)
if shortcut_key in self.env["bindings"]:
self.env["runtime"]["DebugManager"].write_debug_out(
"Skip script with duplicate shortcut:"
+ command_name
+ " "
+ shortcut_key,
debug.DebugLevel.INFO,
)
continue
command_mod = module_utils.import_module(
file_name, sub_command
)
self.env["commands"][section][
file_name.upper()
] = command_mod.command()
self.env["commands"][section][file_name.upper()].initialize(
self.env["commands"][section][command_name] = (
command_mod.command()
)
self.env["commands"][section][command_name].initialize(
self.env, command
)
self.env["bindings"][shortcut_key] = command_name
self.env["rawBindings"][shortcut_key] = shortcut
loaded_script_names.add(script_name)
self.env["runtime"]["DebugManager"].write_debug_out(
"Load script:" + section + "." + file_name.upper(),
"Load script:" + section + "." + command_name,
debug.DebugLevel.INFO,
on_any_level=True,
)
comm_settings = file_name.upper().split("__-__")
if len(comm_settings) == 1:
keys = comm_settings[0]
elif len(comm_settings) == 2:
keys = comm_settings[1]
elif len(comm_settings) > 2:
continue
keys = keys.split("__+__")
shortcut_keys = []
shortcut = []
for key in keys:
if not self.env["runtime"]["InputManager"].is_valid_key(
key.upper()
):
self.env["runtime"]["DebugManager"].write_debug_out(
"invalid key : "
+ key.upper()
+ " script:"
+ file_name,
debug.DebugLevel.WARNING,
)
invalid = True
break
shortcut_keys.append(key.upper())
if invalid:
continue
if "KEY_SCRIPT" not in shortcut_keys:
shortcut_keys.append("KEY_SCRIPT")
shortcut.append(1)
shortcut.append(sorted(shortcut_keys))
self.env["bindings"][str(shortcut)] = file_name.upper()
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"Loading script:" + file_name, debug.DebugLevel.ERROR
@@ -266,6 +297,35 @@ class CommandManager:
str(e), debug.DebugLevel.ERROR
)
continue
return True
def get_script_name(self, file_name):
return file_name.upper().split("__-__", 1)[0]
def get_script_shortcut(self, file_name):
comm_settings = file_name.upper().split("__-__")
if len(comm_settings) == 1:
keys = comm_settings[0]
elif len(comm_settings) == 2:
keys = comm_settings[1]
else:
return None
shortcut_keys = []
for key in keys.split("__+__"):
key = key.upper()
if not key:
return None
if not self.env["runtime"]["InputManager"].is_valid_key(key):
self.env["runtime"]["DebugManager"].write_debug_out(
"invalid key : " + key + " script:" + file_name,
debug.DebugLevel.WARNING,
)
return None
shortcut_keys.append(key)
if "KEY_SCRIPT" not in shortcut_keys:
shortcut_keys.append("KEY_SCRIPT")
return [1, sorted(shortcut_keys)]
def shutdown_commands(self, section):
# Check if the section exists in the commands dictionary
@@ -48,8 +48,7 @@ class DynamicKeyboardLayoutCommand:
)
# Save to the actual config file
configFilePath = settingsManager.get_settings_file()
settingsManager.save_settings(configFilePath)
settingsManager.save_settings()
self.env["runtime"]["OutputManager"].present_text(
f"Keyboard layout set to {self.layoutName}. Please restart Fenrir for this change to take effect."
+10 -3
View File
@@ -126,8 +126,7 @@ class FenrirManager:
self.environment["runtime"]["InputManager"].write_event_buffer()
self.environment["runtime"]["InputManager"].handle_device_grab()
if self.environment["input"]["key_forward"] > 0:
self.environment["input"]["key_forward"] -= 1
self.update_key_forward()
self.environment["runtime"]["CommandManager"].execute_default_trigger(
"onKeyInput"
@@ -260,7 +259,7 @@ class FenrirManager:
)
def detect_shortcut_command(self):
if self.environment["input"]["key_forward"] > 0:
if self.environment["input"]["key_forward"] != 0:
return
if len(self.environment["input"]["prev_input"]) > len(
@@ -324,6 +323,14 @@ class FenrirManager:
)
self.command = ""
def update_key_forward(self):
key_forward = self.environment["input"]["key_forward"]
input_manager = self.environment["runtime"]["InputManager"]
if key_forward == -1 and input_manager.no_key_pressed():
self.environment["input"]["key_forward"] = 1
elif key_forward == 1 and input_manager.no_key_pressed():
self.environment["input"]["key_forward"] = 0
def set_process_name(self, name="fenrir"):
"""Attempts to set the process name to 'fenrir'."""
try:
@@ -163,6 +163,14 @@ class InputManager:
def get_last_event(self):
return self.lastEvent
def record_unmanaged_keypress(self, event_name):
self.lastEvent = {
"event_name": event_name,
"event_state": 1,
}
self.set_last_deepest_input([event_name])
self.lastInputTime = time.time()
def handle_input_event(self, event_data):
if not event_data:
return
+51 -2
View File
@@ -6,6 +6,7 @@
import re
import string
import threading
import time
from fenrirscreenreader.core import debug
@@ -16,6 +17,11 @@ from fenrirscreenreader.utils import line_utils
class OutputManager:
def __init__(self):
self.last_echo = ""
self.interrupt_lock = threading.Lock()
self.interrupt_running = False
self.interrupt_thread = None
self.interrupt_done = None
self.interrupt_wait_timeout = 0.1
def initialize(self, environment):
self.env = environment
@@ -45,7 +51,7 @@ class OutputManager:
announce_capital=False,
flush=True,
):
if text == "":
if text == "" and sound_icon == "":
return
if (
self.env["runtime"]["SettingsManager"].get_setting_as_bool(
@@ -63,6 +69,8 @@ class OutputManager:
"sound_icon found", debug.DebugLevel.INFO
)
return
if text == "":
return
if (len(text) > 1) and (text.strip(string.whitespace) == ""):
return
is_capital = self._should_announce_capital(text, announce_capital)
@@ -278,7 +286,40 @@ class OutputManager:
str(e), debug.DebugLevel.ERROR
)
def interrupt_output(self):
def interrupt_output(self, wait=True):
interrupt_done, started = self.start_interrupt_output()
if wait and started and interrupt_done:
interrupt_done.wait(timeout=self.interrupt_wait_timeout)
def interrupt_output_async(self):
self.start_interrupt_output()
def start_interrupt_output(self):
with self.interrupt_lock:
if self.interrupt_running:
return self.interrupt_done, False
self.interrupt_running = True
self.interrupt_done = threading.Event()
self.interrupt_thread = threading.Thread(
target=self.run_interrupt_output,
args=(self.interrupt_done,),
daemon=True,
)
interrupt_thread = self.interrupt_thread
interrupt_done = self.interrupt_done
interrupt_thread.start()
return interrupt_done, True
def run_interrupt_output(self, interrupt_done):
try:
self.cancel_speech()
finally:
interrupt_done.set()
with self.interrupt_lock:
if self.interrupt_done is interrupt_done:
self.interrupt_running = False
def cancel_speech(self):
try:
self.env["runtime"]["SpeechDriver"].cancel()
self.env["runtime"]["DebugManager"].write_debug_out(
@@ -349,6 +390,14 @@ class OutputManager:
return False
def play_sound(self, sound_icon="", interrupt=True):
aliases = {
"ERROR": "ERRORSCREEN",
"ERRORSOUND": "ERRORSCREEN",
}
normalized_icon = aliases.get(str(sound_icon).upper(), sound_icon)
return self.play_sound_icon(normalized_icon, interrupt)
def play_frequence(self, frequence, duration, interrupt=True):
if not self.env["runtime"]["SettingsManager"].get_setting_as_bool(
"sound", "enabled"
@@ -464,12 +464,6 @@ class RemoteManager:
)
def save_settings(self, setting_config_path=None):
if not setting_config_path:
setting_config_path = self.env["runtime"][
"SettingsManager"
].get_settings_file()
if setting_config_path == "":
return
self.env["runtime"]["SettingsManager"].save_settings(
setting_config_path
)
+159 -80
View File
@@ -42,11 +42,23 @@ fenrir_path = os.path.dirname(currentdir)
class SettingsManager:
system_settings_root = "/etc/fenrirscreenreader/"
system_settings_file = "/etc/fenrirscreenreader/settings/settings.conf"
user_settings_file = (
"~/.local/share/stormux/fenrirscreenreader/settings/settings.conf"
)
user_resource_root = "~/.local/stormux/fenrir/"
system_sound_roots = [
"/usr/share/sounds/fenrir/",
"/usr/share/sounds/fenrirscreenreader/",
]
def __init__(self):
self.settings = settings_data
self.settingArgDict = {}
self.bindingsBackup = None
self.settings_file = ""
self.save_settings_path = ""
def initialize(self, environment):
self.env = environment
@@ -104,12 +116,116 @@ class SettingsManager:
return self.settings_file
def set_settings_file(self, settings_file):
if not os.path.exists(settings_file):
return
if not os.access(settings_file, os.R_OK):
if os.path.exists(settings_file) and not os.access(
settings_file, os.R_OK
):
return
self.settings_file = settings_file
def get_user_settings_file(self):
return os.path.expanduser(self.user_settings_file)
def get_system_settings_file(self):
return self.system_settings_file
def is_user_settings_mode(self):
return os.geteuid() != 0
def get_default_save_settings_file(self):
if self.is_user_settings_mode():
return self.get_user_settings_file()
return self.get_system_settings_file()
def get_bundled_settings_root(self):
return os.path.abspath(os.path.join(fenrir_path, "../../config/")) + "/"
def get_user_resource_root(self):
return os.path.expanduser(self.user_resource_root)
def get_user_script_path(self):
return self.get_user_resource_root()
def get_user_sound_root(self):
return os.path.join(self.get_user_resource_root(), "sounds/")
def get_resource_settings_root(self):
if os.path.exists(self.system_settings_root):
return self.system_settings_root
bundled_settings_root = self.get_bundled_settings_root()
if os.path.exists(bundled_settings_root):
return bundled_settings_root
return ""
def get_sound_theme_roots(self):
candidate_roots = [
self.get_user_sound_root(),
*self.system_sound_roots,
os.path.join(self.get_bundled_settings_root(), "sound/"),
]
sound_roots = []
seen_roots = set()
for root in candidate_roots:
normalized_root = os.path.abspath(os.path.expanduser(root))
if normalized_root in seen_roots:
continue
sound_roots.append(normalized_root + "/")
seen_roots.add(normalized_root)
return sound_roots
def resolve_sound_theme_path(self, theme):
if not theme:
return ""
theme = os.path.expanduser(theme)
if os.path.exists(os.path.join(theme, "soundicons.conf")):
return theme
for sound_root in self.get_sound_theme_roots():
theme_path = os.path.join(sound_root, theme)
if os.path.exists(os.path.join(theme_path, "soundicons.conf")):
return theme_path
return ""
def load_keyboard_layout(self, environment):
settings_root = self.get_resource_settings_root()
keyboard_layout = self.get_setting("keyboard", "keyboard_layout")
if os.path.exists(keyboard_layout):
environment["runtime"]["InputManager"].load_shortcuts(
keyboard_layout
)
return
layout_path = os.path.join(settings_root, "keyboard", keyboard_layout)
if os.path.exists(layout_path):
self.set_setting("keyboard", "keyboard_layout", layout_path)
environment["runtime"]["InputManager"].load_shortcuts(layout_path)
return
layout_path = os.path.join(
settings_root, "keyboard", keyboard_layout + ".conf"
)
if os.path.exists(layout_path):
self.set_setting("keyboard", "keyboard_layout", layout_path)
environment["runtime"]["InputManager"].load_shortcuts(layout_path)
def resolve_settings_file(self, requested_settings_file=None):
if requested_settings_file and os.path.exists(requested_settings_file):
return requested_settings_file
if self.is_user_settings_mode() and os.path.exists(
self.get_user_settings_file()
):
return self.get_user_settings_file()
if os.path.exists(self.get_system_settings_file()):
return self.get_system_settings_file()
bundled_settings_file = os.path.join(
self.get_bundled_settings_root(), "settings/settings.conf"
)
if os.path.exists(bundled_settings_file):
return bundled_settings_file
return ""
def load_settings(self, setting_config_path):
if not os.path.exists(setting_config_path):
return False
@@ -128,7 +244,29 @@ class SettingsManager:
self.set_settings_file(setting_config_path)
return True
def save_settings(self, setting_config_path):
def save_settings(self, setting_config_path=None):
if setting_config_path is None:
if self.save_settings_path:
setting_config_path = self.save_settings_path
else:
setting_config_path = self.get_default_save_settings_file()
# Ensure directory exists for user-local settings
if setting_config_path and setting_config_path.startswith(
os.path.expanduser("~/.local/")
):
config_dir = os.path.dirname(setting_config_path)
if not os.path.exists(config_dir):
try:
os.makedirs(config_dir, mode=0o755, exist_ok=True)
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"save_settings: failed to create directory "
+ config_dir
+ ": "
+ str(e),
debug.DebugLevel.ERROR,
)
# set opt dict here
# save file
try:
@@ -226,7 +364,7 @@ class SettingsManager:
try:
if self.env["runtime"][driverType] is not None:
self.env["runtime"][driverType].shutdown(self.env)
self.env["runtime"][driverType].shutdown()
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"settings_manager load_driver: Error shutting down driver: "
@@ -458,25 +596,16 @@ class SettingsManager:
def init_fenrir_config(
self, cliArgs, fenrir_manager=None, environment=environment.environment
):
settings_root = "/etc/fenrirscreenreader/"
settings_file = cliArgs.setting
sound_root = "/usr/share/sounds/fenrirscreenreader/"
# get fenrir settings root
if not os.path.exists(settings_root):
if os.path.exists(fenrir_path + "/../../config/"):
settings_root = fenrir_path + "/../../config/"
else:
return None
# get settings file
if settings_file is None or not os.path.exists(settings_file):
if os.path.exists(settings_root + "/settings/settings.conf"):
settings_file = settings_root + "/settings/settings.conf"
else:
return None
# get sound themes root
if not os.path.exists(sound_root):
if os.path.exists(fenrir_path + "/../../config/sound/"):
sound_root = fenrir_path + "/../../config/sound/"
self.cliArgs = cliArgs
settings_root = self.get_resource_settings_root()
if not settings_root:
return None
self.save_settings_path = self.get_default_save_settings_file()
settings_file = self.resolve_settings_file(cliArgs.setting)
if not settings_file:
return None
environment["runtime"]["SettingsManager"] = self
environment["runtime"]["SettingsManager"].initialize(environment)
@@ -526,22 +655,11 @@ class SettingsManager:
"screen", "ignore_screen", ",".join(ignore_screens)
)
if not os.path.exists(
self.get_setting("sound", "theme") + "/soundicons.conf"
):
if os.path.exists(sound_root + self.get_setting("sound", "theme")):
self.set_setting(
"sound",
"theme",
sound_root + self.get_setting("sound", "theme"),
)
if os.path.exists(
self.get_setting("sound", "theme") + "/soundicons.conf"
):
environment["runtime"]["SettingsManager"].load_sound_icons(
self.get_setting("sound", "theme"), environment
)
else:
sound_theme_path = self.resolve_sound_theme_path(
self.get_setting("sound", "theme")
)
if sound_theme_path:
self.set_setting("sound", "theme", sound_theme_path)
environment["runtime"]["SettingsManager"].load_sound_icons(
self.get_setting("sound", "theme"), environment
)
@@ -618,6 +736,7 @@ class SettingsManager:
environment["runtime"]["InputManager"] = inputManager.InputManager()
environment["runtime"]["InputManager"].initialize(environment)
self.load_keyboard_layout(environment)
environment["runtime"]["ScreenManager"] = screenManager.ScreenManager()
environment["runtime"]["ScreenManager"].initialize(environment)
@@ -638,46 +757,6 @@ class SettingsManager:
] = diffReviewManager.DiffReviewManager()
environment["runtime"]["DiffReviewManager"].initialize(environment)
if not os.path.exists(
self.get_setting("keyboard", "keyboard_layout")
):
if os.path.exists(
settings_root
+ "keyboard/"
+ self.get_setting("keyboard", "keyboard_layout")
):
self.set_setting(
"keyboard",
"keyboard_layout",
settings_root
+ "keyboard/"
+ self.get_setting("keyboard", "keyboard_layout"),
)
environment["runtime"]["InputManager"].load_shortcuts(
self.get_setting("keyboard", "keyboard_layout")
)
if os.path.exists(
settings_root
+ "keyboard/"
+ self.get_setting("keyboard", "keyboard_layout")
+ ".conf"
):
self.set_setting(
"keyboard",
"keyboard_layout",
settings_root
+ "keyboard/"
+ self.get_setting("keyboard", "keyboard_layout")
+ ".conf",
)
environment["runtime"]["InputManager"].load_shortcuts(
self.get_setting("keyboard", "keyboard_layout")
)
else:
environment["runtime"]["InputManager"].load_shortcuts(
self.get_setting("keyboard", "keyboard_layout")
)
environment["runtime"]["CursorManager"] = cursorManager.CursorManager()
environment["runtime"]["CursorManager"].initialize(environment)
environment["runtime"][
+1 -1
View File
@@ -20,7 +20,7 @@ class sound_driver:
if not self._initialized:
return
self.cancel()
self._is_initialized = False
self._initialized = False
def play_frequence(
self, frequence, duration, adjust_volume=0.0, interrupt=True
+1 -1
View File
@@ -4,5 +4,5 @@
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
version = "2026.05.10"
version = "2026.05.12"
code_name = "testing"
+53 -11
View File
@@ -437,8 +437,8 @@ class driver(inputDriver):
self.refresh_interesting_keys()
passive_grabs = self.build_passive_grabs()
failed_before = self.failed_grabs
for key_name, modifier_mask in passive_grabs:
self.grab_key_name(key_name, modifier_mask)
for key_name, modifier_mask, include_num_lock in passive_grabs:
self.grab_key_name(key_name, modifier_mask, include_num_lock)
self.display.flush()
self.grab_signature = signature
self.write_debug(
@@ -454,9 +454,10 @@ class driver(inputDriver):
def build_passive_grabs(self):
grabs = set()
for fenrir_key in self.env["input"]["fenrir_key"]:
grabs.add((fenrir_key, 0))
grabs.add((fenrir_key, 0, True))
for script_key in self.env["input"]["script_key"]:
grabs.add((script_key, 0))
grabs.add((script_key, 0, True))
grabs.add(("KEY_NUMLOCK", 0, True))
for shortcut in self.env.get("rawBindings", {}).values():
keys = shortcut[1]
expanded_keys = self.expand_special_keys(keys)
@@ -470,12 +471,40 @@ class driver(inputDriver):
final_key = non_modifier_keys[-1]
if "KEY_FENRIR" in keys:
for fenrir_key in self.env["input"]["fenrir_key"]:
grabs.add((fenrir_key, modifier_mask))
grabs.add((fenrir_key, modifier_mask, True))
fenrir_modifier_mask = self.modifier_masks.get(
fenrir_key, 0
)
if fenrir_modifier_mask:
grabs.add(
(
final_key,
modifier_mask | fenrir_modifier_mask,
not final_key.startswith("KEY_KP"),
)
)
elif "KEY_SCRIPT" in keys:
for script_key in self.env["input"]["script_key"]:
grabs.add((script_key, modifier_mask))
grabs.add((script_key, modifier_mask, True))
script_modifier_mask = self.modifier_masks.get(
script_key, 0
)
if script_modifier_mask:
grabs.add(
(
final_key,
modifier_mask | script_modifier_mask,
not final_key.startswith("KEY_KP"),
)
)
else:
grabs.add((final_key, modifier_mask))
grabs.add(
(
final_key,
modifier_mask,
not final_key.startswith("KEY_KP"),
)
)
return grabs
def expand_special_keys(self, keys):
@@ -495,7 +524,9 @@ class driver(inputDriver):
modifier_mask |= self.modifier_masks.get(key_name, 0)
return modifier_mask
def grab_key_name(self, key_name, modifier_mask=0):
def grab_key_name(
self, key_name, modifier_mask=0, include_num_lock=True
):
keysym_names = self.key_name_to_keysym_names(key_name)
for keysym_name in keysym_names:
keysym = XK.string_to_keysym(keysym_name)
@@ -504,7 +535,9 @@ class driver(inputDriver):
keycode = self.display.keysym_to_keycode(keysym)
if not keycode:
continue
for effective_mask in self.optional_modifier_masks(modifier_mask):
for effective_mask in self.optional_modifier_masks(
modifier_mask, include_num_lock
):
try:
self.window.grab_key(
keycode,
@@ -524,9 +557,9 @@ class driver(inputDriver):
debug.DebugLevel.WARNING,
)
def optional_modifier_masks(self, modifier_mask):
def optional_modifier_masks(self, modifier_mask, include_num_lock=True):
optional_masks = [0, X.LockMask]
if self.num_lock_mask:
if include_num_lock and self.num_lock_mask:
optional_masks += [self.num_lock_mask, self.num_lock_mask | X.LockMask]
return {modifier_mask | optional for optional in optional_masks}
@@ -610,6 +643,15 @@ class driver(inputDriver):
self.ungrab_all_devices()
def get_led_state(self, led=0):
try:
pointer = self.root.query_pointer()
mask = getattr(pointer, "mask", 0)
if led == 0:
return bool(self.num_lock_mask and mask & self.num_lock_mask)
if led == 1:
return bool(mask & X.LockMask)
except Exception:
pass
return False
def set_led_state(self, led_dict):
@@ -9,6 +9,7 @@ import getpass
import os
import pty
import shlex
from queue import Full
import signal
import struct
import sys
@@ -201,6 +202,9 @@ class driver(screenDriver):
self.terminal = None
self.p_pid = -1
self.terminal_lock = threading.Lock() # Synchronize terminal operations
self.stdin_interrupt_lock = threading.Lock()
self.stdin_interrupt_running = False
self.stdin_interrupt_thread = None
signal.signal(signal.SIGWINCH, self.handle_sigwinch)
# Runtime configuration storage
@@ -288,6 +292,94 @@ class driver(screenDriver):
msg_bytes = bytes(msg_bytes, "UTF-8")
os.write(screen, msg_bytes)
def interrupt_output_on_stdin_input(self, msg_bytes):
if not msg_bytes:
return
settings_manager = self.env["runtime"]["SettingsManager"]
if not settings_manager.get_setting_as_bool(
"keyboard", "interrupt_on_key_press"
):
return
if settings_manager.get_setting(
"keyboard", "interrupt_on_key_press_filter"
).strip():
return
self.start_stdin_interrupt_thread()
def start_stdin_interrupt_thread(self):
with self.stdin_interrupt_lock:
if self.stdin_interrupt_running:
return
self.stdin_interrupt_running = True
self.stdin_interrupt_thread = threading.Thread(
target=self.run_stdin_interrupt,
daemon=True,
)
self.stdin_interrupt_thread.start()
def run_stdin_interrupt(self):
try:
self.env["runtime"]["OutputManager"].interrupt_output()
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"ptyDriver interrupt_output_on_stdin_input: "
+ str(e),
debug.DebugLevel.ERROR,
)
finally:
with self.stdin_interrupt_lock:
self.stdin_interrupt_running = False
def handle_stdin_input(self, msg_bytes, event_queue):
if self.synthesize_backspace_shortcut(msg_bytes, event_queue):
return
self.record_stdin_keypress(msg_bytes)
self.interrupt_output_on_stdin_input(msg_bytes)
self.inject_text_to_screen(msg_bytes)
def record_stdin_keypress(self, msg_bytes):
if msg_bytes != b"\t":
return
try:
self.env["runtime"]["InputManager"].record_unmanaged_keypress(
"KEY_TAB"
)
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"ptyDriver record_stdin_keypress: " + str(e),
debug.DebugLevel.ERROR,
)
def synthesize_backspace_shortcut(self, msg_bytes, event_queue):
if msg_bytes not in [b"\x7f", b"\x08"]:
return False
if "KEY_FENRIR" not in self.env["input"]["curr_input"]:
return False
event_time = time.time()
for event_state in [1, 0]:
try:
event_queue.put(
{
"Type": FenrirEventType.keyboard_input,
"data": {
"event_name": "KEY_BACKSPACE",
"event_value": 0,
"event_sec": int(event_time),
"event_usec": int((event_time % 1) * 1000000),
"event_state": event_state,
"event_type": 0,
},
},
block=False,
)
except Full:
self.env["runtime"]["DebugManager"].write_debug_out(
"ptyDriver synthesize_backspace_shortcut: Event queue full, dropping backspace events",
debug.DebugLevel.WARNING,
)
return True
def get_session_information(self):
self.env["screen"]["autoIgnoreScreens"] = []
self.env["general"]["prev_user"] = getpass.getuser()
@@ -420,7 +512,7 @@ class driver(screenDriver):
)
break
try:
self.inject_text_to_screen(msg_bytes)
self.handle_stdin_input(msg_bytes, event_queue)
except Exception as e:
self.env["runtime"][
"DebugManager"
@@ -36,7 +36,7 @@ class driver(sound_driver):
self._initialized = _gstreamerAvailable
if not self._initialized:
global _availableError
self.environment["runtime"]["DebugManager"].write_debug_out(
self.env["runtime"]["DebugManager"].write_debug_out(
"Gstreamer not available " + _availableError,
debug.DebugLevel.ERROR,
)
+76
View File
@@ -0,0 +1,76 @@
from unittest.mock import Mock
import pytest
from fenrirscreenreader.commands.commands.forward_keypress import command
from fenrirscreenreader.core.fenrirManager import FenrirManager
@pytest.mark.unit
def test_forward_keypress_arms_when_command_keys_are_still_down():
env = {
"input": {"key_forward": 0},
"runtime": {
"InputManager": Mock(no_key_pressed=Mock(return_value=False)),
"OutputManager": Mock(present_text=Mock()),
},
}
forward_command = command()
forward_command.initialize(env)
forward_command.run()
assert env["input"]["key_forward"] == -1
@pytest.mark.unit
def test_forward_keypress_activates_immediately_when_keyboard_is_idle():
env = {
"input": {"key_forward": 0},
"runtime": {
"InputManager": Mock(no_key_pressed=Mock(return_value=True)),
"OutputManager": Mock(present_text=Mock()),
},
}
forward_command = command()
forward_command.initialize(env)
forward_command.run()
assert env["input"]["key_forward"] == 1
@pytest.mark.unit
def test_forward_keypress_waits_for_command_release_before_forwarding():
input_manager = Mock(no_key_pressed=Mock(return_value=False))
manager = FenrirManager.__new__(FenrirManager)
manager.environment = {
"input": {"key_forward": -1},
"runtime": {"InputManager": input_manager},
}
manager.update_key_forward()
assert manager.environment["input"]["key_forward"] == -1
input_manager.no_key_pressed.return_value = True
manager.update_key_forward()
assert manager.environment["input"]["key_forward"] == 1
@pytest.mark.unit
def test_forward_keypress_stays_active_until_forwarded_key_release():
input_manager = Mock(no_key_pressed=Mock(return_value=False))
manager = FenrirManager.__new__(FenrirManager)
manager.environment = {
"input": {"key_forward": 1},
"runtime": {"InputManager": input_manager},
}
manager.update_key_forward()
assert manager.environment["input"]["key_forward"] == 1
input_manager.no_key_pressed.return_value = True
manager.update_key_forward()
assert manager.environment["input"]["key_forward"] == 0
+122
View File
@@ -0,0 +1,122 @@
from unittest.mock import Mock
import pytest
from fenrirscreenreader.core.commandManager import CommandManager
from fenrirscreenreader.core.settingsManager import SettingsManager
def write_script(path):
path.write_text("#!/bin/sh\n", encoding="utf-8")
path.chmod(0o755)
def build_command_environment(local_path, system_path):
settings_manager = Mock()
settings_manager.get_user_script_path.return_value = str(local_path)
settings_manager.get_setting.return_value = str(system_path)
valid_keys = {
"KEY_A",
"KEY_B",
"KEY_C",
"KEY_D",
"KEY_E",
"KEY_SCRIPT",
}
return {
"commands": {"commands": {}},
"commandsIgnore": {"commands": {}},
"bindings": {
str([1, sorted(["KEY_C", "KEY_SCRIPT"])]): "EXISTING"
},
"rawBindings": {},
"general": {"curr_user": "Username"},
"runtime": {
"SettingsManager": settings_manager,
"DebugManager": Mock(write_debug_out=Mock()),
"InputManager": Mock(
is_valid_key=Mock(side_effect=lambda key: key in valid_keys)
),
},
}
@pytest.mark.unit
def test_local_scripts_load_before_non_conflicting_system_scripts(tmp_path):
local_path = tmp_path / "local"
system_path = tmp_path / "system"
local_path.mkdir()
system_path.mkdir()
write_script(local_path / "local_only__-__KEY_A")
write_script(local_path / "same_name__-__KEY_B")
write_script(system_path / "same_name__-__KEY_D")
write_script(system_path / "system_only__-__KEY_E")
env = build_command_environment(local_path, system_path)
command_manager = CommandManager()
command_manager.initialize = lambda environment: None
command_manager.env = env
command_manager.load_script_commands()
commands = env["commands"]["commands"]
assert "LOCAL_ONLY__-__KEY_A" in commands
assert "SYSTEM_ONLY__-__KEY_E" in commands
assert commands["SAME_NAME__-__KEY_B"].script_path == str(
local_path / "same_name__-__KEY_B"
)
assert "SAME_NAME__-__KEY_D" not in commands
@pytest.mark.unit
def test_system_scripts_skip_existing_shortcut_bindings(tmp_path):
local_path = tmp_path / "local"
system_path = tmp_path / "system"
local_path.mkdir()
system_path.mkdir()
write_script(system_path / "conflicting_key__-__KEY_C")
write_script(system_path / "system_only__-__KEY_E")
env = build_command_environment(local_path, system_path)
command_manager = CommandManager()
command_manager.env = env
command_manager.load_script_commands()
commands = env["commands"]["commands"]
assert "CONFLICTING_KEY__-__KEY_C" not in commands
assert "SYSTEM_ONLY__-__KEY_E" in commands
@pytest.mark.unit
def test_sound_theme_resolution_prefers_local_soundpacks(tmp_path):
manager = SettingsManager()
local_root = tmp_path / "local" / "fenrir"
system_root = tmp_path / "system" / "sounds" / "fenrir"
local_theme = local_root / "sounds" / "default"
system_theme = system_root / "default"
local_theme.mkdir(parents=True)
system_theme.mkdir(parents=True)
(local_theme / "soundicons.conf").write_text(
"Accept=Accept.wav\n", encoding="utf-8"
)
(system_theme / "soundicons.conf").write_text(
"Accept=SystemAccept.wav\n", encoding="utf-8"
)
manager.user_resource_root = str(local_root)
manager.system_sound_roots = [str(system_root)]
assert manager.resolve_sound_theme_path("default") == str(local_theme)
@pytest.mark.unit
def test_sound_theme_resolution_accepts_absolute_theme_path(tmp_path):
manager = SettingsManager()
theme_path = tmp_path / "custom"
theme_path.mkdir()
(theme_path / "soundicons.conf").write_text(
"Accept=Accept.wav\n", encoding="utf-8"
)
assert manager.resolve_sound_theme_path(str(theme_path)) == str(theme_path)
+65
View File
@@ -0,0 +1,65 @@
import importlib.util
from pathlib import Path
from unittest.mock import Mock
import pytest
def load_numlock_command():
command_path = (
Path(__file__).parents[2]
/ "src"
/ "fenrirscreenreader"
/ "commands"
/ "onKeyInput"
/ "80500-numlock.py"
)
spec = importlib.util.spec_from_file_location(
"numlock_command", command_path
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module.command
@pytest.mark.unit
def test_numlock_off_can_be_announced_from_release_event():
output_manager = Mock()
command = load_numlock_command()()
command.initialize(
{
"input": {
"old_num_lock": True,
"new_num_lock": False,
"curr_input": [],
"prev_input": ["KEY_NUMLOCK"],
},
"runtime": {"OutputManager": output_manager},
}
)
command.run()
output_manager.present_text.assert_called_once()
assert output_manager.present_text.call_args.args[0] == "Numlock off"
@pytest.mark.unit
def test_numlock_command_ignores_non_numlock_release():
output_manager = Mock()
command = load_numlock_command()()
command.initialize(
{
"input": {
"old_num_lock": True,
"new_num_lock": False,
"curr_input": [],
"prev_input": ["KEY_KP1"],
},
"runtime": {"OutputManager": output_manager},
}
)
command.run()
output_manager.present_text.assert_not_called()
+150
View File
@@ -0,0 +1,150 @@
import importlib.util
import threading
import time
from pathlib import Path
from unittest.mock import Mock
import pytest
from fenrirscreenreader.core.outputManager import OutputManager
def build_output_manager():
settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = True
settings_manager.get_setting_as_float.return_value = 1.0
sound_driver = Mock()
speech_driver = Mock()
output_manager = OutputManager()
output_manager.env = {
"soundIcons": {
"ACCEPT": "/tmp/Accept.wav",
"ERRORSCREEN": "/tmp/ErrorScreen.wav",
},
"runtime": {
"SettingsManager": settings_manager,
"SoundDriver": sound_driver,
"SpeechDriver": speech_driver,
"DebugManager": Mock(write_debug_out=Mock()),
},
}
return output_manager, sound_driver, speech_driver
def load_key_interrupt_module():
module_path = (
Path(__file__).resolve().parents[2]
/ "src"
/ "fenrirscreenreader"
/ "commands"
/ "onKeyInput"
/ "10000-shut_up.py"
)
spec = importlib.util.spec_from_file_location(
"fenrir_key_interrupt", module_path
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
@pytest.mark.unit
def test_present_text_allows_sound_only_feedback():
output_manager, sound_driver, _speech_driver = build_output_manager()
output_manager.present_text("", sound_icon="Accept", interrupt=False)
sound_driver.play_sound_file.assert_called_once_with(
"/tmp/Accept.wav", False
)
@pytest.mark.unit
def test_play_sound_supports_error_alias():
output_manager, sound_driver, _speech_driver = build_output_manager()
assert output_manager.play_sound("Error") is True
sound_driver.play_sound_file.assert_called_once_with(
"/tmp/ErrorScreen.wav", True
)
@pytest.mark.unit
def test_interrupt_output_async_does_not_block_on_slow_cancel():
output_manager, _sound_driver, speech_driver = build_output_manager()
interrupt_started = threading.Event()
release_interrupt = threading.Event()
def slow_cancel():
interrupt_started.set()
release_interrupt.wait(timeout=1.0)
speech_driver.cancel.side_effect = slow_cancel
start_time = time.monotonic()
output_manager.interrupt_output_async()
elapsed = time.monotonic() - start_time
try:
assert interrupt_started.wait(timeout=0.2)
assert elapsed < 0.2
output_manager.interrupt_output_async()
assert speech_driver.cancel.call_count == 1
finally:
release_interrupt.set()
output_manager.interrupt_thread.join(timeout=1.0)
@pytest.mark.unit
def test_interrupt_output_waits_only_briefly_for_slow_cancel():
output_manager, _sound_driver, speech_driver = build_output_manager()
interrupt_started = threading.Event()
release_interrupt = threading.Event()
def slow_cancel():
interrupt_started.set()
release_interrupt.wait(timeout=1.0)
speech_driver.cancel.side_effect = slow_cancel
start_time = time.monotonic()
output_manager.interrupt_output()
elapsed = time.monotonic() - start_time
try:
assert interrupt_started.wait(timeout=0.2)
assert elapsed < 0.2
output_manager.interrupt_output()
assert speech_driver.cancel.call_count == 1
finally:
release_interrupt.set()
output_manager.interrupt_thread.join(timeout=1.0)
@pytest.mark.unit
def test_key_interrupt_command_uses_nonblocking_interrupt():
module = load_key_interrupt_module()
settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = True
settings_manager.get_setting.return_value = ""
output_manager = Mock()
env = {
"input": {
"curr_input": ["KEY_A"],
"prev_input": [],
},
"runtime": {
"InputManager": Mock(no_key_pressed=Mock(return_value=False)),
"OutputManager": output_manager,
"ScreenManager": Mock(is_screen_change=Mock(return_value=False)),
"SettingsManager": settings_manager,
},
}
command = module.command()
command.initialize(env)
command.run()
output_manager.interrupt_output_async.assert_called_once_with()
output_manager.interrupt_output.assert_not_called()
+177
View File
@@ -1,5 +1,10 @@
import threading
import time
from unittest.mock import Mock
import pytest
from fenrirscreenreader.core.eventData import FenrirEventType
from fenrirscreenreader.screenDriver.ptyDriver import PTYConstants
from fenrirscreenreader.screenDriver.ptyDriver import Terminal
from fenrirscreenreader.screenDriver.ptyDriver import driver as PtyDriver
@@ -52,3 +57,175 @@ def test_optional_float_setting_uses_default_when_missing():
)
== PTYConstants.OUTPUT_READ_TIMEOUT
)
@pytest.mark.unit
def test_pty_stdin_input_interrupts_output_when_all_keys_interrupt_enabled():
pty_driver = PtyDriver()
settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = True
settings_manager.get_setting.return_value = ""
output_manager = Mock()
pty_driver.env = {
"runtime": {
"SettingsManager": settings_manager,
"OutputManager": output_manager,
}
}
pty_driver.interrupt_output_on_stdin_input(b"a")
pty_driver.stdin_interrupt_thread.join(timeout=1.0)
output_manager.interrupt_output.assert_called_once_with()
@pytest.mark.unit
def test_pty_stdin_input_interrupt_does_not_block_input_injection():
pty_driver = PtyDriver()
settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = True
settings_manager.get_setting.return_value = ""
interrupt_started = threading.Event()
release_interrupt = threading.Event()
def slow_interrupt():
interrupt_started.set()
release_interrupt.wait(timeout=1.0)
output_manager = Mock(interrupt_output=Mock(side_effect=slow_interrupt))
pty_driver.env = {
"input": {"curr_input": []},
"runtime": {
"SettingsManager": settings_manager,
"OutputManager": output_manager,
"DebugManager": Mock(write_debug_out=Mock()),
},
}
pty_driver.inject_text_to_screen = Mock()
start_time = time.monotonic()
pty_driver.handle_stdin_input(b"a", Mock())
elapsed = time.monotonic() - start_time
try:
assert interrupt_started.wait(timeout=0.2)
assert elapsed < 0.2
pty_driver.inject_text_to_screen.assert_called_once_with(b"a")
finally:
release_interrupt.set()
pty_driver.stdin_interrupt_thread.join(timeout=1.0)
@pytest.mark.unit
def test_pty_raw_tab_records_recent_tab_keypress():
pty_driver = PtyDriver()
settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = False
input_manager = Mock()
pty_driver.env = {
"input": {"curr_input": []},
"runtime": {
"DebugManager": Mock(write_debug_out=Mock()),
"InputManager": input_manager,
"SettingsManager": settings_manager,
},
}
pty_driver.inject_text_to_screen = Mock()
pty_driver.handle_stdin_input(b"\t", Mock())
input_manager.record_unmanaged_keypress.assert_called_once_with("KEY_TAB")
pty_driver.inject_text_to_screen.assert_called_once_with(b"\t")
@pytest.mark.unit
def test_pty_plain_stdin_does_not_record_tab_keypress():
pty_driver = PtyDriver()
settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = False
input_manager = Mock()
pty_driver.env = {
"input": {"curr_input": []},
"runtime": {
"DebugManager": Mock(write_debug_out=Mock()),
"InputManager": input_manager,
"SettingsManager": settings_manager,
},
}
pty_driver.inject_text_to_screen = Mock()
pty_driver.handle_stdin_input(b"a", Mock())
input_manager.record_unmanaged_keypress.assert_not_called()
pty_driver.inject_text_to_screen.assert_called_once_with(b"a")
@pytest.mark.unit
def test_pty_stdin_input_honors_interrupt_disabled():
pty_driver = PtyDriver()
settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = False
output_manager = Mock()
pty_driver.env = {
"runtime": {
"SettingsManager": settings_manager,
"OutputManager": output_manager,
}
}
pty_driver.interrupt_output_on_stdin_input(b"a")
output_manager.interrupt_output.assert_not_called()
@pytest.mark.unit
def test_pty_stdin_input_leaves_filtered_interrupts_to_key_events():
pty_driver = PtyDriver()
settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = True
settings_manager.get_setting.return_value = "KEY_ENTER"
output_manager = Mock()
pty_driver.env = {
"runtime": {
"SettingsManager": settings_manager,
"OutputManager": output_manager,
}
}
pty_driver.interrupt_output_on_stdin_input(b"a")
output_manager.interrupt_output.assert_not_called()
@pytest.mark.unit
def test_pty_backspace_with_fenrir_key_synthesizes_shortcut_events():
pty_driver = PtyDriver()
event_queue = Mock()
pty_driver.env = {
"input": {"curr_input": ["KEY_FENRIR"]},
}
handled = pty_driver.synthesize_backspace_shortcut(b"\x7f", event_queue)
assert handled is True
assert event_queue.put.call_count == 2
first_event = event_queue.put.call_args_list[0].args[0]
second_event = event_queue.put.call_args_list[1].args[0]
assert first_event["Type"] == FenrirEventType.keyboard_input
assert first_event["data"]["event_name"] == "KEY_BACKSPACE"
assert first_event["data"]["event_state"] == 1
assert second_event["data"]["event_state"] == 0
@pytest.mark.unit
def test_pty_plain_backspace_is_not_synthesized():
pty_driver = PtyDriver()
event_queue = Mock()
pty_driver.env = {
"input": {"curr_input": []},
}
handled = pty_driver.synthesize_backspace_shortcut(b"\x7f", event_queue)
assert handled is False
event_queue.put.assert_not_called()
+92
View File
@@ -8,10 +8,12 @@ for all configurable settings that could cause crashes or accessibility issues.
import pytest
import sys
from pathlib import Path
from unittest.mock import Mock
# Import the settings manager
from fenrirscreenreader.core.settingsData import settings_data
from fenrirscreenreader.core.settingsManager import SettingsManager
from fenrirscreenreader.commands.commands import save_settings
@pytest.mark.unit
@@ -198,3 +200,93 @@ class TestValidationSkipsUnknownSettings:
def test_focus_settings_define_tui_toggle():
"""Focus settings should include the TUI toggle used by on-screen handlers."""
assert settings_data["focus"]["tui"] is False
@pytest.mark.unit
@pytest.mark.settings
class TestSettingsPathSelection:
"""Test root/user settings load and save path selection."""
def setup_method(self):
self.manager = SettingsManager()
def configure_paths(self, tmp_path):
system_root = tmp_path / "etc" / "fenrirscreenreader"
system_file = system_root / "settings" / "settings.conf"
user_file = (
tmp_path
/ "home"
/ "Username"
/ ".local"
/ "share"
/ "stormux"
/ "fenrirscreenreader"
/ "settings"
/ "settings.conf"
)
self.manager.system_settings_root = str(system_root) + "/"
self.manager.system_settings_file = str(system_file)
self.manager.user_settings_file = str(user_file)
return system_file, user_file
def test_non_root_loads_user_settings_when_present(
self, tmp_path, monkeypatch
):
system_file, user_file = self.configure_paths(tmp_path)
system_file.parent.mkdir(parents=True)
system_file.write_text("[general]\n", encoding="utf-8")
user_file.parent.mkdir(parents=True)
user_file.write_text("[general]\n", encoding="utf-8")
monkeypatch.setattr("os.geteuid", lambda: 1000)
assert self.manager.resolve_settings_file() == str(user_file)
def test_non_root_falls_back_to_system_settings(
self, tmp_path, monkeypatch
):
system_file, _user_file = self.configure_paths(tmp_path)
system_file.parent.mkdir(parents=True)
system_file.write_text("[general]\n", encoding="utf-8")
monkeypatch.setattr("os.geteuid", lambda: 1000)
assert self.manager.resolve_settings_file() == str(system_file)
def test_root_uses_system_settings_even_when_user_settings_exists(
self, tmp_path, monkeypatch
):
system_file, user_file = self.configure_paths(tmp_path)
system_file.parent.mkdir(parents=True)
system_file.write_text("[general]\n", encoding="utf-8")
user_file.parent.mkdir(parents=True)
user_file.write_text("[general]\n", encoding="utf-8")
monkeypatch.setattr("os.geteuid", lambda: 0)
assert self.manager.resolve_settings_file() == str(system_file)
def test_save_default_path_follows_effective_user(
self, tmp_path, monkeypatch
):
system_file, user_file = self.configure_paths(tmp_path)
monkeypatch.setattr("os.geteuid", lambda: 1000)
assert self.manager.get_default_save_settings_file() == str(user_file)
monkeypatch.setattr("os.geteuid", lambda: 0)
assert self.manager.get_default_save_settings_file() == str(system_file)
def test_save_settings_command_uses_default_save_path(self):
manager = Mock()
output_manager = Mock()
command = save_settings.command()
command.initialize(
{
"runtime": {
"SettingsManager": manager,
"OutputManager": output_manager,
}
}
)
command.run()
manager.save_settings.assert_called_once_with()
+29
View File
@@ -0,0 +1,29 @@
from unittest.mock import Mock
import pytest
from fenrirscreenreader.core.soundDriver import sound_driver
from fenrirscreenreader.soundDriver import gstreamerDriver
@pytest.mark.unit
def test_base_sound_driver_shutdown_clears_initialized_flag():
driver = sound_driver()
driver.initialize({})
driver.shutdown()
assert driver._initialized is False
@pytest.mark.unit
def test_gstreamer_driver_unavailable_logs_without_crashing(monkeypatch):
monkeypatch.setattr(gstreamerDriver, "_gstreamerAvailable", False)
monkeypatch.setattr(gstreamerDriver, "_availableError", "missing", raising=False)
debug_manager = Mock(write_debug_out=Mock())
driver = gstreamerDriver.driver()
driver.initialize({"runtime": {"DebugManager": debug_manager}})
debug_manager.write_debug_out.assert_called()
assert driver._initialized is False
+33
View File
@@ -0,0 +1,33 @@
from unittest.mock import Mock
import pytest
from fenrirscreenreader.commands.commands import subprocess as subprocess_command
@pytest.mark.unit
def test_script_command_executes_without_shell(monkeypatch):
process = Mock()
process.communicate.return_value = (b"done", b"")
popen = Mock(return_value=process)
monkeypatch.setattr(subprocess_command, "Popen", popen)
output_manager = Mock()
command = subprocess_command.command()
command.initialize(
{
"general": {"curr_user": "Username"},
"runtime": {"OutputManager": output_manager},
},
"/tmp/script with spaces",
)
command._thread_run()
popen.assert_called_once_with(
["/tmp/script with spaces", "Username"],
stdout=subprocess_command.PIPE,
stderr=subprocess_command.PIPE,
)
output_manager.present_text.assert_called_once_with(
"done", sound_icon="", interrupt=False
)
+21
View File
@@ -212,6 +212,27 @@ def test_recent_tab_screen_update_works_without_key_input_snapshot():
assert manager.process_update() == "cuments/"
@pytest.mark.unit
def test_recent_raw_pty_tab_speaks_short_completion_without_capture():
manager, env, input_manager = _build_env(
"cd Documents/".ljust(20), {"x": 13, "y": 0}
)
input_manager.get_last_deepest_input.return_value = ["KEY_TAB"]
input_manager.get_last_event.return_value = None
env["screen"]["old_content_text"] = "cd Docume".ljust(20)
env["screen"]["old_cursor"] = {"x": 9, "y": 0}
env["commandBuffer"]["tabCompletion"]["pending"] = None
_set_screen_update(
env,
"cd Documents/".ljust(20),
{"x": 13, "y": 0},
delta="nts/",
typing=True,
)
assert manager.process_update() == "nts/"
@pytest.mark.unit
def test_large_insertion_echo_speaks_pasted_cursor_text():
large_insertion_module = _load_large_insertion_module()
+32 -5
View File
@@ -167,23 +167,50 @@ def test_x11_build_passive_grabs_for_fenrir_keys_and_shortcuts():
input_manager = Mock(convert_event_name=lambda key: key)
x11.env = {
"input": {
"fenrir_key": ["KEY_KP0", "KEY_CAPSLOCK"],
"fenrir_key": ["KEY_KP0", "KEY_META"],
"script_key": [],
},
"rawBindings": {
"fenrir_combo": [1, ["KEY_FENRIR", "KEY_KP8"]],
"bare_keypad": [1, ["KEY_KP5"]],
"ctrl_keypad": [1, ["KEY_CTRL", "KEY_KP2"]],
"meta_combo": [1, ["KEY_FENRIR", "KEY_BACKSPACE"]],
},
"runtime": {"InputManager": input_manager},
}
grabs = x11.build_passive_grabs()
assert ("KEY_KP0", 0) in grabs
assert ("KEY_CAPSLOCK", 0) in grabs
assert ("KEY_KP5", 0) in grabs
assert ("KEY_KP2", X.ControlMask) in grabs
assert ("KEY_KP0", 0, True) in grabs
assert ("KEY_META", 0, True) in grabs
assert ("KEY_NUMLOCK", 0, True) in grabs
assert ("KEY_KP5", 0, False) in grabs
assert ("KEY_KP2", X.ControlMask, False) in grabs
assert ("KEY_BACKSPACE", X.Mod4Mask, True) in grabs
@pytest.mark.unit
def test_x11_optional_modifier_masks_can_exclude_numlock():
x11 = X11Driver()
x11.num_lock_mask = X.Mod2Mask
masks = x11.optional_modifier_masks(0, include_num_lock=False)
assert X.Mod2Mask not in masks
assert X.Mod2Mask | X.LockMask not in masks
@pytest.mark.unit
def test_x11_get_led_state_reads_lock_modifiers_from_pointer_mask():
x11 = X11Driver()
x11.num_lock_mask = X.Mod2Mask
pointer = Mock(mask=X.Mod2Mask | X.LockMask)
x11.root = Mock()
x11.root.query_pointer.return_value = pointer
assert x11.get_led_state(0) is True
assert x11.get_led_state(1) is True
assert x11.get_led_state(2) is False
@pytest.mark.unit