From bdb9fb5d88a8203a09368c41a063380a9f6f0b3c Mon Sep 17 00:00:00 2001 From: Guilherme Costa Date: Fri, 13 Mar 2026 12:07:29 +0000 Subject: [PATCH 1/5] fix(configfile): fix bugs, optimize parser, add support for normal an inline comments and add unit tests loadWidget: fix bug that would result in a crash --- BlocksScreen/configfile.py | 347 ++++----- BlocksScreen/lib/panels/widgets/loadWidget.py | 4 +- tests/util/test_configfile_unit.py | 664 ++++++++++++++++++ 3 files changed, 844 insertions(+), 171 deletions(-) create mode 100644 tests/util/test_configfile_unit.py diff --git a/BlocksScreen/configfile.py b/BlocksScreen/configfile.py index ed5828a0..904519bb 100644 --- a/BlocksScreen/configfile.py +++ b/BlocksScreen/configfile.py @@ -27,31 +27,29 @@ import configparser import enum -import io import logging -import os import pathlib import re import threading -import typing +from typing import Any from helper_methods import check_file_on_path 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() -T = typing.TypeVar("T") -indentation_size = 4 +_RE_SECTION = re.compile(r"^\s*\[([^]]+)\]") +_RE_OPTION = re.compile(r"^(\w+)[=:]") +_RE_INLINE_COMMENT = re.compile(r"(?<=\w)\s+[#;]") +_RE_SEP_NORMALIZE = re.compile(r"\s*[:=]\s*") class Sentinel(enum.Enum): """Sentinel value to signify missing condition, absence of value""" - MISSING = object + MISSING = object() class ConfigError(Exception): @@ -71,43 +69,50 @@ class BlocksScreenConfig: changes without losing comments or formatting. """ - config = configparser.ConfigParser( - allow_no_value=True, - ) - update_pending: bool = False - _instance = None - - def __init__( - self, configfile: typing.Union[str, pathlib.Path], section: str - ) -> None: + def __init__(self, configfile: str | pathlib.Path, section: str) -> None: """Initialise with the path to the config file and the default section name.""" 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, + comment_prefixes=("#", ";"), + inline_comment_prefixes=None, # handled manually in _parse_file + ) + self.update_pending: bool = False + self.raw_config: list[str] = [] + # RLock: update_option calls add_section/add_option while holding the lock + self.file_lock = threading.RLock() - def __getitem__(self, key: str) -> BlocksScreenConfig: + def __getitem__(self, key: str) -> BlocksScreenConfig | None: """Return a :class:`BlocksScreenConfig` for *key* section (same as ``get_section``).""" return self.get_section(key) - def __contains__(self, key): + def __contains__(self, key: str) -> bool: """Return True if *key* is a section in the underlying ConfigParser.""" return key in self.config - def sections(self) -> typing.List[str]: + def sections(self) -> list[str]: """Returns list of all sections""" 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 section-scoped view sharing this instance's parsed ConfigParser. + + The returned view's read methods (get, has_option, getint, …) operate on + the same data as the parent. Write operations (add_option, update_option, + save_configuration) should still be called on the root object. + """ if not self.config.has_section(section): return fallback - return BlocksScreenConfig(self.configfile, section) + view = BlocksScreenConfig(self.configfile, section) + view.config = self.config # share — reads see the actual parsed data + view.raw_config = self.raw_config # share — _find_section_* helpers work + view.file_lock = self.file_lock # share — thread safety consistent + return view - def get_options(self) -> list: + def get_options(self) -> list[str]: """Get section options""" return self.config.options(self.section) @@ -120,7 +125,7 @@ def has_section(self, section: str) -> bool: Returns: bool: true if section exists, false otherwise """ - return bool(self.config.has_section(section)) + return self.config.has_section(section) def has_option(self, option: str) -> bool: """Check if section has a option @@ -131,58 +136,71 @@ def has_option(self, option: str) -> bool: Returns: bool: true if section exists, false otherwise """ - return bool(self.config.has_option(self.section, option)) + 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.Union[Sentinel, str]: - """Get option value + default: Any = Sentinel.MISSING, + ) -> 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. + parser (type, optional): int, float, or str. Defaults to str. + Do **not** pass ``bool`` — use :meth:`getboolean` instead. + ``get(parser=bool)`` is intercepted and delegated to + ``getboolean`` to avoid the ``bool("false") == True`` pitfall. + default: Returned as-is when the option is absent. + Defaults to Sentinel.MISSING (raises if option not found). Returns: - typing.Union[Sentinel, str]: Requested option. Defaults to the specified default value + Any: Parsed option value, or *default* if the option is absent. """ - return parser( - self.config.get(section=self.section, option=option, fallback=default) - ) + if parser is bool: + 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 getint( self, option: str, - default: typing.Union[Sentinel, int] = Sentinel.MISSING, - ) -> typing.Union[Sentinel, int]: - """Get option value + default: Sentinel | int = Sentinel.MISSING, + ) -> int: + """Get option value as int. Args: option (str): option name - default (typing.Union[Sentinel, int], optional): Default value for specified option. Defaults to Sentinel.MISSING. + default (int, optional): returned as-is when absent. Raises if omitted. Returns: - typing.Union[Sentinel, int]: Requested option. + int: parsed value, or *default* if absent. """ + 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 + default: Sentinel | float = Sentinel.MISSING, + ) -> float: + """Get option value as float. Args: option (str): option name - default (typing.Union[Sentinel, float], optional): Default value for specified option. Defaults to Sentinel.MISSING. + default (float, optional): returned as-is when absent. Raises if omitted. Returns: - typing.Union[Sentinel, float]: _description_ + float: parsed value, or *default* if absent. """ + 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 ) @@ -190,17 +208,19 @@ def getfloat( def getboolean( self, option: str, - default: typing.Union[Sentinel, bool] = Sentinel.MISSING, - ) -> typing.Union[Sentinel, bool]: - """Get option value + default: Sentinel | bool = Sentinel.MISSING, + ) -> bool: + """Get option value as bool. Args: option (str): option name - default (typing.Union[Sentinel, bool], optional): Default value for specified option. Defaults to Sentinel.MISSING. + default (bool, optional): returned as-is when absent. Raises if omitted. Returns: - typing.Union[Sentinel, bool]: _description_ + bool: parsed value, or *default* if absent. """ + 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 ) @@ -208,40 +228,22 @@ def getboolean( def _find_section_index(self, section: str) -> int: """Return the index of the ``[section]`` header 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: + def _find_section_limits(self, section: str) -> tuple[int, int]: """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: + return (section_start, section_end + section_start) + except (configparser.Error, ValueError) as e: raise configparser.Error( f'Error while finding section "{section}" limits on local tracking: {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``.""" - 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}' - ) - def add_section(self, section: str) -> None: """Add a section to configuration file @@ -258,58 +260,49 @@ def add_section(self, section: str) -> None: 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] != "": + if self.raw_config and self.raw_config[-1].strip() != "": self.raw_config.append("") + self.raw_config.extend([sec_string, ""]) self.config.add_section(section) self.update_pending = True except configparser.DuplicateSectionError as e: - logger.error(f'Section "{section}" already exists. {e}') + logger.error('Section "%s" already exists. %s', section, e) except configparser.Error as e: - logger.error(f'Unable to add "{section}" section to configuration: {e}') + logger.error('Unable to add "%s" section to configuration: %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 + """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. + value (str | None, optional): value for the option. ``None`` writes + a value-less option (``allow_no_value=True``). Defaults to None. """ 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("") + _, section_end = self._find_section_limits(section) + raw_line = f"{option}: {value}" if value is not None else f"{option}:" + self.raw_config.insert(section_end, raw_line) self.config.set(section, option, value) self.update_pending = True except configparser.DuplicateOptionError as e: - logger.error(f"Option {option} already present on {section}: {e}") + logger.error("Option %s already present on %s: %s", option, section, e) except configparser.Error as e: logger.error( - f'Unable to add "{option}" option to section "{section}": {e} ' + 'Unable to add "%s" option 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.""" try: @@ -327,7 +320,11 @@ def update_option( self.update_pending = True except Exception 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, + exc_info=True, ) def _find_option_line_index(self, section: str, option: str) -> int: @@ -340,103 +337,113 @@ def _find_option_line_index(self, section: str, option: str) -> int: raise configparser.Error(f'Option "{option}" not found in section "{section}"') def save_configuration(self) -> None: - """Save teh configuration to file""" + """Save the configuration to file. + + ``update_pending`` is only cleared on a successful write so that a + caller can detect a failed save and retry. + """ try: - if not self.update_pending: - return with self.file_lock: + if not self.update_pending: + return 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() + self.update_pending = False except Exception 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""" + def load_config(self) -> None: + """Load configuration file. + + Updates ``raw_config`` in-place so that existing section-view objects + (which share the same list reference) remain valid after a reload. + """ try: + new_raw = self._parse_file() # can raise without corrupting state + self.config.clear() self.raw_config.clear() - self.config.clear() # Reset configparser - self.raw_config, self.raw_dict_config = self._parse_file() + self.raw_config.extend(new_raw) if self.raw_config: self.config.read_file(self.raw_config) except Exception as e: - raise configparser.Error(f"Error loading configuration file: {e}") + raise configparser.Error(f"Error loading configuration file: {e}") from 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. + def _parse_file(self) -> list[str]: + """Read and normalise the config file into a raw line list. - Strips comments, normalises ``=`` to ``:`` separators, deduplicates - sections/options, and ensures the buffer ends with an empty line. + Strips comments, normalises only the **first** ``=``/``:`` separator + to ``: `` (preserving values that contain colons, e.g. URLs or hex + colours), deduplicates sections/options, and ensures the buffer ends + with an empty line. Returns: - A tuple of (raw_lines, dict_representation). + Normalised list of config lines. """ - buffer = [] - dict_buff: typing.Dict = {} - curr_sec: typing.Union[Sentinel, str] = Sentinel.MISSING + buffer: list[str] = [] + seen: dict[str, set[str]] = {} # section → set of seen option names + curr_sec: str | None = None 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: + with self.file_lock: + text = self.configfile.read_text(encoding="utf-8") + for raw in text.splitlines(): + line = raw.strip() + if not line or line.startswith(("#", ";")): continue - line.expandtabs(indentation_size) - re_match = re.search(r"\s*#\s*(.*)(\s*)", line) - if re_match: - line = line[: re_match.start()] + # Strip inline comments (delimiter must be preceded by a word char, + # so values like `color: #ff0000` are not affected). + m = _RE_INLINE_COMMENT.search(line) + if m: + line = line[: m.start()].rstrip() 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] != "": + + # --- Section header --- + # Checked before separator normalisation so that a colon inside + # a section name (e.g. `[a:b]`) is never mangled. + m_sec = _RE_SECTION.match(line) + if m_sec: + sec_name = m_sec.group(1) + if sec_name in seen: + continue # duplicate section — skip + if buffer: + buffer.append("") # blank separator before each new section + seen[sec_name] = set() + curr_sec = sec_name + buffer.append(line) + continue # header fully handled — do not try to parse as option + + # --- Option line --- + # Normalise only the *first* separator so that values containing + # additional ':' or '=' characters (URLs, tokens, regex …) are + # preserved verbatim. + line = _RE_SEP_NORMALIZE.sub(": ", line, count=1) + m_opt = _RE_OPTION.match(line) + if m_opt and curr_sec is not None: + opt_name = m_opt.group(1) + if opt_name in seen[curr_sec]: + continue # duplicate option — skip + seen[curr_sec].add(opt_name) + buffer.append(line) + # Lines matching neither pattern are silently dropped + # (they cannot be round-tripped through configparser anyway). + + if not buffer or buffer[-1] != "": buffer.append("") - return buffer, dict_buff + return buffer except Exception as e: raise configparser.Error( f"Unexpected error while parsing configuration file: {e}" - ) - finally: - self.file_lock.release() + ) from e 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") + wanted_target = DEFAULT_CONFIGFILE_PATH / "BlocksScreen.cfg" + fallback = FALLBACK_CONFIGFILE_PATH / "BlocksScreen.cfg" configfile = ( wanted_target if check_file_on_path(DEFAULT_CONFIGFILE_PATH, "BlocksScreen.cfg") 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..408c5139 --- /dev/null +++ b/tests/util/test_configfile_unit.py @@ -0,0 +1,664 @@ +"""Unit tests for BlocksScreen/configfile.py + +Covers: parsing, read methods, get_section views, add/update/save, get_configparser factory, +comment handling, and thread-safety (RLock reentrancy). +""" + +from __future__ import annotations + +import configparser +import logging +import pathlib +import sys +import textwrap +import threading + +import pytest + +# configfile.py lives in BlocksScreen/ and imports helper_methods from the same dir; +# both resolve when BlocksScreen/ is on sys.path. +_BLOCKSSCREEN = pathlib.Path(__file__).parent.parent.parent / "BlocksScreen" +if str(_BLOCKSSCREEN) not in sys.path: + sys.path.insert(0, str(_BLOCKSSCREEN)) + +# tests/network/conftest.py installs a mock at sys.modules["configfile"] for network +# tests. Popping it here is safe — network modules already bound the mock's symbols +# at their own import time and won't re-import. +sys.modules.pop("configfile", None) + +import configfile as cfmod # noqa: E402 +from configfile import BlocksScreenConfig, ConfigError, Sentinel, get_configparser # noqa: E402 + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +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 + + +# --------------------------------------------------------------------------- +# Sentinel +# --------------------------------------------------------------------------- + + +@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) + + +# --------------------------------------------------------------------------- +# Parsing (_parse_file via load_config) +# --------------------------------------------------------------------------- + + +@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_uses_first_occurrence(self, tmp_path): + cfg = _make_cfg(tmp_path, """ + [server] + host: first + [server] + host: second + """) + assert cfg.get("host") == "first" + + def test_duplicate_option_uses_first_occurrence(self, tmp_path): + cfg = _make_cfg(tmp_path, """ + [server] + host: first + host: second + """) + assert cfg.get("host") == "first" + + 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" + + +# --------------------------------------------------------------------------- +# Read methods (get, getint, getfloat, getboolean, sections, get_options) +# --------------------------------------------------------------------------- + + +@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): + """When the option is absent the default is returned without passing through parser. + Previously parser(default) was called — int('N/A') would raise ValueError.""" + 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") + + +# --------------------------------------------------------------------------- +# get_section / __getitem__ / __contains__ +# --------------------------------------------------------------------------- + + +@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 + + # Reload with new content — view must reflect the updated list + 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) + + +# --------------------------------------------------------------------------- +# add_section +# --------------------------------------------------------------------------- + + +@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 + + +# --------------------------------------------------------------------------- +# add_option +# --------------------------------------------------------------------------- + + +@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") + + +# --------------------------------------------------------------------------- +# update_option +# --------------------------------------------------------------------------- + + +@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 calls add_section/add_option while holding the RLock — must not deadlock.""" + 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" + + +# --------------------------------------------------------------------------- +# save_configuration +# --------------------------------------------------------------------------- + + +@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 + + +# --------------------------------------------------------------------------- +# get_configparser factory +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestGetConfigparser: + 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_CONFIGFILE_PATH", tmp_path / "nonexistent") + monkeypatch.setattr(cfmod, "FALLBACK_CONFIGFILE_PATH", tmp_path) + + 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_CONFIGFILE_PATH", tmp_path / "nonexistent") + monkeypatch.setattr(cfmod, "FALLBACK_CONFIGFILE_PATH", tmp_path) + + with pytest.raises(ConfigError, match=r"\[server\]"): + get_configparser() + + +# --------------------------------------------------------------------------- +# Thread safety +# --------------------------------------------------------------------------- + + +@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") + + +# --------------------------------------------------------------------------- +# Bug fixes (regression tests) +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestEdgeCases: + def test_parse_empty_file_does_not_crash(self, tmp_path): + """Empty (or 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() # must not raise + assert cfg.raw_config == [""] + + 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") From 2ea2cfbdb5d50b1600baf89635e4d28ae424477e Mon Sep 17 00:00:00 2001 From: Guilherme Costa Date: Fri, 10 Apr 2026 17:22:36 +0100 Subject: [PATCH 2/5] refactor: enforce singleton and normalise config separators to ":" --- BlocksScreen/configfile.py | 53 +++++--- tests/util/test_configfile_unit.py | 192 +++++++++++++++++++++-------- 2 files changed, 179 insertions(+), 66 deletions(-) diff --git a/BlocksScreen/configfile.py b/BlocksScreen/configfile.py index 904519bb..b23ecb26 100644 --- a/BlocksScreen/configfile.py +++ b/BlocksScreen/configfile.py @@ -33,15 +33,18 @@ import threading from typing import Any -from helper_methods import check_file_on_path - logger = logging.getLogger(__name__) 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: BlocksScreenConfig | None = None +_singleton_lock = threading.Lock() _RE_SECTION = re.compile(r"^\s*\[([^]]+)\]") -_RE_OPTION = re.compile(r"^(\w+)[=:]") +_RE_OPTION = re.compile(r"^(\w+):") _RE_INLINE_COMMENT = re.compile(r"(?<=\w)\s+[#;]") _RE_SEP_NORMALIZE = re.compile(r"\s*[:=]\s*") @@ -77,6 +80,7 @@ def __init__(self, configfile: str | pathlib.Path, section: str) -> None: allow_no_value=True, comment_prefixes=("#", ";"), inline_comment_prefixes=None, # handled manually in _parse_file + delimiters=(":",), ) self.update_pending: bool = False self.raw_config: list[str] = [] @@ -330,7 +334,7 @@ def update_option( def _find_option_line_index(self, section: str, option: str) -> int: """Find the index of an option line within a specific section.""" start, end = self._find_section_limits(section) - opt_regex = re.compile(rf"^\s*{re.escape(option)}\s*[:=]") + opt_regex = re.compile(rf"^\s*{re.escape(option)}\s*:") for i in range(start + 1, end): if opt_regex.match(self.raw_config[i]): return i @@ -441,17 +445,30 @@ def _parse_file(self) -> list[str]: def get_configparser() -> BlocksScreenConfig: - """Loads configuration from file and returns that configuration""" - wanted_target = DEFAULT_CONFIGFILE_PATH / "BlocksScreen.cfg" - fallback = FALLBACK_CONFIGFILE_PATH / "BlocksScreen.cfg" - configfile = ( - wanted_target - if check_file_on_path(DEFAULT_CONFIGFILE_PATH, "BlocksScreen.cfg") - else fallback - ) - 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 + """Return the singleton :class:`BlocksScreenConfig`, creating it on first call. + + Subsequent calls return the same instance so that only one + :class:`configparser.ConfigParser` is ever created for the application. + Thread-safe via double-checked locking. + """ + global _singleton + if _singleton is not None: + return _singleton + with _singleton_lock: + if _singleton is not None: + return _singleton + 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") + _singleton = config_object + return _singleton + + +def reset_configparser() -> None: + """Reset the singleton instance — for use in tests only.""" + global _singleton + with _singleton_lock: + _singleton = None diff --git a/tests/util/test_configfile_unit.py b/tests/util/test_configfile_unit.py index 408c5139..b21f10f4 100644 --- a/tests/util/test_configfile_unit.py +++ b/tests/util/test_configfile_unit.py @@ -27,7 +27,13 @@ sys.modules.pop("configfile", None) import configfile as cfmod # noqa: E402 -from configfile import BlocksScreenConfig, ConfigError, Sentinel, get_configparser # noqa: E402 +from configfile import ( + BlocksScreenConfig, + ConfigError, + Sentinel, + get_configparser, + reset_configparser, +) # noqa: E402 # --------------------------------------------------------------------------- @@ -35,7 +41,9 @@ # --------------------------------------------------------------------------- -def _make_cfg(tmp_path: pathlib.Path, content: str, section: str = "server") -> BlocksScreenConfig: +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") @@ -67,92 +75,125 @@ def test_missing_is_an_enum_member(self): @pytest.mark.unit class TestParsing: def test_basic_section_and_option(self, tmp_path): - cfg = _make_cfg(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, """ + 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, """ + 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, """ + 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, """ + 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, """ + 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, """ + cfg = _make_cfg( + tmp_path, + """ [server] color: #ff0000 - """) + """, + ) assert cfg.get("color") == "#ff0000" def test_duplicate_section_uses_first_occurrence(self, tmp_path): - cfg = _make_cfg(tmp_path, """ + cfg = _make_cfg( + tmp_path, + """ [server] host: first [server] host: second - """) + """, + ) assert cfg.get("host") == "first" def test_duplicate_option_uses_first_occurrence(self, tmp_path): - cfg = _make_cfg(tmp_path, """ + cfg = _make_cfg( + tmp_path, + """ [server] host: first host: second - """) + """, + ) assert cfg.get("host") == "first" def test_multiple_sections_all_loaded(self, tmp_path): - cfg = _make_cfg(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, """ + cfg = _make_cfg( + tmp_path, + """ [server] host: localhost - """) + """, + ) assert cfg.raw_config[-1] == "" def test_load_config_resets_previous_state(self, tmp_path): @@ -168,26 +209,36 @@ def test_load_config_resets_previous_state(self, tmp_path): 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, """ + 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, """ + 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, """ + cfg = _make_cfg( + tmp_path, + """ [fan:heater_fan] pin: PA8 - """, section="fan:heater_fan") + """, + section="fan:heater_fan", + ) assert cfg.has_section("fan:heater_fan") assert cfg.get("pin") == "PA8" @@ -201,13 +252,16 @@ def test_section_name_with_colon_preserved(self, tmp_path): class TestReadMethods: @pytest.fixture def cfg(self, tmp_path): - return _make_cfg(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 @@ -232,19 +286,25 @@ def test_get_with_int_parser(self, cfg): 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, """ + 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, """ + cfg = _make_cfg( + tmp_path, + """ [server] host: localhost flag: true - """) + """, + ) assert cfg.get("flag", parser=bool) is True def test_getint(self, cfg): @@ -304,12 +364,15 @@ def test_getboolean_raises_when_absent_no_default(self, cfg): class TestGetSection: @pytest.fixture def cfg(self, tmp_path): - return _make_cfg(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") @@ -349,7 +412,9 @@ def test_contains_returns_false_for_missing_section(self, 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_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") @@ -357,7 +422,9 @@ def test_view_raw_config_stays_valid_after_reload(self, tmp_path): assert view.raw_config is cfg.raw_config # same list object # Reload with new content — view must reflect the updated list - cfg_file.write_text("[server]\nhost: second\n[display]\nwidth: 1024\n", encoding="utf-8") + 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) @@ -372,10 +439,13 @@ def test_view_raw_config_stays_valid_after_reload(self, tmp_path): class TestAddSection: @pytest.fixture def cfg(self, tmp_path): - return _make_cfg(tmp_path, """ + return _make_cfg( + tmp_path, + """ [server] host: localhost - """) + """, + ) def test_section_becomes_visible_after_add(self, cfg): cfg.add_section("display") @@ -411,10 +481,13 @@ def test_duplicate_section_does_not_raise(self, cfg): class TestAddOption: @pytest.fixture def cfg(self, tmp_path): - return _make_cfg(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") @@ -458,11 +531,14 @@ def test_none_value_option_is_readable(self, cfg): class TestUpdateOption: @pytest.fixture def cfg(self, tmp_path): - return _make_cfg(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") @@ -552,6 +628,13 @@ def test_pending_preserved_when_write_fails(self, tmp_path): @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") @@ -562,7 +645,9 @@ def test_loads_from_fallback_path(self, tmp_path, monkeypatch): cfg = get_configparser() assert cfg.has_section("server") - def test_raises_config_error_when_server_section_missing(self, tmp_path, monkeypatch): + 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") @@ -572,6 +657,15 @@ def test_raises_config_error_when_server_section_missing(self, tmp_path, monkeyp 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_CONFIGFILE_PATH", tmp_path / "nonexistent") + monkeypatch.setattr(cfmod, "FALLBACK_CONFIGFILE_PATH", tmp_path) + + assert get_configparser() is get_configparser() + # --------------------------------------------------------------------------- # Thread safety @@ -594,7 +688,9 @@ def updater(value: str) -> None: except Exception as exc: errors.append(exc) - threads = [threading.Thread(target=updater, args=(f"host{i}",)) for i in range(20)] + threads = [ + threading.Thread(target=updater, args=(f"host{i}",)) for i in range(20) + ] for t in threads: t.start() for t in threads: From e142f8a20a9f373ca24d3b3a0f1565d67f83519f Mon Sep 17 00:00:00 2001 From: Guilherme Costa Date: Fri, 10 Apr 2026 17:31:43 +0100 Subject: [PATCH 3/5] fix failing test --- tests/util/test_configfile_unit.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/tests/util/test_configfile_unit.py b/tests/util/test_configfile_unit.py index b21f10f4..44498712 100644 --- a/tests/util/test_configfile_unit.py +++ b/tests/util/test_configfile_unit.py @@ -639,8 +639,8 @@ 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_CONFIGFILE_PATH", tmp_path / "nonexistent") - monkeypatch.setattr(cfmod, "FALLBACK_CONFIGFILE_PATH", tmp_path) + 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") @@ -651,8 +651,8 @@ def test_raises_config_error_when_server_section_missing( cfg_file = tmp_path / "BlocksScreen.cfg" cfg_file.write_text("[display]\nwidth: 800\n", encoding="utf-8") - monkeypatch.setattr(cfmod, "DEFAULT_CONFIGFILE_PATH", tmp_path / "nonexistent") - monkeypatch.setattr(cfmod, "FALLBACK_CONFIGFILE_PATH", tmp_path) + 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() @@ -661,8 +661,8 @@ 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_CONFIGFILE_PATH", tmp_path / "nonexistent") - monkeypatch.setattr(cfmod, "FALLBACK_CONFIGFILE_PATH", tmp_path) + monkeypatch.setattr(cfmod, "_DEFAULT_CONFIG", tmp_path / "nonexistent" / "BlocksScreen.cfg") + monkeypatch.setattr(cfmod, "_FALLBACK_CONFIG", cfg_file) assert get_configparser() is get_configparser() From 2b7a8392874730a23f24107b968814986751e158 Mon Sep 17 00:00:00 2001 From: Guilherme Costa Date: Wed, 19 Aug 2026 14:53:55 +0100 Subject: [PATCH 4/5] fix(tests): stop widgets test from corrupting shared configfile module --- tests/widgets/test_slider_page_request_unit.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/tests/widgets/test_slider_page_request_unit.py b/tests/widgets/test_slider_page_request_unit.py index d1eefaf1..72a07891 100644 --- a/tests/widgets/test_slider_page_request_unit.py +++ b/tests/widgets/test_slider_page_request_unit.py @@ -22,8 +22,12 @@ # 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. +# Always replace (not setdefault): tests/util/test_configfile_unit.py may +# already have imported the *real* configfile module by this point in +# collection, and mutating its attributes in place (instead of swapping in +# a fresh stub) would corrupt BlocksScreenConfig for the rest of the session. 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 From 5a7beb4b0d9827a301e3f3d84f38e59ea46a516c Mon Sep 17 00:00:00 2001 From: Guilherme Costa Date: Wed, 23 Sep 2026 11:44:11 +0100 Subject: [PATCH 5/5] fix(configfile): klipper-style parsing, atomic save, round-trip safe values --- BlocksScreen/configfile.py | 446 +++++++----------- tests/util/test_configfile_unit.py | 219 ++++++--- .../widgets/test_slider_page_request_unit.py | 8 +- 3 files changed, 310 insertions(+), 363 deletions(-) diff --git a/BlocksScreen/configfile.py b/BlocksScreen/configfile.py index 0db1c295..bfb63720 100644 --- a/BlocksScreen/configfile.py +++ b/BlocksScreen/configfile.py @@ -23,13 +23,20 @@ # # 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 functools import logging +import os import pathlib import re +import stat +import tempfile import threading from typing import Any @@ -40,106 +47,109 @@ _DEFAULT_CONFIG = DEFAULT_CONFIGFILE_PATH / "BlocksScreen.cfg" _FALLBACK_CONFIG = FALLBACK_CONFIGFILE_PATH / "BlocksScreen.cfg" -_singleton: BlocksScreenConfig | None = None _singleton_lock = threading.Lock() -_RE_SECTION = re.compile(r"^\s*\[([^]]+)\]") -_RE_OPTION = re.compile(r"^(\w+):") -_RE_INLINE_COMMENT = re.compile(r"(?<=\w)\s+[#;]") -_RE_SEP_NORMALIZE = re.compile(r"\s*[:=]\s*") +_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() 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. - """ + +class BlocksScreenConfig: + """Thread-safe ConfigParser wrapper that mirrors the file in ``raw_config``.""" def __init__(self, configfile: str | pathlib.Path, section: str) -> None: - """Initialise with the path to the config file and the default section name.""" + """Bind to *configfile*, reading options from *section*.""" self.configfile = pathlib.Path(configfile) self.section = section self.config = configparser.ConfigParser( allow_no_value=True, + interpolation=None, # as Klipper/Moonraker: '%' in values is literal comment_prefixes=("#", ";"), - inline_comment_prefixes=None, # handled manually in _parse_file + 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 the lock + # RLock: update_option calls add_section/add_option while holding it self.file_lock = threading.RLock() def __getitem__(self, key: str) -> BlocksScreenConfig | None: - """Return a :class:`BlocksScreenConfig` for *key* section (same as ``get_section``).""" + """Return the view for section *key*, or None.""" return self.get_section(key) def __contains__(self, key: str) -> bool: - """Return True if *key* is a section in the underlying ConfigParser.""" + """Return True if *key* is a section.""" return key in self.config def sections(self) -> list[str]: - """Returns list of all sections""" + """Return all section names.""" return self.config.sections() def get_section( self, section: str, fallback: BlocksScreenConfig | None = None ) -> BlocksScreenConfig | None: - """Return a section-scoped view sharing this instance's parsed ConfigParser. - - The returned view's read methods (get, has_option, getint, …) operate on - the same data as the parent. Write operations (add_option, update_option, - save_configuration) should still be called on the root object. - """ + """Return a view of *section* sharing this parser; call writes on the root.""" if not self.config.has_section(section): return fallback - view = BlocksScreenConfig(self.configfile, section) - view.config = self.config # share — reads see the actual parsed data - view.raw_config = self.raw_config # share — _find_section_* helpers work - view.file_lock = self.file_lock # share — thread safety consistent + view = copy.copy(self) + view.section = section return view def get_options(self) -> list[str]: - """Get section options""" + """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 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 True if *option* exists in this section.""" return self.config.has_option(self.section, option) def get( @@ -148,21 +158,9 @@ def get( parser: type = str, default: Any = Sentinel.MISSING, ) -> Any: - """Get option value. - - Args: - option (str): option name - parser (type, optional): int, float, or str. Defaults to str. - Do **not** pass ``bool`` — use :meth:`getboolean` instead. - ``get(parser=bool)`` is intercepted and delegated to - ``getboolean`` to avoid the ``bool("false") == True`` pitfall. - default: Returned as-is when the option is absent. - Defaults to Sentinel.MISSING (raises if option not found). - - Returns: - Any: Parsed option value, or *default* if the option is absent. - """ + """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)) @@ -178,50 +176,20 @@ def getlists( sep: str | tuple[str, ...] = ",", parser: type = str, ) -> list[Any]: - """Get option value parsed as a list - - Args: - option (str): option name - default (Sentinel | list, 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. - """ + """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 None: - if default is not Sentinel.MISSING: - return default - return [] 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: Sentinel | int = Sentinel.MISSING, ) -> int: - """Get option value as int. - - Args: - option (str): option name - default (int, optional): returned as-is when absent. Raises if omitted. - - Returns: - int: parsed value, or *default* if absent. - """ + """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) @@ -231,15 +199,7 @@ def getfloat( option: str, default: Sentinel | float = Sentinel.MISSING, ) -> float: - """Get option value as float. - - Args: - option (str): option name - default (float, optional): returned as-is when absent. Raises if omitted. - - Returns: - float: parsed value, or *default* if absent. - """ + """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( @@ -251,15 +211,7 @@ def getboolean( option: str, default: Sentinel | bool = Sentinel.MISSING, ) -> bool: - """Get option value as bool. - - Args: - option (str): option name - default (bool, optional): returned as-is when absent. Raises if omitted. - - Returns: - bool: parsed value, or *default* if absent. - """ + """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( @@ -267,49 +219,31 @@ def getboolean( ) 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(f"[{section}]") except ValueError as e: - raise configparser.Error(f'Section "{section}" does not exist: {e}') + raise configparser.Error(f'Section "{section}" does not exist') from e def _find_section_limits(self, section: str) -> tuple[int, int]: - """Return ``(start_index, end_index)`` of *section* in ``raw_config``.""" + """Return ``(header, closing_blank)`` indexes 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, section_end + section_start) + start = self._find_section_index(section) + return start, self.raw_config.index("", start) except (configparser.Error, ValueError) as e: - raise configparser.Error( - f'Error while finding section "{section}" limits on local tracking: {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' - ) - if self.raw_config and self.raw_config[-1].strip() != "": - self.raw_config.append("") - self.raw_config.extend([sec_string, ""]) 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('Section "%s" already exists. %s', section, e) - except configparser.Error as e: - logger.error('Unable to add "%s" section to configuration: %s', section, e) + except (configparser.Error, ValueError) as e: + logger.error('Unable to add section "%s": %s', section, e) def add_option( self, @@ -317,26 +251,19 @@ def add_option( option: str, value: str | None = None, ) -> None: - """Add option with a value to a section. - - Args: - section (str): section name - option (str): option name - value (str | None, optional): value for the option. ``None`` writes - a value-less option (``allow_no_value=True``). Defaults to None. - """ + """Add *option* to *section*; None writes a value-less ``option:`` line.""" try: with self.file_lock: - _, section_end = self._find_section_limits(section) - raw_line = f"{option}: {value}" if value is not None else f"{option}:" - self.raw_config.insert(section_end, raw_line) + 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("Option %s already present on %s: %s", option, section, e) - except configparser.Error as e: + except (configparser.Error, ValueError) as e: logger.error( - 'Unable to add "%s" option to section "%s": %s', option, section, e + 'Unable to add option "%s" to section "%s": %s', option, section, e ) def update_option( @@ -345,51 +272,43 @@ def update_option( option: str, 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( - 'Unable to update option "%s" in section "%s": %s', - option, - section, - e, - exc_info=True, + '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 the configuration to file. - - ``update_pending`` is only cleared on a successful write so that a - caller can detect a failed save and retry. - """ + """Write ``raw_config`` atomically; ``update_pending`` stays set on failure.""" try: with self.file_lock: if not self.update_pending: return - self.configfile.write_text("\n".join(self.raw_config), encoding="utf-8") + _atomic_write(self.configfile, "\n".join(self.raw_config)) self.update_pending = False - except Exception as e: + except OSError as e: logger.error( "Unable to save configuration to %s: %s", self.configfile, @@ -398,114 +317,83 @@ def save_configuration(self) -> None: ) def load_config(self) -> None: - """Load configuration file. - - Updates ``raw_config`` in-place so that existing section-view objects - (which share the same list reference) remain valid after a reload. - """ + """(Re)load the file, updating ``raw_config`` in place so views stay valid.""" try: - new_raw = self._parse_file() # can raise without corrupting state + new_raw = self._parse_file() self.config.clear() - self.raw_config.clear() - self.raw_config.extend(new_raw) - if self.raw_config: - self.config.read_file(self.raw_config) - except Exception as e: + 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 and normalise the config file into a raw line list. - - Strips comments, normalises only the **first** ``=``/``:`` separator - to ``: `` (preserving values that contain colons, e.g. URLs or hex - colours), deduplicates sections/options, and ensures the buffer ends - with an empty line. - - Returns: - Normalised list of config lines. - """ - buffer: list[str] = [] - seen: dict[str, set[str]] = {} # section → set of seen option names - curr_sec: str | None = None - try: - with self.file_lock: - text = self.configfile.read_text(encoding="utf-8") - for raw in text.splitlines(): - line = raw.strip() - if not line or line.startswith(("#", ";")): - continue - # Strip inline comments (delimiter must be preceded by a word char, - # so values like `color: #ff0000` are not affected). - m = _RE_INLINE_COMMENT.search(line) - if m: - line = line[: m.start()].rstrip() - if not line: - continue - - # --- Section header --- - # Checked before separator normalisation so that a colon inside - # a section name (e.g. `[a:b]`) is never mangled. - m_sec = _RE_SECTION.match(line) - if m_sec: - sec_name = m_sec.group(1) - if sec_name in seen: - continue # duplicate section — skip - if buffer: - buffer.append("") # blank separator before each new section - seen[sec_name] = set() - curr_sec = sec_name - buffer.append(line) - continue # header fully handled — do not try to parse as option - - # --- Option line --- - # Normalise only the *first* separator so that values containing - # additional ':' or '=' characters (URLs, tokens, regex …) are - # preserved verbatim. - line = _RE_SEP_NORMALIZE.sub(": ", line, count=1) - m_opt = _RE_OPTION.match(line) - if m_opt and curr_sec is not None: - opt_name = m_opt.group(1) - if opt_name in seen[curr_sec]: - continue # duplicate option — skip - seen[curr_sec].add(opt_name) - buffer.append(line) - # Lines matching neither pattern are silently dropped - # (they cannot be round-tripped through configparser anyway). - - if not buffer or buffer[-1] != "": - buffer.append("") - return buffer - except Exception as e: - raise configparser.Error( - f"Unexpected error while parsing configuration file: {e}" - ) from e + """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 :class:`BlocksScreenConfig`, creating it on first call. - - Subsequent calls return the same instance so that only one - :class:`configparser.ConfigParser` is ever created for the application. - Thread-safe via double-checked locking. - """ - global _singleton - if _singleton is not None: - return _singleton + """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: - if _singleton is not None: - return _singleton - 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") - _singleton = config_object - return _singleton + return _load_singleton() def reset_configparser() -> None: - """Reset the singleton instance — for use in tests only.""" - global _singleton + """Drop the singleton; tests only.""" with _singleton_lock: - _singleton = None + _load_singleton.cache_clear() diff --git a/tests/util/test_configfile_unit.py b/tests/util/test_configfile_unit.py index 44498712..eebf67f2 100644 --- a/tests/util/test_configfile_unit.py +++ b/tests/util/test_configfile_unit.py @@ -1,8 +1,4 @@ -"""Unit tests for BlocksScreen/configfile.py - -Covers: parsing, read methods, get_section views, add/update/save, get_configparser factory, -comment handling, and thread-safety (RLock reentrancy). -""" +"""Unit tests for BlocksScreen/configfile.py.""" from __future__ import annotations @@ -15,15 +11,11 @@ import pytest -# configfile.py lives in BlocksScreen/ and imports helper_methods from the same dir; -# both resolve when BlocksScreen/ is on sys.path. _BLOCKSSCREEN = pathlib.Path(__file__).parent.parent.parent / "BlocksScreen" if str(_BLOCKSSCREEN) not in sys.path: sys.path.insert(0, str(_BLOCKSSCREEN)) -# tests/network/conftest.py installs a mock at sys.modules["configfile"] for network -# tests. Popping it here is safe — network modules already bound the mock's symbols -# at their own import time and won't re-import. +# drop tests/network's configfile mock; network modules already bound its symbols sys.modules.pop("configfile", None) import configfile as cfmod # noqa: E402 @@ -36,11 +28,6 @@ ) # noqa: E402 -# --------------------------------------------------------------------------- -# Helpers -# --------------------------------------------------------------------------- - - def _make_cfg( tmp_path: pathlib.Path, content: str, section: str = "server" ) -> BlocksScreenConfig: @@ -52,11 +39,6 @@ def _make_cfg( return cfg -# --------------------------------------------------------------------------- -# Sentinel -# --------------------------------------------------------------------------- - - @pytest.mark.unit class TestSentinel: def test_missing_value_is_an_instance_not_the_class(self): @@ -67,11 +49,6 @@ def test_missing_is_an_enum_member(self): assert isinstance(Sentinel.MISSING, Sentinel) -# --------------------------------------------------------------------------- -# Parsing (_parse_file via load_config) -# --------------------------------------------------------------------------- - - @pytest.mark.unit class TestParsing: def test_basic_section_and_option(self, tmp_path): @@ -150,19 +127,25 @@ def test_hash_in_value_without_leading_space_preserved(self, tmp_path): ) assert cfg.get("color") == "#ff0000" - def test_duplicate_section_uses_first_occurrence(self, tmp_path): + 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") == "first" + 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_uses_first_occurrence(self, tmp_path): + def test_duplicate_option_last_wins(self, tmp_path): cfg = _make_cfg( tmp_path, """ @@ -171,7 +154,39 @@ def test_duplicate_option_uses_first_occurrence(self, tmp_path): host: second """, ) - assert cfg.get("host") == "first" + 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( @@ -243,11 +258,6 @@ def test_section_name_with_colon_preserved(self, tmp_path): assert cfg.get("pin") == "PA8" -# --------------------------------------------------------------------------- -# Read methods (get, getint, getfloat, getboolean, sections, get_options) -# --------------------------------------------------------------------------- - - @pytest.mark.unit class TestReadMethods: @pytest.fixture @@ -334,8 +344,7 @@ def test_get_options_returns_option_names(self, cfg): assert "port" in opts def test_get_default_returned_as_is_not_parsed(self, cfg): - """When the option is absent the default is returned without passing through parser. - Previously parser(default) was called — int('N/A') would raise ValueError.""" + """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" @@ -355,11 +364,6 @@ def test_getboolean_raises_when_absent_no_default(self, cfg): cfg.getboolean("missing") -# --------------------------------------------------------------------------- -# get_section / __getitem__ / __contains__ -# --------------------------------------------------------------------------- - - @pytest.mark.unit class TestGetSection: @pytest.fixture @@ -421,7 +425,6 @@ def test_view_raw_config_stays_valid_after_reload(self, tmp_path): assert view is not None assert view.raw_config is cfg.raw_config # same list object - # Reload with new content — view must reflect the updated list cfg_file.write_text( "[server]\nhost: second\n[display]\nwidth: 1024\n", encoding="utf-8" ) @@ -430,11 +433,6 @@ def test_view_raw_config_stays_valid_after_reload(self, tmp_path): assert any("1024" in line for line in view.raw_config) -# --------------------------------------------------------------------------- -# add_section -# --------------------------------------------------------------------------- - - @pytest.mark.unit class TestAddSection: @pytest.fixture @@ -472,11 +470,6 @@ def test_duplicate_section_does_not_raise(self, cfg): cfg.add_section("server") # must not propagate an exception -# --------------------------------------------------------------------------- -# add_option -# --------------------------------------------------------------------------- - - @pytest.mark.unit class TestAddOption: @pytest.fixture @@ -521,10 +514,12 @@ def test_none_value_option_is_readable(self, cfg): cfg.add_option("server", "flag", None) assert cfg.has_option("flag") - -# --------------------------------------------------------------------------- -# update_option -# --------------------------------------------------------------------------- + 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 @@ -567,16 +562,34 @@ def test_creates_missing_section_and_option(self, cfg): assert view.get("width") == "800" def test_reentrancy_does_not_deadlock(self, tmp_path): - """update_option calls add_section/add_option while holding the RLock — must not deadlock.""" + """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 -# --------------------------------------------------------------------------- -# save_configuration -# --------------------------------------------------------------------------- + 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 @@ -620,10 +633,37 @@ def test_pending_preserved_when_write_fails(self, tmp_path): 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") -# --------------------------------------------------------------------------- -# get_configparser factory -# --------------------------------------------------------------------------- + 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 @@ -639,7 +679,9 @@ 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, "_DEFAULT_CONFIG", tmp_path / "nonexistent" / "BlocksScreen.cfg" + ) monkeypatch.setattr(cfmod, "_FALLBACK_CONFIG", cfg_file) cfg = get_configparser() @@ -651,7 +693,9 @@ def test_raises_config_error_when_server_section_missing( 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, "_DEFAULT_CONFIG", tmp_path / "nonexistent" / "BlocksScreen.cfg" + ) monkeypatch.setattr(cfmod, "_FALLBACK_CONFIG", cfg_file) with pytest.raises(ConfigError, match=r"\[server\]"): @@ -661,15 +705,41 @@ 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, "_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 -# --------------------------------------------------------------------------- -# Thread safety -# --------------------------------------------------------------------------- + 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 @@ -700,20 +770,15 @@ def updater(value: str) -> None: assert cfg.has_option("host") -# --------------------------------------------------------------------------- -# Bug fixes (regression tests) -# --------------------------------------------------------------------------- - - @pytest.mark.unit class TestEdgeCases: def test_parse_empty_file_does_not_crash(self, tmp_path): - """Empty (or comment-only) file must not raise IndexError.""" + """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() # must not raise - assert cfg.raw_config == [""] + 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" diff --git a/tests/widgets/test_slider_page_request_unit.py b/tests/widgets/test_slider_page_request_unit.py index 72a07891..ed1f5e8b 100644 --- a/tests/widgets/test_slider_page_request_unit.py +++ b/tests/widgets/test_slider_page_request_unit.py @@ -19,13 +19,7 @@ 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. -# Always replace (not setdefault): tests/util/test_configfile_unit.py may -# already have imported the *real* configfile module by this point in -# collection, and mutating its attributes in place (instead of swapping in -# a fresh stub) would corrupt BlocksScreenConfig for the rest of the session. +# fresh stubs, not setdefault: patching an already-imported real configfile leaks for _name in ("configfile", "lib.files", "lib.printer"): sys.modules[_name] = types.ModuleType(_name) sys.modules["configfile"].BlocksScreenConfig = MagicMock