diff --git a/install.requires b/install.requires
index f229360..804abb1 100644
--- a/install.requires
+++ b/install.requires
@@ -1 +1,2 @@
requests
+urllib3
diff --git a/pyproject.toml b/pyproject.toml
index d2d3dbd..322b2d5 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -35,7 +35,8 @@ Issues = "https://codeberg.org/AdamWill/rmdepcheck/issues"
Changelog = "https://codeberg.org/AdamWill/rmdepcheck/src/branch/main/CHANGELOG.md"
[project.scripts]
-rmdepcheck = "rmdepcheck:main"
+rmdepcheck = "rmdepcheck.rmdepcheck:main"
+rdcelwrap = "rmdepcheck.rdcelwrap:main"
[build-system]
requires = ["setuptools>=40.6.0", "setuptools-git", "wheel"]
@@ -47,7 +48,7 @@ branch = true
source_pkgs = ["rmdepcheck"]
[tool.coverage.paths]
-source = [".", ".tox/**/site-packages"]
+source = ["src", ".tox/**/site-packages"]
[tool.coverage.report]
show_missing = true
diff --git a/rdclocal.py b/rdclocal.py
new file mode 100755
index 0000000..fa4cde2
--- /dev/null
+++ b/rdclocal.py
@@ -0,0 +1,14 @@
+#!/usr/bin/python3
+
+"""Convenience wrapper for executing rdcelwrap from a git checkout."""
+
+import os
+import sys
+
+# add src subdirectory directory to module import path
+sys.path.insert(0, os.path.join(os.path.dirname(os.path.realpath(__file__)), "src"))
+
+from rmdepcheck.rmdepcheck import main # pylint: disable=wrong-import-position
+
+if __name__ == "__main__":
+ main()
diff --git a/rewlocal.py b/rewlocal.py
new file mode 100755
index 0000000..a9170ec
--- /dev/null
+++ b/rewlocal.py
@@ -0,0 +1,14 @@
+#!/usr/bin/python3
+
+"""Convenience wrapper for executing rdcelwrap from a git checkout."""
+
+import os
+import sys
+
+# add src subdirectory directory to module import path
+sys.path.insert(0, os.path.join(os.path.dirname(os.path.realpath(__file__)), "src"))
+
+from rmdepcheck.rdcelwrap import main # pylint: disable=wrong-import-position
+
+if __name__ == "__main__":
+ main()
diff --git a/src/rmdepcheck/__init__.py b/src/rmdepcheck/__init__.py
new file mode 100644
index 0000000..ae9f5b5
--- /dev/null
+++ b/src/rmdepcheck/__init__.py
@@ -0,0 +1,29 @@
+#!/usr/bin/python3
+
+# Copyright Red Hat
+#
+# This file is part of rmdepcheck.
+#
+# rmdepcheck 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 3 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 .
+#
+# Author(s): Adam Williamson
+
+
+"""RPM package installability and reverse-dependency checks using a
+repository modification strategy (hence 'rm').
+"""
+
+__version__ = "1.0.0"
+
+# vim: set textwidth=100 ts=8 et sw=4:
diff --git a/rdc-el-wrapper.py b/src/rmdepcheck/rdcelwrap.py
old mode 100755
new mode 100644
similarity index 69%
rename from rdc-el-wrapper.py
rename to src/rmdepcheck/rdcelwrap.py
index 30b03d8..cea0dbd
--- a/rdc-el-wrapper.py
+++ b/src/rmdepcheck/rdcelwrap.py
@@ -1,5 +1,3 @@
-#!/usr/bin/python3
-
# Copyright Red Hat
#
# This file is part of rmdepcheck.
@@ -20,7 +18,7 @@
# Author(s): Adam Williamson
-"""Opinionated rmdepcheck wrapper that handles the awkward Enterprise
+"""Opinionated rmdepcheck frontend that handles the awkward Enterprise
Linux case, where we have to test against multiple shipped variants
with complex lookaside relationships, all filtered from a shared build
root.
@@ -32,26 +30,32 @@ import argparse
import ast
import os
import pathlib
-import platform
import re
import shutil
import subprocess
import sys
import tempfile
-from functools import partial
-from urllib.parse import urlparse
from requests import Session
from requests.adapters import HTTPAdapter
-from requests.packages.urllib3.util.retry import Retry
+from urllib3.util.retry import Retry
+
+from rmdepcheck.shared import (
+ DNFARGS,
+ SUBPCAPTURE,
+ check_arch,
+ url_check,
+ check_utils,
+ get_source_packages,
+ get_base_repoclosure,
+ get_modified_and_new_repoclosure,
+ parse_display_exit,
+)
VrDict = dict[str, str]
-DNFTEMP = tempfile.TemporaryDirectory(prefix="rdcelndnf", dir="/var/tmp")
-DNFARGS = ["dnf", "--setopt", f"cachedir={DNFTEMP.name}", "-q", "--disablerepo=*"]
HERE = os.path.abspath(os.path.dirname(__file__))
SESSION = Session()
-SUBPCAPTURE = partial(subprocess.run, capture_output=True, text=True, check=False)
def split_put(put: pathlib.Path, repotemp: pathlib.Path, vrs: VrDict) -> set[str]:
@@ -74,7 +78,11 @@ def split_put(put: pathlib.Path, repotemp: pathlib.Path, vrs: VrDict) -> set[str
pkgs = {}
for variant, repo in vrs.items():
args = DNFARGS + [
- "--repofrompath", f"{variant},{repo}", "repoquery", "--queryformat", "%{name}\n"
+ "--repofrompath",
+ f"{variant},{repo}",
+ "repoquery",
+ "--queryformat",
+ "%{name}\n",
]
res = SUBPCAPTURE(args)
res.check_returncode()
@@ -84,7 +92,7 @@ def split_put(put: pathlib.Path, repotemp: pathlib.Path, vrs: VrDict) -> set[str
for _file in os.listdir(put):
if _file.endswith(".rpm"):
pkgname = _file.rsplit("-", 2)[0]
- for (variant, pkgnames) in pkgs.items():
+ for variant, pkgnames in pkgs.items():
if pkgname in pkgnames:
populated.add(variant)
os.makedirs(repotemp / variant, exist_ok=True)
@@ -136,33 +144,9 @@ def get_val():
return ast.literal_eval(match.group(1))
-def url_check(arg: str) -> str:
+def url_check_http(arg: str) -> str:
"""Check arg is an http or https URL."""
- parsed = urlparse(arg)
- if parsed.scheme in ("http", "https"):
- return arg
- if parsed.scheme:
- raise ValueError(f"Unsupported URL scheme {parsed.scheme} in {arg}")
- raise ValueError(f"No URL scheme in {arg}")
-
-
-def check_arch(arg: str) -> str:
- """Use host arch if no value passed."""
- if arg:
- return arg
- return platform.machine()
-
-
-def check_utils() -> None:
- """Check required utilities are installed."""
- missing = []
- for prog in (("dnf", "--version"), ("createrepo", "--version")):
- try:
- subprocess.run(prog, stdout=subprocess.DEVNULL, check=True)
- except FileNotFoundError:
- missing.append(prog[0])
- if missing:
- sys.exit("Please install missing required utilities: " + " ".join(missing))
+ return url_check(arg, fileok=False)
def parse_args() -> argparse.Namespace:
@@ -172,7 +156,7 @@ def parse_args() -> argparse.Namespace:
)
parser.add_argument(
"compose",
- type=url_check,
+ type=url_check_http,
help="The URL(s) of a compose to discover variant repos from and check against",
)
parser.add_argument(
@@ -186,10 +170,15 @@ def parse_args() -> argparse.Namespace:
default="",
help="Arch to operate on. Must match arch of passed repository",
)
+ parser.add_argument(
+ "--json",
+ action="store_true",
+ help="Output results in JSON rather than human-readable format",
+ )
return parser.parse_args()
-def main() -> None:
+def main() -> None: # pylint: disable=too-many-locals
"""Main loop."""
try:
# set up requests session with retries
@@ -197,18 +186,22 @@ def main() -> None:
total=5,
backoff_factor=0.5,
status_forcelist=[502, 503, 504],
- allowed_methods={'GET'},
+ allowed_methods={"GET"},
)
- SESSION.mount('https://', HTTPAdapter(max_retries=retries))
+ SESSION.mount("https://", HTTPAdapter(max_retries=retries))
- # check DNF is installed
- check_utils()
+ # check DNF and createrepo are installed
+ check_utils((("dnf", "--version"), ("createrepo", "--version")))
# set up args
args = parse_args()
if f"--forcearch={args.arch}" not in DNFARGS:
DNFARGS.append(f"--forcearch={args.arch}")
+ # repoclosure output strings
+ basercs = ""
+ modrcs = ""
+ newrcs = ""
# find variants, do the repo split
vrs = get_variant_repos(args.compose, args.arch)
val = get_val()
@@ -216,41 +209,40 @@ def main() -> None:
repotemp = pathlib.Path(repotd)
variants = split_put(args.repo, repotemp, vrs)
- rets = set()
for variant in variants:
+ baserepo = vrs[variant]
# figure out what other variants to pull in per val
depvars = set()
for pair in val:
if pair[0] == variant:
depvars.add(pair[1])
- # put the compose repos for all variants in basearg, even
- # ones we didn't 'populate' locally
- basearg = vrs[variant]
- ncbasearg = ",".join(vrs[var] for var in depvars)
- newarg = str(repotemp / variant)
- # put depended-on variant-split PUT repos that actually
- # exist in addrepos
+ # put the compose repos for all depended-on variants in
+ # ncbaserepos, even ones we didn't 'populate' locally
+ ncbaserepos = [vrs[var] for var in depvars]
+ # first newrepo is the PUT repo for this variant
+ newrepos = [str(repotemp / variant)]
+ # additional repos are depended-on variant-split PUT
+ # repos that actually exist
# variants contains only variants we 'populated'
- addarg = ",".join(str(repotemp / var) for var in depvars if var in variants)
- # run rmdepcheck, track return code
- # FIXME: what to do about JSON mode? capture output, combine?
- rargs = [f"{HERE}/rmdepcheck.py", "--arch", args.arch]
- if addarg:
- rargs.extend(("--addrepos", addarg))
- if ncbasearg:
- rargs.extend(("--ncbaserepos", ncbasearg))
- rargs.extend((basearg, newarg))
- rets.add(subprocess.run(rargs, check=False).returncode)
+ newrepos.extend(str(repotemp / var) for var in depvars if var in variants)
- # we exit the sum of all *unique* return codes
- sys.exit(sum(rets))
+ baserc = get_base_repoclosure([baserepo], ncbaserepos, [])
+ basercs = f"{basercs}\n{baserc}"
+
+ # get the modified rpmclosure output
+ sources = get_source_packages([newrepos[0]])
+ modrc, newrc = get_modified_and_new_repoclosure(
+ [baserepo], ncbaserepos, [], newrepos, sources
+ )
+ modrcs = f"{modrcs}\n{modrc}"
+ newrcs = f"{newrcs}\n{newrc}"
+
+ # output
+ parse_display_exit(basercs, modrcs, newrcs, "the tested packages", args.json, False)
except KeyboardInterrupt:
sys.stderr.write("Interrupted, exiting...\n")
sys.exit(1)
-if __name__ == "__main__": # pragma: no cover
- main()
-
# vim: set textwidth=100 ts=8 et sw=4:
diff --git a/src/rmdepcheck/rmdepcheck.py b/src/rmdepcheck/rmdepcheck.py
new file mode 100644
index 0000000..0b7f3f6
--- /dev/null
+++ b/src/rmdepcheck/rmdepcheck.py
@@ -0,0 +1,193 @@
+# Copyright Red Hat
+#
+# This file is part of rmdepcheck.
+#
+# rmdepcheck 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 3 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 .
+#
+# Author(s): Adam Williamson
+
+
+"""RPM package installability and reverse-dependency checks using a
+repository modification strategy (hence 'rm').
+"""
+
+# Standard libraries
+
+import argparse
+import sys
+
+from rmdepcheck.shared import (
+ DNFARGS,
+ check_arch,
+ url_check,
+ check_utils,
+ get_source_packages,
+ get_base_repoclosure,
+ get_modified_and_new_repoclosure,
+ parse_display_exit,
+)
+
+
+def url_check_file(arg: str) -> str:
+ """Check arg is a file, http or https URL. Automatically converts
+ local paths to file:// URIs."""
+ return url_check(arg, fileok=True)
+
+
+def comma_url(arg: str) -> list[str]:
+ """Check arg is a comma-separated list of URLs and return them
+ all as a list. If arg is the empty string, return empty list.
+ """
+ if arg == "":
+ return []
+ split = arg.split(",")
+ for item in split:
+ try:
+ url_check_file(item)
+ except ValueError as err:
+ newerr = str(err) + f"from {arg}"
+ raise ValueError(newerr) from err
+ return split
+
+
+def comma_list(arg: str) -> list[str]:
+ """Handle a comma-separated list, return as a list."""
+ if arg == "":
+ return []
+ return arg.split(",")
+
+
+def parse_args() -> argparse.Namespace:
+ """Parse arguments with argparse."""
+ parser = argparse.ArgumentParser(
+ description=("Reverse dependency check implemented as a repoclosure diff.")
+ )
+ parser.add_argument(
+ "--addrepos",
+ type=comma_url,
+ default="",
+ help="The URL(s) of additional repositories containing new packages to be tested "
+ "(comma-separated). This is mainly intended for multilib cases: i.e. for testing "
+ "x86_64 package sets it should contain the matching i686 packages. It will be "
+ "available to the repoclosure check, but will be ignored by the installability check",
+ )
+ parser.add_argument(
+ "--ncbaserepos",
+ type=comma_url,
+ default="",
+ help="The URL(s) of non-checked base repositories to compare against (comma-separated). "
+ "These repositories *will* be modified as part of testing, but they will not be checked "
+ "for repoclosure. They should be repositories whose packages would be replaced with those "
+ "from the new package repo(s), and which we want to be available to the dependency solver "
+ "when checking the repoclosure of other repositories, but which we are for some reason not "
+ "interested in checking the repoclosure of - e.g. the buildroot for Fedora ELN",
+ )
+ parser.add_argument(
+ "--nmbaserepos",
+ type=comma_url,
+ default="",
+ help="The URL(s) of non-modified base repositories to compare against (comma-separated). "
+ "These repositories *will not* be modified as part of testing. They should be repositories "
+ "whose packages would *not* be replaced with those from the new package repo(s) - e.g. "
+ "the frozen release repository for a stable release, which is never changed",
+ )
+ parser.add_argument(
+ "--json",
+ action="store_true",
+ help="Output results in JSON rather than human-readable format",
+ )
+ parser.add_argument(
+ "--onlyerrors",
+ action="store_true",
+ help="Only print messages about problems caused, not about things that would be fixed",
+ )
+ parser.add_argument(
+ "--removes",
+ action="store_true",
+ help="Alternative mode: test removal (only) of all binary packages in the baserepos built "
+ "from the source package(s) specified as the final argument (a comma-separated list)",
+ )
+ parser.add_argument(
+ "--arch",
+ type=check_arch,
+ default="",
+ help="Arch to operate on. Must match arch of passed repositories. Defaults to host arch",
+ )
+ parser.add_argument(
+ "baserepos",
+ type=comma_url,
+ help="The URL(s) of base repositories to compare against (comma-separated). "
+ "These repositories *will* be modified as part of testing. They should be "
+ "repositories whose packages would be replaced with those from the new package "
+ "repo(s) - e.g. the main repository for a development release, or the updates repository "
+ "for a stable release",
+ )
+ # I wanted to do this with parse_known_args, but it messes up --help. aw
+ if "--removes" in sys.argv:
+ parser.add_argument(
+ "removes",
+ type=comma_list,
+ help="A comma-separated list of source packages to test the removal of",
+ )
+ else:
+ parser.add_argument(
+ "repo",
+ metavar="repo_or_removes",
+ type=url_check_file,
+ help="The URL of the repo containing the main set of new packages to be tested, "
+ "or a comma-separated list of source packages to test the removal of (if --removes "
+ "is passed)",
+ )
+ args = parser.parse_args()
+ if args.removes:
+ args.repo = ""
+ else:
+ args.removes = ""
+ return args
+
+
+def main() -> None:
+ """Main loop."""
+ try:
+ check_utils((("dnf", "--version"),))
+ args = parse_args()
+ if f"--forcearch={args.arch}" not in DNFARGS:
+ DNFARGS.append(f"--forcearch={args.arch}")
+ if args.removes:
+ newrepos = []
+ sources = args.removes
+ iut = "the specified source package removals"
+ else:
+ newrepos = [args.repo] + args.addrepos
+ # find source package(s) of our tested repo(s)
+ # whether to include addrepos is arguable, but should usually be moot
+ sources = get_source_packages(newrepos)
+ iut = "the tested packages"
+
+ baseraw = get_base_repoclosure(args.baserepos, args.ncbaserepos, args.nmbaserepos)
+
+ # get the modified rpmclosure output
+ modraw, newraw = get_modified_and_new_repoclosure(
+ args.baserepos, args.ncbaserepos, args.nmbaserepos, newrepos, sources
+ )
+
+ # output
+ parse_display_exit(baseraw, modraw, newraw, iut, args.json, args.onlyerrors)
+
+ except KeyboardInterrupt:
+ sys.stderr.write("Interrupted, exiting...\n")
+ sys.exit(1)
+
+
+# vim: set textwidth=100 ts=8 et sw=4:
diff --git a/rmdepcheck.py b/src/rmdepcheck/shared.py
old mode 100755
new mode 100644
similarity index 51%
rename from rmdepcheck.py
rename to src/rmdepcheck/shared.py
index 41e9c06..049d374
--- a/rmdepcheck.py
+++ b/src/rmdepcheck/shared.py
@@ -1,5 +1,3 @@
-#!/usr/bin/python3
-
# Copyright Red Hat
#
# This file is part of rmdepcheck.
@@ -21,12 +19,10 @@
"""RPM package installability and reverse-dependency checks using a
-repository modification strategy (hence 'rm').
+repository modification strategy (hence 'rm') This file contains things
+shared between the two main scripts.
"""
-# Standard libraries
-
-import argparse
import hashlib
import json
import os
@@ -50,10 +46,61 @@ DNFTEMP = tempfile.TemporaryDirectory(prefix="rmdepcheck", dir="/var/tmp")
DNFARGS = ["dnf", "--setopt", f"cachedir={DNFTEMP.name}", "-q", "--disablerepo=*"]
SUBPCAPTURE = partial(subprocess.run, capture_output=True, text=True, check=False)
SUBPCAPTCHECK = partial(subprocess.run, capture_output=True, text=True, check=True)
-SUBPCHECK = partial(subprocess.run, check=True)
+SUBPCHECK = partial(subprocess.run, check=True) # pylint: disable=invalid-name
REPOHASHES: dict[str, str] = {}
+def check_arch(arg: str) -> str:
+ """Use host arch if no value passed."""
+ if arg:
+ return arg
+ return platform.machine()
+
+
+def url_check(arg: str, fileok: bool) -> str:
+ """Check arg is a file, http or https URL. Automatically converts
+ local paths to file:// URIs."""
+ # If it's an existing local path, convert to file:// URI
+ if os.path.exists(arg):
+ if fileok:
+ return f"file://{os.path.abspath(arg)}"
+ raise ValueError("Value must be an HTTP or HTTPS URL")
+
+ parsed = urlparse(arg)
+ schemes = ["http", "https"]
+ if fileok:
+ schemes.append("file")
+ if parsed.scheme in schemes:
+ return arg
+ if parsed.scheme:
+ raise ValueError(f"Unsupported URL scheme {parsed.scheme} in {arg}")
+ raise ValueError(f"No URL scheme in {arg}")
+
+
+def check_utils(utils: tuple[tuple[str, ...], ...]) -> None:
+ """Check required utilities are installed."""
+ missing = []
+ for prog in utils:
+ try:
+ subprocess.run(prog, stdout=subprocess.DEVNULL, check=True)
+ except FileNotFoundError:
+ missing.append(prog[0])
+ if missing:
+ sys.exit("Please install missing required utilities: " + " ".join(missing))
+
+
+def get_source_packages(repos: Iterable[str]) -> set[str]:
+ """Finds and returns the source package names for all packages in
+ the repositories specified, as a set.
+ """
+ gspargs = list(DNFARGS)
+ for repo in repos:
+ gspargs.extend(["--repofrompath", f"{hash_repo(repo)},{repo}"])
+ gspargs.extend(["repoquery", "--qf", "%{sourcerpm} "])
+ srpms = SUBPCAPTCHECK(gspargs).stdout.split()
+ return {srpm.rsplit("-", 2)[0] for srpm in srpms}
+
+
def hash_repo(repo: str) -> str:
"""Generate a hash for the repo name, stash it in a dict so we can
map back out later, and return it. This is so we can show the repo
@@ -138,8 +185,7 @@ def get_modified_and_new_repoclosure(
# figure out what to exclude
excludes = []
args = queryargs + ["--repofrompath", f"{hash_repo(mrepo)},{mrepo}"]
- out = SUBPCAPTURE(args).stdout
- for line in out.splitlines():
+ for line in SUBPCAPTURE(args).stdout.splitlines():
# sname, nevr
elems = line.split(",")
if len(elems) != 2:
@@ -174,18 +220,6 @@ def get_modified_and_new_repoclosure(
return (mod, new)
-def get_source_packages(repos: Iterable[str]) -> set[str]:
- """Finds and returns the source package names for all packages in
- the repositories specified, as a set.
- """
- args = list(DNFARGS)
- for repo in repos:
- args.extend(["--repofrompath", f"{hash_repo(repo)},{repo}"])
- args.extend(["repoquery", "--qf", "%{sourcerpm} "])
- srpms = SUBPCAPTCHECK(args).stdout.split()
- return {srpm.rsplit("-", 2)[0] for srpm in srpms}
-
-
def handle_preexisting(fixederrors: list[DepTuple], newrc: list[DepTuple]) -> list[DepTuple]:
"""Catch cases where an error 'moved' from baserc to newrc - a
package in the set under test previously had a dependency issue,
@@ -218,228 +252,62 @@ def handle_preexisting(fixederrors: list[DepTuple], newrc: list[DepTuple]) -> li
return preexisting
-def url_check(arg: str) -> str:
- """Check arg is a file, http or https URL. Automatically converts
- local paths to file:// URIs."""
- # If it's an existing local path, convert to file:// URI
- if os.path.exists(arg):
- return f"file://{os.path.abspath(arg)}"
-
- parsed = urlparse(arg)
- if parsed.scheme in ("http", "https", "file"):
- return arg
- if parsed.scheme:
- raise ValueError(f"Unsupported URL scheme {parsed.scheme} in {arg}")
- raise ValueError(f"No URL scheme in {arg}")
-
-
-def comma_url(arg: str) -> list[str]:
- """Check arg is a comma-separated list of URLs and return them
- all as a list. If arg is the empty string, return empty list.
+# pylint: disable-next=too-many-branches,too-many-arguments,too-many-positional-arguments
+def parse_display_exit(
+ baseraw: str, modraw: str, newraw: str, iut: str, dojson: bool, onlyerrors: bool
+) -> None:
+ """Take the raw rmdepcheck output strings, parse them, produce
+ human-readable or JSON output, and exit with an appropriate code.
"""
- if arg == "":
- return []
- split = arg.split(",")
- for item in split:
- try:
- url_check(item)
- except ValueError as err:
- newerr = str(err) + f"from {arg}"
- raise ValueError(newerr) from err
- return split
+ exitcode = 0
+ baserc = parse_repoclosure(baseraw)
+ modrc = parse_repoclosure(modraw)
+ newrc = []
+ if newraw:
+ newrc = parse_repoclosure(newraw)
+ newerrors = [dep for dep in modrc if dep not in baserc]
+ fixederrors = [dep for dep in baserc if dep not in modrc]
+ preexisting = handle_preexisting(fixederrors, newrc)
-def comma_list(arg: str) -> list[str]:
- """Handle a comma-separated list, return as a list."""
- if arg == "":
- return []
- return arg.split(",")
-
-
-def check_arch(arg: str) -> str:
- """Use host arch if no value passed."""
- if arg:
- return arg
- return platform.machine()
-
-
-def parse_args() -> argparse.Namespace:
- """Parse arguments with argparse."""
- parser = argparse.ArgumentParser(
- description=("Reverse dependency check implemented as a repoclosure diff.")
- )
- parser.add_argument(
- "--addrepos",
- type=comma_url,
- default="",
- help="The URL(s) of additional repositories containing new packages to be tested "
- "(comma-separated). This is mainly intended for multilib cases: i.e. for testing "
- "x86_64 package sets it should contain the matching i686 packages. It will be "
- "available to the repoclosure check, but will be ignored by the installability check",
- )
- parser.add_argument(
- "--ncbaserepos",
- type=comma_url,
- default="",
- help="The URL(s) of non-checked base repositories to compare against (comma-separated). "
- "These repositories *will* be modified as part of testing, but they will not be checked "
- "for repoclosure. They should be repositories whose packages would be replaced with those "
- "from the new package repo(s), and which we want to be available to the dependency solver "
- "when checking the repoclosure of other repositories, but which we are for some reason not "
- "interested in checking the repoclosure of - e.g. the buildroot for Fedora ELN",
- )
- parser.add_argument(
- "--nmbaserepos",
- type=comma_url,
- default="",
- help="The URL(s) of non-modified base repositories to compare against (comma-separated). "
- "These repositories *will not* be modified as part of testing. They should be repositories "
- "whose packages would *not* be replaced with those from the new package repo(s) - e.g. "
- "the frozen release repository for a stable release, which is never changed",
- )
- parser.add_argument(
- "--json",
- action="store_true",
- help="Output results in JSON rather than human-readable format",
- )
- parser.add_argument(
- "--onlyerrors",
- action="store_true",
- help="Only print messages about problems caused, not about things that would be fixed",
- )
- parser.add_argument(
- "--removes",
- action="store_true",
- help="Alternative mode: test removal (only) of all binary packages in the baserepos built "
- "from the source package(s) specified as the final argument (a comma-separated list)",
- )
- parser.add_argument(
- "--arch",
- type=check_arch,
- default="",
- help="Arch to operate on. Must match arch of passed repositories. Defaults to host arch",
- )
- parser.add_argument(
- "baserepos",
- type=comma_url,
- help="The URL(s) of base repositories to compare against (comma-separated). "
- "These repositories *will* be modified as part of testing. They should be "
- "repositories whose packages would be replaced with those from the new package "
- "repo(s) - e.g. the main repository for a development release, or the updates repository "
- "for a stable release",
- )
- # I wanted to do this with parse_known_args, but it messes up --help. aw
- if "--removes" in sys.argv:
- parser.add_argument(
- "removes",
- type=comma_list,
- help="A comma-separated list of source packages to test the removal of",
- )
- else:
- parser.add_argument(
- "repo",
- metavar="repo_or_removes",
- type=url_check,
- help="The URL of the repo containing the main set of new packages to be tested, "
- "or a comma-separated list of source packages to test the removal of (if --removes "
- "is passed)",
- )
- args = parser.parse_args()
- if args.removes:
- args.repo = ""
- else:
- args.removes = ""
- return args
-
-
-def check_dnf() -> None:
- """Check DNF is installed."""
- try:
- subprocess.run(("dnf", "--version"), stdout=subprocess.DEVNULL, check=True)
- except FileNotFoundError:
- sys.exit("Please install missing required utilities: dnf")
-
-
-def main() -> None:
- """Main loop."""
- try:
- check_dnf()
- exitcode = 0
- args = parse_args()
- if f"--forcearch={args.arch}" not in DNFARGS:
- DNFARGS.append(f"--forcearch={args.arch}")
- if args.removes:
- newrepos = []
- sources = args.removes
- iut = "the specified source package removals"
+ # output
+ if dojson:
+ jsonout = {}
+ if newerrors:
+ if dojson:
+ jsonout["newerrors"] = [list(err) for err in newerrors]
else:
- newrepos = [args.repo] + args.addrepos
- # find source package(s) of our tested repo(s)
- # whether to include addrepos is arguable, but should usually be moot
- sources = get_source_packages(newrepos)
- iut = "the tested packages"
+ print(f"Dependencies of other packages that would be BROKEN by {iut}:")
+ format_rc_errors(newerrors)
+ exitcode += 1
+ if newrc:
+ if dojson:
+ jsonout["installability"] = [list(err) for err in newrc]
+ else:
+ print("")
+ print("New dependency problems in the tested packages themselves:")
+ format_rc_errors(newrc)
+ exitcode += 2
+ if preexisting:
+ if dojson:
+ jsonout["installability_preexisting"] = [list(err) for err in preexisting]
+ else:
+ print("")
+ print("Pre-existing dependency problems in the tested packages themselves:")
+ format_rc_errors(preexisting)
+ exitcode += 4
+ if fixederrors:
+ if dojson:
+ jsonout["fixederrors"] = [list(err) for err in fixederrors]
+ elif not onlyerrors:
+ print("")
+ print(f"Dependencies of other packages that would be FIXED by {iut}:")
+ format_rc_errors(fixederrors)
+ if dojson:
+ json.dump(jsonout, sys.stdout, indent=4)
+ sys.stdout.write("\n")
- baseraw = get_base_repoclosure(args.baserepos, args.ncbaserepos, args.nmbaserepos)
- baserc = parse_repoclosure(baseraw)
+ sys.exit(exitcode)
- # get the modified rpmclosure output
- modraw, newraw = get_modified_and_new_repoclosure(
- args.baserepos, args.ncbaserepos, args.nmbaserepos, newrepos, sources
- )
- modrc = parse_repoclosure(modraw)
- newrc = []
- if newraw:
- newrc = parse_repoclosure(newraw)
-
- newerrors = [dep for dep in modrc if dep not in baserc]
- fixederrors = [dep for dep in baserc if dep not in modrc]
- preexisting = handle_preexisting(fixederrors, newrc)
-
- # output
- if args.json:
- jsonout = {}
- if newerrors:
- if args.json:
- jsonout["newerrors"] = [list(err) for err in newerrors]
- else:
- print(f"Dependencies of other packages that would be BROKEN by {iut}:")
- format_rc_errors(newerrors)
- exitcode += 1
- if newrc:
- if args.json:
- jsonout["installability"] = [list(err) for err in newrc]
- else:
- print("")
- print("New dependency problems in the tested packages themselves:")
- format_rc_errors(newrc)
- exitcode += 2
- if preexisting:
- if args.json:
- jsonout["installability_preexisting"] = [list(err) for err in preexisting]
- else:
- print("")
- print("Pre-existing dependency problems in the tested packages themselves:")
- format_rc_errors(preexisting)
- exitcode += 4
- if fixederrors:
- if args.json:
- jsonout["fixederrors"] = [list(err) for err in fixederrors]
- elif not args.onlyerrors:
- print("")
- print(f"Dependencies of other packages that would be FIXED by {iut}:")
- format_rc_errors(fixederrors)
- if args.json:
- json.dump(jsonout, sys.stdout, indent=4)
- sys.stdout.write("\n")
-
- sys.exit(exitcode)
-
- except KeyboardInterrupt:
- sys.stderr.write("Interrupted, exiting...\n")
- sys.exit(1)
-
-
-if __name__ == "__main__": # pragma: no cover
- main()
# vim: set textwidth=100 ts=8 et sw=4:
diff --git a/tests/test_rmdepcheck.py b/tests/test_rmdepcheck.py
index 0a0906e..9b5f0ab 100644
--- a/tests/test_rmdepcheck.py
+++ b/tests/test_rmdepcheck.py
@@ -26,8 +26,21 @@ import os
import sys
from unittest import mock
+from rmdepcheck.shared import (
+ check_arch,
+ url_check,
+ check_utils,
+ get_source_packages,
+ hash_repo,
+ parse_repoclosure,
+ format_rc_errors,
+ get_base_repoclosure,
+ get_modified_and_new_repoclosure,
+ handle_preexisting,
+)
+from rmdepcheck import rmdepcheck
+
import pytest
-import rmdepcheck
HERE = os.path.abspath(os.path.dirname(__file__))
TESTDATA = f"{HERE}/testdata"
@@ -37,7 +50,7 @@ REPOS = f"{TESTDATA}/repos"
def test_parse_repoclosure():
with open(f"{TESTDATA}/test_parse_repoclosure.txt", "r", encoding="utf-8") as fh:
rctext = fh.read()
- assert rmdepcheck.parse_repoclosure(rctext) == [
+ assert parse_repoclosure(rctext) == [
("python3-wxnatpy-0.4.0-13.fc42.noarch", "baserepo0", "python(abi) = 3.13"),
("python3-wxnatpy-0.4.0-13.fc42.noarch", "baserepo0", "python3.13dist(wxpython) >= 4"),
("python3-wxnatpy-0.4.0-13.fc42.noarch", "baserepo0", "python3.13dist(xnat) >= 0.3.3"),
@@ -52,7 +65,7 @@ def test_format_rc_errors(capsys):
("python3-wxnatpy-0.4.0-13.fc42.noarch", "baserepo0", "python3.13dist(xnat) >= 0.3.3"),
("python3-x3dh-1.0.4-3.fc43.noarch", "baserepo1", "python3.14dist(pydantic) >= 1.7.4"),
]
- rmdepcheck.format_rc_errors(errs)
+ format_rc_errors(errs)
captured = capsys.readouterr()
with open(f"{TESTDATA}/test_format_rc_errors.txt", "r", encoding="utf-8") as fh:
exptext = fh.read()
@@ -63,8 +76,8 @@ def test_get_base_repoclosure():
repo = f"file://{REPOS}/base"
with open(f"{TESTDATA}/test_get_base_repoclosure.txt", "r", encoding="utf-8") as testfh:
expected = testfh.read()
- expected = expected.replace("{HASH}", rmdepcheck.hash_repo(repo))
- ret = rmdepcheck.get_base_repoclosure([repo], [], [])
+ expected = expected.replace("{HASH}", hash_repo(repo))
+ ret = get_base_repoclosure([repo], [], [])
assert ret == expected
@@ -73,11 +86,11 @@ def test_get_modified_and_new_repoclosure():
nrepo = f"file://{REPOS}/new"
with open(f"{TESTDATA}/test_get_modified_repoclosure.txt", "r", encoding="utf-8") as testfh:
expectedmod = testfh.read()
- expectedmod = expectedmod.replace("{HASH}", rmdepcheck.hash_repo(brepo))
+ expectedmod = expectedmod.replace("{HASH}", hash_repo(brepo))
with open(f"{TESTDATA}/test_get_new_repoclosure.txt", "r", encoding="utf-8") as testfh:
expectednew = testfh.read()
- expectednew = expectednew.replace("{HASH}", rmdepcheck.hash_repo(nrepo))
- ret = rmdepcheck.get_modified_and_new_repoclosure(
+ expectednew = expectednew.replace("{HASH}", hash_repo(nrepo))
+ ret = get_modified_and_new_repoclosure(
[brepo], [], [], [nrepo], ["aaa", "ccc", "eee", "fff", "ggg"]
)
assert ret == (expectedmod, expectednew)
@@ -93,13 +106,13 @@ def test_get_modified_and_new_repoclosure_commasafe():
qamock.stdout = "ccc\nbbb,bbb-0:1.0-1.x86_64\nccc,ccc-0:1.0-1.x86_64"
rcmock = mock.Mock()
rcmock.stdout = ""
- with mock.patch("rmdepcheck.SUBPCAPTURE", side_effect=[qamock, rcmock]):
+ with mock.patch("rmdepcheck.shared.SUBPCAPTURE", side_effect=[qamock, rcmock]):
# this shouldn't raise an exception
- rmdepcheck.get_modified_and_new_repoclosure(["file:///foo/bar"], [], [], [], ["ccc"])
+ get_modified_and_new_repoclosure(["file:///foo/bar"], [], [], [], ["ccc"])
def test_get_source_packages():
- sources = rmdepcheck.get_source_packages([f"file://{REPOS}/new"])
+ sources = get_source_packages([f"file://{REPOS}/new"])
assert sources == {"111", "222", "aaa", "ccc", "eee", "fff", "ggg"}
@@ -119,7 +132,7 @@ def test_handle_preexisting():
]
# this should null out the 'matching' errors (the first three in
# each list), but leave the 'non-matching' ones (the others)
- ret = rmdepcheck.handle_preexisting(fixederrors, newrc)
+ ret = handle_preexisting(fixederrors, newrc)
assert fixederrors == [
("someotherpackage-1.0-1.eln155.x86_64", "newrepo", "someotherdep"),
]
@@ -137,21 +150,25 @@ def test_handle_preexisting():
def test_url_check():
- rmdepcheck.url_check("file:///foo/bar")
- rmdepcheck.url_check("https://www.some.where")
- rmdepcheck.url_check("http://some.where.insecure")
+ url_check("file:///foo/bar", fileok=True)
+ with pytest.raises(ValueError):
+ url_check("file:///foo/bar", fileok=False)
+ url_check("https://www.some.where", fileok=True)
+ url_check("http://some.where.insecure", fileok=True)
# Test automatic conversion of existing local paths to file:// URIs
- result = rmdepcheck.url_check(REPOS)
+ result = url_check(REPOS, fileok=True)
assert result == f"file://{REPOS}"
+ with pytest.raises(ValueError):
+ url_check(REPOS, fileok=False)
# Test with relative path that exists
- result = rmdepcheck.url_check("tests")
+ result = url_check("tests", fileok=True)
assert result.startswith("file://")
assert result.endswith("/tests")
with pytest.raises(ValueError):
- rmdepcheck.url_check("ftp://1997.called")
+ url_check("ftp://1997.called", fileok=True)
# Non-existent paths without scheme should still raise ValueError
with pytest.raises(ValueError):
- rmdepcheck.url_check("whatisthis")
+ url_check("whatisthis", fileok=True)
def test_comma_url():
@@ -175,20 +192,20 @@ def test_comma_list():
@mock.patch("platform.machine", return_value="foobar")
def test_check_arch(_):
- assert rmdepcheck.check_arch("x86_64") == "x86_64"
- assert rmdepcheck.check_arch("") == "foobar"
+ assert check_arch("x86_64") == "x86_64"
+ assert check_arch("") == "foobar"
@mock.patch("subprocess.run", autospec=True)
-def test_check_dnf(mock_run):
- rmdepcheck.check_dnf()
+def test_check_utils(mock_run):
+ check_utils((("dnf", "--version"),))
mock_run.side_effect = FileNotFoundError
with pytest.raises(SystemExit) as excinfo:
- rmdepcheck.check_dnf()
+ check_utils((("dnf", "--version"),))
assert excinfo.value.code == "Please install missing required utilities: dnf"
-@mock.patch("rmdepcheck.check_dnf", side_effect=KeyboardInterrupt)
+@mock.patch("rmdepcheck.rmdepcheck.check_utils", side_effect=KeyboardInterrupt)
def test_ctrl_c(_, capsys):
with pytest.raises(SystemExit) as excinfo:
rmdepcheck.main()
@@ -316,3 +333,6 @@ def test_e2e_unfixed(output, capsys):
exptext = fh.read()
exptext = exptext.replace("{REPOS}", REPOS)
assert captured.out == exptext
+
+
+# vim: set textwidth=100 ts=8 et sw=4:
diff --git a/tox.ini b/tox.ini
index d96cc57..0135d40 100644
--- a/tox.ini
+++ b/tox.ini
@@ -7,8 +7,6 @@ deps =
-r{toxinidir}/tests.requires
commands =
coverage run -m pytest {posargs}
-setenv =
- PYTHONPATH = {toxinidir}
[testenv:ci]
sitepackages = false
deps =
@@ -16,11 +14,12 @@ deps =
skip_install = true
ignore_errors = true
commands =
- black --check rmdepcheck.py tests/test_rmdepcheck.py tests/testdata/repos/mkrepos.py
+ black --check rdclocal.py rewlocal.py src/rmdepcheck tests/test_rmdepcheck.py tests/testdata/repos/mkrepos.py
coverage combine
coverage report
coverage xml
diff-cover coverage.xml --fail-under=100
diff-quality --violations=pylint --fail-under=100
- mypy rmdepcheck.py
-
+ mypy src/rmdepcheck
+setenv =
+ PYTHONPATH = {toxinidir}/src
diff --git a/tox.requires b/tox.requires
index 70d78d6..c25f561 100644
--- a/tox.requires
+++ b/tox.requires
@@ -4,3 +4,6 @@ diff-cover
mypy
pylint
pytest-cov
+requests
+types-requests
+urllib3