1
0
mirror of https://github.com/openSUSE/osc.git synced 2024-11-09 22:36:14 +01:00

Replace core.findpacs() with Package.from_paths() and Package.from_paths_nofail()

The original findpacs() was returning either [Package]
or ([Package], [str]) depending on the `fatal` option.
This confused pylint and it was returning false-positives:
E1101: Instance of 'list' has no '...' member (no-member)
This commit is contained in:
Daniel Mach 2022-09-13 16:34:54 +02:00
parent c7370522cc
commit c8f4487201
35 changed files with 214 additions and 55 deletions

View File

@ -1130,7 +1130,7 @@ def main(apiurl, opts, argv):
if not old_pkg_dir.startswith('/') and not opts.offline:
data = [prj, pacname, repo, arch]
if old_pkg_dir == '_link':
p = core.findpacs([os.curdir])[0]
p = core.Package(os.curdir)
if not p.islink():
raise oscerr.WrongOptions('package is not a link')
data[0] = p.linkinfo.project

View File

@ -1395,7 +1395,7 @@ class Osc(cmdln.Cmdln):
elif len(args) <= 2:
# try using the working copy at hand
p = findpacs([os.curdir])[0]
p = core.Package(os.curdir)
src_project = p.prjname
src_package = p.name
if self.options.apiurl and self.options.apiurl != p.apiurl:
@ -1627,7 +1627,7 @@ Please submit there instead, or use --nodevelproject to force direct submission.
elif len(args) <= 2:
# try using the working copy at hand
p = findpacs([os.curdir])[0]
p = core.Package(os.curdir)
src_project = p.prjname
src_package = p.name
if len(args) == 0 and p.islink():
@ -2742,7 +2742,7 @@ Please submit there instead, or use --nodevelproject to force direct submission.
rev = None
if len(args) == 0:
p = findpacs([os.curdir])[0]
p = core.Package(os.curdir)
project = p.prjname
package = p.name
apiurl = p.apiurl
@ -3834,7 +3834,7 @@ Please submit there instead, or use --nodevelproject to force direct submission.
specfile = opts.specfile
else:
specfile = None
pacs = findpacs(args)
pacs = Package.from_paths(args)
for p in pacs:
p.read_meta_from_spec(specfile)
p.update_package_meta()
@ -3886,7 +3886,7 @@ Please submit there instead, or use --nodevelproject to force direct submission.
pacs = None
if not opts.link or not len(args) == 2:
pacs = findpacs(args)
pacs = Package.from_paths(args)
if opts.link:
query = {'rev': 'latest'}
@ -4663,7 +4663,7 @@ Please submit there instead, or use --nodevelproject to force direct submission.
for st, filename in sorted(states, key=cmp_to_key(compare)):
lines.append(statfrmt(st, os.path.normpath(os.path.join(p.dir, filename))))
else:
p = findpacs([arg])[0]
p = Package(arg)
for st, filename in sorted(p.get_status(opts.show_excluded, *excl_states), key=cmp_to_key(compare)):
lines.append(statfrmt(st, os.path.normpath(os.path.join(p.dir, filename))))
if lines:
@ -4754,7 +4754,7 @@ Please submit there instead, or use --nodevelproject to force direct submission.
'\'do_package_tracking\' is enabled in the configuration file', file=sys.stderr)
sys.exit(1)
pacs = findpacs(args)
pacs = Package.from_paths(args)
for p in pacs:
todo = list(set(p.filenamelist + p.filenamelist_unvers + p.to_be_added))
for filename in todo:
@ -4857,7 +4857,7 @@ Please submit there instead, or use --nodevelproject to force direct submission.
prj.commit(msg=msg, skip_local_service_run=skip_local_service_run, verbose=opts.verbose, can_branch=can_branch)
args.remove(arg)
pacs, no_pacs = findpacs(args, fatal=False)
pacs, no_pacs = Package.from_paths_nofail(args)
for pac in pacs.copy():
if pac.scm_url:
@ -5009,7 +5009,7 @@ Please submit there instead, or use --nodevelproject to force direct submission.
print_request_list(prj.apiurl, prj.name)
args.sort()
pacs = findpacs(args, progress_obj=self.download_progress)
pacs = Package.from_paths(args, progress_obj=self.download_progress)
if opts.revision and len(args) == 1:
rev, dummy = parseRevisionOption(opts.revision)
@ -5123,7 +5123,7 @@ Please submit there instead, or use --nodevelproject to force direct submission.
print(statfrmt('D', getTransActPath(i)))
args.remove(i)
prj.write_packages()
pacs = findpacs(args)
pacs = Package.from_paths(args)
for p in pacs:
if not p.todo:
@ -5172,7 +5172,7 @@ Please submit there instead, or use --nodevelproject to force direct submission.
self.argparse_error("Incorrect number of arguments.")
args = parseargs(args)
pacs = findpacs(args)
pacs = Package.from_paths(args)
for p in pacs:
for filename in p.todo:
@ -7012,7 +7012,7 @@ Please submit there instead, or use --nodevelproject to force direct submission.
"""
args = parseargs(args)
pacs = findpacs(args)
pacs = Package.from_paths(args)
for p in pacs:
print(p.info())
@ -8957,8 +8957,8 @@ Please submit there instead, or use --nodevelproject to force direct submission.
raise oscerr.WrongArgs("Dest file '%s' already exists" % dest)
if os.path.isdir(dest):
dest = os.path.join(dest, os.path.basename(source))
src_pkg = findpacs([source])
tgt_pkg = findpacs([dest])
src_pkg = Package(source)
tgt_pkg = Package(dest)
if not src_pkg:
raise oscerr.NoWorkingCopy("Error: \"%s\" is not located in an osc working copy." % os.path.abspath(source))
if not tgt_pkg:
@ -9081,7 +9081,7 @@ Please submit there instead, or use --nodevelproject to force direct submission.
Note: this only works for package working copies
"""
files = opts.file
pacs = findpacs(files)
pacs = Package.from_paths(files)
for p in pacs:
if not p.todo:
p.todo = p.filenamelist + p.to_be_added

View File

@ -1208,6 +1208,37 @@ class Package:
def __lt__(self, other):
return (self.name, self.prjname, self.apiurl) < (other.name, other.prjname, other.apiurl)
@classmethod
def from_paths(cls, paths, progress_obj=None):
"""
Return a list of Package objects from working copies in given paths.
"""
packages = []
for path in paths:
package = cls(path, progress_obj)
if package in packages:
raise oscerr.PackageExists(package.prjname, package.name, "Duplicate package")
packages.append(package)
return packages
@classmethod
def from_paths_nofail(cls, paths, progress_obj=None):
"""
Return a list of Package objects from working copies in given paths
and a list of strings with paths that do not contain Package working copies.
"""
packages = []
failed_to_load = []
for path in paths:
try:
package = cls(path, progress_obj)
if package in packages:
raise oscerr.PackageExists(package.prjname, package.name, "Duplicate package")
packages.append(package)
except oscerr.NoWorkingCopy:
failed_to_load.append(path)
return packages, failed_to_load
def wc_check(self):
dirty_files = []
if self.scm_url:
@ -3208,7 +3239,7 @@ def expand_proj_pack(args, idx=0, howmany=0):
If args[idx] does not exist, an implicit '.' is assumed.
If not enough elements up to idx exist, an error is raised.
See also parseargs(args), slash_split(args), findpacs(args)
See also parseargs(args), slash_split(args), Package.from_paths(args)
All these need unification, somehow.
"""
@ -3246,44 +3277,15 @@ def expand_proj_pack(args, idx=0, howmany=0):
def findpacs(files, progress_obj=None, fatal=True):
"""collect Package objects belonging to the given files
and make sure each Package is returned only once"""
pacs = []
no_pacs = []
for f in files:
try:
p = filedir_to_pac(f, progress_obj)
except oscerr.OscBaseError as e:
if fatal:
raise e
no_pacs.append(f)
continue
known = None
for i in pacs:
if i.name == p.name and i.prjname == p.prjname:
known = i
break
if known:
i.merge(p)
else:
pacs.append(p)
if not fatal:
return pacs, no_pacs
return pacs
def filedir_to_pac(f, progress_obj=None):
"""Takes a working copy path, or a path to a file inside a working copy,
and returns a Package object instance
If the argument was a filename, add it onto the "todo" list of the Package """
if os.path.isdir(f):
wd = f
p = Package(wd, progress_obj=progress_obj)
else:
wd = os.path.dirname(f) or os.curdir
p = Package(wd, progress_obj=progress_obj)
p.todo = [os.path.basename(f)]
return p
import warnings
warnings.warn(
"osc.core.findpacs() is deprecated. "
"Use osc.core.Package.from_paths() or osc.core.Package.from_paths_nofail() instead.",
DeprecationWarning
)
if fatal:
return Package.from_paths(files, progress_obj)
return Package.from_paths_nofail(files, progress_obj)
def read_filemeta(dir):

7
tests/fixtures/packages/oscrc vendored Normal file
View File

@ -0,0 +1,7 @@
[general]
apiurl = http://localhost
[http://localhost]
user=Admin
pass=opensuse
allow_http=1

View File

@ -0,0 +1 @@
http://example.com

View File

@ -0,0 +1 @@
<directory name="pkgA" rev="1" srcmd5="d41d8cd98f00b204e9800998ecf8427e" vrev="1" />

View File

@ -0,0 +1 @@
pkgA

View File

@ -0,0 +1 @@
projectA

View File

@ -0,0 +1 @@
http://example.com

View File

@ -0,0 +1 @@
<directory name="pkgA" rev="1" srcmd5="d41d8cd98f00b204e9800998ecf8427e" vrev="1" />

View File

@ -0,0 +1 @@
pkgB

View File

@ -0,0 +1 @@
projectA

View File

@ -0,0 +1 @@
http://localhost

View File

@ -0,0 +1 @@
<directory name="pkgA" rev="1" srcmd5="d41d8cd98f00b204e9800998ecf8427e" vrev="1" />

View File

@ -0,0 +1 @@
1.0

View File

@ -0,0 +1 @@
pkgA

View File

@ -0,0 +1 @@
projectA

View File

@ -0,0 +1 @@
http://localhost

View File

@ -0,0 +1 @@
<directory name="pkgA" rev="1" srcmd5="d41d8cd98f00b204e9800998ecf8427e" vrev="1" />

View File

@ -0,0 +1 @@
1.0

View File

@ -0,0 +1 @@
pkgB

View File

@ -0,0 +1 @@
projectA

View File

@ -0,0 +1 @@
http://localhost

View File

@ -0,0 +1 @@
<directory name="pkgA" rev="1" srcmd5="d41d8cd98f00b204e9800998ecf8427e" vrev="1" />

View File

@ -0,0 +1 @@
1.0

View File

@ -0,0 +1 @@
pkgA

View File

@ -0,0 +1 @@
projectB

View File

@ -0,0 +1 @@
http://localhost

View File

@ -0,0 +1 @@
<directory name="pkgA" rev="1" srcmd5="d41d8cd98f00b204e9800998ecf8427e" vrev="1" />

View File

@ -0,0 +1 @@
1.0

View File

@ -0,0 +1 @@
pkgB

View File

@ -0,0 +1 @@
projectB

View File

@ -1,6 +1,13 @@
import os
import unittest
import osc.core
import osc.oscerr
from .common import OscTestCase
FIXTURES_DIR = os.path.join(os.path.dirname(__file__), "fixtures", "packages")
class PackageMock(osc.core.Package):
@ -58,5 +65,118 @@ class TestPackage(unittest.TestCase):
self.assertIn(p2, packages)
class TestPackageFromPaths(OscTestCase):
def _get_fixtures_dir(self):
return FIXTURES_DIR
def test_single_package(self):
paths = ["projectA/pkgA"]
paths = [os.path.join(self.tmpdir, 'osctest', i) for i in paths]
pacs = osc.core.Package.from_paths(paths)
self.assertEqual(len(pacs), 1)
pac = pacs[0]
self.assertEqual(pac.name, "pkgA")
self.assertEqual(pac.prjname, "projectA")
self.assertEqual(pac.apiurl, "http://localhost")
def test_duplicates(self):
paths = ["projectA/pkgA", "projectA/pkgA"]
paths = [os.path.join(self.tmpdir, 'osctest', i) for i in paths]
self.assertRaises(osc.oscerr.PackageExists, osc.core.Package.from_paths, paths)
def test_two_packages(self):
paths = ["projectA/pkgA", "projectA/pkgB"]
paths = [os.path.join(self.tmpdir, 'osctest', i) for i in paths]
pacs = osc.core.Package.from_paths(paths)
self.assertEqual(len(pacs), 2)
pac = pacs[0]
self.assertEqual(pac.name, "pkgA")
self.assertEqual(pac.prjname, "projectA")
self.assertEqual(pac.apiurl, "http://localhost")
pac = pacs[1]
self.assertEqual(pac.name, "pkgB")
self.assertEqual(pac.prjname, "projectA")
self.assertEqual(pac.apiurl, "http://localhost")
def test_two_projects(self):
paths = ["projectA/pkgA", "projectA/pkgB", "projectB/pkgA", "projectB/pkgB"]
paths = [os.path.join(self.tmpdir, 'osctest', i) for i in paths]
pacs = osc.core.Package.from_paths(paths)
self.assertEqual(len(pacs), 4)
pac = pacs[0]
self.assertEqual(pac.name, "pkgA")
self.assertEqual(pac.prjname, "projectA")
self.assertEqual(pac.apiurl, "http://localhost")
pac = pacs[1]
self.assertEqual(pac.name, "pkgB")
self.assertEqual(pac.prjname, "projectA")
self.assertEqual(pac.apiurl, "http://localhost")
pac = pacs[2]
self.assertEqual(pac.name, "pkgA")
self.assertEqual(pac.prjname, "projectB")
self.assertEqual(pac.apiurl, "http://localhost")
pac = pacs[3]
self.assertEqual(pac.name, "pkgB")
self.assertEqual(pac.prjname, "projectB")
self.assertEqual(pac.apiurl, "http://localhost")
def test_two_apiurls(self):
paths = ["projectA/pkgA", "projectA/pkgB", "projectA-different-apiurl/pkgA", "projectA-different-apiurl/pkgB"]
paths = [os.path.join(self.tmpdir, 'osctest', i) for i in paths]
pacs = osc.core.Package.from_paths(paths)
self.assertEqual(len(pacs), 4)
pac = pacs[0]
self.assertEqual(pac.name, "pkgA")
self.assertEqual(pac.prjname, "projectA")
self.assertEqual(pac.apiurl, "http://localhost")
pac = pacs[1]
self.assertEqual(pac.name, "pkgB")
self.assertEqual(pac.prjname, "projectA")
self.assertEqual(pac.apiurl, "http://localhost")
pac = pacs[2]
self.assertEqual(pac.name, "pkgA")
self.assertEqual(pac.prjname, "projectA")
self.assertEqual(pac.apiurl, "http://example.com")
pac = pacs[3]
self.assertEqual(pac.name, "pkgB")
self.assertEqual(pac.prjname, "projectA")
self.assertEqual(pac.apiurl, "http://example.com")
def test_invalid_package(self):
paths = ["projectA/pkgA", "projectA"]
paths = [os.path.join(self.tmpdir, 'osctest', i) for i in paths]
self.assertRaises(osc.oscerr.NoWorkingCopy, osc.core.Package.from_paths, paths)
def test_nofail(self):
# valid package, invalid package, nonexistent package
paths = ["projectA/pkgA", "projectA", "does-not-exist"]
paths = [os.path.join(self.tmpdir, 'osctest', i) for i in paths]
pacs, nopacs = osc.core.Package.from_paths_nofail(paths)
self.assertEqual(len(pacs), 1)
pac = pacs[0]
self.assertEqual(pac.name, "pkgA")
self.assertEqual(pac.prjname, "projectA")
self.assertEqual(pac.apiurl, "http://localhost")
expected = [
os.path.join(self.tmpdir, "osctest", "projectA"),
os.path.join(self.tmpdir, "osctest", "does-not-exist"),
]
self.assertEqual(nopacs, expected)
if __name__ == "__main__":
unittest.main()