pip-compile: Use the same newline string already found in relevant files (if impossible use LF), or override with --force-lf-newlines

piptools/scripts/compile.py

@@ -49,6 +49,26 @@ def _get_default_option(option_name: str) -> Any:
     return getattr(default_values, option_name)
 
 
+def _existing_linesep(*filenames: str) -> str:
+    """
+    Check files in order for an existing linesep and return it, if possible.
+    Otherwise, return LF ("\n").
+    """
+    linesep = "\n"
+    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:
+            linesep = "\r\n"
+            break
+        elif b"\n" in existing_text:
+            break
+    return linesep
+
+
 @click.command(context_settings={"help_option_names": ("-h", "--help")})
 @click.version_option(**version_option_kwargs)
 @click.pass_context
@@ -165,6 +185,12 @@ def _get_default_option(option_name: str) -> Any:
         "Will be derived from input file otherwise."
     ),
 )
+@click.option(
+    "--force-lf-newlines",
+    is_flag=True,
+    default=False,
+    help="Always use LF newlines, rather than auto-detecting from existing files.",
+)
 @click.option(
     "--allow-unsafe/--no-allow-unsafe",
     is_flag=True,
@@ -279,6 +305,7 @@ def cli(
     upgrade: bool,
     upgrade_packages: Tuple[str, ...],
     output_file: Union[LazyFile, IO[Any], None],
+    force_lf_newlines: bool,
     allow_unsafe: bool,
     strip_extras: bool,
     generate_hashes: bool,
@@ -515,6 +542,10 @@ def cli(
 
     log.debug("")
 
+    linesep = (
+        "\n" if force_lf_newlines else _existing_linesep(output_file.name, *src_files)
+    )
+
     ##
     # Output
     ##
@@ -534,6 +565,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,70 @@ 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"),
+    ),
+)
+def test_force_lf_newlines(runner, gen_hashes, annotate_options):
+    opts = (*annotate_options, "--force-lf-newlines")
+    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 "\n" in txt
+    assert "\r\n" 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_newlines_from_output_or_input(runner, linesep, must_exclude):
+    with open("requirements.in", "wb") as req_in:
+        req_in.write(f"six{linesep}".encode())
+
+    runner.invoke(cli, ["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
+
+    # Now that we have good output,
+    # see that it's preserved when we have bad input:
+
+    with open("requirements.in", "wb") as req_in:
+        req_in.write(f"six{must_exclude}".encode())
+
+    runner.invoke(cli, ["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,