def ensure_dir(dir_path):
"""Ensure a directory with 755 permissions mode."""
- if not os.path.isdir(dir_path):
- try:
- os.makedirs(dir_path, 0o755)
- except OSError as e:
- # Make sure that the error was that the directory was created
- # by a different (concurrent) worker. If not, raise the error.
- if e.errno != errno.EEXIST:
- raise
+ try:
+ os.makedirs(dir_path, 0o755)
+ except OSError as e:
+ # If the directory already existed, don't raise the error.
+ if e.errno != errno.EEXIST:
+ raise
def _get_conf_base(cfg_root, uuid, ensure_conf_dir):
self.execute = self.execute_p.start()
self.makedirs = mock.patch('os.makedirs').start()
- self.isdir = mock.patch('os.path.isdir').start()
- self.isdir.return_value = False
self.rmtree = mock.patch('shutil.rmtree').start()
self.external_process = mock.patch(
getgrgid.assert_called_once_with(self.EGID)
@mock.patch('os.makedirs')
- @mock.patch('os.path.exists', return_value=False)
- def test_ensure_dir_no_fail_if_exists(self, path_exists, makedirs):
+ def test_ensure_dir_no_fail_if_exists(self, makedirs):
error = OSError()
error.errno = errno.EEXIST
makedirs.side_effect = error
utils.ensure_dir("/etc/create/concurrently")
+ @mock.patch('os.makedirs')
+ def test_ensure_dir_calls_makedirs(self, makedirs):
+ utils.ensure_dir("/etc/create/directory")
+ makedirs.assert_called_once_with("/etc/create/directory", 0o755)
+
class TestUnixDomainHttpConnection(base.BaseTestCase):
def test_connect(self):