Produits Entr'ouvert » Passerelle

# Copyright (C) 2018  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import os

import contextlib

import tempfile

import stat

@contextlib.contextmanager

def atomic_write(filepath, **kwargs):

    '''Return a file descriptor to a temporary file using NamedTemporaryFile

       which will be atomically renamed to filepath if possible.

       Atomic renaming is only possible on the same filesystem, so the

       temporary file will be created in the same directory as the target file

       You can pass any possible argument to NamedTemporaryFile with kwargs.

    '''

    target_dir = kwargs.get('dir') or os.path.dirname(filepath)

    if not os.path.exists(target_dir):

        os.makedirs(target_dir)

    fd = tempfile.NamedTemporaryFile(dir=target_dir, delete=False)

    try:

        with fd:

            yield fd

            fd.flush()

            os.fsync(fd.fileno())

        os.rename(fd.name, filepath)

        # set read only permission for owner and group

        os.chmod(filepath, stat.S_IRUSR|stat.S_IRGRP)

    except Exception:

        os.unlink(fd.name)

        raise

# Copyright (C) 2018  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import os

import stat

import pytest

from passerelle.utils.files import atomic_write

def test_atomic_write(tmpdir):

    filepath = str(tmpdir.join('test'))

    with pytest.raises(Exception):

        with atomic_write(filepath) as fd:

            fd.write('coucou')

            raise Exception()

    assert not os.path.exists(filepath)

    assert os.listdir(str(tmpdir)) == []

    with atomic_write(filepath) as fd:

        fd.write('coucou')

    assert os.path.exists(filepath)

    # make sure only owner and group can read file

    assert bool(os.stat(filepath).st_mode & stat.S_IRUSR)

    assert bool(os.stat(filepath).st_mode & stat.S_IRGRP)

    with open(filepath) as fd:

        assert fd.read() == 'coucou'

def test_atomic_write_in_subdirectory(tmpdir):

    filepath = str(tmpdir.join('subdir', 'test1'))

    with atomic_write(filepath) as fd:

        fd.write('coucou')

    with open(filepath) as fd:

        assert fd.read() == 'coucou'