diff --git a/BlocksScreen/configfile.py b/BlocksScreen/configfile.py index 9d3a3bc6..bfb63720 100644 --- a/BlocksScreen/configfile.py +++ b/BlocksScreen/configfile.py @@ -23,202 +23,185 @@ # # SPDX-License-Identifier: AGPL-3.0-or-later +"""Klipper-style access to BlocksScreen.cfg.""" + from __future__ import annotations import configparser +import copy import enum -import io +import functools import logging import os import pathlib import re +import stat +import tempfile import threading -import typing - -from helper_methods import check_file_on_path +from typing import Any logger = logging.getLogger(__name__) -HOME_DIR = os.path.expanduser("~/") -WORKING_DIR = os.getcwd() -DEFAULT_CONFIGFILE_PATH = pathlib.Path(HOME_DIR, "printer_data", "config") -FALLBACK_CONFIGFILE_PATH = pathlib.Path(WORKING_DIR) +DEFAULT_CONFIGFILE_PATH = pathlib.Path.home() / "printer_data" / "config" +FALLBACK_CONFIGFILE_PATH = pathlib.Path.cwd() +_DEFAULT_CONFIG = DEFAULT_CONFIGFILE_PATH / "BlocksScreen.cfg" +_FALLBACK_CONFIG = FALLBACK_CONFIGFILE_PATH / "BlocksScreen.cfg" + +_singleton_lock = threading.Lock() -T = typing.TypeVar("T") -indentation_size = 4 +_RE_SECTION = re.compile(r"^\[([^]]+)\]") +_RE_SEP = re.compile(r"\s*[:=]\s*") +# Moonraker rule: whitespace + '#'/';' starts a comment, a backslash escapes it +_RE_INLINE_COMMENT = re.compile(r"\s+[#;].*$") +_RE_ESCAPE = re.compile(r"(\s)([#;])") +_RE_UNESCAPE = re.compile(r"(\s)\\([#;])") class Sentinel(enum.Enum): - """Sentinel value to signify missing condition, absence of value""" + """Marks an absent default.""" - MISSING = object + MISSING = object() class ConfigError(Exception): - """Exception raised when Configfile errors exist""" + """Raised when the configuration file is unusable.""" - def __init__(self, msg) -> None: - """Store the error message on both the exception and the ``msg`` attribute.""" + def __init__(self, msg: str) -> None: + """Store *msg* on the exception and as ``msg``.""" super().__init__(msg) self.msg = msg -class BlocksScreenConfig: - """Thread-safe wrapper around :class:`configparser.ConfigParser` with raw-text tracking. +def _option_line(option: str, value: str | None) -> str: + """Build an ``option: value`` line, escaping comment chars so it round-trips.""" + if value is None: + return f"{option}:" + if "\n" in value or "\r" in value: + raise ValueError(f'value for "{option}" must be a single line') + escaped = _RE_ESCAPE.sub(r"\1\\\2", value) + return f"{option}: {escaped}".rstrip() + + +def _atomic_write(path: pathlib.Path, text: str) -> None: + """Write through a temp file and rename so a power cut never truncates *path*.""" + target = path.resolve() # keep a symlinked config pointing at its target + fd, tmp = tempfile.mkstemp(dir=target.parent, prefix=f".{target.name}.") + tmp_path = pathlib.Path(tmp) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(text) + f.flush() + os.fsync(f.fileno()) + if target.exists(): + tmp_path.chmod(stat.S_IMODE(target.stat().st_mode)) # mkstemp makes 0600 + tmp_path.replace(target) + finally: + tmp_path.unlink(missing_ok=True) - Maintains a ``raw_config`` list that mirrors the on-disk file so that - ``add_section``, ``add_option``, and ``update_option`` can write back - changes without losing comments or formatting. - """ - config = configparser.ConfigParser( - allow_no_value=True, - ) - update_pending: bool = False - _instance = None +class BlocksScreenConfig: + """Thread-safe ConfigParser wrapper that mirrors the file in ``raw_config``.""" - def __init__( - self, configfile: typing.Union[str, pathlib.Path], section: str - ) -> None: - """Initialise with the path to the config file and the default section name.""" + def __init__(self, configfile: str | pathlib.Path, section: str) -> None: + """Bind to *configfile*, reading options from *section*.""" self.configfile = pathlib.Path(configfile) self.section = section - self.raw_config: typing.List[str] = [] - self.raw_dict_config: typing.Dict = {} - self.file_lock = threading.Lock() # Thread safety for future work + self.config = configparser.ConfigParser( + allow_no_value=True, + interpolation=None, # as Klipper/Moonraker: '%' in values is literal + comment_prefixes=("#", ";"), + inline_comment_prefixes=None, # stripped in _parse_file + delimiters=(":",), + ) + self.update_pending: bool = False + self.raw_config: list[str] = [] + # RLock: update_option calls add_section/add_option while holding it + self.file_lock = threading.RLock() - def __getitem__(self, key: str) -> BlocksScreenConfig: - """Return a :class:`BlocksScreenConfig` for *key* section (same as ``get_section``).""" + def __getitem__(self, key: str) -> BlocksScreenConfig | None: + """Return the view for section *key*, or None.""" return self.get_section(key) - def __contains__(self, key): - """Return True if *key* is a section in the underlying ConfigParser.""" + def __contains__(self, key: str) -> bool: + """Return True if *key* is a section.""" return key in self.config - def sections(self) -> typing.List[str]: - """Returns list of all sections""" + def sections(self) -> list[str]: + """Return all section names.""" return self.config.sections() def get_section( - self, section: str, fallback: typing.Optional[T] = None - ) -> BlocksScreenConfig: - """Get configfile section""" + self, section: str, fallback: BlocksScreenConfig | None = None + ) -> BlocksScreenConfig | None: + """Return a view of *section* sharing this parser; call writes on the root.""" if not self.config.has_section(section): return fallback - return BlocksScreenConfig(self.configfile, section) + view = copy.copy(self) + view.section = section + return view - def get_options(self) -> list: - """Get section options""" + def get_options(self) -> list[str]: + """Return the option names of this section.""" return self.config.options(self.section) def has_section(self, section: str) -> bool: - """Check if config file has a section - - Args: - section (str): section name - - Returns: - bool: true if section exists, false otherwise - """ - return bool(self.config.has_section(section)) + """Return True if *section* exists.""" + return self.config.has_section(section) def has_option(self, option: str) -> bool: - """Check if section has a option - - Args: - option (str): option name - - Returns: - bool: true if section exists, false otherwise - """ - return bool(self.config.has_option(self.section, option)) + """Return True if *option* exists in this section.""" + return self.config.has_option(self.section, option) def get( self, option: str, parser: type = str, - default: typing.Union[Sentinel, str, T] = Sentinel.MISSING, - ) -> typing.Any: - """Get option value - - Args: - option (str): option name - parser (type, optional): bool, float, int. Defaults to str. - default (typing.Union[Sentinel, str, T], optional): Default value for specified option. Defaults to Sentinel.MISSING. - - Returns: - typing.Union[Sentinel, str]: Requested option. Defaults to the specified default value - """ - value = self.config.get(section=self.section, option=option, fallback=default) - if value is Sentinel.MISSING: - return Sentinel.MISSING - return parser(value) + default: Any = Sentinel.MISSING, + ) -> Any: + """Return *option* cast by *parser*; *default* comes back unparsed.""" + if parser is bool: + # bool("false") is True + return self.getboolean(option, default=default) + try: + return parser(self.config.get(section=self.section, option=option)) + except (configparser.NoOptionError, configparser.NoSectionError): + if default is Sentinel.MISSING: + raise + return default def getlists( self, option: str, - default: Sentinel | list[T] = Sentinel.MISSING, + default: Sentinel | list[Any] = Sentinel.MISSING, sep: str | tuple[str, ...] = ",", parser: type = str, - ) -> list[str | float | int]: - """Get option value parsed as a list - - Args: - option (str): option name - default (Sentinel | list[T], optional): Default value when option missing. - sep (str | tuple[str, ...], optional): Single separator string or tuple - of separators. Defaults to ",". - parser (type, optional): Type to cast each list item. Defaults to str. - - Returns: - list: Parsed list of values, or *default* when option is missing. - """ + ) -> list[Any]: + """Return *option* split on *sep*, cast by *parser*; empty gives *default*.""" raw = self.config.get(section=self.section, option=option, fallback=None) - if raw is Sentinel.MISSING: - return default if default is not Sentinel else [] if not raw: - if default is not Sentinel.MISSING: - return default - return [] - - if isinstance(sep, str): - items = [item.strip() for item in raw.split(sep)] - else: - pattern = "|".join(re.escape(s) for s in sep) - items = [item.strip() for item in re.split(pattern, raw)] - - return [parser(item) for item in items if item] + return [] if default is Sentinel.MISSING else default + seps = (sep,) if isinstance(sep, str) else sep + items = re.split("|".join(map(re.escape, seps)), raw) + return [parser(item.strip()) for item in items if item.strip()] def getint( self, option: str, - default: typing.Union[Sentinel, int] = Sentinel.MISSING, - ) -> typing.Union[Sentinel, int]: - """Get option value - - Args: - option (str): option name - default (typing.Union[Sentinel, int], optional): Default value for specified option. Defaults to Sentinel.MISSING. - - Returns: - typing.Union[Sentinel, int]: Requested option. - """ + default: Sentinel | int = Sentinel.MISSING, + ) -> int: + """Return *option* as int; *default* when absent, else raise.""" + if default is Sentinel.MISSING: + return self.config.getint(section=self.section, option=option) return self.config.getint(section=self.section, option=option, fallback=default) def getfloat( self, option: str, - default: typing.Union[Sentinel, float] = Sentinel.MISSING, - ) -> typing.Union[Sentinel, float]: - """Get the value for the specified option - - Args: - option (str): option name - default (typing.Union[Sentinel, float], optional): Default value for specified option. Defaults to Sentinel.MISSING. - - Returns: - typing.Union[Sentinel, float]: _description_ - """ + default: Sentinel | float = Sentinel.MISSING, + ) -> float: + """Return *option* as float; *default* when absent, else raise.""" + if default is Sentinel.MISSING: + return self.config.getfloat(section=self.section, option=option) return self.config.getfloat( section=self.section, option=option, fallback=default ) @@ -226,261 +209,191 @@ def getfloat( def getboolean( self, option: str, - default: typing.Union[Sentinel, bool] = Sentinel.MISSING, - ) -> typing.Union[Sentinel, bool]: - """Get option value - - Args: - option (str): option name - default (typing.Union[Sentinel, bool], optional): Default value for specified option. Defaults to Sentinel.MISSING. - - Returns: - typing.Union[Sentinel, bool]: _description_ - """ + default: Sentinel | bool = Sentinel.MISSING, + ) -> bool: + """Return *option* as bool; *default* when absent, else raise.""" + if default is Sentinel.MISSING: + return self.config.getboolean(section=self.section, option=option) return self.config.getboolean( section=self.section, option=option, fallback=default ) def _find_section_index(self, section: str) -> int: - """Return the index of the ``[section]`` header line in ``raw_config``.""" + """Return the index of the ``[section]`` line in ``raw_config``.""" try: - return self.raw_config.index("[" + section + "]") + return self.raw_config.index(f"[{section}]") except ValueError as e: - raise configparser.Error(f'Section "{section}" does not exist: {e}') - - def _find_section_limits(self, section: str) -> typing.Tuple: - """Return ``(start_index, end_index)`` of *section* in ``raw_config``.""" - try: - section_start = self._find_section_index(section) - buffer = self.raw_config[section_start:] - section_end = buffer.index("") - return (section_start, int(section_end + section_start)) - except configparser.Error as e: - raise configparser.Error( - f'Error while finding section "{section}" limits on local tracking: {e}' - ) + raise configparser.Error(f'Section "{section}" does not exist') from e - def _find_option_index( - self, section: str, option: str - ) -> typing.Union[Sentinel, int, None]: - """Return the index of the *option* line within *section* in ``raw_config``.""" + def _find_section_limits(self, section: str) -> tuple[int, int]: + """Return ``(header, closing_blank)`` indexes of *section* in ``raw_config``.""" try: - start, end = self._find_section_limits(section) - section_buffer = self.raw_config[start:][:end] - for index in range(len(section_buffer)): - if "[" + option + "]" in section_buffer[index]: - return start + index - raise configparser.Error( - f'Cannot find option "{option}" in section "{section}"' - ) - except configparser.Error as e: - raise configparser.Error( - f'Unable to find option "{option}" in section "{section}": {e}' - ) + start = self._find_section_index(section) + return start, self.raw_config.index("", start) + except (configparser.Error, ValueError) as e: + raise configparser.Error(f'Cannot locate section "{section}": {e}') from e def add_section(self, section: str) -> None: - """Add a section to configuration file - - Args: - section (str): section name - - Raises: - configparser.DuplicateSectionError: Exception thrown when section is duplicated - """ + """Add an empty *section*; logs instead of raising when it exists.""" try: with self.file_lock: - sec_string = f"[{section}]" - if sec_string in self.raw_config: - raise configparser.DuplicateSectionError( - f'Section "{sec_string}" already exists' - ) - config = self.raw_config - if config and config[-1].strip() != "": - config.append("") - config.extend([sec_string, ""]) - updated_config = "\n".join(config) - self.raw_config = updated_config.splitlines() - if self.raw_config[-1] != "": - self.raw_config.append("") self.config.add_section(section) + if self.raw_config and self.raw_config[-1]: + self.raw_config.append("") + self.raw_config.extend([f"[{section}]", ""]) self.update_pending = True - except configparser.DuplicateSectionError as e: - logger.error(f'Section "{section}" already exists. {e}') - except configparser.Error as e: - logger.error(f'Unable to add "{section}" section to configuration: {e}') + except (configparser.Error, ValueError) as e: + logger.error('Unable to add section "%s": %s', section, e) def add_option( self, section: str, option: str, - value: typing.Union[str, None] = None, + value: str | None = None, ) -> None: - """Add option with a value to a section - - Args: - section (str): section name - option (str): option name - value (typing.Union[str, None], optional): value for the specified option. Defaults to None. - """ + """Add *option* to *section*; None writes a value-less ``option:`` line.""" try: with self.file_lock: - section_start, section_end = self._find_section_limits(section) - config = self.raw_config.copy() - opt_string = f"{option}: {value}" - config.insert(section_end, opt_string) - updated_config = "\n".join(config) - self.raw_config = updated_config.splitlines() - if self.raw_config[-1] != "": - self.raw_config.append("") + if self.config.has_option(section, option): + raise configparser.DuplicateOptionError(section, option) + line = _option_line(option, value) + _, end = self._find_section_limits(section) self.config.set(section, option, value) + self.raw_config.insert(end, line) self.update_pending = True - except configparser.DuplicateOptionError as e: - logger.error(f"Option {option} already present on {section}: {e}") - except configparser.Error as e: + except (configparser.Error, ValueError) as e: logger.error( - f'Unable to add "{option}" option to section "{section}": {e} ' + 'Unable to add option "%s" to section "%s": %s', option, section, e ) def update_option( self, section: str, option: str, - value: typing.Any, + value: Any, ) -> None: - """Update an existing option's value in both raw tracking and configparser.""" + """Set *option* in *section*, creating either when missing.""" try: with self.file_lock: if not self.config.has_section(section): self.add_section(section) - if not self.config.has_option(section, option): self.add_option(section, option, str(value)) return - - line_idx = self._find_option_line_index(section, option) - self.raw_config[line_idx] = f"{option}: {value}" + line = _option_line(option, str(value)) + idx = self._find_option_line_index(section, option) self.config.set(section, option, str(value)) + self.raw_config[idx] = line self.update_pending = True - except Exception as e: + except (configparser.Error, ValueError) as e: logger.error( - f'Unable to update option "{option}" in section "{section}": {e}' + 'Unable to update option "%s" in section "%s": %s', option, section, e ) def _find_option_line_index(self, section: str, option: str) -> int: - """Find the index of an option line within a specific section.""" + """Return the ``raw_config`` index of *option* inside *section*.""" start, end = self._find_section_limits(section) - opt_regex = re.compile(rf"^\s*{re.escape(option)}\s*[:=]") + # optionxform lowercases keys, so match the file's spelling case-insensitively + opt_regex = re.compile(rf"^{re.escape(option)}(?::|$)", re.IGNORECASE) for i in range(start + 1, end): if opt_regex.match(self.raw_config[i]): return i raise configparser.Error(f'Option "{option}" not found in section "{section}"') def save_configuration(self) -> None: - """Save teh configuration to file""" + """Write ``raw_config`` atomically; ``update_pending`` stays set on failure.""" try: - if not self.update_pending: - return with self.file_lock: - self.configfile.write_text("\n".join(self.raw_config), encoding="utf-8") - sio = io.StringIO() - sio.writelines(self.raw_config) - self.config.write(sio) - sio.close() - except Exception as e: + if not self.update_pending: + return + _atomic_write(self.configfile, "\n".join(self.raw_config)) + self.update_pending = False + except OSError as e: logger.error( - f"ERROR: Unable to save new configuration, something went wrong while saving updated configuration. {e}" + "Unable to save configuration to %s: %s", + self.configfile, + e, + exc_info=True, ) - finally: - self.update_pending = False - def load_config(self): - """Load configuration file""" - try: - self.raw_config.clear() - self.config.clear() # Reset configparser - self.raw_config, self.raw_dict_config = self._parse_file() - if self.raw_config: - self.config.read_file(self.raw_config) - except Exception as e: - raise configparser.Error(f"Error loading configuration file: {e}") - - def _parse_file(self) -> typing.Tuple[typing.List[str], typing.Dict]: - """Read and normalise the config file into a raw line list and a nested dict. - - Strips comments, normalises ``=`` to ``:`` separators, deduplicates - sections/options, and ensures the buffer ends with an empty line. - - Returns: - A tuple of (raw_lines, dict_representation). - """ - buffer = [] - dict_buff: typing.Dict = {} - curr_sec: typing.Union[Sentinel, str] = Sentinel.MISSING + def load_config(self) -> None: + """(Re)load the file, updating ``raw_config`` in place so views stay valid.""" try: - self.file_lock.acquire(blocking=False) - f = self.configfile.read_text(encoding="utf-8") - for line in f.splitlines(): - line = line.strip() - if not line: - continue - line.expandtabs(indentation_size) - re_match = re.search(r"\s*#\s*(.*)(\s*)", line) - if re_match: - line = line[: re_match.start()] - if not line: - continue - # remove leading and trailing white spaces - line = re.sub(r"\s*([:=])\s*", r"\1 ", line) - line = re.sub(r"=", r":", line) - # find the beginning of sections - section_match = re.compile(r"[^\s]*\[([^]]+)\]") - match_sec = re.match(section_match, line) # - if match_sec: - sec_name = re.sub(r"[\[*\]]", r"", line) - if sec_name not in dict_buff.keys(): - if buffer: - buffer.extend( - [""] - ) # REFACTOR: Just add some line separation between sections - dict_buff.update({sec_name: {}}) - curr_sec = sec_name - else: - continue - option_match = re.compile(r"^(?:\w+)([:*])(?:.*)") - match_opt = re.match(option_match, line) - if match_opt: - option_name, value = line.split(":", maxsplit=1) - if option_name not in dict_buff.get(curr_sec, {}).keys(): - if curr_sec in dict_buff.keys(): - section: dict = dict_buff.get(curr_sec, {}) - section.update({option_name: value}) - dict_buff.update({curr_sec: section}) - else: - continue - buffer.append(line) - if buffer[-1] != "": - buffer.append("") - return buffer, dict_buff - except Exception as e: - raise configparser.Error( - f"Unexpected error while parsing configuration file: {e}" - ) - finally: - self.file_lock.release() - - -def get_configparser() -> BlocksScreenConfig: - """Loads configuration from file and returns that configuration""" - wanted_target = os.path.join(DEFAULT_CONFIGFILE_PATH, "BlocksScreen.cfg") - fallback = os.path.join(WORKING_DIR, "BlocksScreen.cfg") - configfile = ( - wanted_target - if check_file_on_path(DEFAULT_CONFIGFILE_PATH, "BlocksScreen.cfg") - else fallback - ) + new_raw = self._parse_file() + self.config.clear() + self.config.read_file(_RE_UNESCAPE.sub(r"\1\2", ln) for ln in new_raw) + self.raw_config[:] = new_raw + except (OSError, ValueError, configparser.Error) as e: + raise configparser.Error(f"Error loading configuration file: {e}") from e + + def _parse_file(self) -> list[str]: + """Read the file into normalised lines, one blank-ended block per section.""" + with self.file_lock: + text = self.configfile.read_text(encoding="utf-8") + blocks: dict[str | None, list[str]] = {None: []} + opt_index: dict[tuple[str, str], int] = {} + pending: list[str] = [] # full-line comments, kept with the next line + sec: str | None = None + for lineno, line in enumerate(map(str.strip, text.splitlines()), 1): + if not line: + continue + if line[0] in "#;": + pending.append(line) + continue + # before separator handling so '[fan:x]' keeps its colon + if m_sec := _RE_SECTION.match(line): + sec = m_sec.group(1) + if sec in blocks: + blocks[sec].extend(pending) # Klipper merges repeated sections + else: + blocks[sec] = [*pending, f"[{sec}]"] + pending = [] + continue + name, *rest = _RE_SEP.split(line, maxsplit=1) + name = _RE_INLINE_COMMENT.sub("", name) + if sec is None or not name: + logger.warning( + "%s:%d ignored: no section or option name", self.configfile, lineno + ) + continue + opt = name + if rest: + # on the value alone, so 'color: #ff0000' is not a comment + value = _RE_INLINE_COMMENT.sub("", rest[0]) + opt = f"{name}: {value}" if value else f"{name}:" + block = blocks[sec] + block.extend(pending) + pending = [] + key = (sec, self.config.optionxform(name)) + if key in opt_index: + block[opt_index[key]] = opt # Klipper: the last duplicate wins + else: + opt_index[key] = len(block) + block.append(opt) + blocks[sec].extend(pending) + return [ln for block in blocks.values() if block for ln in (*block, "")] or [""] + + +@functools.cache +def _load_singleton() -> BlocksScreenConfig: + """Build the process-wide config; raises are not cached, so callers can retry.""" + configfile = _DEFAULT_CONFIG if _DEFAULT_CONFIG.exists() else _FALLBACK_CONFIG config_object = BlocksScreenConfig(configfile=configfile, section="server") config_object.load_config() if not config_object.has_section("server"): logger.error("Error loading configuration file for the application.") raise ConfigError("Section [server] is missing from configuration") return config_object + + +def get_configparser() -> BlocksScreenConfig: + """Return the singleton config, loading it on first call.""" + # functools.cache alone can build twice when two threads race the first call + with _singleton_lock: + return _load_singleton() + + +def reset_configparser() -> None: + """Drop the singleton; tests only.""" + with _singleton_lock: + _load_singleton.cache_clear() diff --git a/BlocksScreen/lib/panels/widgets/loadWidget.py b/BlocksScreen/lib/panels/widgets/loadWidget.py index 1119bd14..5a54dd89 100644 --- a/BlocksScreen/lib/panels/widgets/loadWidget.py +++ b/BlocksScreen/lib/panels/widgets/loadWidget.py @@ -42,7 +42,9 @@ def __init__( else: try: - loading_config = config.loading + loading_config = config["loading"] + if loading_config is None: + raise KeyError("missing [loading] section") animation_path = loading_config.get( str(initial_anim_type.name), ) diff --git a/tests/util/test_configfile_unit.py b/tests/util/test_configfile_unit.py new file mode 100644 index 00000000..eebf67f2 --- /dev/null +++ b/tests/util/test_configfile_unit.py @@ -0,0 +1,825 @@ +"""Unit tests for BlocksScreen/configfile.py.""" + +from __future__ import annotations + +import configparser +import logging +import pathlib +import sys +import textwrap +import threading + +import pytest + +_BLOCKSSCREEN = pathlib.Path(__file__).parent.parent.parent / "BlocksScreen" +if str(_BLOCKSSCREEN) not in sys.path: + sys.path.insert(0, str(_BLOCKSSCREEN)) + +# drop tests/network's configfile mock; network modules already bound its symbols +sys.modules.pop("configfile", None) + +import configfile as cfmod # noqa: E402 +from configfile import ( + BlocksScreenConfig, + ConfigError, + Sentinel, + get_configparser, + reset_configparser, +) # noqa: E402 + + +def _make_cfg( + tmp_path: pathlib.Path, content: str, section: str = "server" +) -> BlocksScreenConfig: + """Write *content* to a temp file, load it, and return the config object.""" + cfg_file = tmp_path / "test.cfg" + cfg_file.write_text(textwrap.dedent(content).strip(), encoding="utf-8") + cfg = BlocksScreenConfig(cfg_file, section) + cfg.load_config() + return cfg + + +@pytest.mark.unit +class TestSentinel: + def test_missing_value_is_an_instance_not_the_class(self): + assert Sentinel.MISSING.value is not object + assert isinstance(Sentinel.MISSING.value, object) + + def test_missing_is_an_enum_member(self): + assert isinstance(Sentinel.MISSING, Sentinel) + + +@pytest.mark.unit +class TestParsing: + def test_basic_section_and_option(self, tmp_path): + cfg = _make_cfg( + tmp_path, + """ + [server] + host: localhost + """, + ) + assert cfg.has_section("server") + assert cfg.get("host") == "localhost" + + def test_equals_separator_normalised_to_colon(self, tmp_path): + cfg = _make_cfg( + tmp_path, + """ + [server] + host = localhost + """, + ) + assert cfg.get("host") == "localhost" + + def test_full_line_hash_comment_ignored(self, tmp_path): + cfg = _make_cfg( + tmp_path, + """ + # this is a comment + [server] + host: localhost + """, + ) + assert cfg.has_section("server") + assert not cfg.has_section("# this is a comment") + + def test_full_line_semicolon_comment_ignored(self, tmp_path): + cfg = _make_cfg( + tmp_path, + """ + ; this is also a comment + [server] + host: localhost + """, + ) + assert cfg.has_section("server") + assert not cfg.has_section("; this is also a comment") + + def test_inline_hash_comment_stripped(self, tmp_path): + cfg = _make_cfg( + tmp_path, + """ + [server] + host: localhost # inline comment + """, + ) + assert cfg.get("host") == "localhost" + + def test_inline_semicolon_comment_stripped(self, tmp_path): + cfg = _make_cfg( + tmp_path, + """ + [server] + host: localhost ; inline comment + """, + ) + assert cfg.get("host") == "localhost" + + def test_hash_in_value_without_leading_space_preserved(self, tmp_path): + """#ff0000 has no space before # so must not be stripped as a comment.""" + cfg = _make_cfg( + tmp_path, + """ + [server] + color: #ff0000 + """, + ) + assert cfg.get("color") == "#ff0000" + + def test_duplicate_section_merges_like_klipper(self, tmp_path): + cfg = _make_cfg( + tmp_path, + """ + [server] + host: first + port: 1 + [display] + width: 800 + [server] + host: second + """, + ) + assert cfg.get("host") == "second" + assert cfg.get("port") == "1" + assert cfg.get_section("display").get_options() == ["width"] + assert cfg.raw_config.count("[server]") == 1 + + def test_duplicate_option_last_wins(self, tmp_path): + cfg = _make_cfg( + tmp_path, + """ + [server] + host: first + host: second + """, + ) + assert cfg.get("host") == "second" + assert [ln for ln in cfg.raw_config if ln.startswith("host")] == [ + "host: second" + ] + + def test_case_duplicate_option_loads(self, tmp_path): + cfg = _make_cfg(tmp_path, "[server]\nHost: first\nhost: second\n") + assert cfg.get("host") == "second" + cfg.update_option("server", "host", "third") + assert [ln for ln in cfg.raw_config if ln.lower().startswith("host")] == [ + "host: third" + ] + + def test_dash_dot_and_bare_keys_kept(self, tmp_path): + cfg = _make_cfg(tmp_path, "[server]\napi-key: a\nlog.level: b\nflag\n") + assert cfg.get("api-key") == "a" + assert cfg.get("log.level") == "b" + assert cfg.has_option("flag") + + def test_header_inline_comment_stripped(self, tmp_path): + cfg = _make_cfg(tmp_path, "[server] # main\nhost: localhost\n") + cfg.add_option("server", "port", "7125") + assert "[server]" in cfg.raw_config + assert cfg.getint("port") == 7125 + + def test_option_before_any_section_ignored(self, tmp_path): + cfg = _make_cfg(tmp_path, "orphan: 1\n[server]\nhost: localhost\n") + assert cfg.sections() == ["server"] + assert not any("orphan" in ln for ln in cfg.raw_config) + + def test_full_line_comments_kept_in_raw_config(self, tmp_path): + cfg = _make_cfg(tmp_path, "# top\n[server]\n; why\nhost: localhost\n") + assert cfg.raw_config[:4] == ["# top", "[server]", "; why", "host: localhost"] + + def test_multiple_sections_all_loaded(self, tmp_path): + cfg = _make_cfg( + tmp_path, + """ + [server] + host: localhost + [display] + width: 800 + """, + ) + assert cfg.has_section("server") + assert cfg.has_section("display") + + def test_raw_config_always_ends_with_empty_line(self, tmp_path): + cfg = _make_cfg( + tmp_path, + """ + [server] + host: localhost + """, + ) + assert cfg.raw_config[-1] == "" + + def test_load_config_resets_previous_state(self, tmp_path): + cfg_file = tmp_path / "test.cfg" + cfg_file.write_text("[server]\nhost: localhost\n", encoding="utf-8") + cfg = BlocksScreenConfig(cfg_file, "server") + cfg.load_config() + assert cfg.get("host") == "localhost" + + cfg_file.write_text("[server]\nhost: remotehost\n", encoding="utf-8") + cfg.load_config() + assert cfg.get("host") == "remotehost" + + def test_value_with_embedded_colon_preserved(self, tmp_path): + """Values containing ':' after the key separator must not be mangled (e.g. URLs).""" + cfg = _make_cfg( + tmp_path, + """ + [server] + url: http://localhost:7125 + """, + ) + assert cfg.get("url") == "http://localhost:7125" + + def test_value_with_embedded_equals_preserved(self, tmp_path): + """Values containing '=' after the key separator must not be mangled (e.g. tokens).""" + cfg = _make_cfg( + tmp_path, + """ + [server] + token: abc=def==ghi + """, + ) + assert cfg.get("token") == "abc=def==ghi" + + def test_section_name_with_colon_preserved(self, tmp_path): + """Section names containing ':' must not be mangled by separator normalisation.""" + cfg = _make_cfg( + tmp_path, + """ + [fan:heater_fan] + pin: PA8 + """, + section="fan:heater_fan", + ) + assert cfg.has_section("fan:heater_fan") + assert cfg.get("pin") == "PA8" + + +@pytest.mark.unit +class TestReadMethods: + @pytest.fixture + def cfg(self, tmp_path): + return _make_cfg( + tmp_path, + """ + [server] + host: localhost + port: 7125 + ratio: 1.5 + enabled: true + """, + ) + + def test_has_section_present(self, cfg): + assert cfg.has_section("server") is True + + def test_has_section_absent(self, cfg): + assert cfg.has_section("missing") is False + + def test_has_option_present(self, cfg): + assert cfg.has_option("host") is True + + def test_has_option_absent(self, cfg): + assert cfg.has_option("missing") is False + + def test_get_returns_string(self, cfg): + assert cfg.get("host") == "localhost" + + def test_get_returns_default_when_absent(self, cfg): + assert cfg.get("missing", default="fallback") == "fallback" + + def test_get_with_int_parser(self, cfg): + assert cfg.get("port", parser=int) == 7125 + + def test_get_with_bool_parser_false_value(self, tmp_path): + """get(parser=bool) must not treat 'false' as True (bool('false') pitfall).""" + cfg = _make_cfg( + tmp_path, + """ + [server] + host: localhost + flag: false + """, + ) + assert cfg.get("flag", parser=bool) is False + + def test_get_with_bool_parser_true_value(self, tmp_path): + cfg = _make_cfg( + tmp_path, + """ + [server] + host: localhost + flag: true + """, + ) + assert cfg.get("flag", parser=bool) is True + + def test_getint(self, cfg): + assert cfg.getint("port") == 7125 + + def test_getint_default_when_absent(self, cfg): + assert cfg.getint("missing", default=42) == 42 + + def test_getfloat(self, cfg): + assert cfg.getfloat("ratio") == pytest.approx(1.5) + + def test_getfloat_default_when_absent(self, cfg): + assert cfg.getfloat("missing", default=0.5) == pytest.approx(0.5) + + def test_getboolean_true(self, cfg): + assert cfg.getboolean("enabled") is True + + def test_getboolean_default_when_absent(self, cfg): + assert cfg.getboolean("missing", default=False) is False + + def test_sections_returns_list_of_names(self, cfg): + assert "server" in cfg.sections() + + def test_get_options_returns_option_names(self, cfg): + opts = cfg.get_options() + assert "host" in opts + assert "port" in opts + + def test_get_default_returned_as_is_not_parsed(self, cfg): + """An absent option returns the default unparsed, so int("N/A") never runs.""" + result = cfg.get("missing", parser=int, default="N/A") + assert result == "N/A" + + def test_getint_raises_when_absent_no_default(self, cfg): + """getint() with no default must raise, not silently return Sentinel.MISSING.""" + with pytest.raises((configparser.NoOptionError, configparser.NoSectionError)): + cfg.getint("missing") + + def test_getfloat_raises_when_absent_no_default(self, cfg): + """getfloat() with no default must raise, not silently return Sentinel.MISSING.""" + with pytest.raises((configparser.NoOptionError, configparser.NoSectionError)): + cfg.getfloat("missing") + + def test_getboolean_raises_when_absent_no_default(self, cfg): + """getboolean() with no default must raise, not silently return Sentinel.MISSING.""" + with pytest.raises((configparser.NoOptionError, configparser.NoSectionError)): + cfg.getboolean("missing") + + +@pytest.mark.unit +class TestGetSection: + @pytest.fixture + def cfg(self, tmp_path): + return _make_cfg( + tmp_path, + """ + [server] + host: localhost + [display] + width: 800 + """, + ) + + def test_view_reads_actual_values(self, cfg): + view = cfg.get_section("server") + assert view is not None + assert view.get("host") == "localhost" + + def test_view_shares_the_configparser_instance(self, cfg): + view = cfg.get_section("display") + assert view.config is cfg.config + + def test_view_uses_correct_section(self, cfg): + view = cfg.get_section("display") + assert view.section == "display" + assert view.getint("width") == 800 + + def test_returns_none_when_section_missing(self, cfg): + assert cfg.get_section("nonexistent") is None + + def test_returns_custom_fallback_when_section_missing(self, cfg): + sentinel = object() + assert cfg.get_section("nonexistent", fallback=sentinel) is sentinel + + def test_getitem_returns_working_view(self, cfg): + view = cfg["server"] + assert view is not None + assert view.get("host") == "localhost" + + def test_getitem_returns_none_for_missing_section(self, cfg): + assert cfg["nonexistent"] is None + + def test_contains_returns_true_for_existing_section(self, cfg): + assert "server" in cfg + + def test_contains_returns_false_for_missing_section(self, cfg): + assert "nonexistent" not in cfg + + def test_view_raw_config_stays_valid_after_reload(self, tmp_path): + """load_config must update raw_config in-place so existing views remain valid.""" + cfg_file = tmp_path / "test.cfg" + cfg_file.write_text( + "[server]\nhost: first\n[display]\nwidth: 800\n", encoding="utf-8" + ) + cfg = BlocksScreenConfig(cfg_file, "server") + cfg.load_config() + view = cfg.get_section("display") + assert view is not None + assert view.raw_config is cfg.raw_config # same list object + + cfg_file.write_text( + "[server]\nhost: second\n[display]\nwidth: 1024\n", encoding="utf-8" + ) + cfg.load_config() + assert view.raw_config is cfg.raw_config # still the same list object + assert any("1024" in line for line in view.raw_config) + + +@pytest.mark.unit +class TestAddSection: + @pytest.fixture + def cfg(self, tmp_path): + return _make_cfg( + tmp_path, + """ + [server] + host: localhost + """, + ) + + def test_section_becomes_visible_after_add(self, cfg): + cfg.add_section("display") + assert cfg.has_section("display") + + def test_section_header_appears_in_raw_config(self, cfg): + cfg.add_section("display") + assert "[display]" in cfg.raw_config + + def test_sets_update_pending(self, cfg): + cfg.add_section("display") + assert cfg.update_pending is True + + def test_raw_config_ends_with_empty_line_after_add(self, cfg): + cfg.add_section("display") + assert cfg.raw_config[-1] == "" + + def test_duplicate_section_logs_error_not_raises(self, cfg, caplog): + with caplog.at_level(logging.ERROR): + cfg.add_section("server") # already exists + assert "already exists" in caplog.text.lower() + + def test_duplicate_section_does_not_raise(self, cfg): + cfg.add_section("server") # must not propagate an exception + + +@pytest.mark.unit +class TestAddOption: + @pytest.fixture + def cfg(self, tmp_path): + return _make_cfg( + tmp_path, + """ + [server] + host: localhost + """, + ) + + def test_option_becomes_visible_after_add(self, cfg): + cfg.add_option("server", "port", "7125") + assert cfg.has_option("port") + + def test_added_option_is_readable(self, cfg): + cfg.add_option("server", "port", "7125") + assert cfg.getint("port") == 7125 + + def test_option_line_appears_in_raw_config(self, cfg): + cfg.add_option("server", "port", "7125") + assert any("port: 7125" in line for line in cfg.raw_config) + + def test_sets_update_pending(self, cfg): + cfg.add_option("server", "port", "7125") + assert cfg.update_pending is True + + def test_option_inserted_before_section_separator(self, cfg): + cfg.add_option("server", "port", "7125") + port_idx = next(i for i, l in enumerate(cfg.raw_config) if "port" in l) + assert cfg.raw_config[port_idx + 1] == "" + + def test_none_value_writes_valueless_option_line(self, cfg): + """add_option(value=None) must write 'flag:' not 'flag: None'.""" + cfg.add_option("server", "flag", None) + assert not any("None" in line for line in cfg.raw_config) + assert any(line.strip() == "flag:" for line in cfg.raw_config) + + def test_none_value_option_is_readable(self, cfg): + """Value-less options must be detectable via has_option after add.""" + cfg.add_option("server", "flag", None) + assert cfg.has_option("flag") + + def test_existing_option_not_duplicated(self, cfg, caplog): + with caplog.at_level(logging.ERROR): + cfg.add_option("server", "host", "other") + assert cfg.get("host") == "localhost" + assert sum(ln.startswith("host") for ln in cfg.raw_config) == 1 + assert cfg.update_pending is False + + +@pytest.mark.unit +class TestUpdateOption: + @pytest.fixture + def cfg(self, tmp_path): + return _make_cfg( + tmp_path, + """ + [server] + host: localhost + port: 7125 + """, + ) + + def test_updates_existing_option_value(self, cfg): + cfg.update_option("server", "host", "remotehost") + assert cfg.get("host") == "remotehost" + + def test_updated_value_reflected_in_raw_config(self, cfg): + cfg.update_option("server", "host", "remotehost") + assert any("host: remotehost" in line for line in cfg.raw_config) + + def test_old_value_absent_from_raw_config(self, cfg): + cfg.update_option("server", "host", "remotehost") + assert not any("host: localhost" in line for line in cfg.raw_config) + + def test_sets_update_pending(self, cfg): + cfg.update_option("server", "host", "remotehost") + assert cfg.update_pending is True + + def test_creates_missing_option(self, cfg): + cfg.update_option("server", "timeout", "30") + assert cfg.get("timeout") == "30" + + def test_creates_missing_section_and_option(self, cfg): + cfg.update_option("display", "width", "800") + assert cfg.has_section("display") + view = cfg.get_section("display") + assert view.get("width") == "800" + + def test_reentrancy_does_not_deadlock(self, tmp_path): + """update_option re-enters add_section/add_option under the RLock.""" + cfg = _make_cfg(tmp_path, "[server]\nhost: localhost\n") + cfg.update_option("brandnew", "key", "value") + assert cfg.has_section("brandnew") + assert cfg.get_section("brandnew").get("key") == "value" + + def test_newline_in_value_rejected(self, cfg): + cfg.update_option("server", "host", "evil\n[injected]") + assert cfg.get("host") == "localhost" + assert "[injected]" not in cfg.raw_config + assert cfg.update_pending is False + + +@pytest.mark.unit +class TestValueRoundTrip: + @pytest.mark.parametrize( + "value", ["Printer #2", "my pass ;1 #2", "#ff0000", "100%", "%(x)s", "a=b:c"] + ) + def test_value_survives_save_and_reload(self, tmp_path, value): + cfg = _make_cfg(tmp_path, "[server]\nhost: localhost\n") + cfg.update_option("server", "password", value) + cfg.save_configuration() + cfg.load_config() + assert cfg.get("password") == value + + def test_percent_in_file_is_literal(self, tmp_path): + cfg = _make_cfg(tmp_path, "[server]\npassword: 50%off\n") + assert cfg.get("password") == "50%off" + + +@pytest.mark.unit +class TestSaveConfiguration: + def test_writes_updated_value_to_file(self, tmp_path): + cfg_file = tmp_path / "test.cfg" + cfg_file.write_text("[server]\nhost: localhost\n", encoding="utf-8") + cfg = BlocksScreenConfig(cfg_file, "server") + cfg.load_config() + cfg.update_option("server", "host", "newhost") + cfg.save_configuration() + assert "host: newhost" in cfg_file.read_text(encoding="utf-8") + + def test_clears_update_pending_after_save(self, tmp_path): + cfg_file = tmp_path / "test.cfg" + cfg_file.write_text("[server]\nhost: localhost\n", encoding="utf-8") + cfg = BlocksScreenConfig(cfg_file, "server") + cfg.load_config() + cfg.update_option("server", "host", "newhost") + cfg.save_configuration() + assert cfg.update_pending is False + + def test_skips_write_when_not_pending(self, tmp_path): + original = "[server]\nhost: localhost\n" + cfg_file = tmp_path / "test.cfg" + cfg_file.write_text(original, encoding="utf-8") + cfg = BlocksScreenConfig(cfg_file, "server") + cfg.load_config() + assert cfg.update_pending is False + cfg.save_configuration() + assert cfg_file.read_text(encoding="utf-8") == original + + def test_pending_preserved_when_write_fails(self, tmp_path): + """A failed write must leave update_pending True so the caller can retry.""" + cfg_file = tmp_path / "test.cfg" + cfg_file.write_text("[server]\nhost: localhost\n", encoding="utf-8") + cfg = BlocksScreenConfig(cfg_file, "server") + cfg.load_config() + cfg.update_pending = True + cfg.configfile = pathlib.Path("/nonexistent_dir/file.cfg") + cfg.save_configuration() + assert cfg.update_pending is True + + def test_save_keeps_mode_and_leaves_no_temp_file(self, tmp_path): + cfg_file = tmp_path / "test.cfg" + cfg_file.write_text("[server]\nhost: localhost\n", encoding="utf-8") + cfg_file.chmod(0o644) + cfg = BlocksScreenConfig(cfg_file, "server") + cfg.load_config() + cfg.update_option("server", "host", "newhost") + cfg.save_configuration() + assert cfg_file.stat().st_mode & 0o777 == 0o644 + assert [f.name for f in tmp_path.iterdir()] == ["test.cfg"] + + def test_save_through_symlink_keeps_link(self, tmp_path): + target = tmp_path / "real.cfg" + target.write_text("[server]\nhost: localhost\n", encoding="utf-8") + link = tmp_path / "test.cfg" + link.symlink_to(target) + cfg = BlocksScreenConfig(link, "server") + cfg.load_config() + cfg.update_option("server", "host", "newhost") + cfg.save_configuration() + assert link.is_symlink() + assert "host: newhost" in target.read_text(encoding="utf-8") + + def test_full_line_comments_survive_save(self, tmp_path): + cfg_file = tmp_path / "test.cfg" + cfg_file.write_text("[server]\n# keep me\nhost: localhost\n", encoding="utf-8") + cfg = BlocksScreenConfig(cfg_file, "server") + cfg.load_config() + cfg.update_option("server", "host", "newhost") + cfg.save_configuration() + assert "# keep me" in cfg_file.read_text(encoding="utf-8") + + +@pytest.mark.unit +class TestGetConfigparser: + @pytest.fixture(autouse=True) + def _reset_singleton(self): + """Reset the module-level singleton before and after each test.""" + reset_configparser() + yield + reset_configparser() + + def test_loads_from_fallback_path(self, tmp_path, monkeypatch): + cfg_file = tmp_path / "BlocksScreen.cfg" + cfg_file.write_text("[server]\nhost: localhost\n", encoding="utf-8") + + monkeypatch.setattr( + cfmod, "_DEFAULT_CONFIG", tmp_path / "nonexistent" / "BlocksScreen.cfg" + ) + monkeypatch.setattr(cfmod, "_FALLBACK_CONFIG", cfg_file) + + cfg = get_configparser() + assert cfg.has_section("server") + + def test_raises_config_error_when_server_section_missing( + self, tmp_path, monkeypatch + ): + cfg_file = tmp_path / "BlocksScreen.cfg" + cfg_file.write_text("[display]\nwidth: 800\n", encoding="utf-8") + + monkeypatch.setattr( + cfmod, "_DEFAULT_CONFIG", tmp_path / "nonexistent" / "BlocksScreen.cfg" + ) + monkeypatch.setattr(cfmod, "_FALLBACK_CONFIG", cfg_file) + + with pytest.raises(ConfigError, match=r"\[server\]"): + get_configparser() + + def test_returns_same_instance_on_repeated_calls(self, tmp_path, monkeypatch): + cfg_file = tmp_path / "BlocksScreen.cfg" + cfg_file.write_text("[server]\nhost: localhost\n", encoding="utf-8") + + monkeypatch.setattr( + cfmod, "_DEFAULT_CONFIG", tmp_path / "nonexistent" / "BlocksScreen.cfg" + ) + monkeypatch.setattr(cfmod, "_FALLBACK_CONFIG", cfg_file) + + assert get_configparser() is get_configparser() + + def test_concurrent_first_calls_build_one_instance(self, tmp_path, monkeypatch): + cfg_file = tmp_path / "BlocksScreen.cfg" + cfg_file.write_text("[server]\nhost: localhost\n", encoding="utf-8") + monkeypatch.setattr(cfmod, "_DEFAULT_CONFIG", tmp_path / "missing.cfg") + monkeypatch.setattr(cfmod, "_FALLBACK_CONFIG", cfg_file) + barrier = threading.Barrier(8) + results: list[BlocksScreenConfig] = [] + + def worker() -> None: + barrier.wait() + results.append(get_configparser()) + + threads = [threading.Thread(target=worker) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + assert len({id(r) for r in results}) == 1 + + def test_failed_load_is_retried_not_cached(self, tmp_path, monkeypatch): + cfg_file = tmp_path / "BlocksScreen.cfg" + cfg_file.write_text("[display]\nwidth: 800\n", encoding="utf-8") + monkeypatch.setattr(cfmod, "_DEFAULT_CONFIG", tmp_path / "missing.cfg") + monkeypatch.setattr(cfmod, "_FALLBACK_CONFIG", cfg_file) + with pytest.raises(ConfigError): + get_configparser() + cfg_file.write_text("[server]\nhost: localhost\n", encoding="utf-8") + assert get_configparser().get("host") == "localhost" + + +@pytest.mark.unit +class TestThreadSafety: + def test_concurrent_updates_do_not_corrupt_state(self, tmp_path): + cfg_file = tmp_path / "test.cfg" + cfg_file.write_text("[server]\nhost: localhost\n", encoding="utf-8") + cfg = BlocksScreenConfig(cfg_file, "server") + cfg.load_config() + + errors: list[Exception] = [] + + def updater(value: str) -> None: + try: + cfg.update_option("server", "host", value) + except Exception as exc: + errors.append(exc) + + threads = [ + threading.Thread(target=updater, args=(f"host{i}",)) for i in range(20) + ] + for t in threads: + t.start() + for t in threads: + t.join() + + assert not errors + assert cfg.has_option("host") + + +@pytest.mark.unit +class TestEdgeCases: + def test_parse_empty_file_does_not_crash(self, tmp_path): + """A comment-only file must not raise IndexError.""" + cfg_file = tmp_path / "test.cfg" + cfg_file.write_text("# only a comment\n", encoding="utf-8") + cfg = BlocksScreenConfig(cfg_file, "server") + cfg.load_config() + assert cfg.raw_config == ["# only a comment", ""] + + def test_parse_completely_empty_file_does_not_crash(self, tmp_path): + cfg_file = tmp_path / "test.cfg" + cfg_file.write_text("", encoding="utf-8") + cfg = BlocksScreenConfig(cfg_file, "server") + cfg.load_config() + assert cfg.raw_config == [""] + + def test_get_missing_option_without_default_raises(self, tmp_path): + """get() with no default must raise NoOptionError, not silently return Sentinel.MISSING.""" + cfg = _make_cfg(tmp_path, "[server]\nhost: localhost\n") + with pytest.raises(configparser.NoOptionError): + cfg.get("nonexistent") + + def test_load_config_missing_file_raises(self, tmp_path): + """load_config on a non-existent file must raise configparser.Error, not crash silently.""" + cfg = BlocksScreenConfig(tmp_path / "nonexistent.cfg", "server") + with pytest.raises(configparser.Error): + cfg.load_config() + + def test_save_configuration_toctou_safe(self, tmp_path): + """update_pending check and file write happen atomically under the lock.""" + cfg_file = tmp_path / "test.cfg" + cfg_file.write_text("[server]\nhost: localhost\n", encoding="utf-8") + cfg = BlocksScreenConfig(cfg_file, "server") + cfg.load_config() + cfg.update_option("server", "host", "newhost") + + errors: list[Exception] = [] + + def saver() -> None: + try: + cfg.save_configuration() + except Exception as exc: + errors.append(exc) + + threads = [threading.Thread(target=saver) for _ in range(10)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert not errors + assert "host: newhost" in cfg_file.read_text(encoding="utf-8") diff --git a/tests/widgets/test_slider_page_request_unit.py b/tests/widgets/test_slider_page_request_unit.py index d1eefaf1..ed1f5e8b 100644 --- a/tests/widgets/test_slider_page_request_unit.py +++ b/tests/widgets/test_slider_page_request_unit.py @@ -19,11 +19,9 @@ import pytest -# printTab/controlTab import configfile, lib.files, lib.printer purely for -# type hints and passthrough attribute assignment -- stub them so importing -# the tab classes doesn't require the real config/websocket/printer stack. +# fresh stubs, not setdefault: patching an already-imported real configfile leaks for _name in ("configfile", "lib.files", "lib.printer"): - sys.modules.setdefault(_name, types.ModuleType(_name)) + sys.modules[_name] = types.ModuleType(_name) sys.modules["configfile"].BlocksScreenConfig = MagicMock sys.modules["configfile"].get_configparser = MagicMock(return_value=MagicMock()) sys.modules["lib.files"].Files = MagicMock