Add helper functions for work with temp dirs and files

* Add function get_tempname required for create temporary filename
It has optional arguments: dir and prefix:
dir setup a directory where this file will be created
prefix setup a prefix value of the file

Usage:

    from octane.util import tempfile

    tmp_name = tempfile.get_tempname()

* Adding of contextmanager temp_dir is required for creating of tempdir
before code block and removing it upon completion of the block.

Usage:

    from octane.util import tempfile

    with tempfile.temp_dir() as my_tmp_dir:
        # code block
        ...

Change-Id: I20b8a0109a85ef41708fa8c9e7e050b391ed039e
This commit is contained in:
Sergey Abramov 2016-06-09 13:06:00 +03:00
parent edc56bf7dc
commit a15d657e65
4 changed files with 94 additions and 6 deletions

View File

@ -0,0 +1,53 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import mock
import pytest
from octane.util import tempfile
@pytest.mark.parametrize("dir", ["dir_1", "dir_2", None])
@pytest.mark.parametrize("prefix", ["prefix_1", "prefix_2", None])
def test_get_tempname(mocker, dir, prefix):
fd = mock.Mock()
tmp_file_name = mock.Mock()
mock_mkstemp = mocker.patch(
"tempfile.mkstemp",
return_value=(fd, tmp_file_name))
os_close_mock = mocker.patch("os.close")
assert tmp_file_name == tempfile.get_tempname(dir, prefix)
if prefix:
mock_mkstemp.assert_called_once_with(dir=dir, prefix=prefix)
else:
mock_mkstemp.assert_called_once_with(dir=dir)
os_close_mock.assert_called_once_with(fd)
@pytest.mark.parametrize("is_exception", [True, False])
def test_temp_dir(mocker, is_exception):
class TestException(Exception):
pass
temp_dir_name = mock.Mock()
mkdtemp_mock = mocker.patch("tempfile.mkdtemp", return_value=temp_dir_name)
rm_tree_mock = mocker.patch("shutil.rmtree")
if is_exception:
with pytest.raises(TestException):
with tempfile.temp_dir():
raise TestException
else:
with tempfile.temp_dir():
pass
mkdtemp_mock.assert_called_once_with()
rm_tree_mock.assert_called_once_with(temp_dir_name)

View File

@ -10,6 +10,8 @@
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
import contextlib
import io
import itertools

View File

@ -10,22 +10,19 @@
# License for the specific language governing permissions and limitations
# under the License.
import os
import shutil
import tempfile
import yaml
import contextlib
from octane.util import helpers
from octane.util import tempfile
@contextlib.contextmanager
def set_astute_password(auth_context):
fd, tmp_file_name = tempfile.mkstemp(
dir="/etc/fuel",
prefix=".astute.yaml.octane")
os.close(fd)
tmp_file_name = tempfile.get_tempname(
dir="/etc/fuel", prefix=".astute.yaml.octane")
shutil.copy2("/etc/fuel/astute.yaml", tmp_file_name)
try:
data = helpers.get_astute_dict()

36
octane/util/tempfile.py Normal file
View File

@ -0,0 +1,36 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
import contextlib
import os
import shutil
import tempfile
def get_tempname(dir=None, prefix=None):
kwargs = {}
if prefix is not None:
kwargs["prefix"] = prefix
fd, tmp_file_name = tempfile.mkstemp(dir=dir, **kwargs)
os.close(fd)
return tmp_file_name
@contextlib.contextmanager
def temp_dir():
temp_dir = tempfile.mkdtemp()
try:
yield temp_dir
finally:
shutil.rmtree(temp_dir)