From af3553b69f5655d06b1d473faffed51035887e40 Mon Sep 17 00:00:00 2001 From: Jonas Jelten Date: Wed, 9 Sep 2026 16:37:38 +0200 Subject: [PATCH] feat(pathops): support line replacements and appending in ensure_contents Add matcher argument to update matching lines in an existing file instead of requiring the full contents. This supports backreferences, and configurable handling when there's no match. A str matcher is compiled with re.MULTILINE; pass a precompiled re.Pattern for other flags. --- pathops/CHANGELOG.md | 1 + pathops/src/charmlibs/pathops/_functions.py | 106 +++++++++++++- pathops/tests/unit/test_functions.py | 148 +++++++++++++++++++- 3 files changed, 247 insertions(+), 8 deletions(-) diff --git a/pathops/CHANGELOG.md b/pathops/CHANGELOG.md index b1c7fc11b..333aac27c 100644 --- a/pathops/CHANGELOG.md +++ b/pathops/CHANGELOG.md @@ -1,5 +1,6 @@ # Unreleased +`ensure_contents` now supports line replacements: pass a `matcher` to substitute each match with `source` (backreferences supported), optionally appending `source` when nothing matches. `ContainerPath` now provides `is_relative_to` and `with_stem`, matching `pathlib.Path` on Python 3.9+. `with_stem` is also part of `PathProtocol`; `is_relative_to` is not, because `pathlib`'s signature doesn't settle until Python 3.12. `ContainerPath.match` now accepts a `str | os.PathLike[str]` pattern, matching `pathlib.Path.match` on Python 3.12+. diff --git a/pathops/src/charmlibs/pathops/_functions.py b/pathops/src/charmlibs/pathops/_functions.py index 1fc84a81e..75a0b341a 100644 --- a/pathops/src/charmlibs/pathops/_functions.py +++ b/pathops/src/charmlibs/pathops/_functions.py @@ -17,6 +17,7 @@ from __future__ import annotations import pathlib +import re import typing from . import _constants, _fileinfo @@ -25,7 +26,7 @@ if typing.TYPE_CHECKING: import os - from typing import BinaryIO, TextIO + from typing import BinaryIO, Literal, TextIO from ops import pebble from typing_extensions import TypeIs @@ -37,19 +38,38 @@ def ensure_contents( path: str | os.PathLike[str] | PathProtocol, source: bytes | str | BinaryIO | TextIO, *, + matcher: str | re.Pattern[str] | None = None, + no_match: Literal['ignore', 'append', 'replace'] = 'ignore', mode: int = _constants.DEFAULT_WRITE_MODE, user: str | None = None, group: str | None = None, ) -> bool: - """Ensure ``source`` can be read from ``path``. Return True if any changes were made. + r"""Ensure ``source`` can be read from ``path``. Return True if any changes were made. Ensure that ``path`` exists, contains ``source``, has the correct permissions (``mode``), and has the correct file ownership (``user`` and ``group``). + If ``matcher`` is provided, ``source`` is instead the replacement text: every + substring matching ``matcher`` is replaced by ``source``. If the file doesn't + match, ``no_match`` decides what happens; if the file doesn't exist, ``source`` + is written. In this mode the file is read and written as UTF-8 text, and + ``source`` must not end with a newline: the matched line's newline is preserved + automatically, and a ``source`` written whole (file creation, append, replace) + is newline-terminated automatically. + Args: path: A local or remote filesystem path. source: The desired contents in ``str`` or ``bytes`` form, or an object with a ``.read()`` method which returns a ``str`` or ``bytes`` object. + matcher: A regular expression to search for in the file's current contents. + A ``str`` is compiled with :const:`re.MULTILINE`, so ``^`` and ``$`` match at + line boundaries. For any other flags -- for example a pattern that spans + multiple lines -- pass a precompiled :class:`re.Pattern` instead, which is + used as-is with its own flags. + no_match: What to do if ``matcher`` is provided but the file doesn't match: + ``'ignore'`` (default) to leave the file unchanged, ``'append'`` to append + ``source`` to the file, or ``'replace'`` to write the file with ``source``. + Ignored if the file doesn't exist, in which case ``source`` is written. mode: The desired file permissions. user: The desired file owner, or ``None`` to not change the owner. group: The desired group, or ``None`` to not change the group. @@ -61,28 +81,100 @@ def ensure_contents( LookupError: if the user or group is unknown. NotADirectoryError: if the parent exists as a non-directory file. PermissionError: if the user does not have permissions for the operation. + ValueError: if ``matcher`` is provided and ``source`` ends with a newline + (literal or ``\n`` escape); or if ``no_match`` is ``'append'`` or + ``'replace'``, the file doesn't match, and ``source`` contains + backreferences (there is no match to expand them against). + UnicodeDecodeError: if ``matcher`` is provided and the file's contents are not + valid UTF-8. :class:`PebbleConnectionError`: if the remote Pebble client cannot be reached. """ if _is_str_pathlike(path): path = LocalPath(path) - source = _as_bytes(source) + if matcher is not None: + if not isinstance(matcher, re.Pattern): + matcher = re.compile(matcher, re.MULTILINE) + template = _as_text(source) + if template.endswith(('\n', '\\n')): + raise ValueError( + 'in matcher mode, source must not end with a newline: ' + "the matched line's newline is preserved automatically" + ) + # a source written whole is a single newline-terminated line + source_bytes = template.encode() + b'\n' + else: + source_bytes = _as_bytes(source) + template = None try: info = _get_fileinfo(path) except FileNotFoundError: - pass # file doesn't exist, so writing is required - else: # check if metadata and contents already match + current = None # file doesn't exist, so writing is required + else: + current = path.read_bytes() if ( (info.permissions == mode) and (user is None or info.user == user) and (group is None or info.group == group) - and (path.read_bytes() == source) + and _contents_match(current, source_bytes, matcher, template, no_match) ): return False # everything matches, so writing is not required path.parent.mkdir(parents=True, exist_ok=True) - path.write_bytes(source, mode=mode, user=user, group=group) + if current is None: + desired = source_bytes + elif matcher is not None and template is not None: + desired = _edit_contents(current, source_bytes, matcher, template, no_match) + else: + desired = source_bytes + path.write_bytes(desired, mode=mode, user=user, group=group) return True +def _edit_contents( + current: bytes, + source: bytes, + matcher: re.Pattern[str], + template: str, + no_match: Literal['ignore', 'append', 'replace'], +) -> bytes: + """The desired contents, given the file's current contents and a matcher.""" + current_text = current.decode() + if matcher.search(current_text): + return matcher.sub(template, current_text).encode() + if no_match == 'ignore': + return current + if re.search(r'\\[1-9]|\\g<', template): + raise ValueError('source contains backreferences but the file has no match') + if no_match == 'append': + # newline-terminate the file so the appended line is its own line + return current.removesuffix(b'\n') + b'\n' + source + return source + + +def _contents_match( + current: bytes, + source: bytes, + matcher: re.Pattern[str] | None, + template: str | None, + no_match: Literal['ignore', 'append', 'replace'], +) -> bool: + """Whether the file's current contents already match the desired contents.""" + if matcher is None or template is None: + return current == source + current_text = current.decode() + if not matcher.search(current_text): + # no match: the desired contents are source appended, or the file unchanged + return current + source == current if no_match == 'append' else no_match == 'ignore' + return matcher.sub(template, current_text).encode() == current + + +def _as_text(source: str | bytes | BinaryIO | TextIO) -> str: + if isinstance(source, str): + return source + if isinstance(source, bytes): + return source.decode() + return _as_text(source.read()) + + def _is_str_pathlike(obj: object) -> TypeIs[str | os.PathLike[str]]: return isinstance(obj, str) or hasattr(obj, '__fspath__') diff --git a/pathops/tests/unit/test_functions.py b/pathops/tests/unit/test_functions.py index 1e190809f..e486e90a9 100644 --- a/pathops/tests/unit/test_functions.py +++ b/pathops/tests/unit/test_functions.py @@ -16,6 +16,7 @@ from __future__ import annotations +import pathlib import typing import ops @@ -23,7 +24,7 @@ from ops import pebble import utils -from charmlibs.pathops import ContainerPath +from charmlibs.pathops import ContainerPath, LocalPath, ensure_contents from charmlibs.pathops._functions import _get_fileinfo if typing.TYPE_CHECKING: @@ -47,3 +48,148 @@ def test_get_fileinfo_reraises_unhandled_pebble_errors( monkeypatch.setattr(container, 'list_files', mock) with pytest.raises(error): _get_fileinfo(ContainerPath('/', container=container)) + + +@pytest.mark.parametrize('path_type', [str, pathlib.Path, LocalPath]) +def test_ensure_contents_matcher_replaces( + tmp_path: pathlib.Path, + path_type: type[str] | type[pathlib.Path] | type[LocalPath], +): + path = tmp_path / 'config' + path.write_text('user=root\nport=80\n') + changed = ensure_contents( + path_type(path), + 'user=guest', + matcher=r'^user=.*$', + ) + assert changed + assert path.read_text() == 'user=guest\nport=80\n' + # idempotent + assert not ensure_contents(path_type(path), 'user=guest', matcher=r'^user=.*$') + + +def test_ensure_contents_matcher_replaces_all_matches(tmp_path: pathlib.Path): + path = tmp_path / 'config' + path.write_text('user=a\nuser=b\n') + changed = ensure_contents(path, 'user=c', matcher=r'^user=.*$') + assert changed + assert path.read_text() == 'user=c\nuser=c\n' + + +def test_ensure_contents_matcher_appends_when_no_match(tmp_path: pathlib.Path): + path = tmp_path / 'config' + path.write_text('port=80\n') + changed = ensure_contents(path, 'user=guest', matcher=r'^user=.*$', no_match='append') + assert changed + assert path.read_text() == 'port=80\nuser=guest\n' + # appending an already-present line is a no-op + assert not ensure_contents(path, 'user=guest', matcher=r'^user=.*$', no_match='append') + + +def test_ensure_contents_matcher_append_terminates_unterminated_file(tmp_path: pathlib.Path): + path = tmp_path / 'config' + path.write_text('port=80') # no trailing newline + changed = ensure_contents(path, 'user=guest', matcher=r'^user=.*$', no_match='append') + assert changed + assert path.read_text() == 'port=80\nuser=guest\n' + + +def test_ensure_contents_matcher_no_match_leaves_file_unchanged(tmp_path: pathlib.Path): + path = tmp_path / 'config' + path.write_text('port=80\n') + assert not ensure_contents(path, 'user=guest', matcher=r'^user=.*$') + assert path.read_text() == 'port=80\n' + + +def test_ensure_contents_matcher_no_match_replace(tmp_path: pathlib.Path): + path = tmp_path / 'config' + path.write_text('port=80\n') + changed = ensure_contents(path, 'user=guest', matcher=r'^user=.*$', no_match='replace') + assert changed + assert path.read_text() == 'user=guest\n' + + +def test_ensure_contents_matcher_creates_missing_file(tmp_path: pathlib.Path): + path = tmp_path / 'subdir' / 'config' + changed = ensure_contents(path, 'user=guest', matcher=r'^user=.*$') + assert changed + assert path.read_text() == 'user=guest\n' + + +def test_ensure_contents_matcher_accepts_pattern(tmp_path: pathlib.Path): + import re + + path = tmp_path / 'config' + path.write_text('user=root\n') + changed = ensure_contents(path, 'user=guest', matcher=re.compile(r'^user=.*$')) + assert changed + assert path.read_text() == 'user=guest\n' + + +def test_ensure_contents_matcher_spans_multiple_lines(tmp_path: pathlib.Path): + import re + + path = tmp_path / 'config' + path.write_text('# begin\nold=1\nold=2\n# end\nport=80\n') + matcher = re.compile(r'# begin\n.*\n# end', re.DOTALL) + changed = ensure_contents(path, '# begin\nnew=3\n# end', matcher=matcher) + assert changed + assert path.read_text() == '# begin\nnew=3\n# end\nport=80\n' + # idempotent + assert not ensure_contents(path, '# begin\nnew=3\n# end', matcher=matcher) + + +def test_ensure_contents_matcher_pattern_without_multiline(tmp_path: pathlib.Path): + import re + + # without re.MULTILINE, ^ only matches at the start of the file + path = tmp_path / 'config' + path.write_text('port=80\nuser=root\n') + changed = ensure_contents( + path, + 'user=guest', + matcher=re.compile(r'^user=.*$'), + no_match='append', + ) + assert changed + # no match, so source was appended + assert path.read_text() == 'port=80\nuser=root\nuser=guest\n' + + +def test_ensure_contents_matcher_supports_backreferences(tmp_path: pathlib.Path): + path = tmp_path / 'config' + path.write_text('user=root:admin\nport=80\n') + changed = ensure_contents(path, r'user=\1:wheel', matcher=r'^user=(\S+):(\S+)$') + assert changed + assert path.read_text() == 'user=root:wheel\nport=80\n' + + +def test_ensure_contents_matcher_named_backreferences(tmp_path: pathlib.Path): + path = tmp_path / 'config' + path.write_text('user=root\n') + changed = ensure_contents(path, r'user=\g (managed)', matcher=r'^user=(?P\S+)$') + assert changed + assert path.read_text() == 'user=root (managed)\n' + + +def test_ensure_contents_matcher_backrefs_no_match_raises(tmp_path: pathlib.Path): + path = tmp_path / 'config' + path.write_text('port=80\n') + with pytest.raises(ValueError, match='no match'): + ensure_contents(path, r'user=\1', matcher=r'^user=(\S+)$', no_match='append') + + +def test_ensure_contents_matcher_backrefs_no_match_ignored(tmp_path: pathlib.Path): + # with the default ignore behaviour there is nothing to expand, so no error + path = tmp_path / 'config' + path.write_text('port=80\n') + assert not ensure_contents(path, r'user=\1', matcher=r'^user=(\S+)$') + assert path.read_text() == 'port=80\n' + + +@pytest.mark.parametrize('newline', ['\n', r'\n']) +def test_ensure_contents_matcher_source_newline_raises(tmp_path: pathlib.Path, newline: str): + path = tmp_path / 'config' + path.write_text('user=root\n') + with pytest.raises(ValueError, match='must not end with a newline'): + ensure_contents(path, 'user=guest' + newline, matcher=r'^user=.*$')