#
# sysutils.py
#
# Copyright (C) 2009-2019 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# Red Hat Author(s): Martin Gracik <mgracik@redhat.com>
#
__all__ = ["joinpaths", "touch", "replace", "chown_", "chmod_", "remove",
"linktree"]
import sys
import os
import re
import fileinput
import pwd
import grp
import glob
import shutil
import shlex
from configparser import ConfigParser
from pylorax.executils import runcmd
[docs]def joinpaths(*args, **kwargs):
path = os.path.sep.join(args)
if kwargs.get("follow_symlinks"):
return os.path.realpath(path)
else:
return path
[docs]def touch(fname):
with open(fname, "w") as f:
f.write("")
[docs]def replace(fname, find, sub):
fin = fileinput.input(fname, inplace=1)
pattern = re.compile(find)
for line in fin:
line = pattern.sub(sub, line)
sys.stdout.write(line)
fin.close()
[docs]def chown_(path, user=None, group=None, recursive=False):
uid = gid = -1
if user is not None:
uid = pwd.getpwnam(user)[2]
if group is not None:
gid = grp.getgrnam(group)[2]
for fname in glob.iglob(path):
os.chown(fname, uid, gid)
if recursive and os.path.isdir(fname):
for nested in os.listdir(fname):
nested = joinpaths(fname, nested)
chown_(nested, user, group, recursive)
[docs]def chmod_(path, mode, recursive=False):
for fname in glob.iglob(path):
os.chmod(fname, mode)
if recursive and os.path.isdir(fname):
for nested in os.listdir(fname):
nested = joinpaths(fname, nested)
chmod_(nested, mode, recursive)
def cpfile(src, dst):
shutil.copy2(src, dst)
if os.path.isdir(dst):
dst = joinpaths(dst, os.path.basename(src))
return dst
def mvfile(src, dst):
if os.path.isdir(dst):
dst = joinpaths(dst, os.path.basename(src))
os.rename(src, dst)
return dst
[docs]def remove(target):
if os.path.isdir(target) and not os.path.islink(target):
shutil.rmtree(target)
else:
os.unlink(target)
[docs]def linktree(src, dst):
runcmd(["/bin/cp", "-alx", src, dst])
def unquote(s):
return ' '.join(shlex.split(s))
class UnquotingConfigParser(ConfigParser):
"""A ConfigParser, only with unquoting of the values."""
# pylint: disable=arguments-differ
def get(self, *args, **kwargs):
ret = super().get(*args, **kwargs)
if ret:
ret = unquote(ret)
return ret
def flatconfig(filename):
"""Use UnquotingConfigParser to read a flat config file (without
section headers) by adding a section header.
"""
with open (filename, 'r') as conffh:
conftext = "[main]\n" + conffh.read()
config = UnquotingConfigParser()
config.read_string(conftext)
return config['main']
def read_tail(path, size):
"""Read up to `size` kibibytes from the end of a file"""
# NOTE: In py3 text files are unicode, not bytes so we have to open it as bytes
with open(path, "rb") as f:
return _read_file_end(f, size)
def _read_file_end(f, size):
"""Read the end of a file
This skips to the next line to avoid starting in the middle of a unicode character.
And returns "" in the case of a UnicodeDecodeError
"""
f.seek(0, 2)
end = f.tell()
if end < 1024 * size:
f.seek(0, 0)
else:
f.seek(end - (1024 * size))
data = f.read()
try:
# Find the first newline in the block
newline = min(1+data.find(b'\n'), len(data))
text = data[newline:].decode("UTF-8")
except UnicodeDecodeError:
return ""
return text