diff --git a/pyproject.toml b/pyproject.toml index 249f513..dae4f85 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,7 +31,7 @@ classifiers = [ "Topic :: Software Development :: Libraries" ] dependencies = [ - "openjd-model >= 0.8,< 0.9", + "openjd-model >= 0.9,< 0.10", "pywin32 >= 307; platform_system == 'Windows'", "psutil >= 5.9,< 7.3; platform_system == 'Windows'", ] diff --git a/src/openjd/sessions/_embedded_files.py b/src/openjd/sessions/_embedded_files.py index ceb416f..ca09957 100644 --- a/src/openjd/sessions/_embedded_files.py +++ b/src/openjd/sessions/_embedded_files.py @@ -1,6 +1,7 @@ # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. import os +import re import stat from contextlib import contextmanager from dataclasses import dataclass @@ -37,8 +38,40 @@ def _open_context(*args: Any, **kwargs: Any) -> Generator[int, None, None]: os.close(fd) +# Regex to match LF not preceded by CR (for CRLF conversion) +_LF_NOT_CRLF = re.compile(r"(? str: + """Convert line endings based on the specified mode. + + Args: + data: The string data to convert + end_of_line: One of None, "AUTO", "LF", or "CRLF" + + Returns: + The data with converted line endings + """ + if end_of_line is None or end_of_line == "AUTO": + # AUTO: use OS native line endings + if os.name == "nt": + # Windows: ensure CRLF + return _LF_NOT_CRLF.sub("\r\n", data) + # POSIX: ensure LF + return data.replace("\r\n", "\n") + elif end_of_line == "LF": + return data.replace("\r\n", "\n") + elif end_of_line == "CRLF": + return _LF_NOT_CRLF.sub("\r\n", data) + return data + + def write_file_for_user( - filename: Path, data: str, user: Optional[SessionUser], additional_permissions: int = 0 + filename: Path, + data: str, + user: Optional[SessionUser], + additional_permissions: int = 0, + end_of_line: Optional[str] = None, ) -> None: # File should only be r/w by the owner, by default @@ -47,17 +80,22 @@ def write_file_for_user( # O_CREAT - create if it does not exist # O_TRUNC - truncate the file. If we overwrite an existing file, then we # need to clear its contents. + # O_BINARY - (Windows only) prevent automatic \n to \r\n conversion # O_EXCL (intentionally not present) - fail if file exists # - We exclude this 'cause we expect to be writing the same embedded file # into the same location repeatedly with different contents as we run # multiple Tasks in the same Session. 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 + flags |= getattr(os, "O_BINARY", 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) with _open_context(filename, flags, mode=mode) as fd: - os.write(fd, data.encode("utf-8")) + os.write(fd, converted_data.encode("utf-8")) if os.name == "posix": if user is not None: @@ -227,8 +265,16 @@ def _materialize_file( execute_permissions |= stat.S_IXUSR | (stat.S_IXGRP if self._user is not None else 0) data = file.data.resolve(symtab=symtab) + # Get endOfLine setting if present + end_of_line = file.endOfLine.value if file.endOfLine else None # Create the file as r/w owner, and optionally group - write_file_for_user(filename, data, self._user, additional_permissions=execute_permissions) + write_file_for_user( + filename, + data, + self._user, + additional_permissions=execute_permissions, + end_of_line=end_of_line, + ) self._logger.info( f"Wrote: {file.name} -> {str(filename)}", diff --git a/src/openjd/sessions/_runner_base.py b/src/openjd/sessions/_runner_base.py index 98aaf38..2708393 100644 --- a/src/openjd/sessions/_runner_base.py +++ b/src/openjd/sessions/_runner_base.py @@ -464,7 +464,7 @@ def _run_action( else: time_limit: Optional[timedelta] = default_timeout if action.timeout: - time_limit = timedelta(seconds=action.timeout) + time_limit = timedelta(seconds=action.timeout) # type: ignore[arg-type] self._run(command, time_limit) def _cancel( diff --git a/src/openjd/sessions/_runner_env_script.py b/src/openjd/sessions/_runner_env_script.py index 25c7f2a..2ce1962 100644 --- a/src/openjd/sessions/_runner_env_script.py +++ b/src/openjd/sessions/_runner_env_script.py @@ -207,7 +207,7 @@ def cancel( method = NotifyCancelMethod(terminate_delay=timedelta(seconds=30)) else: method = NotifyCancelMethod( - terminate_delay=timedelta(seconds=model_cancel_method.notifyPeriodInSeconds) + terminate_delay=timedelta(seconds=model_cancel_method.notifyPeriodInSeconds) # type: ignore[arg-type] ) # Note: If the given time_limit is less than that in the method, then the time_limit will be what's used. diff --git a/src/openjd/sessions/_runner_step_script.py b/src/openjd/sessions/_runner_step_script.py index 1411cd4..a4d2e65 100644 --- a/src/openjd/sessions/_runner_step_script.py +++ b/src/openjd/sessions/_runner_step_script.py @@ -143,7 +143,7 @@ def cancel( method = NotifyCancelMethod(terminate_delay=timedelta(seconds=120)) else: method = NotifyCancelMethod( - terminate_delay=timedelta(seconds=model_cancel_method.notifyPeriodInSeconds) + terminate_delay=timedelta(seconds=model_cancel_method.notifyPeriodInSeconds) # type: ignore[arg-type] ) # Note: If the given time_limit is less than that in the method, then the time_limit will be what's used. diff --git a/test/openjd/sessions/test_embedded_files.py b/test/openjd/sessions/test_embedded_files.py index ae5b1ed..9d70c87 100644 --- a/test/openjd/sessions/test_embedded_files.py +++ b/test/openjd/sessions/test_embedded_files.py @@ -5,7 +5,7 @@ import uuid from dataclasses import dataclass from pathlib import Path -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch from openjd.sessions._os_checker import is_posix, is_windows import pytest @@ -19,6 +19,9 @@ from openjd.model.v2023_09 import ( EmbeddedFileTypes as EmbeddedFileTypes_2023_09, ) +from openjd.model.v2023_09 import ( + EndOfLine as EndOfLine_2023_09, +) from openjd.sessions._embedded_files import EmbeddedFiles, EmbeddedFilesScope from openjd.sessions._session_user import PosixSessionUser, WindowsSessionUser @@ -365,6 +368,119 @@ def test_changes_owner(self, tmp_path: Path, windows_user: WindowsSessionUser) - result_contents = file.read() assert result_contents == testdata, "File contents are as expected" + class TestEndOfLine: + """Tests for endOfLine handling in embedded files.""" + + @pytest.mark.parametrize( + "end_of_line,input_data,expected_bytes", + [ + pytest.param( + EndOfLine_2023_09.LF, + "line1\nline2\nline3", + b"line1\nline2\nline3", + id="LF-only", + ), + pytest.param( + EndOfLine_2023_09.LF, + "line1\r\nline2\r\nline3", + b"line1\nline2\nline3", + id="LF-converts-CRLF", + ), + pytest.param( + EndOfLine_2023_09.CRLF, + "line1\nline2\nline3", + b"line1\r\nline2\r\nline3", + id="CRLF-converts-LF", + ), + pytest.param( + EndOfLine_2023_09.CRLF, + "line1\r\nline2\r\nline3", + b"line1\r\nline2\r\nline3", + id="CRLF-preserves-CRLF", + ), + ], + ) + def test_end_of_line_conversion( + self, + tmp_path: Path, + end_of_line: EndOfLine_2023_09, + input_data: str, + expected_bytes: bytes, + ) -> None: + # Test that endOfLine correctly converts line endings + + # GIVEN + test_obj = EmbeddedFiles( + logger=MagicMock(), scope=EmbeddedFilesScope.STEP, session_files_directory=tmp_path + ) + test_file = EmbeddedFileText_2023_09( + name="Foo", + type=EmbeddedFileTypes_2023_09.TEXT, + data=DataString_2023_09(input_data), + endOfLine=end_of_line, + ) + filename = tmp_path / uuid.uuid4().hex + symtab = SymbolTable() + + # WHEN + test_obj._materialize_file(filename, test_file, symtab) + + # THEN + with open(filename, "rb") as file: + result_bytes = file.read() + assert result_bytes == expected_bytes + + @pytest.mark.skipif(is_windows(), reason="Cannot simulate POSIX file I/O on Windows") + def test_auto_uses_lf_on_posix(self, tmp_path: Path) -> None: + # Test that AUTO mode uses LF on POSIX systems + + # GIVEN + test_obj = EmbeddedFiles( + logger=MagicMock(), scope=EmbeddedFilesScope.STEP, session_files_directory=tmp_path + ) + test_file = EmbeddedFileText_2023_09( + name="Foo", + type=EmbeddedFileTypes_2023_09.TEXT, + data=DataString_2023_09("line1\r\nline2"), + endOfLine=EndOfLine_2023_09.AUTO, + ) + filename = tmp_path / uuid.uuid4().hex + symtab = SymbolTable() + + # WHEN + test_obj._materialize_file(filename, test_file, symtab) + + # THEN + with open(filename, "rb") as file: + result_bytes = file.read() + assert result_bytes == b"line1\nline2" + + @pytest.mark.skipif(not is_windows(), reason="Windows-specific test") + def test_auto_uses_crlf_on_windows(self, tmp_path: Path) -> None: + # Test that AUTO mode uses CRLF on Windows systems + + # GIVEN + test_obj = EmbeddedFiles( + logger=MagicMock(), scope=EmbeddedFilesScope.STEP, session_files_directory=tmp_path + ) + test_file = EmbeddedFileText_2023_09( + name="Foo", + type=EmbeddedFileTypes_2023_09.TEXT, + data=DataString_2023_09("line1\nline2"), + endOfLine=EndOfLine_2023_09.AUTO, + ) + filename = tmp_path / uuid.uuid4().hex + symtab = SymbolTable() + + # WHEN + with patch("os.name", "nt"): + test_obj._materialize_file(filename, test_file, symtab) + + # THEN + with open(filename, "rb") as file: + result_bytes = file.read() + assert result_bytes == b"line1\r\nline2" + class TestMaterialize: """Tests for EmbeddedFiles.materialize()"""