feat: add a verbatim write option to write_file_for_user

src/openjd/sessions/_embedded_files.py

@@ -47,10 +47,14 @@ def _convert_line_endings(data: str, end_of_line: Optional[str]) -> str:
 
     Args:
         data: The string data to convert
-        end_of_line: One of None, "AUTO", "LF", or "CRLF"
+        end_of_line: One of None, "AUTO", "LF", or "CRLF". Any other value is
+            rejected rather than silently leaving the data unconverted.
 
     Returns:
         The data with converted line endings
+
+    Raises:
+        ValueError: if ``end_of_line`` is not a recognized value.
     """
     if end_of_line is None or end_of_line == "AUTO":
         # AUTO: use OS native line endings
@@ -63,7 +67,10 @@ def _convert_line_endings(data: str, end_of_line: Optional[str]) -> str:
         return data.replace("\r\n", "\n")
     elif end_of_line == "CRLF":
         return _LF_NOT_CRLF.sub("\r\n", data)
-    return data
+    raise ValueError(
+        f"Unrecognized end_of_line value {end_of_line!r}. Expected one of "
+        "None, 'AUTO', 'LF', or 'CRLF'."
+    )
 
 
 def chown_group(path: Path, group: str) -> None:
@@ -142,7 +149,47 @@ def write_file_for_user(
     user: Optional[SessionUser],
     additional_permissions: int = 0,
     end_of_line: Optional[str] = None,
+    *,
+    verbatim: bool = False,
 ) -> None:
+    """Write ``data`` to ``filename`` with owner-only read/write permissions,
+    optionally extended to a session user's group.
+
+    Arguments:
+        filename: Absolute path of the file to create or overwrite.
+        data: The text to write. Encoded as UTF-8 by this function.
+        user: If given, the session user whose group gains read/write access.
+        additional_permissions: Extra mode bits to set, masked to the owner
+            triad, and extended to the group triad only when ``user`` is given
+            on POSIX.
+        end_of_line: One of ``None``, ``"AUTO"``, ``"LF"``, or ``"CRLF"``.
+            ``None`` and ``"AUTO"`` select the host's native line ending. Any
+            other value is rejected.
+        verbatim: Write ``data`` exactly as given. The content is not escaped,
+            not re-interpreted, and its line endings are not converted:
+            ``_convert_line_endings`` is not called at all, and the bytes on
+            disk are ``data.encode("utf-8")`` unmodified. This is for a caller
+            writing opaque content it does not own and must not alter, such as
+            a script fetched from a third party. This is a writer-level option
+            only and deliberately not an ``openjd-model`` ``EndOfLine`` enum
+            value -- that enum is a specification surface describing what a job
+            template may request, and verbatim is not a line-ending mode a
+            template can ask for. It is a statement by the calling code that
+            the bytes it holds are already final.
+
+    Raises:
+        ValueError: if ``verbatim`` is True and ``end_of_line`` is not None
+            (the two requests contradict each other), or if ``end_of_line``
+            is not a recognized value.
+        OSError: if the file could not be created or written, or if the final
+            path component is a symbolic link.
+    """
+    if verbatim and end_of_line is not None:
+        raise ValueError(
+            f"verbatim=True writes data unmodified, so end_of_line={end_of_line!r} "
+            "cannot also be applied. Pass one or the other."
+        )
+
     # File should only be r/w by the owner, by default
 
     # flags:
@@ -166,17 +213,18 @@ def write_file_for_user(
     #      while still allowing the file to be rewritten across Tasks.
     flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
     # On Windows, use O_BINARY to prevent automatic line ending conversion
-    # since we handle line endings explicitly via _convert_line_endings
+    # since line endings are handled explicitly here via _convert_line_endings,
+    # when they are converted at all (see 'verbatim').
     flags |= getattr(os, "O_BINARY", 0)
     # O_NOFOLLOW is not defined on Windows; getattr(..., 0) makes this a no-op there.
     flags |= getattr(os, "O_NOFOLLOW", 0)
     # mode:
     #  S_IRUSR - Read by owner
     #  S_IWUSR - Write by owner
     mode = stat.S_IRUSR | stat.S_IWUSR | (additional_permissions & stat.S_IRWXU)
-    converted_data = _convert_line_endings(data, end_of_line)
+    data_to_write = data if verbatim else _convert_line_endings(data, end_of_line)
     with _open_context(filename, flags, mode=mode) as fd:
-        os.write(fd, converted_data.encode("utf-8"))
+        os.write(fd, data_to_write.encode("utf-8"))
 
     if os.name == "posix":
         if user is not None:

test/openjd/sessions_v0/test_embedded_files.py

@@ -6,6 +6,7 @@
 import uuid
 from dataclasses import dataclass
 from pathlib import Path
+from typing import Optional
 from unittest.mock import MagicMock, patch
 from openjd.sessions._os_checker import is_posix, is_windows
 import pytest
@@ -26,6 +27,7 @@
 from openjd.sessions._embedded_files import (
     EmbeddedFiles,
     EmbeddedFilesScope,
+    _convert_line_endings,
     _validate_embedded_filename,
     chown_group,
     write_file_for_user,
@@ -599,6 +601,219 @@ def test_auto_uses_crlf_on_windows(self, tmp_path: Path) -> None:
                 result_bytes = file.read()
             assert result_bytes == b"line1\r\nline2"
 
+    class TestVerbatimWrite:
+        """Tests for write_file_for_user()'s verbatim option."""
+
+        def test_verbatim_preserves_mixed_line_endings(self, tmp_path: Path) -> None:
+            # verbatim=True must write the input bytes exactly, on POSIX and
+            # Windows alike. A mixed LF/CRLF body is rewritten by every
+            # end_of_line mode, so it fails if verbatim stops bypassing the
+            # conversion.
+
+            # GIVEN
+            filename = tmp_path / uuid.uuid4().hex
+            data = "line1\nline2\r\nline3\n"
+
+            # WHEN
+            write_file_for_user(filename, data, user=None, verbatim=True)
+
+            # THEN
+            with open(filename, "rb") as file:
+                result_bytes = file.read()
+            assert result_bytes == b"line1\nline2\r\nline3\n"
+
+        @pytest.mark.skipif(not is_posix(), reason="posix file mode bits")
+        def test_verbatim_matches_non_verbatim_permissions(self, tmp_path: Path) -> None:
+            # The verbatim branch changes only which string is encoded, so the
+            # resulting file mode must be identical to the default path's.
+
+            # GIVEN
+            verbatim_file = tmp_path / uuid.uuid4().hex
+            converted_file = tmp_path / uuid.uuid4().hex
+
+            # WHEN
+            write_file_for_user(
+                verbatim_file,
+                "data",
+                user=None,
+                additional_permissions=stat.S_IXUSR,
+                verbatim=True,
+            )
+            write_file_for_user(
+                converted_file, "data", user=None, additional_permissions=stat.S_IXUSR
+            )
+
+            # THEN
+            verbatim_mode = os.stat(verbatim_file).st_mode & 0o777
+            converted_mode = os.stat(converted_file).st_mode & 0o777
+            assert verbatim_mode == converted_mode
+            # Asserted absolutely as well, so the equality cannot hold by both
+            # paths regressing together.
+            assert verbatim_mode == (stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
+
+        @pytest.mark.skipif(not is_posix(), reason="O_NOFOLLOW is posix-specific")
+        def test_verbatim_still_refuses_to_follow_symlink(self, tmp_path: Path) -> None:
+            # The verbatim branch must go through the same O_NOFOLLOW-guarded
+            # writer, not a convenience writer that follows symlinks.
+
+            # GIVEN
+            target = tmp_path / "target.txt"
+            target.write_text("original-contents")
+            link = tmp_path / "link.txt"
+            link.symlink_to(target)
+
+            # WHEN / THEN
+            with pytest.raises(OSError):
+                write_file_for_user(link, "overwritten", user=None, verbatim=True)
+            # AND the symlink target is untouched
+            assert target.read_text() == "original-contents"
+
+        @pytest.mark.parametrize(
+            "end_of_line",
+            [
+                pytest.param("AUTO", id="AUTO"),
+                pytest.param("LF", id="LF"),
+                pytest.param("CRLF", id="CRLF"),
+            ],
+        )
+        def test_verbatim_with_explicit_end_of_line_raises(
+            self, tmp_path: Path, end_of_line: str
+        ) -> None:
+            # verbatim=True and an explicit end_of_line are contradictory
+            # requests. Neither silently wins.
+
+            # GIVEN
+            filename = tmp_path / uuid.uuid4().hex
+
+            # WHEN
+            with pytest.raises(ValueError) as excinfo:
+                write_file_for_user(
+                    filename, "data", user=None, end_of_line=end_of_line, verbatim=True
+                )
+
+            # THEN
+            assert "verbatim" in str(excinfo.value)
+            assert "end_of_line" in str(excinfo.value)
+            # AND the guard ran before the O_TRUNC open, so the destination was
+            # not created or truncated on the way to raising.
+            assert not filename.exists()
+
+        def test_verbatim_with_explicit_none_end_of_line_is_allowed(self, tmp_path: Path) -> None:
+            # Negative control for the guard: it fires on a non-None
+            # end_of_line, not on the keyword being present at all.
+
+            # GIVEN
+            filename = tmp_path / uuid.uuid4().hex
+
+            # WHEN
+            write_file_for_user(filename, "a\nb\r\nc\n", user=None, end_of_line=None, verbatim=True)
+
+            # THEN
+            with open(filename, "rb") as file:
+                result_bytes = file.read()
+            assert result_bytes == b"a\nb\r\nc\n"
+
+        @pytest.mark.parametrize(
+            "end_of_line,expected_bytes",
+            [
+                pytest.param("LF", b"a\nb\nc\n", id="LF"),
+                pytest.param("CRLF", b"a\r\nb\r\nc\r\n", id="CRLF"),
+            ],
+        )
+        def test_non_verbatim_still_converts(
+            self, tmp_path: Path, end_of_line: str, expected_bytes: bytes
+        ) -> None:
+            # Negative control: conversion still happens when verbatim is not
+            # requested. AUTO is covered by TestEndOfLine, whose expectation is
+            # platform-dependent.
+
+            # GIVEN
+            filename = tmp_path / uuid.uuid4().hex
+
+            # WHEN
+            write_file_for_user(filename, "a\nb\r\nc\n", user=None, end_of_line=end_of_line)
+
+            # THEN
+            with open(filename, "rb") as file:
+                result_bytes = file.read()
+            assert result_bytes == expected_bytes
+
+    class TestConvertLineEndings:
+        """Tests for _convert_line_endings()'s handling of unrecognized values."""
+
+        @pytest.mark.parametrize(
+            "end_of_line",
+            [
+                pytest.param("crlf", id="case-wrong-crlf"),
+                pytest.param("Lf", id="case-wrong-Lf"),
+                pytest.param("", id="empty-string"),
+                pytest.param("typo", id="typo"),
+                pytest.param("VERBATIM", id="not-a-mode"),
+            ],
+        )
+        def test_rejects_unrecognized_value(self, end_of_line: str) -> None:
+            # An unrecognized end_of_line must be rejected, not silently
+            # ignored. Previously each of these returned the data unconverted
+            # with no error, so a case-wrong value looked like it worked.
+
+            # WHEN
+            with pytest.raises(ValueError) as excinfo:
+                _convert_line_endings("a\nb\r\nc\n", end_of_line)
+
+            # THEN
+            assert repr(end_of_line) in str(excinfo.value)
+
+        @pytest.mark.parametrize(
+            "end_of_line",
+            [
+                pytest.param(None, id="None"),
+                pytest.param("AUTO", id="AUTO"),
+            ],
+        )
+        def test_accepts_native_values(self, end_of_line: Optional[str]) -> None:
+            # Negative control: None and AUTO still select the host's native
+            # line ending and do not raise.
+
+            # GIVEN
+            expected = "a\r\nb\r\nc\r\n" if is_windows() else "a\nb\nc\n"
+
+            # WHEN
+            result = _convert_line_endings("a\nb\r\nc\n", end_of_line)
+
+            # THEN
+            assert result == expected
+
+        @pytest.mark.parametrize(
+            "end_of_line,expected",
+            [
+                pytest.param("LF", "a\nb\nc\n", id="LF"),
+                pytest.param("CRLF", "a\r\nb\r\nc\r\n", id="CRLF"),
+            ],
+        )
+        def test_accepts_explicit_values(self, end_of_line: str, expected: str) -> None:
+            # Negative control: the explicit modes still convert and do not
+            # raise, so the strict check is not over-broad.
+
+            # WHEN
+            result = _convert_line_endings("a\nb\r\nc\n", end_of_line)
+
+            # THEN
+            assert result == expected
+
+        def test_write_file_for_user_rejects_unrecognized_value(self, tmp_path: Path) -> None:
+            # End to end through the writer: the conversion happens before the
+            # O_TRUNC open, so a rejected value leaves no file behind.
+
+            # GIVEN
+            filename = tmp_path / uuid.uuid4().hex
+
+            # WHEN
+            with pytest.raises(ValueError):
+                write_file_for_user(filename, "a\nb\r\nc\n", user=None, end_of_line="crlf")
+
+            # THEN
+            assert not filename.exists()
+
     class TestMaterialize:
         """Tests for EmbeddedFiles.materialize()"""