# GNU Library Public License for more details.
#
+from __future__ import print_function
+
import argparse
import base64
import errno
def acquire(self):
assert not self.fd
- self.fd = file(self.fn, 'w')
+ self.fd = open(self.fn, 'w')
fcntl.lockf(self.fd, fcntl.LOCK_EX)
def release(self):
"""
def __str__(self):
- doc = self.__doc__.strip()
- return ': '.join([doc] + [str(a) for a in self.args])
+ doc = _bytes2str(self.__doc__.strip())
+ return ': '.join([doc] + [_bytes2str(a) for a in self.args])
class MountError(Error):
"""
Detect whether systemd is running
"""
- with file('/proc/1/comm', 'rb') as i:
- for line in i:
- if 'systemd' in line:
- return True
- return False
+ with open('/proc/1/comm', 'r') as f:
+ return 'systemd' in f.read()
def is_upstart():
Detect whether upstart is running
"""
(out, err, _) = command(['init', '--version'])
- if 'upstart' in out:
- return True
- return False
+ return 'upstart' in out
def maybe_mkdir(*a, **kw):
os.unlink(*a)
try:
os.mkdir(*a, **kw)
- except OSError, e:
+ except OSError as e:
if e.errno == errno.EEXIST:
pass
else:
def which(executable):
"""find the location of an executable"""
- if 'PATH' in os.environ:
- envpath = os.environ['PATH']
- else:
- envpath = os.defpath
+ envpath = os.environ.get('PATH') or os.defpath
PATH = envpath.split(os.pathsep)
locations = PATH + [
This returns the output of the command and the return code of the
process in a tuple: (output, returncode).
"""
- arguments = _get_command_executable(arguments)
+
+ arguments = list(map(_bytes2str, _get_command_executable(arguments)))
+
LOG.info('Running command: %s' % ' '.join(arguments))
process = subprocess.Popen(
arguments,
stderr=subprocess.PIPE,
**kwargs)
out, err = process.communicate()
- return out, err, process.returncode
+
+ return _bytes2str(out), _bytes2str(err), process.returncode
+
+
+def _bytes2str(string):
+ return string.decode('utf-8') if isinstance(string, bytes) else string
def command_check_call(arguments):
Check if the given device is mounted.
"""
dev = os.path.realpath(dev)
- with file('/proc/mounts', 'rb') as proc_mounts:
+ with open('/proc/mounts', 'rb') as proc_mounts:
for line in proc_mounts:
fields = line.split()
if len(fields) < 3:
if mounts_dev.startswith('/') and os.path.exists(mounts_dev):
mounts_dev = os.path.realpath(mounts_dev)
if mounts_dev == dev:
- return path
+ return _bytes2str(path)
return None
:raises: TruncatedLineError or TooManyLinesError
:return: Content of the line, or None if line isn't valid.
"""
+ line = _bytes2str(line)
+
if line[-1:] != '\n':
raise TruncatedLineError(line)
line = line[:-1]
"""
path = os.path.join(parent, name)
try:
- line = file(path, 'rb').read()
+ line = open(path, 'rb').read()
except IOError as e:
if e.errno == errno.ENOENT:
return None
"""
path = os.path.join(parent, name)
tmp = '{path}.{pid}.tmp'.format(path=path, pid=os.getpid())
- with file(tmp, 'wb') as tmp_file:
- tmp_file.write(text + '\n')
+ with open(tmp, 'wb') as tmp_file:
+ tmp_file.write(text.encode('utf-8') + b'\n')
os.fsync(tmp_file.fileno())
path_set_context(tmp)
os.rename(tmp, path)
pwd.getpwnam(CEPH_PREF_USER)
return CEPH_PREF_USER
except KeyError:
- print "No such user: " + CEPH_PREF_USER
+ print("No such user:", CEPH_PREF_USER)
sys.exit(2)
else:
try:
grp.getgrnam(CEPH_PREF_GROUP)
return CEPH_PREF_GROUP
except KeyError:
- print "No such group: " + CEPH_PREF_GROUP
+ print("No such group:", CEPH_PREF_GROUP)
sys.exit(2)
else:
try:
error = subprocess.CalledProcessError(ret, cmd)
error.output = out + err
raise error
- return out
+ return _bytes2str(out)
def get_conf(cluster, variable):
if dev:
return dev
- if isinstance(key, types.TupleType):
+ if isinstance(key, tuple):
# legacy, before lockbox
assert os.path.exists(key[0])
keypath = key[0]
# isn't too thorough.
lba_size = 4096
size = 33 * lba_size
- with file(dev, 'wb') as dev_file:
+ with open(dev, 'wb') as dev_file:
dev_file.seek(-size, os.SEEK_END)
- dev_file.write(size * '\0')
+ dev_file.write(size * b'\0')
command_check_call(
[
' (ceph-osd will resize and allocate)',
self.name,
getattr(self.args, self.name))
- with file(getattr(self.args, self.name), 'wb') as space_file:
+ with open(getattr(self.args, self.name), 'wb') as space_file:
pass
LOG.debug('%s is file %s',
try:
fsid = get_fsid(cluster)
except Error as e:
- if e.message != 'getting cluster uuid from configuration failed':
+ if 'getting cluster uuid from configuration failed' not in str(e):
raise e
no_fsid.append(cluster)
else:
LOG.debug('Marking with init system %s', init)
init_path = os.path.join(path, init)
- with file(init_path, 'w'):
+ with open(init_path, 'w'):
path_set_context(init_path)
# remove markers for others, just in case.
)
except Exception as e:
- print >> sys.stderr, '{prog}: {msg}'.format(
- prog=args.prog,
- msg=e,
+ print(
+ '{prog}: {msg}'.format(prog=args.prog, msg=e),
+ file=sys.stderr
)
+
err = True
finally:
def is_swap(dev):
dev = os.path.realpath(dev)
- with file('/proc/swaps', 'rb') as proc_swaps:
+ with open('/proc/swaps', 'rb') as proc_swaps:
for line in proc_swaps.readlines()[1:]:
fields = line.split()
if len(fields) < 3:
def get_oneliner(base, name):
path = os.path.join(base, name)
if os.path.isfile(path):
- with open(path, 'r') as _file:
- return _file.readline().rstrip()
+ with open(path, 'rb') as _file:
+ return _bytes2str(_file.readline().rstrip())
return None
uuid_map = {}
space_map = {}
- for base, parts in sorted(partmap.iteritems()):
+ for base, parts in sorted(partmap.items()):
for p in parts:
dev = get_dev_path(p)
part_uuid = get_partition_uuid(dev)
str(uuid_map) + ", space_map = " + str(space_map))
devices = []
- for base, parts in sorted(partmap.iteritems()):
+ for base, parts in sorted(partmap.items()):
if parts:
disk = {'path': get_dev_path(base)}
partitions = []
else:
selected_devices = devices
if args.format == 'json':
- print json.dumps(selected_devices)
+ print(json.dumps(selected_devices))
else:
output = list_format_plain(selected_devices)
if output:
- print output
+ print(output)
###########################
raise Error('not a block device', path)
base = get_dev_name(disk)
- with file(SUPPRESS_PREFIX + base, 'w') as f: # noqa
+ with open(SUPPRESS_PREFIX + base, 'w') as f: # noqa
pass
LOG.info('set suppress flag on %s', base)
import unittest
from ceph_disk import main
+try:
+ import builtins
+except:
+ import __builtin__ as builtins
+
def fail_to_mount(dev, fstype, options):
raise main.MountError(dev + " mount fail")
main.PTYPE['plain']['osd']['ready']: 'plain',
main.PTYPE['luks']['osd']['ready']: 'luks',
}
- for (ptype, type) in ptype2type.iteritems():
+ for (ptype, type) in ptype2type.items():
for holders in ((), ("dm_0",), ("dm_0", "dm_1")):
dev = {
'dmcrypt': {
main.PTYPE['plain']['journal']['ready']: 'plain',
main.PTYPE['luks']['journal']['ready']: 'luks',
}
- for (ptype, type) in ptype2type.iteritems():
+ for (ptype, type) in ptype2type.items():
for holders in ((), ("dm_0",)):
dev = {
'path': '/dev/Xda2',
main.PTYPE['plain']['osd']['ready']: 'plain',
main.PTYPE['luks']['osd']['ready']: 'LUKS',
}
- for (partition_type, type) in partition_type2type.iteritems():
+ for (partition_type, type) in partition_type2type.items():
#
# dmcrypt data partition with one holder
#
def setup_class(self):
main.setup_logging(verbose=True, log_stdout=False)
- @patch('__builtin__.open')
+ @patch('{0}.open'.format(builtins.__name__))
def test_main_deactivate(self, mock_open):
data = tempfile.mkdtemp()
main.setup_statedir(data)
projects / ceph.git / commitdiff