diff --git a/dist/package/openSUSE-release-tools.spec b/dist/package/openSUSE-release-tools.spec index 987119de6..78e12748b 100644 --- a/dist/package/openSUSE-release-tools.spec +++ b/dist/package/openSUSE-release-tools.spec @@ -462,6 +462,7 @@ exit 0 %{_bindir}/osrt-git-openqa-maintenance %{_bindir}/osrt-repo2fileprovides %{_bindir}/osrt-requestfinder +%{_bindir}/osrt-sle_check %{_bindir}/osrt-totest-manager %{_datadir}/%{source_dir} %exclude %{_datadir}/%{source_dir}/abichecker diff --git a/sle_check.py b/sle_check.py new file mode 100755 index 000000000..422e28545 --- /dev/null +++ b/sle_check.py @@ -0,0 +1,526 @@ +#!/usr/bin/python3 +"""Gitea ReviewBot: gate products/SLES productcompose maintainership. + +Reviews products/SLES PRs. For each PR it fetches the SLFO _maintainership.json once, validates +that every maintainer login is a confirmed OBS account, and — if the PR changes +000productcompose/default.productcompose — resolves the binaries it adds to their source packages +and reports sources with no maintainer (orphans). Both checks decline the review (aggregated), with +the reason in the review body and the logs. OBS access uses the osc library; Gitea access uses the +platform API. +""" + +import logging +import re +import sys +import traceback +import urllib.error + +import requests + +from collections import namedtuple + +import ReviewBot + +from osc import conf +from osc.core import http_GET, makeurl, xml_parse + +DEFAULT_ALLOW_REPOS = "" +PRODUCTCOMPOSE_PATH = "000productcompose/default.productcompose" +SOURCE_PROJECT = "SUSE:SLFO:Main" +MAINTAINERSHIP_OWNER = "products" +MAINTAINERSHIP_REPO = "SLFO" +MAINTAINERSHIP_REF = "slfo-main" +MAINTAINERSHIP_FILE = "_maintainership.json" +PERSON_BATCH_SIZE = 50 +PUBLISHED_BATCH_SIZE = 50 + +# A maintainer login safe to embed in an XPath string literal. +_LOGIN_RE = re.compile(r"^[A-Za-z0-9_.@-]+$") +# A binary name safe to embed in an XPath string literal (published-binary fallback query). +# \Z (not $) so a trailing newline is rejected — the barrier must not rely on upstream stripping. +_BINARY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._+-]*\Z") +# An added productcompose package line "+ - name" (name stops at whitespace or '#'). +_ADDED_BINARY_RE = re.compile(r"^\+\s+-\s+([A-Za-z0-9][^\s#]*)", re.MULTILINE) + +OrphanResult = namedtuple("OrphanResult", ["orphans", "failed", "checked"]) +UserCheckResult = namedtuple("UserCheckResult", ["confirmed", "invalid", "not_found"]) + + +# --------------------------------------------------------------------------- # +# Pure functions (no I/O) +# --------------------------------------------------------------------------- # + + +def extract_file_diff(diff_text, path): + """Return the section of a (multi-file) unified diff for a single file ("" if absent).""" + target = f"diff --git a/{path} b/{path}" + section = [] + capturing = False + for line in diff_text.splitlines(keepends=True): + if line.startswith("diff --git "): + capturing = line.rstrip("\r\n") == target + if capturing: + section.append(line) + return "".join(section) + + +def parse_added_binaries(file_diff_text): + """Return sorted unique binary names added in a single file's diff section.""" + return sorted(set(_ADDED_BINARY_RE.findall(file_diff_text))) + + +def added_binaries_from_diff(diff_text): + """Return sorted unique binaries added to the productcompose in a PR diff.""" + return parse_added_binaries(extract_file_diff(diff_text, PRODUCTCOMPOSE_PATH)) + + +def build_source_map(root): + """Build a {binary -> base source package} map from view=info&parse=1 XML. + + Root tag-agnostic; multibuild flavors (graphviz:qt6) are stripped to the base name. + """ + result = {} + for sourceinfo in root.iter("sourceinfo"): + pkg = sourceinfo.get("package") + if not pkg: + continue + pkg = pkg.split(":", 1)[0] + for subpacks in sourceinfo.findall("subpacks"): + binary = (subpacks.text or "").strip() + if binary: + result[binary] = pkg + return result + + +def resolve_sources(binaries, source_map): + """Return (sorted unique resolved sources, binaries that did not resolve).""" + sources = set() + failed = [] + for binary in binaries: + source = source_map.get(binary) + if source: + sources.add(source) + else: + failed.append(binary) + return sorted(sources), failed + + +def merge_fallback_sources(sources, failed, resolved): + """Fold a fallback ``{binary -> [source]}`` map into already-resolved sources. + + Returns ``(sorted_sources, still_failed)`` where ``still_failed`` lists the failed binaries + that the fallback did not resolve. ``resolved`` may map one binary to several sources. + """ + merged = set(sources) + still_failed = [] + for binary in failed: + pkgs = resolved.get(binary) + if pkgs: + merged.update(pkgs) + else: + still_failed.append(binary) + return sorted(merged), still_failed + + +def normalize_maintainership(data): + """Return a {package -> {"users": [...], "groups": [...]}} map. + + Mirrors check_bugowner.py format detection (1.0 vs legacy unversioned). Raises ValueError on an + obs-maintainers header with an unsupported version/document. + """ + if {"header", "project", "packages"} <= set(data.keys()): + header = data["header"] + if {"document", "version"} == set(header.keys()) and header.get( + "document" + ) == "obs-maintainers": + if header["version"] == "1.0": + return data["packages"] + raise ValueError( + f"Unsupported maintainership file version {header['version']}" + ) + raise ValueError( + f"Unsupported maintainership file format {header.get('document')!r}" + ) + return {pkg: {"users": users or [], "groups": []} for pkg, users in data.items()} + + +def is_orphan(db, pkg): + """True if pkg has no maintainer (missing, falsy, or empty users AND groups).""" + entry = db.get(pkg) + if not entry: + return True + return not ((entry.get("users") or []) or (entry.get("groups") or [])) + + +def find_orphans(sources, db): + """Return the sorted source packages from ``sources`` that have no maintainer.""" + return sorted(src for src in sources if is_orphan(db, src)) + + +def extract_users(db): + """Return sorted unique maintainer logins from a normalized DB (groups ignored). + + Raises ValueError on an empty/non-string login or one unsafe to embed in an XPath literal. + """ + seen = set() + for pkg, entry in db.items(): + for login in entry.get("users") or []: + if not isinstance(login, str) or not login: + raise ValueError(f"invalid user entry {login!r} in package {pkg!r}") + if not _LOGIN_RE.match(login): + raise ValueError( + f"login {login!r} contains characters unsafe for XPath" + ) + seen.add(login) + return sorted(seen) + + +def build_person_match(users): + """Return an OBS /search/person XPath match predicate for a batch of logins.""" + return "(" + " or ".join(f"@login='{u}'" for u in users) + ")" + + +def build_binary_source_match(names, project): + """Return an OBS /search/published/binary/id XPath match for a batch of binary names. + + Assumes ``names`` are pre-filtered safe for embedding in an XPath string literal. + """ + return ( + "(" + " or ".join(f"@name='{n}'" for n in names) + ")" + + f" and @project='{project}'" + ) + + +def parse_published_sources(root, project): + """Return {binary -> sorted[source pkgs]} from a /search/published/binary/id response. + + Only ```` rows whose ``project`` attr exactly equals ``project`` are kept (so nested + projects like SUSE:SLFO:Main:Staging:V are excluded); multibuild ``:flavor`` is stripped from + the ``package`` attr; rows missing ``name`` or ``package`` are skipped. A binary may map to + more than one source. + """ + grouped = {} + for binary in root.iter("binary"): + if binary.get("project") != project: + continue + name = binary.get("name") + pkg = binary.get("package") + if not name or not pkg: + continue + pkg = pkg.split(":", 1)[0] + grouped.setdefault(name, set()).add(pkg) + return {name: sorted(pkgs) for name, pkgs in grouped.items()} + + +def parse_persons(root): + """Return {login: state} from an OBS /search/person response (no-login records skipped).""" + return { + person.findtext("login"): person.findtext("state") + for person in root.findall("person") + if person.findtext("login") + } + + +def classify_users(users, found): + """Classify each login as confirmed / invalid (other state) / not_found (absent).""" + confirmed = [] + invalid = [] + not_found = [] + for user in users: + if user not in found: + not_found.append(user) + elif found[user] == "confirmed": + confirmed.append(user) + else: + invalid.append(user) + return UserCheckResult(confirmed, invalid, not_found) + + +def find_orphan_sources(binaries, source_info_root, db, fallback_resolver=None): + """Resolve added binaries to sources and return the orphans among them. + + Binaries the static ``view=info`` map cannot resolve are, when ``fallback_resolver`` is given, + passed to it (a ``list[str] -> {binary: [source]}`` callable) so dynamically-named binaries + (KMPs, livepatches, python-flavor subpackages) still get orphan-checked. Only genuinely + unresolvable binaries remain in ``failed``. + """ + source_map = build_source_map(source_info_root) + sources, failed = resolve_sources(binaries, source_map) + if failed and fallback_resolver: + resolved = fallback_resolver(failed) + sources, failed = merge_fallback_sources(sources, failed, resolved) + return OrphanResult(find_orphans(sources, db), failed, len(sources)) + + +def _bullets(items): + return "\n".join(f"- {i}" for i in items) + + +def build_declined_message(orphans, invalid_users, not_found_users): + """Build the aggregated decline message ("" if nothing to report).""" + sections = [] + if orphans: + sections.append( + f"New source packages with no maintainer in {MAINTAINERSHIP_FILE}:\n\n" + + _bullets(orphans) + ) + if not_found_users: + sections.append( + "Maintainer logins not found in OBS:\n\n" + + _bullets(sorted(not_found_users)) + ) + if invalid_users: + sections.append( + "Maintainer logins that are not confirmed OBS accounts:\n\n" + + _bullets(sorted(invalid_users)) + ) + return "\n\n".join(sections) + + +def build_accepted_message(unresolved): + """Build the acceptance message, noting any unresolved (unchecked) binaries.""" + if unresolved: + return ( + "Maintainership check passed.\n\n" + "Warning - binaries that could not be resolved to a source package " + "(not checked):\n\n" + _bullets(sorted(unresolved)) + ) + return "Maintainership check passed." + + +# --------------------------------------------------------------------------- # +# Thin I/O wrappers (mocked in tests) +# --------------------------------------------------------------------------- # + + +def fetch_pr_diff(platform, owner, repo, pr_id): + """Return the unified diff text of a Gitea pull request.""" + return platform.get_path(f"repos/{owner}/{repo}/pulls/{pr_id}.diff").text + + +def fetch_maintainership(platform, owner, repo, ref): + """Fetch and normalize _maintainership.json from a Gitea repo ref (raw API).""" + data = platform.get_path( + f"repos/{owner}/{repo}/raw/{MAINTAINERSHIP_FILE}?ref={ref}" + ).json() + return normalize_maintainership(data) + + +def fetch_source_info(apiurl, project): + """Return the parsed root Element of /source/?view=info&parse=1.""" + url = makeurl(apiurl, ["source", project], {"view": "info", "parse": "1"}) + return xml_parse(http_GET(url)).getroot() + + +def query_persons(apiurl, users): + """Return {login: state} for a batch of logins via OBS /search/person.""" + url = makeurl(apiurl, ["search", "person"], {"match": build_person_match(users)}) + return parse_persons(xml_parse(http_GET(url)).getroot()) + + +def query_published_sources(apiurl, names, project, batch_size=PUBLISHED_BATCH_SIZE): + """Resolve binary names to their source packages via the published-binary index. + + Names unsafe to embed in an XPath string literal are dropped (they stay unresolved). Safe + names are queried in batches, filtered server-side to ``project``; results are merged into one + ``{binary -> [sources]}`` dict. + """ + safe = [name for name in names if _BINARY_RE.match(name)] + resolved = {} + for offset in range(0, len(safe), batch_size): + batch = safe[offset:offset + batch_size] + url = makeurl( + apiurl, + ["search", "published", "binary", "id"], + {"match": build_binary_source_match(batch, project)}, + ) + root = xml_parse(http_GET(url)).getroot() + resolved.update(parse_published_sources(root, project)) + return resolved + + +def check_user_validity(db, apiurl, batch_size=PERSON_BATCH_SIZE): + """Validate every maintainer login in the DB against OBS, in batches.""" + users = extract_users(db) + confirmed = [] + invalid = [] + not_found = [] + for offset in range(0, len(users), batch_size): + batch = users[offset:offset + batch_size] + result = classify_users(batch, query_persons(apiurl, batch)) + confirmed.extend(result.confirmed) + invalid.extend(result.invalid) + not_found.extend(result.not_found) + return UserCheckResult(confirmed, invalid, not_found) + + +# --------------------------------------------------------------------------- # +# Bot +# --------------------------------------------------------------------------- # + + +class SleCheckBot(ReviewBot.ReviewBot): + """A review bot that gates productcompose maintainership on products/SLES PRs.""" + + # ReviewBot.apiurl setter (ReviewBot.py:173-176) unconditionally accesses self.scm and + # self.platform. Tests construct SleCheckBot via object.__new__() to bypass __init__, then + # assign bot.apiurl directly — at that point neither scm nor platform exists yet, causing + # AttributeError. This override guards both accesses with getattr so the setter is safe on + # a partially-initialised instance while remaining behaviourally identical in production + # (scm and platform are always present after ReviewBot.__init__ completes). + # Confirmed against ReviewBot.py:106,114,138-139,149,157,173-176 and plat/gitea.py:296. + @property + def apiurl(self): + return self._apiurl + + @apiurl.setter + def apiurl(self, url): + self._apiurl = url + if getattr(self, "scm", None) is not None and self.scm.name == "OSC": + self.scm.apiurl = url + if getattr(self, "platform", None) is not None and self.platform.name == "OBS": + self.platform.apiurl = url + + def __init__(self, *args, **kwargs): + ReviewBot.ReviewBot.__init__(self, *args, **kwargs) + + conf.get_config() + + self.apiurl = conf.config["apiurl"] + + self.allowed_repositories = [] + + # This is heavily dependent on the GITEA platform + if self.platform.name != "GITEA": + raise Exception("Unsupported platform: this bot is only supported on Gitea") + + @staticmethod + def get_request_from_src_rev(requests_list, src_rev): + for request in requests_list: + if request.actions[0].src_rev == src_rev: + return request + return None + + def check_source_submission( + self, src_owner, src_project, src_rev, target_owner, target_package + ): + self.logger.info( + f"Checking {src_owner}/{src_project} -> {target_owner}/{target_package}" + ) + + try: + result, message = self.run_check( + src_owner, src_project, src_rev, target_owner, target_package + ) + except ( + requests.exceptions.RequestException, + urllib.error.HTTPError, + urllib.error.URLError, + ): + self.logger.warning( + "transient error, will retry next run:\n%s", traceback.format_exc() + ) + return None + except Exception: + self.review_messages["declined"] = ( + f"Unhandled exception:\n\n```{traceback.format_exc()}```" + ) + return False + + if result: + self.review_messages["accepted"] = message or "OK" + return result + + def run_check(self, src_owner, src_project, src_rev, target_owner, target_package): + """Run both checks. Returns (result, message): True/False/None and an accept message.""" + request = self.get_request_from_src_rev(self.requests, src_rev) + if not request: + self.logger.warning(f"Request for src_rev {src_rev} not found") + return None, None + + if f"{request._owner}/{request._repo}" not in self.allowed_repositories: + self.logger.info( + f"{request._owner}/{request._repo} is not in the allowed repositories list" + ) + return None, None + + db = fetch_maintainership( + self.platform, MAINTAINERSHIP_OWNER, MAINTAINERSHIP_REPO, MAINTAINERSHIP_REF + ) + + # 1. User-validity check (always). + users = check_user_validity(db, self.apiurl) + + # 2. Orphan check (only if the PR changes the productcompose). + binaries = added_binaries_from_diff( + fetch_pr_diff(self.platform, request._owner, request._repo, request._pr_id) + ) + if binaries: + orphans = find_orphan_sources( + binaries, + fetch_source_info(self.apiurl, SOURCE_PROJECT), + db, + fallback_resolver=lambda names: query_published_sources( + self.apiurl, names, SOURCE_PROJECT + ), + ) + else: + orphans = OrphanResult([], [], 0) + + if orphans.failed: + self.logger.warning( + "Binaries with no source package (not checked): %s", + ", ".join(orphans.failed), + ) + if orphans.orphans: + self.logger.warning( + "Orphan source packages (no maintainer): %s", ", ".join(orphans.orphans) + ) + if users.not_found: + self.logger.warning( + "Maintainer logins not found in OBS: %s", + ", ".join(sorted(users.not_found)), + ) + if users.invalid: + self.logger.warning( + "Maintainer logins not confirmed: %s", ", ".join(sorted(users.invalid)) + ) + + if orphans.orphans or users.invalid or users.not_found: + self.review_messages["declined"] = build_declined_message( + orphans.orphans, users.invalid, users.not_found + ) + return False, None + + return True, build_accepted_message(orphans.failed) + + +class CommandLineInterface(ReviewBot.CommandLineInterface): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.clazz = SleCheckBot + + def get_optparser(self): + parser = super().get_optparser() + + parser.add_option( + "--git-allow-repos", + default=DEFAULT_ALLOW_REPOS, + help="allowed git repositories (e.g. products/SLFO,products/SLES)", + ) + + return parser + + def setup_checker(self): + instance = super().setup_checker() + + instance.allowed_repositories = [r for r in self.options.git_allow_repos.split(",") if r] + + return instance + + +if __name__ == "__main__": + app = CommandLineInterface() + logging.basicConfig(level=logging.INFO) + + sys.exit(app.main()) diff --git a/tests/sle_check_tests.py b/tests/sle_check_tests.py new file mode 100644 index 000000000..c1ae92919 --- /dev/null +++ b/tests/sle_check_tests.py @@ -0,0 +1,626 @@ +import logging +import unittest +import urllib.error +from unittest import mock +from xml.etree import ElementTree as ET + +import requests as requests_lib + +import sle_check +from sle_check import ( + added_binaries_from_diff, + build_accepted_message, + build_binary_source_match, + build_declined_message, + build_person_match, + build_source_map, + check_user_validity, + classify_users, + extract_file_diff, + extract_users, + find_orphan_sources, + find_orphans, + is_orphan, + merge_fallback_sources, + normalize_maintainership, + parse_added_binaries, + parse_persons, + parse_published_sources, + query_published_sources, + resolve_sources, +) + +SOURCEINFO_XML = """ + + libguestfs + libguestfs-appliance + + + systemd + udev + + + graphviz-gd + graphviz-gnome + + + ignored + +""" + +PERSONS_XML = """ + aliceconfirmed + boblocked + confirmed +""" + +PUBLISHED_XML = """ + + + + + + + + +""" + +MULTI_FILE_DIFF = ( + "diff --git a/other.yaml b/other.yaml\n" + "--- a/other.yaml\n" + "+++ b/other.yaml\n" + "@@\n" + "+ - not-a-binary\n" + "diff --git a/000productcompose/default.productcompose" + " b/000productcompose/default.productcompose\n" + "--- a/000productcompose/default.productcompose\n" + "+++ b/000productcompose/default.productcompose\n" + "@@\n" + " context\n" + "+ - libguestfs-appliance\n" + "+ - systemd # comment\n" + "+ - graphviz-gd# nospace\n" + "- - removed\n" +) + + +class ParseAddedBinariesTests(unittest.TestCase): + def test_added_only_and_comment_strip(self): + patch = " context\n+ - aaa\n+ - bbb # c\n+ - ccc#d\n- - removed\n" + self.assertEqual(parse_added_binaries(patch), ["aaa", "bbb", "ccc"]) + + def test_dedup_and_sort(self): + self.assertEqual( + parse_added_binaries("+ - b\n+ - a\n+ - b\n"), ["a", "b"] + ) + + def test_empty(self): + self.assertEqual(parse_added_binaries(""), []) + + +class ExtractFileDiffTests(unittest.TestCase): + def test_isolates_target_file(self): + self.assertEqual( + added_binaries_from_diff(MULTI_FILE_DIFF), + ["graphviz-gd", "libguestfs-appliance", "systemd"], + ) + + def test_absent_file(self): + self.assertEqual( + extract_file_diff( + "diff --git a/x b/x\n+ - y\n", sle_check.PRODUCTCOMPOSE_PATH + ), + "", + ) + self.assertEqual(added_binaries_from_diff("diff --git a/x b/x\n+ - y\n"), []) + + def test_crlf_line_endings(self): + crlf_diff = MULTI_FILE_DIFF.replace("\n", "\r\n") + self.assertEqual( + added_binaries_from_diff(crlf_diff), + ["graphviz-gd", "libguestfs-appliance", "systemd"], + ) + + +class BuildSourceMapTests(unittest.TestCase): + def setUp(self): + self.m = build_source_map(ET.fromstring(SOURCEINFO_XML)) + + def test_maps_subpacks(self): + self.assertEqual(self.m["libguestfs-appliance"], "libguestfs") + self.assertEqual(self.m["udev"], "systemd") + + def test_strips_flavor(self): + self.assertEqual(self.m["graphviz-gd"], "graphviz") + self.assertEqual(self.m["graphviz-gnome"], "graphviz") + + def test_skips_empty_package(self): + self.assertNotIn("ignored", self.m) + + +class ResolveSourcesTests(unittest.TestCase): + def test_hits_and_misses(self): + sm = {"libguestfs-appliance": "libguestfs", "udev": "systemd"} + self.assertEqual( + resolve_sources(["libguestfs-appliance", "udev", "nope"], sm), + (["libguestfs", "systemd"], ["nope"]), + ) + + def test_dedup(self): + self.assertEqual(resolve_sources(["a", "b"], {"a": "s", "b": "s"}), (["s"], [])) + + +class NormalizeMaintainershipTests(unittest.TestCase): + def test_one_dot_o(self): + data = { + "header": {"document": "obs-maintainers", "version": "1.0"}, + "project": "p", + "packages": {"foo": {"users": ["a"], "groups": []}}, + } + self.assertEqual(normalize_maintainership(data)["foo"]["users"], ["a"]) + + def test_legacy(self): + db = normalize_maintainership({"foo": ["a"], "bar": []}) + self.assertEqual(db["foo"], {"users": ["a"], "groups": []}) + self.assertEqual(db["bar"], {"users": [], "groups": []}) + + def test_unsupported_version(self): + with self.assertRaises(ValueError): + normalize_maintainership( + { + "header": {"document": "obs-maintainers", "version": "2.0"}, + "project": "p", + "packages": {}, + } + ) + + def test_bad_document(self): + with self.assertRaises(ValueError): + normalize_maintainership( + { + "header": {"document": "other", "version": "1.0"}, + "project": "p", + "packages": {}, + } + ) + + +class OrphanPredicateTests(unittest.TestCase): + def setUp(self): + self.db = { + "hu": {"users": ["a"], "groups": []}, + "hg": {"users": [], "groups": ["g"]}, + "e": {"users": [], "groups": []}, + "n": None, + } + + def test_is_orphan(self): + self.assertFalse(is_orphan(self.db, "hu")) + self.assertFalse(is_orphan(self.db, "hg")) + self.assertTrue(is_orphan(self.db, "e")) + self.assertTrue(is_orphan(self.db, "n")) + self.assertTrue(is_orphan(self.db, "missing")) + + def test_find_orphans_sorted(self): + self.assertEqual( + find_orphans(["hu", "e", "missing"], self.db), ["e", "missing"] + ) + + +class ExtractUsersTests(unittest.TestCase): + def test_unique_sorted_groups_ignored(self): + db = { + "a": {"users": ["alice", "bob"], "groups": ["g"]}, + "b": {"users": ["alice"], "groups": []}, + } + self.assertEqual(extract_users(db), ["alice", "bob"]) + + def test_unsafe_login(self): + with self.assertRaises(ValueError): + extract_users({"a": {"users": ["bad login"], "groups": []}}) + + def test_empty_login(self): + with self.assertRaises(ValueError): + extract_users({"a": {"users": [""], "groups": []}}) + + +class PersonTests(unittest.TestCase): + def test_build_person_match(self): + self.assertEqual(build_person_match(["a", "b"]), "(@login='a' or @login='b')") + + def test_parse_persons(self): + self.assertEqual( + parse_persons(ET.fromstring(PERSONS_XML)), + {"alice": "confirmed", "bob": "locked"}, + ) + + def test_classify_users(self): + r = classify_users( + ["alice", "bob", "carol"], {"alice": "confirmed", "bob": "locked"} + ) + self.assertEqual(r.confirmed, ["alice"]) + self.assertEqual(r.invalid, ["bob"]) + self.assertEqual(r.not_found, ["carol"]) + + +class BuildBinarySourceMatchTests(unittest.TestCase): + def test_ors_names_and_ands_project(self): + self.assertEqual( + build_binary_source_match(["a", "b"], "SUSE:SLFO:Main"), + "(@name='a' or @name='b') and @project='SUSE:SLFO:Main'", + ) + + def test_single_name(self): + self.assertEqual( + build_binary_source_match(["crash-kmp-rt"], "SUSE:SLFO:Main"), + "(@name='crash-kmp-rt') and @project='SUSE:SLFO:Main'", + ) + + +class ParsePublishedSourcesTests(unittest.TestCase): + def setUp(self): + self.m = parse_published_sources( + ET.fromstring(PUBLISHED_XML), "SUSE:SLFO:Main" + ) + + def test_maps_binary_to_source(self): + self.assertEqual(self.m["crash-kmp-rt"], ["crash"]) + self.assertEqual(self.m["python313-libmodulemd"], ["libmodulemd"]) + + def test_strips_flavor(self): + self.assertEqual(self.m["graphviz-gd"], ["graphviz"]) + + def test_excludes_other_project_exact_match(self): + # crash-kmp-rt in SUSE:SLFO:Main:Staging:V must not leak in. + self.assertEqual(self.m["crash-kmp-rt"], ["crash"]) + + def test_groups_multiple_sources_sorted(self): + self.assertEqual(self.m["dual"], ["src-a", "src-b"]) + + def test_skips_rows_missing_attrs(self): + # Row with no @package (name "no-package") is dropped. + self.assertNotIn("no-package", self.m) + # Row with no @name is dropped (no key, and its package never appears). + self.assertNotIn(None, self.m) + self.assertNotIn("no-name", [s for sources in self.m.values() for s in sources]) + + +class MergeFallbackSourcesTests(unittest.TestCase): + def test_folds_resolved_and_reports_still_failed(self): + sources, still_failed = merge_fallback_sources( + ["libguestfs"], + ["crash-kmp-rt", "python313-libmodulemd", "brand-new"], + {"crash-kmp-rt": ["crash"], "python313-libmodulemd": ["libmodulemd"]}, + ) + self.assertEqual(sources, ["crash", "libguestfs", "libmodulemd"]) + self.assertEqual(still_failed, ["brand-new"]) + + def test_dedups_against_existing_sources(self): + sources, still_failed = merge_fallback_sources( + ["crash"], ["crash-kmp-rt"], {"crash-kmp-rt": ["crash"]} + ) + self.assertEqual(sources, ["crash"]) + self.assertEqual(still_failed, []) + + def test_multiple_sources_per_binary(self): + sources, still_failed = merge_fallback_sources( + [], ["dual"], {"dual": ["src-a", "src-b"]} + ) + self.assertEqual(sources, ["src-a", "src-b"]) + self.assertEqual(still_failed, []) + + def test_empty_resolved_leaves_all_failed(self): + sources, still_failed = merge_fallback_sources(["x"], ["a", "b"], {}) + self.assertEqual(sources, ["x"]) + self.assertEqual(still_failed, ["a", "b"]) + + +class QueryPublishedSourcesTests(unittest.TestCase): + @staticmethod + def _fake_xml_parse(_response): + return ET.ElementTree(ET.fromstring(PUBLISHED_XML)) + + def test_batches_by_batch_size(self): + with ( + mock.patch.object(sle_check, "http_GET") as http_get, + mock.patch.object( + sle_check, "xml_parse", side_effect=self._fake_xml_parse + ), + mock.patch.object(sle_check, "makeurl", return_value="http://url"), + ): + result = query_published_sources( + "http://api", ["a", "b", "c"], "SUSE:SLFO:Main", batch_size=1 + ) + self.assertEqual(http_get.call_count, 3) + # Merged across batches, resolved from the fixture. + self.assertEqual(result["crash-kmp-rt"], ["crash"]) + + def test_unsafe_name_never_reaches_query(self): + captured = [] + + def fake_makeurl(_apiurl, _path, query): + captured.append(query["match"]) + return "http://url" + + with ( + mock.patch.object(sle_check, "http_GET"), + mock.patch.object( + sle_check, "xml_parse", side_effect=self._fake_xml_parse + ), + mock.patch.object(sle_check, "makeurl", side_effect=fake_makeurl), + ): + query_published_sources( + "http://api", + ["crash-kmp-rt", "evil' or '1'='1"], + "SUSE:SLFO:Main", + ) + joined = " ".join(captured) + self.assertIn("crash-kmp-rt", joined) + self.assertNotIn("evil", joined) + + def test_trailing_newline_name_never_reaches_query(self): + # Python '$' matches before a trailing '\n'; the safety barrier must not rely on that. + with ( + mock.patch.object(sle_check, "http_GET") as http_get, + mock.patch.object(sle_check, "xml_parse"), + mock.patch.object(sle_check, "makeurl"), + ): + result = query_published_sources( + "http://api", ["crash\n"], "SUSE:SLFO:Main" + ) + http_get.assert_not_called() + self.assertEqual(result, {}) + + def test_all_unsafe_makes_no_request(self): + with ( + mock.patch.object(sle_check, "http_GET") as http_get, + mock.patch.object(sle_check, "xml_parse"), + mock.patch.object(sle_check, "makeurl"), + ): + result = query_published_sources( + "http://api", ["bad'name", ""], "SUSE:SLFO:Main" + ) + http_get.assert_not_called() + self.assertEqual(result, {}) + + +class FindOrphanSourcesTests(unittest.TestCase): + def test_orphans_and_failed(self): + db = {"libguestfs": {"users": ["a"], "groups": []}} + result = find_orphan_sources( + ["libguestfs-appliance", "udev", "nobin"], ET.fromstring(SOURCEINFO_XML), db + ) + self.assertEqual(result.orphans, ["systemd"]) + self.assertEqual(result.failed, ["nobin"]) + self.assertEqual(result.checked, 2) + + def test_fallback_resolves_dynamic_binary(self): + db = {"libguestfs": {"users": ["a"], "groups": []}} + seen = {} + + def resolver(names): + seen["names"] = list(names) + return {"crash-kmp-rt": ["crash"]} + + result = find_orphan_sources( + ["libguestfs-appliance", "crash-kmp-rt", "brand-new"], + ET.fromstring(SOURCEINFO_XML), + db, + fallback_resolver=resolver, + ) + # Resolver is asked only for the binaries view=info could not resolve. + self.assertEqual(sorted(seen["names"]), ["brand-new", "crash-kmp-rt"]) + # crash-kmp-rt now resolves to orphan source "crash". + self.assertEqual(result.orphans, ["crash"]) + # brand-new remains genuinely unresolved. + self.assertEqual(result.failed, ["brand-new"]) + self.assertEqual(result.checked, 2) + + def test_fallback_exception_propagates(self): + # Fail-closed: a fallback error must surface (to check_source_submission's handler), + # never be swallowed into a pass that lets an orphan through this gating bot. + def resolver(names): + raise urllib.error.HTTPError("u", 500, "err", {}, None) + + with self.assertRaises(urllib.error.HTTPError): + find_orphan_sources( + ["brand-new"], + ET.fromstring(SOURCEINFO_XML), + {}, + fallback_resolver=resolver, + ) + + def test_fallback_not_called_when_all_resolved(self): + db = {"libguestfs": {"users": ["a"], "groups": []}, "systemd": {"users": ["a"], "groups": []}} + called = [] + result = find_orphan_sources( + ["libguestfs-appliance", "udev"], + ET.fromstring(SOURCEINFO_XML), + db, + fallback_resolver=lambda names: called.append(names) or {}, + ) + self.assertEqual(called, []) + self.assertEqual(result.failed, []) + + +class CheckUserValidityTests(unittest.TestCase): + def _fake_query(self, apiurl, batch): + table = {"alice": "confirmed", "bob": "locked"} + return {u: table[u] for u in batch if u in table} + + def test_aggregates(self): + db = { + "p1": {"users": ["alice", "carol"], "groups": []}, + "p2": {"users": ["bob"], "groups": []}, + } + with mock.patch.object( + sle_check, "query_persons", side_effect=self._fake_query + ): + r = check_user_validity(db, "http://api") + self.assertEqual(r.confirmed, ["alice"]) + self.assertEqual(r.invalid, ["bob"]) + self.assertEqual(r.not_found, ["carol"]) + + def test_batching(self): + db = {"p1": {"users": ["alice", "bob", "carol"], "groups": []}} + with mock.patch.object( + sle_check, "query_persons", side_effect=self._fake_query + ) as q: + check_user_validity(db, "http://api", batch_size=1) + self.assertEqual(q.call_count, 3) + + +class MessageTests(unittest.TestCase): + def test_declined(self): + self.assertEqual( + build_declined_message(["src1"], ["bob"], ["carol"]), + "New source packages with no maintainer in _maintainership.json:\n\n- src1\n\n" + "Maintainer logins not found in OBS:\n\n- carol\n\n" + "Maintainer logins that are not confirmed OBS accounts:\n\n- bob", + ) + + def test_accepted_clean(self): + self.assertEqual(build_accepted_message([]), "Maintainership check passed.") + + def test_accepted_with_unresolved(self): + msg = build_accepted_message(["x"]) + self.assertIn("could not be resolved", msg) + self.assertIn("- x", msg) + + +class _Req: + def __init__(self): + self.actions = [mock.Mock(src_rev="HEAD")] + self._owner = "products" + self._repo = "SLES" + self._pr_id = 1 + + +def _make_bot(allowed=("products/SLES",)): + bot = object.__new__(sle_check.SleCheckBot) + bot.logger = logging.getLogger("test") + bot.review_messages = {} + bot.allowed_repositories = list(allowed) + bot.apiurl = "http://api" + bot.platform = mock.Mock() + bot.requests = [_Req()] + return bot + + +class RunCheckTests(unittest.TestCase): + def test_not_allowed_repo(self): + self.assertEqual( + _make_bot(allowed=[]).run_check("o", "p", "HEAD", "to", "tp"), (None, None) + ) + + def test_decline_aggregated(self): + db = {"libguestfs": {"users": ["alice"], "groups": []}} + with ( + mock.patch.object(sle_check, "fetch_maintainership", return_value=db), + mock.patch.object( + sle_check, + "check_user_validity", + return_value=sle_check.UserCheckResult([], [], ["carol"]), + ), + mock.patch.object(sle_check, "fetch_pr_diff", return_value=MULTI_FILE_DIFF), + mock.patch.object( + sle_check, + "fetch_source_info", + return_value=ET.fromstring(SOURCEINFO_XML), + ), + ): + bot = _make_bot() + result, message = bot.run_check("o", "p", "HEAD", "to", "tp") + self.assertFalse(result) + self.assertIsNone(message) + self.assertIn("systemd", bot.review_messages["declined"]) + self.assertIn("graphviz", bot.review_messages["declined"]) + self.assertIn("carol", bot.review_messages["declined"]) + + def test_decline_via_published_fallback(self): + # crash-kmp-rt is a build-time KMP name absent from the view=info map; only the + # published-binary fallback resolves it, to orphan source "crash". + diff = ( + "diff --git a/000productcompose/default.productcompose" + " b/000productcompose/default.productcompose\n" + "--- a/000productcompose/default.productcompose\n" + "+++ b/000productcompose/default.productcompose\n" + "@@\n" + "+ - crash-kmp-rt\n" + ) + with ( + mock.patch.object(sle_check, "fetch_maintainership", return_value={}), + mock.patch.object( + sle_check, + "check_user_validity", + return_value=sle_check.UserCheckResult([], [], []), + ), + mock.patch.object(sle_check, "fetch_pr_diff", return_value=diff), + mock.patch.object( + sle_check, + "fetch_source_info", + return_value=ET.fromstring(SOURCEINFO_XML), + ), + mock.patch.object( + sle_check, + "query_published_sources", + return_value={"crash-kmp-rt": ["crash"]}, + ) as qps, + ): + bot = _make_bot() + result, message = bot.run_check("o", "p", "HEAD", "to", "tp") + self.assertFalse(result) + self.assertIsNone(message) + self.assertIn("crash", bot.review_messages["declined"]) + qps.assert_called_once_with(bot.apiurl, ["crash-kmp-rt"], sle_check.SOURCE_PROJECT) + + def test_accept_clean_untouched(self): + with ( + mock.patch.object(sle_check, "fetch_maintainership", return_value={}), + mock.patch.object( + sle_check, + "check_user_validity", + return_value=sle_check.UserCheckResult([], [], []), + ), + mock.patch.object( + sle_check, + "fetch_pr_diff", + return_value="diff --git a/x b/x\n+ - y\n", + ), + mock.patch.object(sle_check, "fetch_source_info") as fsi, + ): + bot = _make_bot() + result, message = bot.run_check("o", "p", "HEAD", "to", "tp") + self.assertTrue(result) + self.assertEqual(message, "Maintainership check passed.") + fsi.assert_not_called() + + +class CheckSourceSubmissionTests(unittest.TestCase): + def test_transient_requests(self): + bot = _make_bot() + bot.run_check = mock.Mock(side_effect=requests_lib.exceptions.ConnectionError()) + self.assertIsNone(bot.check_source_submission("o", "p", "r", "to", "tp")) + + def test_transient_urllib(self): + bot = _make_bot() + bot.run_check = mock.Mock(side_effect=urllib.error.URLError("x")) + self.assertIsNone(bot.check_source_submission("o", "p", "r", "to", "tp")) + + def test_unexpected_declines(self): + bot = _make_bot() + bot.run_check = mock.Mock(side_effect=RuntimeError("boom")) + self.assertFalse(bot.check_source_submission("o", "p", "r", "to", "tp")) + self.assertIn("Unhandled exception", bot.review_messages["declined"]) + + def test_success_accepts(self): + bot = _make_bot() + bot.run_check = mock.Mock(return_value=(True, "Maintainership check passed.")) + self.assertTrue(bot.check_source_submission("o", "p", "r", "to", "tp")) + self.assertEqual( + bot.review_messages["accepted"], "Maintainership check passed." + ) + + +if __name__ == "__main__": + unittest.main()