feat(pathops): support line replacements and appending in ensure_contents

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+.

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__')
 

pathops/tests/unit/test_functions.py

@@ -16,14 +16,15 @@
 
 from __future__ import annotations
 
+import pathlib
 import typing
 
 import ops
 import pytest
 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<name> (managed)', matcher=r'^user=(?P<name>\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=.*$')