mirrors/github.com_openSUSE_osc
1
0
Fork 0
mirror of https://github.com/openSUSE/osc.git synced 2024-11-12 23:56:13 +01:00
Code

Merge pull request #1142 from dmach/fix-pylint-errors-and-warnings-2

Browse Source 
Fix whitespace issues in osc.util
This commit is contained in:
Daniel Mach 2022-09-12 16:08:09 +02:00 committed by GitHub
commit 86e4be9119
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 87 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
osc/util/ar.py
View File

@ -25,8 +25,10 @@ from io import BytesIO
if not hasattr(os, 'SEEK_SET'):
os.SEEK_SET = 0
class ArError(Exception):
"""Base class for all ar related errors"""
def __init__(self, fn, msg):
super().__init__()
self.file = fn
@ -35,8 +37,10 @@ class ArError(Exception):
def __str__(self):
return 'ar error: %s' % self.msg
class ArHdr:
"""Represents an ar header entry"""
def __init__(self, fn, date, uid, gid, mode, size, fmag, off):
self.file = fn.strip()
self.date = date.strip()
@ -54,8 +58,10 @@ class ArHdr:
def __str__(self):
return '%16s %d' % (self.file, self.size)
class ArFile(BytesIO):
"""Represents a file which resides in the archive"""
def __init__(self, fn, uid, gid, mode, buf):
super().__init__(buf)
self.name = fn
@ -63,7 +69,7 @@ class ArFile(BytesIO):
self.gid = gid
self.mode = mode
def saveTo(self, dir = None):
def saveTo(self, dir=None):
"""
writes file to dir/filename if dir isn't specified the current
working dir is used. Additionally it tries to set the owner/group
@ -79,7 +85,7 @@ class ArFile(BytesIO):
if uid != os.geteuid() or os.geteuid() != 0:
uid = -1
gid = self.gid
if not gid in os.getgroups() or os.getegid() != 0:
if gid not in os.getgroups() or os.getegid() != 0:
gid = -1
os.chown(fn, uid, gid)
@ -87,6 +93,7 @@ class ArFile(BytesIO):
return '%s %s %s %s' % (self.name, self.uid,
self.gid, self.mode)
class Ar:
"""
Represents an ar archive (only GNU format is supported).
@ -96,7 +103,7 @@ class Ar:
hdr_pat = re.compile(b'^(.{16})(.{12})(.{6})(.{6})(.{8})(.{10})(.{2})',
re.DOTALL)
def __init__(self, fn = None, fh = None):
def __init__(self, fn=None, fh=None):
if fn is None and fh is None:
raise ValueError('either \'fn\' or \'fh\' must be is not None')
if fh is not None:

11
osc/util/archquery.py
View File

@ -9,20 +9,21 @@ from . import packagequery
class ArchError(packagequery.PackageError):
pass
class ArchQuery(packagequery.PackageQuery, packagequery.PackageQueryResult):
def __init__(self, fh):
self.__file = fh
self.__path = os.path.abspath(fh.name)
self.fields = {}
#self.magic = None
#self.pkgsuffix = 'pkg.tar.gz'
# self.magic = None
# self.pkgsuffix = 'pkg.tar.gz'
self.pkgsuffix = b'arch'
def read(self, all_tags=True, self_provides=True, *extra_tags):
# all_tags and *extra_tags are currently ignored
f = open(self.__path, 'rb')
#self.magic = f.read(5)
#if self.magic == '\375\067zXZ':
# self.magic = f.read(5)
# if self.magic == '\375\067zXZ':
# self.pkgsuffix = 'pkg.tar.xz'
fn = open('/dev/null', 'wb')
pipe = subprocess.Popen(['tar', '-O', '-xf', self.__path, '.PKGINFO'], stdout=subprocess.PIPE, stderr=fn).stdout
@ -135,7 +136,7 @@ class ArchQuery(packagequery.PackageQuery, packagequery.PackageQueryResult):
return None
@staticmethod
def query(filename, all_tags = False, *extra_tags):
def query(filename, all_tags=False, *extra_tags):
f = open(filename, 'rb')
archq = ArchQuery(f)
archq.read(all_tags, *extra_tags)

43
osc/util/cpio.py
View File

@ -26,19 +26,24 @@ if not hasattr(os, 'SEEK_SET'):
# format implementation is based on src/copyin.c and src/util.c (see cpio sources)
class CpioError(Exception):
"""base class for all cpio related errors"""
def __init__(self, fn, msg):
super().__init__()
self.file = fn
self.msg = msg
def __str__(self):
return '%s: %s' % (self.file, self.msg)
class CpioHdr:
"""
Represents a cpio header ("New" portable format and CRC format).
"""
def __init__(self, mgc, ino, mode, uid, gid, nlink, mtime, filesize,
dev_maj, dev_min, rdev_maj, rdev_min, namesize, checksum,
off=-1, filename=b''):
@ -72,6 +77,7 @@ class CpioHdr:
def __str__(self):
return "%s %s %s %s" % (self.filename, self.filesize, self.namesize, self.dataoff)
class CpioRead:
"""
Represents a cpio archive.
@ -81,8 +87,8 @@ class CpioRead:
# supported formats - use name -> mgc mapping to increase readabilty
sfmt = {
'newascii': b'070701',
}
'newascii': b'070701',
}
# header format
hdr_fmt = '6s8s8s8s8s8s8s8s8s8s8s8s8s8s'
@ -131,7 +137,7 @@ class CpioRead:
if uid != os.geteuid() or os.geteuid() != 1:
uid = -1
gid = hdr.gid
if not gid in os.getgroups() or os.getegid() != -1:
if gid not in os.getgroups() or os.getegid() != -1:
gid = -1
os.chown(fn, uid, gid)
@ -149,7 +155,7 @@ class CpioRead:
self._init_datastructs()
data = self.__file.read(6)
self.format = data
if not self.format in self.sfmt.values():
if self.format not in self.sfmt.values():
raise CpioError(self.filename, '\'%s\' is not a supported cpio format' % self.format)
pos = 0
while len(data) != 0:
@ -170,7 +176,7 @@ class CpioRead:
self.hdrs.append(hdr)
pos += hdr.filesize + self._calc_padding(hdr.filesize)
def copyin_file(self, filename, dest = None, new_fn = None):
def copyin_file(self, filename, dest=None, new_fn=None):
"""
copies filename to dest.
If dest is None the file will be stored in $PWD/filename. If dest points
@ -184,7 +190,7 @@ class CpioRead:
fn = new_fn or filename
self._copyin_file(hdr, dest, fn)
def copyin(self, dest = None):
def copyin(self, dest=None):
"""
extracts the cpio archive to dest.
If dest is None $PWD will be used.
@ -193,6 +199,7 @@ class CpioRead:
for h in self.hdrs:
self._copyin_file(h, dest, h.filename)
class CpioWrite:
"""cpio archive small files in memory, using new style portable header format"""
@ -207,20 +214,20 @@ class CpioWrite:
mode = perms | type
c = bytearray()
c.extend(b'070701') # magic
c.extend(b'%08X' % 0) # inode
c.extend(b'%08X' % mode) # mode
c.extend(b'%08X' % 0) # uid
c.extend(b'%08X' % 0) # gid
c.extend(b'%08X' % 0) # nlink
c.extend(b'%08X' % 0) # mtime
c.extend(b'070701') # magic
c.extend(b'%08X' % 0) # inode
c.extend(b'%08X' % mode) # mode
c.extend(b'%08X' % 0) # uid
c.extend(b'%08X' % 0) # gid
c.extend(b'%08X' % 0) # nlink
c.extend(b'%08X' % 0) # mtime
c.extend(b'%08X' % filesize)
c.extend(b'%08X' % 0) # major
c.extend(b'%08X' % 0) # minor
c.extend(b'%08X' % 0) # rmajor
c.extend(b'%08X' % 0) # rminor
c.extend(b'%08X' % 0) # major
c.extend(b'%08X' % 0) # minor
c.extend(b'%08X' % 0) # rmajor
c.extend(b'%08X' % 0) # rminor
c.extend(b'%08X' % namesize)
c.extend(b'%08X' % 0) # checksum
c.extend(b'%08X' % 0) # checksum
c.extend(name + b'\0')
c.extend(b'\0' * (len(c) % 4))

15
osc/util/debquery.py
View File

@ -23,8 +23,7 @@ except ImportError:
HAVE_ZSTD = False
if (not hasattr(itertools, 'zip_longest')
and hasattr(itertools, 'izip_longest')):
if (not hasattr(itertools, 'zip_longest') and hasattr(itertools, 'izip_longest')):
# python2 case
itertools.zip_longest = itertools.izip_longest
@ -32,11 +31,14 @@ if (not hasattr(itertools, 'zip_longest')
class DebError(packagequery.PackageError):
pass
class DebQuery(packagequery.PackageQuery, packagequery.PackageQueryResult):
default_tags = (b'package', b'version', b'release', b'epoch',
default_tags = (
b'package', b'version', b'release', b'epoch',
b'architecture', b'description', b'provides', b'depends',
b'pre_depends', b'conflicts', b'breaks')
b'pre_depends', b'conflicts', b'breaks'
)
def __init__(self, fh):
self.__file = fh
@ -45,7 +47,7 @@ class DebQuery(packagequery.PackageQuery, packagequery.PackageQueryResult):
self.fields = {}
def read(self, all_tags=False, self_provides=True, *extra_tags):
arfile = ar.Ar(fh = self.__file)
arfile = ar.Ar(fh=self.__file)
arfile.read()
debbin = arfile.get_file(b'debian-binary')
if debbin is None:
@ -190,7 +192,7 @@ class DebQuery(packagequery.PackageQuery, packagequery.PackageQueryResult):
return DebQuery.filename(self.name(), self.epoch(), self.version(), self.release(), self.arch())
@staticmethod
def query(filename, all_tags = False, *extra_tags):
def query(filename, all_tags=False, *extra_tags):
f = open(filename, 'rb')
debq = DebQuery(f)
debq.read(all_tags, *extra_tags)
@ -249,6 +251,7 @@ class DebQuery(packagequery.PackageQuery, packagequery.PackageQueryResult):
else:
return b'%s_%s_%s.deb' % (name, version, arch)
if __name__ == '__main__':
try:
debq = DebQuery.query(sys.argv[1])

2
osc/util/helper.py
View File

@ -66,7 +66,7 @@ def format_table(rows, headers):
templ = ' '.join(tpltpl) + '\n'
out = templ.format(*headers)
out += templ.format(*['-'*m for m in maxlens])
out += templ.format(*['-' * m for m in maxlens])
for r in rows:
out += templ.format(*r)
return out

9
osc/util/packagequery.py
View File

@ -5,11 +5,13 @@ from .helper import decode_it
class PackageError(Exception):
"""base class for all package related errors"""
def __init__(self, fname, msg):
super().__init__()
self.fname = fname
self.msg = msg
class PackageQueries(dict):
"""Dict of package name keys and package query values. When assigning a
package query, to a name, the package is evaluated to see if it matches the
@ -40,16 +42,18 @@ class PackageQueries(dict):
if (architecture in [self.wanted_architecture, 'noarch', 'all', 'any']
or self.wanted_architecture in self.architectureMap.get(architecture,
[])):
[])):
current_query = self.get(name)
# if current query does not exist or is older than this new query
if current_query is None or current_query.vercmp(query) <= 0:
super().__setitem__(name, query)
class PackageQuery:
"""abstract base class for all package types"""
def read(self, all_tags = False, *extra_tags):
def read(self, all_tags=False, *extra_tags):
"""Returns a PackageQueryResult instance"""
raise NotImplementedError
@ -97,6 +101,7 @@ class PackageQuery:
class PackageQueryResult:
"""abstract base class that represents the result of a package query"""
def name(self):
raise NotImplementedError

9
osc/util/repodata.py
View File

@ -13,6 +13,7 @@ from . import packagequery
def namespace(name):
return "{http://linux.duke.edu/metadata/%s}" % name
OPERATOR_BY_FLAGS = {
"EQ": "=",
"LE": "<=",
@ -21,6 +22,7 @@ OPERATOR_BY_FLAGS = {
"GT": ">"
}
def primaryPath(directory):
"""Returns path to the primary repository data file.
@ -45,6 +47,7 @@ def primaryPath(directory):
return primaryPath
def queries(directory):
"""Returns a list of RepoDataQueries constructed from the repodata under
the directory.
@ -122,8 +125,7 @@ class RepoDataQueryResult(packagequery.PackageQueryResult):
entries = []
if collectionElement is not None:
for entryElement in collectionElement.findall(namespace("rpm") +
"entry"):
for entryElement in collectionElement.findall(namespace("rpm") + "entry"):
entry = self.__parseEntry(entryElement)
entries.append(entry)
@ -199,8 +201,7 @@ class RepoDataQueryResult(packagequery.PackageQueryResult):
release = None
else:
release = self.release()
return rpmquery.RpmQuery.filename(self.name(), None,
self.version(), release, self.arch())
return rpmquery.RpmQuery.filename(self.name(), None, self.version(), release, self.arch())
def gettag(self, tag):
# implement me, if needed

35
osc/util/rpmquery.py
View File

@ -10,14 +10,18 @@ from .helper import decode_it
def cmp(a, b):
return (a > b) - (a < b)
class RpmError(packagequery.PackageError):
pass
class RpmHeaderError(RpmError):
pass
class RpmHeader:
"""corresponds more or less to the indexEntry_s struct"""
def __init__(self, offset, length):
self.offset = offset
# length of the data section (without length of indexEntries)
@ -39,11 +43,13 @@ class RpmHeader:
def __len__(self):
return len(self.entries)
class RpmHeaderEntry:
"""corresponds to the entryInfo_s struct (except the data attribute)"""
# each element represents an int
ENTRY_SIZE = 16
def __init__(self, tag, type, offset, count):
self.tag = tag
self.type = type
@ -51,6 +57,7 @@ class RpmHeaderEntry:
self.count = count
self.data = None
class RpmQuery(packagequery.PackageQuery, packagequery.PackageQueryResult):
LEAD_SIZE = 96
LEAD_MAGIC = 0xedabeedb
@ -63,15 +70,16 @@ class RpmQuery(packagequery.PackageQuery, packagequery.PackageQueryResult):
SENSE_STRONG = 1 << 27
default_tags = (1000, 1001, 1002, 1003, 1004, 1022, 1005, 1020,
1047, 1112, 1113, # provides
1049, 1048, 1050, # requires
1054, 1053, 1055, # conflicts
1090, 1114, 1115, # obsoletes
1156, 1158, 1157, # oldsuggests
5046, 5047, 5048, # recommends
5049, 5051, 5050, # suggests
5052, 5053, 5054, # supplements
default_tags = (
1000, 1001, 1002, 1003, 1004, 1022, 1005, 1020,
1047, 1112, 1113, # provides
1049, 1048, 1050, # requires
1054, 1053, 1055, # conflicts
1090, 1114, 1115, # obsoletes
1156, 1158, 1157, # oldsuggests
5046, 5047, 5048, # recommends
5049, 5051, 5050, # suggests
5052, 5053, 5054, # supplements
5055, 5056, 5057 # enhances
)
@ -108,9 +116,10 @@ class RpmQuery(packagequery.PackageQuery, packagequery.PackageQueryResult):
data = self.__file.read(self.header.length)
for i in self.header:
if i.tag in self.default_tags + extra_tags or all_tags:
try: # this may fail for -debug* packages
try: # this may fail for -debug* packages
self.__read_data(i, data)
except: pass
except:
pass
return self
def __read_lead(self):
@ -310,7 +319,7 @@ class RpmQuery(packagequery.PackageQuery, packagequery.PackageQueryResult):
entry = rpmq.gettag(1004)
if entry is None:
return None
return ''.join([ "%02x" % x for x in struct.unpack('16B', entry.data) ])
return ''.join(["%02x" % x for x in struct.unpack('16B', entry.data)])
@staticmethod
def rpmvercmp(ver1, ver2):
@ -375,6 +384,7 @@ class RpmQuery(packagequery.PackageQuery, packagequery.PackageQueryResult):
def filename(name, epoch, version, release, arch):
return b'%s-%s-%s.%s.rpm' % (name, version, release, arch)
def unpack_string(data, encoding=None):
"""unpack a '\\0' terminated string from data"""
idx = data.find(b'\0')
@ -385,6 +395,7 @@ def unpack_string(data, encoding=None):
data = data.decode(encoding)
return data
if __name__ == '__main__':
try:
rpmq = RpmQuery.query(sys.argv[1])

1
osc/util/safewriter.py
View File

@ -6,6 +6,7 @@ class SafeWriter:
the str is encoded with the "encoding" encoding.
All getattr, setattr calls are passed through to the "writer" instance.
"""
def __init__(self, writer, encoding='unicode_escape'):
self._writer = writer
self._encoding = encoding