# License for the specific language governing permissions and limitations
# under the License.
+import os
+
from oslo_config import cfg
from six import moves
def all_children_active():
return all(pm.active for pm in self._child_processes)
+ for pm in self._child_processes:
+ directory = os.path.dirname(pm.get_pid_file_name())
+ self.assertEqual(0o755, os.stat(directory).st_mode & 0o777)
+
# we need to allow extra_time for the check process to happen
# and properly execute action over the gone processes under
# high load conditions
# License for the specific language governing permissions and limitations
# under the License.
-import os
-
import mock
import socket
import testtools
EGID = 456
EGNAME = 'group'
- @mock.patch.object(os.path, 'isdir', return_value=False)
- @mock.patch.object(os, 'makedirs')
- def test_ensure_dir_not_exist(self, makedirs, isdir):
- utils.ensure_dir('/the')
- isdir.assert_called_once_with('/the')
- makedirs.assert_called_once_with('/the', 0o755)
-
- @mock.patch.object(os.path, 'isdir', return_value=True)
- @mock.patch.object(os, 'makedirs')
- def test_ensure_dir_exist(self, makedirs, isdir):
- utils.ensure_dir('/the')
- isdir.assert_called_once_with('/the')
- self.assertFalse(makedirs.called)
-
@mock.patch('os.geteuid', return_value=EUID)
@mock.patch('pwd.getpwuid', return_value=FakeUser(EUNAME))
def test_is_effective_user_id(self, getpwuid, geteuid):