move utils.io.atomic_writing to contents.fileio

IPython/html/services/contents/tests/test_fileio.py

@@ -0,0 +1,131 b''
+# encoding: utf-8
+"""Tests for file IO"""
+
+# Copyright (c) IPython Development Team.
+# Distributed under the terms of the Modified BSD License.
+
+import io as stdlib_io
+import os.path
+import stat
+
+import nose.tools as nt
+
+from IPython.testing.decorators import skip_win32
+from ..fileio import atomic_writing
+
+from IPython.utils.tempdir import TemporaryDirectory
+
+umask = 0
+
+def test_atomic_writing():
+    class CustomExc(Exception): pass
+
+    with TemporaryDirectory() as td:
+        f1 = os.path.join(td, 'penguin')
+        with stdlib_io.open(f1, 'w') as f:
+            f.write(u'Before')
+        
+        if os.name != 'nt':
+            os.chmod(f1, 0o701)
+            orig_mode = stat.S_IMODE(os.stat(f1).st_mode)
+
+        f2 = os.path.join(td, 'flamingo')
+        try:
+            os.symlink(f1, f2)
+            have_symlink = True
+        except (AttributeError, NotImplementedError, OSError):
+            # AttributeError: Python doesn't support it
+            # NotImplementedError: The system doesn't support it
+            # OSError: The user lacks the privilege (Windows)
+            have_symlink = False
+
+        with nt.assert_raises(CustomExc):
+            with atomic_writing(f1) as f:
+                f.write(u'Failing write')
+                raise CustomExc
+
+        # Because of the exception, the file should not have been modified
+        with stdlib_io.open(f1, 'r') as f:
+            nt.assert_equal(f.read(), u'Before')
+
+        with atomic_writing(f1) as f:
+            f.write(u'Overwritten')
+
+        with stdlib_io.open(f1, 'r') as f:
+            nt.assert_equal(f.read(), u'Overwritten')
+
+        if os.name != 'nt':
+            mode = stat.S_IMODE(os.stat(f1).st_mode)
+            nt.assert_equal(mode, orig_mode)
+
+        if have_symlink:
+            # Check that writing over a file preserves a symlink
+            with atomic_writing(f2) as f:
+                f.write(u'written from symlink')
+            
+            with stdlib_io.open(f1, 'r') as f:
+                nt.assert_equal(f.read(), u'written from symlink')
+
+def _save_umask():
+    global umask
+    umask = os.umask(0)
+    os.umask(umask)
+
+def _restore_umask():
+    os.umask(umask)
+
+@skip_win32
[e-mail]_setup(_save_umask, _restore_umask)
+def test_atomic_writing_umask():
+    with TemporaryDirectory() as td:
+        os.umask(0o022)
+        f1 = os.path.join(td, '1')
+        with atomic_writing(f1) as f:
+            f.write(u'1')
+        mode = stat.S_IMODE(os.stat(f1).st_mode)
+        nt.assert_equal(mode, 0o644, '{:o} != 644'.format(mode))
+
+        os.umask(0o057)
+        f2 = os.path.join(td, '2')
+        with atomic_writing(f2) as f:
+            f.write(u'2')
+        mode = stat.S_IMODE(os.stat(f2).st_mode)
+        nt.assert_equal(mode, 0o620, '{:o} != 620'.format(mode))
+
+
+def test_atomic_writing_newlines():
+    with TemporaryDirectory() as td:
+        path = os.path.join(td, 'testfile')
+        
+        lf = u'a\nb\nc\n'
+        plat = lf.replace(u'\n', os.linesep)
+        crlf = lf.replace(u'\n', u'\r\n')
+        
+        # test default
+        with stdlib_io.open(path, 'w') as f:
+            f.write(lf)
+        with stdlib_io.open(path, 'r', newline='') as f:
+            read = f.read()
+        nt.assert_equal(read, plat)
+        
+        # test newline=LF
+        with stdlib_io.open(path, 'w', newline='\n') as f:
+            f.write(lf)
+        with stdlib_io.open(path, 'r', newline='') as f:
+            read = f.read()
+        nt.assert_equal(read, lf)
+        
+        # test newline=CRLF
+        with atomic_writing(path, newline='\r\n') as f:
+            f.write(lf)
+        with stdlib_io.open(path, 'r', newline='') as f:
+            read = f.read()
+        nt.assert_equal(read, crlf)
+        
+        # test newline=no convert
+        text = u'crlf\r\ncr\rlf\n'
+        with atomic_writing(path, newline='') as f:
+            f.write(text)
+        with stdlib_io.open(path, 'r', newline='') as f:
+            read = f.read()
+        nt.assert_equal(read, text)

IPython/html/services/contents/fileio.py

@@ -1,174 +1,256 b''
 """
 Utilities for file-based Contents/Checkpoints managers.
 """
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 import base64
 from contextlib import contextmanager
 import errno
 import io
 import os
 import shutil
+import tempfile
 
 from tornado.web import HTTPError
 
 from IPython.html.utils import (
     to_api_path,
     to_os_path,
 )
 from IPython import nbformat
-from IPython.utils.io import atomic_writing
 from IPython.utils.py3compat import str_to_unicode
 
 
+def _copy_metadata(src, dst):
+    """Copy the set of metadata we want for atomic_writing.
+    
+    Permission bits and flags. We'd like to copy file ownership as well, but we
+    can't do that.
+    """
+    shutil.copymode(src, dst)
+    st = os.stat(src)
+    if hasattr(os, 'chflags') and hasattr(st, 'st_flags'):
+        os.chflags(dst, st.st_flags)
+
+@contextmanager
+def atomic_writing(path, text=True, encoding='utf-8', **kwargs):
+    """Context manager to write to a file only if the entire write is successful.
+    
+    This works by creating a temporary file in the same directory, and renaming
+    it over the old file if the context is exited without an error. If other
+    file names are hard linked to the target file, this relationship will not be
+    preserved.
+    
+    On Windows, there is a small chink in the atomicity: the target file is
+    deleted before renaming the temporary file over it. This appears to be
+    unavoidable.
+    
+    Parameters
+    ----------
+    path : str
+      The target file to write to.
+     
+    text : bool, optional
+      Whether to open the file in text mode (i.e. to write unicode). Default is
+      True.
+    
+    encoding : str, optional
+      The encoding to use for files opened in text mode. Default is UTF-8.
+     
+    **kwargs
+      Passed to :func:`io.open`.
+    """
+    # realpath doesn't work on Windows: http://bugs.python.org/issue9949
+    # Luckily, we only need to resolve the file itself being a symlink, not
+    # any of its directories, so this will suffice:
+    if os.path.islink(path):
+        path = os.path.join(os.path.dirname(path), os.readlink(path))
+
+    dirname, basename = os.path.split(path)
+    tmp_dir = tempfile.mkdtemp(prefix=basename, dir=dirname)
+    tmp_path = os.path.join(tmp_dir, basename)
+    if text:
+        fileobj = io.open(tmp_path, 'w', encoding=encoding, **kwargs)
+    else:
+        fileobj = io.open(tmp_path, 'wb', **kwargs)
+
+    try:
+        yield fileobj
+    except:
+        fileobj.close()
+        shutil.rmtree(tmp_dir)
+        raise
+
+    # Flush to disk
+    fileobj.flush()
+    os.fsync(fileobj.fileno())
+
+    # Written successfully, now rename it
+    fileobj.close()
+
+    # Copy permission bits, access time, etc.
+    try:
+        _copy_metadata(path, tmp_path)
+    except OSError:
+        # e.g. the file didn't already exist. Ignore any failure to copy metadata
+        pass
+
+    if os.name == 'nt' and os.path.exists(path):
+        # Rename over existing file doesn't work on Windows
+        os.remove(path)
+
+    os.rename(tmp_path, path)
+    shutil.rmtree(tmp_dir)
+
+
 class FileManagerMixin(object):
     """
     Mixin for ContentsAPI classes that interact with the filesystem.
 
     Provides facilities for reading, writing, and copying both notebooks and
     generic files.
 
     Shared by FileContentsManager and FileCheckpoints.
 
     Note
     ----
     Classes using this mixin must provide the following attributes:
 
     root_dir : unicode
         A directory against against which API-style paths are to be resolved.
 
     log : logging.Logger
     """
 
     @contextmanager
     def open(self, os_path, *args, **kwargs):
         """wrapper around io.open that turns permission errors into 403"""
         with self.perm_to_403(os_path):
             with io.open(os_path, *args, **kwargs) as f:
                 yield f
 
     @contextmanager
     def atomic_writing(self, os_path, *args, **kwargs):
         """wrapper around atomic_writing that turns permission errors to 403"""
         with self.perm_to_403(os_path):
             with atomic_writing(os_path, *args, **kwargs) as f:
                 yield f
 
     @contextmanager
     def perm_to_403(self, os_path=''):
         """context manager for turning permission errors into 403."""
         try:
             yield
         except (OSError, IOError) as e:
             if e.errno in {errno.EPERM, errno.EACCES}:
                 # make 403 error message without root prefix
                 # this may not work perfectly on unicode paths on Python 2,
                 # but nobody should be doing that anyway.
                 if not os_path:
                     os_path = str_to_unicode(e.filename or 'unknown file')
                 path = to_api_path(os_path, root=self.root_dir)
                 raise HTTPError(403, u'Permission denied: %s' % path)
             else:
                 raise
 
     def _copy(self, src, dest):
         """copy src to dest
 
         like shutil.copy2, but log errors in copystat
         """
         shutil.copyfile(src, dest)
         try:
             shutil.copystat(src, dest)
         except OSError:
             self.log.debug("copystat on %s failed", dest, exc_info=True)
 
     def _get_os_path(self, path):
         """Given an API path, return its file system path.
 
         Parameters
         ----------
         path : string
             The relative API path to the named file.
 
         Returns
         -------
         path : string
             Native, absolute OS path to for a file.
 
         Raises
         ------
         404: if path is outside root
         """
         root = os.path.abspath(self.root_dir)
         os_path = to_os_path(path, root)
         if not (os.path.abspath(os_path) + os.path.sep).startswith(root):
             raise HTTPError(404, "%s is outside root contents directory" % path)
         return os_path
 
     def _read_notebook(self, os_path, as_version=4):
         """Read a notebook from an os path."""
         with self.open(os_path, 'r', encoding='utf-8') as f:
             try:
                 return nbformat.read(f, as_version=as_version)
             except Exception as e:
                 raise HTTPError(
                     400,
                     u"Unreadable Notebook: %s %r" % (os_path, e),
                 )
 
     def _save_notebook(self, os_path, nb):
         """Save a notebook to an os_path."""
         with self.atomic_writing(os_path, encoding='utf-8') as f:
             nbformat.write(nb, f, version=nbformat.NO_CONVERT)
 
     def _read_file(self, os_path, format):
         """Read a non-notebook file.
 
         os_path: The path to be read.
         format:
           If 'text', the contents will be decoded as UTF-8.
           If 'base64', the raw bytes contents will be encoded as base64.
           If not specified, try to decode as UTF-8, and fall back to base64
         """
         if not os.path.isfile(os_path):
             raise HTTPError(400, "Cannot read non-file %s" % os_path)
 
         with self.open(os_path, 'rb') as f:
             bcontent = f.read()
 
         if format is None or format == 'text':
             # Try to interpret as unicode if format is unknown or if unicode
             # was explicitly requested.
             try:
                 return bcontent.decode('utf8'), 'text'
             except UnicodeError:
                 if format == 'text':
                     raise HTTPError(
                         400,
                         "%s is not UTF-8 encoded" % os_path,
                         reason='bad format',
                     )
         return base64.encodestring(bcontent).decode('ascii'), 'base64'
 
     def _save_file(self, os_path, content, format):
         """Save content of a generic file."""
         if format not in {'text', 'base64'}:
             raise HTTPError(
                 400,
                 "Must specify format of file contents as 'text' or 'base64'",
             )
         try:
             if format == 'text':
                 bcontent = content.encode('utf8')
             else:
                 b64_bytes = content.encode('ascii')
                 bcontent = base64.decodestring(b64_bytes)
         except Exception as e:
             raise HTTPError(
                 400, u'Encoding error saving %s: %s' % (os_path, e)
             )
 
         with self.atomic_writing(os_path, text=False) as f:
             f.write(bcontent)

IPython/utils/io.py

@@ -1,322 +1,246 b''
 # encoding: utf-8
 """
 IO related utilities.
 """
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 from __future__ import print_function
 from __future__ import absolute_import
 
 
 import codecs
 from contextlib import contextmanager
 import io
 import os
 import shutil
 import sys
 import tempfile
 import warnings
 from .capture import CapturedIO, capture_output
 from .py3compat import string_types, input, PY3
 
 
 class IOStream:
 
     def __init__(self,stream, fallback=None):
         if not hasattr(stream,'write') or not hasattr(stream,'flush'):
             if fallback is not None:
                 stream = fallback
             else:
                 raise ValueError("fallback required, but not specified")
         self.stream = stream
         self._swrite = stream.write
 
         # clone all methods not overridden:
         def clone(meth):
             return not hasattr(self, meth) and not meth.startswith('_')
         for meth in filter(clone, dir(stream)):
             setattr(self, meth, getattr(stream, meth))
 
     def __repr__(self):
         cls = self.__class__
         tpl = '{mod}.{cls}({args})'
         return tpl.format(mod=cls.__module__, cls=cls.__name__, args=self.stream)
 
     def write(self,data):
         try:
             self._swrite(data)
         except:
             try:
                 # print handles some unicode issues which may trip a plain
                 # write() call.  Emulate write() by using an empty end
                 # argument.
                 print(data, end='', file=self.stream)
             except:
                 # if we get here, something is seriously broken.
                 print('ERROR - failed to write data to stream:', self.stream,
                       file=sys.stderr)
 
     def writelines(self, lines):
         if isinstance(lines, string_types):
             lines = [lines]
         for line in lines:
             self.write(line)
 
     # This class used to have a writeln method, but regular files and streams
     # in Python don't have this method. We need to keep this completely
     # compatible so we removed it.
 
     @property
     def closed(self):
         return self.stream.closed
 
     def close(self):
         pass
 
 # setup stdin/stdout/stderr to sys.stdin/sys.stdout/sys.stderr
 devnull = open(os.devnull, 'w') 
 stdin = IOStream(sys.stdin, fallback=devnull)
 stdout = IOStream(sys.stdout, fallback=devnull)
 stderr = IOStream(sys.stderr, fallback=devnull)
 
 class IOTerm:
     """ Term holds the file or file-like objects for handling I/O operations.
 
     These are normally just sys.stdin, sys.stdout and sys.stderr but for
     Windows they can can replaced to allow editing the strings before they are
     displayed."""
 
     # In the future, having IPython channel all its I/O operations through
     # this class will make it easier to embed it into other environments which
     # are not a normal terminal (such as a GUI-based shell)
     def __init__(self, stdin=None, stdout=None, stderr=None):
         mymodule = sys.modules[__name__]
         self.stdin  = IOStream(stdin, mymodule.stdin)
         self.stdout = IOStream(stdout, mymodule.stdout)
         self.stderr = IOStream(stderr, mymodule.stderr)
 
 
 class Tee(object):
     """A class to duplicate an output stream to stdout/err.
 
     This works in a manner very similar to the Unix 'tee' command.
 
     When the object is closed or deleted, it closes the original file given to
     it for duplication.
     """
     # Inspired by:
     # http://mail.python.org/pipermail/python-list/2007-May/442737.html
 
     def __init__(self, file_or_name, mode="w", channel='stdout'):
         """Construct a new Tee object.
 
         Parameters
         ----------
         file_or_name : filename or open filehandle (writable)
           File that will be duplicated
 
         mode : optional, valid mode for open().
           If a filename was give, open with this mode.
 
         channel : str, one of ['stdout', 'stderr']
         """
         if channel not in ['stdout', 'stderr']:
             raise ValueError('Invalid channel spec %s' % channel)
 
         if hasattr(file_or_name, 'write') and hasattr(file_or_name, 'seek'):
             self.file = file_or_name
         else:
             self.file = open(file_or_name, mode)
         self.channel = channel
         self.ostream = getattr(sys, channel)
         setattr(sys, channel, self)
         self._closed = False
 
     def close(self):
         """Close the file and restore the channel."""
         self.flush()
         setattr(sys, self.channel, self.ostream)
         self.file.close()
         self._closed = True
 
     def write(self, data):
         """Write data to both channels."""
         self.file.write(data)
         self.ostream.write(data)
         self.ostream.flush()
 
     def flush(self):
         """Flush both channels."""
         self.file.flush()
         self.ostream.flush()
 
     def __del__(self):
         if not self._closed:
             self.close()
 
 
 def ask_yes_no(prompt, default=None, interrupt=None):
     """Asks a question and returns a boolean (y/n) answer.
 
     If default is given (one of 'y','n'), it is used if the user input is
     empty. If interrupt is given (one of 'y','n'), it is used if the user
     presses Ctrl-C. Otherwise the question is repeated until an answer is
     given.
 
     An EOF is treated as the default answer.  If there is no default, an
     exception is raised to prevent infinite loops.
 
     Valid answers are: y/yes/n/no (match is not case sensitive)."""
 
     answers = {'y':True,'n':False,'yes':True,'no':False}
     ans = None
     while ans not in answers.keys():
         try:
             ans = input(prompt+' ').lower()
             if not ans:  # response was an empty string
                 ans = default
         except KeyboardInterrupt:
             if interrupt:
                 ans = interrupt
         except EOFError:
             if default in answers.keys():
                 ans = default
                 print()
             else:
                 raise
 
     return answers[ans]
 
 
 def temp_pyfile(src, ext='.py'):
     """Make a temporary python file, return filename and filehandle.
 
     Parameters
     ----------
     src : string or list of strings (no need for ending newlines if list)
       Source code to be written to the file.
 
     ext : optional, string
       Extension for the generated file.
 
     Returns
     -------
     (filename, open filehandle)
       It is the caller's responsibility to close the open file and unlink it.
     """
     fname = tempfile.mkstemp(ext)[1]
     f = open(fname,'w')
     f.write(src)
     f.flush()
     return fname, f
 
-def _copy_metadata(src, dst):
-    """Copy the set of metadata we want for atomic_writing.
-    
-    Permission bits and flags. We'd like to copy file ownership as well, but we
-    can't do that.
-    """
-    shutil.copymode(src, dst)
-    st = os.stat(src)
-    if hasattr(os, 'chflags') and hasattr(st, 'st_flags'):
-        os.chflags(dst, st.st_flags)
-
-@contextmanager
-def atomic_writing(path, text=True, encoding='utf-8', **kwargs):
-    """Context manager to write to a file only if the entire write is successful.
-    
-    This works by creating a temporary file in the same directory, and renaming
-    it over the old file if the context is exited without an error. If other
-    file names are hard linked to the target file, this relationship will not be
-    preserved.
-    
-    On Windows, there is a small chink in the atomicity: the target file is
-    deleted before renaming the temporary file over it. This appears to be
-    unavoidable.
-    
-    Parameters
-    ----------
-    path : str
-      The target file to write to.
-     
-    text : bool, optional
-      Whether to open the file in text mode (i.e. to write unicode). Default is
-      True.
-    
-    encoding : str, optional
-      The encoding to use for files opened in text mode. Default is UTF-8.
-     
-    **kwargs
-      Passed to :func:`io.open`.
-    """
-    # realpath doesn't work on Windows: http://bugs.python.org/issue9949
-    # Luckily, we only need to resolve the file itself being a symlink, not
-    # any of its directories, so this will suffice:
-    if os.path.islink(path):
-        path = os.path.join(os.path.dirname(path), os.readlink(path))
-
-    dirname, basename = os.path.split(path)
-    tmp_dir = tempfile.mkdtemp(prefix=basename, dir=dirname)
-    tmp_path = os.path.join(tmp_dir, basename)
-    if text:
-        fileobj = io.open(tmp_path, 'w', encoding=encoding, **kwargs)
-    else:
-        fileobj = io.open(tmp_path, 'wb', **kwargs)
-
-    try:
-        yield fileobj
-    except:
-        fileobj.close()
-        shutil.rmtree(tmp_dir)
-        raise
-
-    # Flush to disk
-    fileobj.flush()
-    os.fsync(fileobj.fileno())
-
-    # Written successfully, now rename it
-    fileobj.close()
-
-    # Copy permission bits, access time, etc.
-    try:
-        _copy_metadata(path, tmp_path)
-    except OSError:
-        # e.g. the file didn't already exist. Ignore any failure to copy metadata
-        pass
-
-    if os.name == 'nt' and os.path.exists(path):
-        # Rename over existing file doesn't work on Windows
-        os.remove(path)
-
-    os.rename(tmp_path, path)
-    shutil.rmtree(tmp_dir)
-
+def atomic_writing(*args, **kwargs):
+    """DEPRECATED: moved to IPython.html.services.contents.fileio"""
+    warn("IPython.utils.io.atomic_writing has moved to IPython.html.services.contents.fileio")
+    from IPython.html.services.contents.fileio import atomic_writing
+    return atomic_writing(*args, **kwargs)
 
 def raw_print(*args, **kw):
     """Raw print to sys.__stdout__, otherwise identical interface to print()."""
 
     print(*args, sep=kw.get('sep', ' '), end=kw.get('end', '\n'),
           file=sys.__stdout__)
     sys.__stdout__.flush()
 
 
 def raw_print_err(*args, **kw):
     """Raw print to sys.__stderr__, otherwise identical interface to print()."""
 
     print(*args, sep=kw.get('sep', ' '), end=kw.get('end', '\n'),
           file=sys.__stderr__)
     sys.__stderr__.flush()
 
 
 # Short aliases for quick debugging, do NOT use these in production code.
 rprint = raw_print
 rprinte = raw_print_err
 
 
 def unicode_std_stream(stream='stdout'):
     """DEPRECATED, moved to jupyter_nbconvert.utils.io"""
     warn("IPython.utils.io.unicode_std_stream has moved to jupyter_nbconvert.utils.io")
     from jupyter_nbconvert.utils.io import unicode_std_stream
     return unicode_std_stream(stream)

IPython/utils/tests/test_io.py

@@ -1,201 +1,87 b''
 # encoding: utf-8
 """Tests for io.py"""
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 from __future__ import print_function
 from __future__ import absolute_import
 
 import io as stdlib_io
 import os.path
 import stat
 import sys
 
 from subprocess import Popen, PIPE
 import unittest
 
 import nose.tools as nt
 
 from IPython.testing.decorators import skipif, skip_win32
-from IPython.utils.io import (Tee, capture_output, unicode_std_stream,
-                              atomic_writing,
-                              )
+from IPython.utils.io import Tee, capture_output
 from IPython.utils.py3compat import doctest_refactor_print, PY3
 from IPython.utils.tempdir import TemporaryDirectory
 
 if PY3:
     from io import StringIO
 else:
     from StringIO import StringIO
 
 
 def test_tee_simple():
     "Very simple check with stdout only"
     chan = StringIO()
     text = 'Hello'
     tee = Tee(chan, channel='stdout')
     print(text, file=chan)
     nt.assert_equal(chan.getvalue(), text+"\n")
 
 
 class TeeTestCase(unittest.TestCase):
 
     def tchan(self, channel, check='close'):
         trap = StringIO()
         chan = StringIO()
         text = 'Hello'
         
         std_ori = getattr(sys, channel)
         setattr(sys, channel, trap)
 
         tee = Tee(chan, channel=channel)
         print(text, end='', file=chan)
         setattr(sys, channel, std_ori)
         trap_val = trap.getvalue()
         nt.assert_equal(chan.getvalue(), text)
         if check=='close':
             tee.close()
         else:
             del tee
 
     def test(self):
         for chan in ['stdout', 'stderr']:
             for check in ['close', 'del']:
                 self.tchan(chan, check)
 
 def test_io_init():
     """Test that io.stdin/out/err exist at startup"""
     for name in ('stdin', 'stdout', 'stderr'):
         cmd = doctest_refactor_print("from IPython.utils import io;print io.%s.__class__"%name)
         p = Popen([sys.executable, '-c', cmd],
                     stdout=PIPE)
         p.wait()
         classname = p.stdout.read().strip().decode('ascii')
         # __class__ is a reference to the class object in Python 3, so we can't
         # just test for string equality.
         assert 'IPython.utils.io.IOStream' in classname, classname
 
 def test_capture_output():
     """capture_output() context works"""
     
     with capture_output() as io:
         print('hi, stdout')
         print('hi, stderr', file=sys.stderr)
     
     nt.assert_equal(io.stdout, 'hi, stdout\n')
     nt.assert_equal(io.stderr, 'hi, stderr\n')
 
 
-def test_atomic_writing():
-    class CustomExc(Exception): pass
-
-    with TemporaryDirectory() as td:
-        f1 = os.path.join(td, 'penguin')
-        with stdlib_io.open(f1, 'w') as f:
-            f.write(u'Before')
-        
-        if os.name != 'nt':
-            os.chmod(f1, 0o701)
-            orig_mode = stat.S_IMODE(os.stat(f1).st_mode)
-
-        f2 = os.path.join(td, 'flamingo')
-        try:
-            os.symlink(f1, f2)
-            have_symlink = True
-        except (AttributeError, NotImplementedError, OSError):
-            # AttributeError: Python doesn't support it
-            # NotImplementedError: The system doesn't support it
-            # OSError: The user lacks the privilege (Windows)
-            have_symlink = False
-
-        with nt.assert_raises(CustomExc):
-            with atomic_writing(f1) as f:
-                f.write(u'Failing write')
-                raise CustomExc
-
-        # Because of the exception, the file should not have been modified
-        with stdlib_io.open(f1, 'r') as f:
-            nt.assert_equal(f.read(), u'Before')
-
-        with atomic_writing(f1) as f:
-            f.write(u'Overwritten')
-
-        with stdlib_io.open(f1, 'r') as f:
-            nt.assert_equal(f.read(), u'Overwritten')
-
-        if os.name != 'nt':
-            mode = stat.S_IMODE(os.stat(f1).st_mode)
-            nt.assert_equal(mode, orig_mode)
-
-        if have_symlink:
-            # Check that writing over a file preserves a symlink
-            with atomic_writing(f2) as f:
-                f.write(u'written from symlink')
-            
-            with stdlib_io.open(f1, 'r') as f:
-                nt.assert_equal(f.read(), u'written from symlink')
-
-def _save_umask():
-    global umask
-    umask = os.umask(0)
-    os.umask(umask)
-
-def _restore_umask():
-    os.umask(umask)
-
-@skip_win32
[e-mail]_setup(_save_umask, _restore_umask)
-def test_atomic_writing_umask():
-    with TemporaryDirectory() as td:
-        os.umask(0o022)
-        f1 = os.path.join(td, '1')
-        with atomic_writing(f1) as f:
-            f.write(u'1')
-        mode = stat.S_IMODE(os.stat(f1).st_mode)
-        nt.assert_equal(mode, 0o644, '{:o} != 644'.format(mode))
-
-        os.umask(0o057)
-        f2 = os.path.join(td, '2')
-        with atomic_writing(f2) as f:
-            f.write(u'2')
-        mode = stat.S_IMODE(os.stat(f2).st_mode)
-        nt.assert_equal(mode, 0o620, '{:o} != 620'.format(mode))
-
-
-def test_atomic_writing_newlines():
-    with TemporaryDirectory() as td:
-        path = os.path.join(td, 'testfile')
-        
-        lf = u'a\nb\nc\n'
-        plat = lf.replace(u'\n', os.linesep)
-        crlf = lf.replace(u'\n', u'\r\n')
-        
-        # test default
-        with stdlib_io.open(path, 'w') as f:
-            f.write(lf)
-        with stdlib_io.open(path, 'r', newline='') as f:
-            read = f.read()
-        nt.assert_equal(read, plat)
-        
-        # test newline=LF
-        with stdlib_io.open(path, 'w', newline='\n') as f:
-            f.write(lf)
-        with stdlib_io.open(path, 'r', newline='') as f:
-            read = f.read()
-        nt.assert_equal(read, lf)
-        
-        # test newline=CRLF
-        with atomic_writing(path, newline='\r\n') as f:
-            f.write(lf)
-        with stdlib_io.open(path, 'r', newline='') as f:
-            read = f.read()
-        nt.assert_equal(read, crlf)
-        
-        # test newline=no convert
-        text = u'crlf\r\ncr\rlf\n'
-        with atomic_writing(path, newline='') as f:
-            f.write(text)
-        with stdlib_io.open(path, 'r', newline='') as f:
-            read = f.read()
-        nt.assert_equal(read, text)