almalinux/lorax
13
0
Fork 0
Code Issues Pull Requests Releases Activity

Update pylorax.api.projects for DNF usage

Browse Source 
And adjust tests to match.
This commit is contained in:
Brian C. Lane 2018-05-02 10:00:58 -07:00
parent e1be281fbb
commit 9085840401
2 changed files with 144 additions and 187 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

233
src/pylorax/api/projects.py
View File

@ -17,10 +17,9 @@
import logging
log = logging.getLogger("lorax-composer")
import dnf
import time
from yum.Errors import YumBaseError
TIME_FORMAT = "%Y-%m-%dT%H:%M:%S"
@ -56,80 +55,80 @@ def api_changelog(changelog):
return entry
def yaps_to_project(yaps):
"""Extract the details from a YumAvailablePackageSqlite object
def pkg_to_project(pkg):
"""Extract the details from a hawkey.Package object
:param yaps: Yum object with package details
:type yaps: YumAvailablePackageSqlite
:param pkgs: hawkey.Package object with package details
:type pkgs: hawkey.Package
:returns: A dict with the name, summary, description, and url.
:rtype: dict
upstream_vcs is hard-coded to UPSTREAM_VCS
"""
return {"name": yaps.name,
"summary": yaps.summary,
"description": yaps.description,
"homepage": yaps.url,
return {"name": pkg.name,
"summary": pkg.summary,
"description": pkg.description,
"homepage": pkg.url,
"upstream_vcs": "UPSTREAM_VCS"}
def yaps_to_project_info(yaps):
"""Extract the details from a YumAvailablePackageSqlite object
def pkg_to_project_info(pkg):
"""Extract the details from a hawkey.Package object
:param yaps: Yum object with package details
:type yaps: YumAvailablePackageSqlite
:param pkg: hawkey.Package object with package details
:type pkg: hawkey.Package
:returns: A dict with the project details, as well as epoch, release, arch, build_time, changelog, ...
:rtype: dict
metadata entries are hard-coded to {}
"""
build = {"epoch": int(yaps.epoch),
"release": yaps.release,
"arch": yaps.arch,
"build_time": api_time(yaps.buildtime),
"changelog": api_changelog(yaps.returnChangelog()),
build = {"epoch": pkg.epoch,
"release": pkg.release,
"arch": pkg.arch,
"build_time": api_time(pkg.buildtime),
"changelog": "CHANGELOG_NEEDED", # XXX Not in hawkey.Package
"build_config_ref": "BUILD_CONFIG_REF",
"build_env_ref": "BUILD_ENV_REF",
"metadata": {},
"source": {"license": yaps.license,
"version": yaps.version,
"source": {"license": pkg.license,
"version": pkg.version,
"source_ref": "SOURCE_REF",
"metadata": {}}}
return {"name": yaps.name,
"summary": yaps.summary,
"description": yaps.description,
"homepage": yaps.url,
return {"name": pkg.name,
"summary": pkg.summary,
"description": pkg.description,
"homepage": pkg.url,
"upstream_vcs": "UPSTREAM_VCS",
"builds": [build]}
def tm_to_dep(tm):
"""Extract the info from a TransactionMember object
def pkg_to_dep(pkg):
"""Extract the info from a hawkey.Package object
:param tm: A Yum transaction object
:type tm: TransactionMember
:param pkg: A hawkey.Package object
:type pkg: hawkey.Package
:returns: A dict with name, epoch, version, release, arch
:rtype: dict
"""
return {"name": tm.name,
"epoch": int(tm.epoch),
"version": tm.version,
"release": tm.release,
"arch": tm.arch}
return {"name": pkg.name,
"epoch": pkg.epoch,
"version": pkg.version,
"release": pkg.release,
"arch": pkg.arch}
def yaps_to_module(yaps):
"""Extract the name from a YumAvailablePackageSqlite object
def proj_to_module(proj):
"""Extract the name from a project_info dict
:param yaps: Yum object with package details
:type yaps: YumAvailablePackageSqlite
:param pkg: dict with package details
:type pkg: dict
:returns: A dict with name, and group_type
:rtype: dict
group_type is hard-coded to "rpm"
"""
return {"name": yaps.name,
return {"name": proj["name"],
"group_type": "rpm"}
@ -150,122 +149,126 @@ def dep_nevra(dep):
"""Return the name-epoch:version-release.arch"""
return dep["name"]+"-"+dep_evra(dep)
def projects_list(yb):
def projects_list(dbo):
"""Return a list of projects
:param yb: yum base object
:type yb: YumBase
:param dbo: dnf base object
:type dbo: dnf.Base
:returns: List of project info dicts with name, summary, description, homepage, upstream_vcs
:rtype: list of dicts
"""
try:
ybl = yb.doPackageLists(pkgnarrow="available", showdups=False)
except YumBaseError as e:
raise ProjectsError("There was a problem listing projects: %s" % str(e))
finally:
yb.closeRpmDB()
return sorted(map(yaps_to_project, ybl.available), key=lambda p: p["name"].lower())
return projects_info(dbo, None)
def projects_info(yb, project_names):
def projects_info(dbo, project_names):
"""Return details about specific projects
:param yb: yum base object
:type yb: YumBase
:param dbo: dnf base object
:type dbo: dnf.Base
:param project_names: List of names of projects to get info about
:type project_names: str
:returns: List of project info dicts with yaps_to_project as well as epoch, version, release, etc.
:returns: List of project info dicts with pkg_to_project as well as epoch, version, release, etc.
:rtype: list of dicts
If project_names is None it will return the full list of available packages
"""
try:
ybl = yb.doPackageLists(pkgnarrow="available", patterns=project_names, showdups=False)
except YumBaseError as e:
raise ProjectsError("There was a problem with info for %s: %s" % (project_names, str(e)))
finally:
yb.closeRpmDB()
return sorted(map(yaps_to_project_info, ybl.available), key=lambda p: p["name"].lower())
if project_names:
pkgs = dbo.sack.query().available().filter(name__glob=project_names)
else:
pkgs = dbo.sack.query().available()
return sorted(map(pkg_to_project_info, pkgs), key=lambda p: p["name"].lower())
def projects_depsolve(yb, project_names):
def projects_depsolve(dbo, project_names):
"""Return the dependencies for a list of projects
:param yb: yum base object
:type yb: YumBase
:param dbo: dnf base object
:type dbo: dnf.Base
:param project_names: The projects to find the dependencies for
:type project_names: List of Strings
:returns: NEVRA's of the project and its dependencies
:rtype: list of dicts
"""
# This resets the transaction
dbo.reset(goal=True)
for p in project_names:
try:
dbo.install(p)
except dnf.exceptions.MarkingError:
raise ProjectsError("No match for %s" % p)
try:
# This resets the transaction
yb.closeRpmDB()
for p in project_names:
yb.install(pattern=p)
(rc, msg) = yb.buildTransaction()
if rc not in [0, 1, 2]:
raise ProjectsError("There was a problem depsolving %s: %s" % (project_names, msg))
yb.tsInfo.makelists()
deps = sorted(map(tm_to_dep, yb.tsInfo.installed + yb.tsInfo.depinstalled), key=lambda p: p["name"].lower())
except YumBaseError as e:
dbo.resolve()
except dnf.exceptions.DepsolveError as e:
raise ProjectsError("There was a problem depsolving %s: %s" % (project_names, str(e)))
finally:
yb.closeRpmDB()
return deps
if len(dbo.transaction) == 0:
raise ProjectsError("No packages installed for %s" % project_names)
return sorted(map(pkg_to_dep, dbo.transaction.install_set), key=lambda p: p["name"].lower())
def estimate_size(packages, block_size=4096):
"""Estimate the installed size of a package list
:param packages: The packages to be installed
:type packages: list of TransactionMember objects
:type packages: list of hawkey.Package objects
:param block_size: The block size to use for rounding up file sizes.
:type block_size: int
:returns: The estimated size of installed packages
:rtype: int
Estimating actual requirements is difficult without the actual file sizes, which
yum doesn't provide access to. So use the file count and block size to estimate
dnf doesn't provide access to. So use the file count and block size to estimate
a minimum size for each package.
"""
installed_size = 0
for p in packages:
installed_size += len(p.po.filelist) * block_size
installed_size += p.po.installedsize
installed_size += len(p.files) * block_size
installed_size += p.installsize
return installed_size
def projects_depsolve_with_size(yb, project_names, with_core=True):
def projects_depsolve_with_size(dbo, project_names, with_core=True):
"""Return the dependencies and installed size for a list of projects
:param yb: yum base object
:type yb: YumBase
:param dbo: dnf base object
:type dbo: dnf.Base
:param project_names: The projects to find the dependencies for
:type project_names: List of Strings
:returns: installed size and a list of NEVRA's of the project and its dependencies
:rtype: tuple of (int, list of dicts)
"""
# This resets the transaction
dbo.reset(goal=True)
for p in project_names:
try:
dbo.install(p)
except dnf.exceptions.MarkingError:
raise ProjectsError("No match for %s" % p)
if with_core:
dbo.group_install("core", ['mandatory', 'default', 'optional'])
try:
# This resets the transaction
yb.closeRpmDB()
for p in project_names:
yb.install(pattern=p)
if with_core:
yb.selectGroup("core", group_package_types=['mandatory', 'default', 'optional'])
(rc, msg) = yb.buildTransaction()
if rc not in [0, 1, 2]:
raise ProjectsError("There was a problem depsolving %s: %s" % (project_names, msg))
yb.tsInfo.makelists()
installed_size = estimate_size(yb.tsInfo.installed + yb.tsInfo.depinstalled)
deps = sorted(map(tm_to_dep, yb.tsInfo.installed + yb.tsInfo.depinstalled), key=lambda p: p["name"].lower())
except YumBaseError as e:
dbo.resolve()
except dnf.exceptions.DepsolveError as e:
raise ProjectsError("There was a problem depsolving %s: %s" % (project_names, str(e)))
finally:
yb.closeRpmDB()
if len(dbo.transaction) == 0:
raise ProjectsError("No packages installed for %s" % project_names)
installed_size = estimate_size(dbo.transaction.install_set)
deps = sorted(map(pkg_to_dep, dbo.transaction.install_set), key=lambda p: p["name"].lower())
return (installed_size, deps)
def modules_list(yb, module_names):
def modules_list(dbo, module_names):
"""Return a list of modules
:param yb: yum base object
:type yb: YumBase
:param dbo: dnf base object
:type dbo: dnf.Base
:param offset: Number of modules to skip
:type limit: int
:param limit: Maximum number of modules to return
@ -275,37 +278,27 @@ def modules_list(yb, module_names):
Modules don't exist in RHEL7 so this only returns projects
and sets the type to "rpm"
"""
try:
ybl = yb.doPackageLists(pkgnarrow="available", patterns=module_names, showdups=False)
except YumBaseError as e:
raise ProjectsError("There was a problem listing modules: %s" % str(e))
finally:
yb.closeRpmDB()
return sorted(map(yaps_to_module, ybl.available), key=lambda p: p["name"].lower())
# TODO - Figure out what to do with this for Fedora 'modules'
projs = projects_info(dbo, module_names)
return sorted(map(proj_to_module, projs), key=lambda p: p["name"].lower())
def modules_info(yb, module_names):
def modules_info(dbo, module_names):
"""Return details about a module, including dependencies
:param yb: yum base object
:type yb: YumBase
:param dbo: dnf base object
:type dbo: dnf.Base
:param module_names: Names of the modules to get info about
:type module_names: str
:returns: List of dicts with module details and dependencies.
:rtype: list of dicts
"""
try:
# Get the info about each module
ybl = yb.doPackageLists(pkgnarrow="available", patterns=module_names, showdups=False)
except YumBaseError as e:
raise ProjectsError("There was a problem with info for %s: %s" % (module_names, str(e)))
finally:
yb.closeRpmDB()
modules = projects_info(dbo, module_names)
modules = sorted(map(yaps_to_project, ybl.available), key=lambda p: p["name"].lower())
# Add the dependency info to each one
for module in modules:
module["dependencies"] = projects_depsolve(yb, [module["name"]])
module["dependencies"] = projects_depsolve(dbo, [module["name"]])
return modules

98
tests/pylorax/test_projects.py
View File

@ -15,23 +15,19 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import os
import mock
import time
import shutil
import tempfile
import time
import unittest
from yum.Errors import YumBaseError
from pylorax.api.config import configure, make_yum_dirs
from pylorax.api.projects import api_time, api_changelog, yaps_to_project, yaps_to_project_info
from pylorax.api.projects import tm_to_dep, yaps_to_module, projects_list, projects_info, projects_depsolve
from pylorax.api.config import configure, make_dnf_dirs
from pylorax.api.projects import api_time, api_changelog, pkg_to_project, pkg_to_project_info, pkg_to_dep
from pylorax.api.projects import proj_to_module, projects_list, projects_info, projects_depsolve
from pylorax.api.projects import modules_list, modules_info, ProjectsError, dep_evra, dep_nevra
from pylorax.api.yumbase import get_base_object
from pylorax.api.dnfbase import get_base_object
class Yaps(object):
"""Test class for yaps tests"""
class Package(object):
"""Test class for hawkey.Package tests"""
name = "name"
summary = "summary"
description = "description"
@ -43,26 +39,13 @@ class Yaps(object):
license = "license"
version = "version"
def returnChangelog(self):
return [[0, 1, "Heavy!"]]
class TM(object):
"""Test class for tm test"""
name = "name"
epoch = 1
version = "version"
release = "release"
arch = "arch"
class ProjectsTest(unittest.TestCase):
@classmethod
def setUpClass(self):
self.tmp_dir = tempfile.mkdtemp(prefix="lorax.test.repo.")
self.config = configure(root_dir=self.tmp_dir, test_config=True)
make_yum_dirs(self.config)
self.yb = get_base_object(self.config)
make_dnf_dirs(self.config)
self.dbo = get_base_object(self.config)
os.environ["TZ"] = "UTC"
time.tzset()
@ -82,22 +65,22 @@ class ProjectsTest(unittest.TestCase):
def test_api_changelog_missing_text_entry(self):
self.assertEqual(api_changelog([('now', 'atodorov')]), '')
def test_yaps_to_project(self):
def test_pkg_to_project(self):
result = {"name":"name",
"summary":"summary",
"description":"description",
"homepage":"url",
"upstream_vcs":"UPSTREAM_VCS"}
y = Yaps()
self.assertEqual(yaps_to_project(y), result)
pkg = Package()
self.assertEqual(pkg_to_project(pkg), result)
def test_yaps_to_project_info(self):
def test_pkg_to_project_info(self):
build = {"epoch":1,
"release":"release",
"arch":"arch",
"build_time":"1985-10-27T01:00:00",
"changelog":"Heavy!",
"changelog":"CHANGELOG_NEEDED",
"build_config_ref": "BUILD_CONFIG_REF",
"build_env_ref": "BUILD_ENV_REF",
"metadata": {},
@ -113,25 +96,25 @@ class ProjectsTest(unittest.TestCase):
"upstream_vcs":"UPSTREAM_VCS",
"builds": [build]}
y = Yaps()
self.assertEqual(yaps_to_project_info(y), result)
pkg = Package()
self.assertEqual(pkg_to_project_info(pkg), result)
def test_tm_to_dep(self):
def test_pkg_to_dep(self):
result = {"name":"name",
"epoch":1,
"version":"version",
"release":"release",
"arch":"arch"}
tm = TM()
self.assertEqual(tm_to_dep(tm), result)
pkg = Package()
self.assertEqual(pkg_to_dep(pkg), result)
def test_yaps_to_module(self):
def test_proj_to_module(self):
result = {"name":"name",
"group_type":"rpm"}
y = Yaps()
self.assertEqual(yaps_to_module(y), result)
proj = pkg_to_project(Package())
self.assertEqual(proj_to_module(proj), result)
def test_dep_evra(self):
dep = {"arch": "noarch",
@ -158,60 +141,41 @@ class ProjectsTest(unittest.TestCase):
self.assertEqual(dep_nevra(dep), "basesystem-10.0-7.el7.noarch")
def test_projects_list(self):
projects = projects_list(self.yb)
projects = projects_list(self.dbo)
self.assertEqual(len(projects) > 10, True)
def test_projects_list_yum_raises_exception(self):
with self.assertRaises(ProjectsError):
with mock.patch.object(self.yb, 'doPackageLists', side_effect=YumBaseError('TESTING')):
projects_list(self.yb)
def test_projects_info(self):
projects = projects_info(self.yb, ["bash"])
projects = projects_info(self.dbo, ["bash"])
self.assertEqual(projects[0]["name"], "bash")
self.assertEqual(projects[0]["builds"][0]["source"]["license"], "GPLv3+")
def test_projects_info_yum_raises_exception(self):
with self.assertRaises(ProjectsError):
with mock.patch.object(self.yb, 'doPackageLists', side_effect=YumBaseError('TESTING')):
projects_info(self.yb, ["bash"])
def test_projects_depsolve(self):
deps = projects_depsolve(self.yb, ["bash"])
deps = projects_depsolve(self.dbo, ["bash"])
self.assertEqual(deps[0]["name"], "basesystem")
def test_projects_depsolve_fail(self):
with self.assertRaises(ProjectsError):
projects_depsolve(self.yb, ["nada-package"])
projects_depsolve(self.dbo, ["nada-package"])
def test_modules_list(self):
modules = modules_list(self.yb, None)
def test_modules_list_all(self):
modules = modules_list(self.dbo, None)
self.assertEqual(len(modules) > 10, True)
self.assertEqual(modules[0]["group_type"], "rpm")
modules = modules_list(self.yb, ["g*"])
def test_modules_list_glob(self):
modules = modules_list(self.dbo, ["g*"])
self.assertEqual(modules[0]["name"].startswith("g"), True)
def test_modules_list_yum_raises_exception(self):
with self.assertRaises(ProjectsError):
with mock.patch.object(self.yb, 'doPackageLists', side_effect=YumBaseError('TESTING')):
modules_list(self.yb, None)
def test_modules_info(self):
modules = modules_info(self.yb, ["bash"])
modules = modules_info(self.dbo, ["bash"])
print(modules)
self.assertEqual(modules[0]["name"], "bash")
self.assertEqual(modules[0]["dependencies"][0]["name"], "basesystem")
def test_modules_info_yum_raises_exception(self):
with self.assertRaises(ProjectsError):
with mock.patch.object(self.yb, 'doPackageLists', side_effect=YumBaseError('TESTING')):
modules_info(self.yb, ["bash"])
class ConfigureTest(unittest.TestCase):
@classmethod