Coverage for nova/privsep/path.py: 96%
56 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-04-17 15:08 +0000
« prev ^ index » next coverage.py v7.6.12, created at 2025-04-17 15:08 +0000
1# Copyright 2016 Red Hat, Inc
2# Copyright 2017 Rackspace Australia
3#
4# Licensed under the Apache License, Version 2.0 (the "License"); you may
5# not use this file except in compliance with the License. You may obtain
6# a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13# License for the specific language governing permissions and limitations
14# under the License.
16"""Routines that bypass file-system checks."""
18import errno
19import os
20import shutil
22from oslo_utils import fileutils
24from nova import exception
25import nova.privsep
28@nova.privsep.sys_admin_pctxt.entrypoint
29def writefile(path, mode, content):
30 if not os.path.exists(os.path.dirname(path)):
31 raise exception.FileNotFound(file_path=path)
32 with open(path, mode) as f:
33 f.write(content)
36@nova.privsep.sys_admin_pctxt.entrypoint
37def chown(
38 path: str, uid: int = -1, gid: int = -1, recursive: bool = False,
39) -> None:
40 if not os.path.exists(path):
41 raise exception.FileNotFound(file_path=path)
43 if not recursive or os.path.isfile(path):
44 return os.chown(path, uid, gid)
46 for root, dirs, files in os.walk(path):
47 os.chown(root, uid, gid)
48 for item in dirs:
49 os.chown(os.path.join(root, item), uid, gid)
50 for item in files:
51 os.chown(os.path.join(root, item), uid, gid)
54@nova.privsep.sys_admin_pctxt.entrypoint
55def makedirs(path):
56 fileutils.ensure_tree(path)
59@nova.privsep.sys_admin_pctxt.entrypoint
60def chmod(path, mode):
61 if not os.path.exists(path):
62 raise exception.FileNotFound(file_path=path)
63 os.chmod(path, mode)
66@nova.privsep.sys_admin_pctxt.entrypoint
67def move_tree(source_path: str, dest_path: str) -> None:
68 shutil.move(source_path, dest_path)
71@nova.privsep.sys_admin_pctxt.entrypoint
72def utime(path):
73 if not os.path.exists(path):
74 raise exception.FileNotFound(file_path=path)
76 # NOTE(mikal): the old version of this used execute(touch, ...), which
77 # would apparently fail on shared storage when multiple instances were
78 # being launched at the same time. If we see failures here, we might need
79 # to wrap this in a try / except.
80 os.utime(path, None)
83@nova.privsep.sys_admin_pctxt.entrypoint
84def rmdir(path):
85 if not os.path.exists(path):
86 raise exception.FileNotFound(file_path=path)
87 os.rmdir(path)
90@nova.privsep.sys_admin_pctxt.entrypoint
91def last_bytes(path, num):
92 """Return num bytes from the end of the file, and remaining byte count.
94 :param path: The file to read
95 :param num: The number of bytes to return
97 :returns: (data, remaining)
98 """
99 with open(path, 'rb') as f:
100 try:
101 f.seek(-num, os.SEEK_END)
102 except IOError as e:
103 # seek() fails with EINVAL when trying to go before the start of
104 # the file. It means that num is larger than the file size, so
105 # just go to the start.
106 if e.errno == errno.EINVAL: 106 ↛ 109line 106 didn't jump to line 109 because the condition on line 106 was always true
107 f.seek(0, os.SEEK_SET)
108 else:
109 raise
111 remaining = f.tell()
112 return (f.read(), remaining)