Add --newline=[LF|CRLF|native|preserve] to compile, to override the line separator characters written (#1652)

Co-authored-by: Thomas Grainger <tagrain@gmail.com>
Co-authored-by: Sorin Sbarnea <sorin.sbarnea@gmail.com>

Author: 3 people

piptools/scripts/compile.py

@@ -49,6 +49,35 @@ def _get_default_option(option_name: str) -> Any:
     return getattr(default_values, option_name)
 
 
+def _determine_linesep(
+    strategy: str = "preserve", filenames: Tuple[str, ...] = ()
+) -> str:
+    """
+    Determine and return linesep string for OutputWriter to use.
+    Valid strategies: "LF", "CRLF", "native", "preserve"
+    When preserving, files are checked in order for existing newlines.
+    """
+    if strategy == "preserve":
+        for fname in filenames:
+            try:
+                with open(fname, "rb") as existing_file:
+                    existing_text = existing_file.read()
+            except FileNotFoundError:
+                continue
+            if b"\r\n" in existing_text:
+                strategy = "CRLF"
+                break
+            elif b"\n" in existing_text:
+                strategy = "LF"
+                break
+    return {
+        "native": os.linesep,
+        "LF": "\n",
+        "CRLF": "\r\n",
+        "preserve": "\n",
+    }[strategy]
+
+
 @click.command(context_settings={"help_option_names": ("-h", "--help")})
 @click.version_option(**version_option_kwargs)
 @click.pass_context
@@ -165,6 +194,12 @@ def _get_default_option(option_name: str) -> Any:
         "Will be derived from input file otherwise."
     ),
 )
+@click.option(
+    "--newline",
+    type=click.Choice(("LF", "CRLF", "native", "preserve"), case_sensitive=False),
+    default="preserve",
+    help="Override the newline control characters used",
+)
 @click.option(
     "--allow-unsafe/--no-allow-unsafe",
     is_flag=True,
@@ -279,6 +314,7 @@ def cli(
     upgrade: bool,
     upgrade_packages: Tuple[str, ...],
     output_file: Union[LazyFile, IO[Any], None],
+    newline: str,
     allow_unsafe: bool,
     strip_extras: bool,
     generate_hashes: bool,
@@ -515,6 +551,10 @@ def cli(
 
     log.debug("")
 
+    linesep = _determine_linesep(
+        strategy=newline, filenames=(output_file.name, *src_files)
+    )
+
     ##
     # Output
     ##
@@ -534,6 +574,7 @@ def cli(
         index_urls=repository.finder.index_urls,
         trusted_hosts=repository.finder.trusted_hosts,
         format_control=repository.finder.format_control,
+        linesep=linesep,
         allow_unsafe=allow_unsafe,
         find_links=repository.finder.find_links,
         emit_find_links=emit_find_links,

piptools/writer.py

@@ -1,3 +1,4 @@
+import io
 import os
 import re
 import sys
@@ -98,6 +99,7 @@ def __init__(
         index_urls: Iterable[str],
         trusted_hosts: Iterable[str],
         format_control: FormatControl,
+        linesep: str,
         allow_unsafe: bool,
         find_links: List[str],
         emit_find_links: bool,
@@ -117,6 +119,7 @@ def __init__(
         self.index_urls = index_urls
         self.trusted_hosts = trusted_hosts
         self.format_control = format_control
+        self.linesep = linesep
         self.allow_unsafe = allow_unsafe
         self.find_links = find_links
         self.emit_find_links = emit_find_links
@@ -257,14 +260,25 @@ def write(
         hashes: Optional[Dict[InstallRequirement, Set[str]]],
     ) -> None:
 
-        for line in self._iter_lines(results, unsafe_requirements, markers, hashes):
-            if self.dry_run:
-                # Bypass the log level to always print this during a dry run
-                log.log(line)
-            else:
-                log.info(line)
-                self.dst_file.write(unstyle(line).encode())
-                self.dst_file.write(os.linesep.encode())
+        if not self.dry_run:
+            dst_file = io.TextIOWrapper(
+                self.dst_file,
+                encoding="utf8",
+                newline=self.linesep,
+                line_buffering=True,
+            )
+        try:
+            for line in self._iter_lines(results, unsafe_requirements, markers, hashes):
+                if self.dry_run:
+                    # Bypass the log level to always print this during a dry run
+                    log.log(line)
+                else:
+                    log.info(line)
+                    dst_file.write(unstyle(line))
+                    dst_file.write("\n")
+        finally:
+            if not self.dry_run:
+                dst_file.detach()
 
     def _format_requirement(
         self,

tests/test_cli_compile.py

@@ -936,6 +936,88 @@ def test_generate_hashes_with_annotations(runner):
     )
 
 
+@pytest.mark.network
+@pytest.mark.parametrize("gen_hashes", (True, False))
+@pytest.mark.parametrize(
+    "annotate_options",
+    (
+        ("--no-annotate",),
+        ("--annotation-style", "line"),
+        ("--annotation-style", "split"),
+    ),
+)
+@pytest.mark.parametrize(
+    ("nl_options", "must_include", "must_exclude"),
+    (
+        pytest.param(("--newline", "lf"), "\n", "\r\n", id="LF"),
+        pytest.param(("--newline", "crlf"), "\r\n", "\n", id="CRLF"),
+        pytest.param(
+            ("--newline", "native"),
+            os.linesep,
+            {"\n": "\r\n", "\r\n": "\n"}[os.linesep],
+            id="native",
+        ),
+    ),
+)
+def test_override_newline(
+    runner, gen_hashes, annotate_options, nl_options, must_include, must_exclude
+):
+    opts = annotate_options + nl_options
+    if gen_hashes:
+        opts += ("--generate-hashes",)
+
+    with open("requirements.in", "w") as req_in:
+        req_in.write("six==1.15.0\n")
+        req_in.write("setuptools\n")
+        req_in.write("pip-tools @ git+https://github.com/jazzband/pip-tools\n")
+
+    runner.invoke(cli, [*opts, "requirements.in"])
+    with open("requirements.txt", "rb") as req_txt:
+        txt = req_txt.read().decode()
+
+    assert must_include in txt
+
+    if must_exclude in must_include:
+        txt = txt.replace(must_include, "")
+    assert must_exclude not in txt
+
+    # Do it again, with --newline=preserve:
+
+    opts = annotate_options + ("--newline", "preserve")
+    if gen_hashes:
+        opts += ("--generate-hashes",)
+
+    runner.invoke(cli, [*opts, "requirements.in"])
+    with open("requirements.txt", "rb") as req_txt:
+        txt = req_txt.read().decode()
+
+    assert must_include in txt
+
+    if must_exclude in must_include:
+        txt = txt.replace(must_include, "")
+    assert must_exclude not in txt
+
+
+@pytest.mark.network
+@pytest.mark.parametrize(
+    ("linesep", "must_exclude"),
+    (pytest.param("\n", "\r\n", id="LF"), pytest.param("\r\n", "\n", id="CRLF")),
+)
+def test_preserve_newline_from_input(runner, linesep, must_exclude):
+    with open("requirements.in", "wb") as req_in:
+        req_in.write(f"six{linesep}".encode())
+
+    runner.invoke(cli, ["--newline=preserve", "requirements.in"])
+    with open("requirements.txt", "rb") as req_txt:
+        txt = req_txt.read().decode()
+
+    assert linesep in txt
+
+    if must_exclude in linesep:
+        txt = txt.replace(linesep, "")
+    assert must_exclude not in txt
+
+
 @pytest.mark.network
 def test_generate_hashes_with_split_style_annotations(runner):
     with open("requirements.in", "w") as fp:

tests/test_writer.py

@@ -42,6 +42,7 @@ def writer(tmpdir_cwd):
             index_urls=[],
             trusted_hosts=[],
             format_control=FormatControl(set(), set()),
+            linesep="\n",
             allow_unsafe=False,
             find_links=[],
             emit_find_links=True,