wip #1566

isort/api.py

@@ -318,7 +318,7 @@ def sort_file(
                 tmp_file = source_file.path.with_suffix(source_file.path.suffix + ".isorted")
                 try:
                     with tmp_file.open(
-                        "w", encoding=source_file.encoding, newline=""
+                        "w", encoding=source_file.encoding, newline=source_file.newline
                     ) as output_stream:
                         shutil.copymode(filename, tmp_file)
                         changed = sort_stream(
@@ -333,7 +333,7 @@ def sort_file(
                         if show_diff or ask_to_apply:
                             source_file.stream.seek(0)
                             with tmp_file.open(
-                                encoding=source_file.encoding, newline=""
+                                encoding=source_file.encoding, newline=source_file.newline
                             ) as tmp_out:
                                 show_unified_diff(
                                     file_input=source_file.stream.read(),

isort/io.py

@@ -1,10 +1,13 @@
 """Defines any IO utilities used by isort"""
+import io
 import re
 import tokenize
 from contextlib import contextmanager
 from io import BytesIO, StringIO, TextIOWrapper
 from pathlib import Path
-from typing import Callable, Iterator, NamedTuple, TextIO, Union
+from typing import BinaryIO
+from typing import Iterator, NamedTuple, TextIO, Union
+from typing import Tuple
 
 from isort.exceptions import UnsupportedEncoding
 
@@ -15,50 +18,45 @@ class File(NamedTuple):
     stream: TextIO
     path: Path
     encoding: str
+    newline: str
 
     @staticmethod
-    def detect_encoding(filename: str, readline: Callable[[], bytes]):
-        try:
-            return tokenize.detect_encoding(readline)[0]
-        except Exception:
-            raise UnsupportedEncoding(filename)
+    def decode_bytes(buffer: BinaryIO) -> Tuple[TextIO, str, str]:
+        encoding, lines = tokenize.detect_encoding(buffer.readline)
+        if not lines:
+            newline = "\n"
+        elif b"\r\n" == lines[0][-2:]:
+            newline = "\r\n"
+        elif b"\r" == lines[0][-1:]:
+            newline = "\r"
+        else:
+            newline = "\n"
+
+        buffer.seek(0)
+        text = io.TextIOWrapper(buffer, encoding, line_buffering=True)
+        return text, encoding, newline
 
     @staticmethod
     def from_contents(contents: str, filename: str) -> "File":
-        encoding = File.detect_encoding(filename, BytesIO(contents.encode("utf-8")).readline)
-        return File(StringIO(contents), path=Path(filename).resolve(), encoding=encoding)
+        text, encoding, newline = File.decode_bytes(BytesIO(contents.encode("utf-8")))
+        return File(StringIO(contents), path=Path(filename).resolve(), encoding=encoding, newline=newline)
 
     @property
     def extension(self):
         return self.path.suffix.lstrip(".")
 
-    @staticmethod
-    def _open(filename):
-        """Open a file in read only mode using the encoding detected by
-        detect_encoding().
-        """
-        buffer = open(filename, "rb")
-        try:
-            encoding = File.detect_encoding(filename, buffer.readline)
-            buffer.seek(0)
-            text = TextIOWrapper(buffer, encoding, line_buffering=True, newline="")
-            text.mode = "r"  # type: ignore
-            return text
-        except Exception:
-            buffer.close()
-            raise
-
     @staticmethod
     @contextmanager
     def read(filename: Union[str, Path]) -> Iterator["File"]:
         file_path = Path(filename).resolve()
-        stream = None
+        buffer = None
         try:
-            stream = File._open(file_path)
-            yield File(stream=stream, path=file_path, encoding=stream.encoding)
+            buffer = open(filename, "rb")
+            stream, encoding, newline = File.decode_bytes(buffer)
+            yield File(stream=stream, path=file_path, encoding=encoding, newline=newline)
         finally:
-            if stream is not None:
-                stream.close()
+            if buffer is not None:
+                buffer.close()
 
 
 class _EmptyIO(StringIO):

tests/unit/test_api.py

@@ -2,6 +2,7 @@
 import os
 import sys
 from io import StringIO
+from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -10,14 +11,15 @@
 from isort.settings import Config
 
 imperfect_content = "import b\nimport a\n"
-fixed_content = "import a\nimport b\n"
+fixed_content = f"import a\nimport b\n"
 fixed_diff = "+import a\n import b\n-import a\n"
 
 
 @pytest.fixture
-def imperfect(tmpdir) -> None:
+def imperfect(tmpdir) -> Path:
     imperfect_file = tmpdir.join("test_needs_changes.py")
-    imperfect_file.write_text(imperfect_content, "utf8")
+    with open(imperfect_file, mode="w", encoding="utf-8", newline=os.linesep) as f:
+        f.write(imperfect_content)
     return imperfect_file