diff --git a/elliott/elliottlib/cli/__main__.py b/elliott/elliottlib/cli/__main__.py index 1840ccbeeb..e3bbf8b7f5 100644 --- a/elliott/elliottlib/cli/__main__.py +++ b/elliott/elliottlib/cli/__main__.py @@ -75,6 +75,7 @@ from elliottlib.cli.verify_attached_operators_cli import verify_attached_operators_cli from elliottlib.cli.verify_cvp_cli import verify_cvp_cli from elliottlib.cli.verify_image_grades_cli import verify_image_grades_cli +from elliottlib.cli.verify_kernel_tag_cli import verify_kernel_tag_cli from elliottlib.cli.verify_payload import verify_payload from elliottlib.cli.verify_signatures_cli import verify_signatures_cli from elliottlib.exceptions import ElliottFatalError @@ -327,6 +328,7 @@ def poll_signed(runtime, minutes, advisory, default_advisory_type, noop): cli.add_command(process_release_from_fbc_bugs_cli) cli.add_command(verify_signatures_cli) cli.add_command(verify_image_grades_cli) +cli.add_command(verify_kernel_tag_cli) # ----------------------------------------------------------------------------- # CLI Entry point diff --git a/elliott/elliottlib/cli/verify_kernel_tag_cli.py b/elliott/elliottlib/cli/verify_kernel_tag_cli.py new file mode 100644 index 0000000000..4b2895f15d --- /dev/null +++ b/elliott/elliottlib/cli/verify_kernel_tag_cli.py @@ -0,0 +1,309 @@ +import asyncio +import json +import logging +import re +from dataclasses import dataclass, field +from typing import Optional + +import click +import koji +import requests +from artcommonlib.assembly import assembly_config_struct +from artcommonlib.constants import BREW_DOWNLOAD_URL, BREW_HUB + +from elliottlib import brew +from elliottlib.cli.common import cli, click_coroutine +from elliottlib.errata_async import AsyncErrataAPI + +LOGGER = logging.getLogger(__name__) + +KERNEL_TAG_ADVISORY_TYPES = ("image", "rhcos") + + +@dataclass +class KernelBuildInfo: + nvr: str + has_stop_ship: bool = False + + +@dataclass +class AdvisoryKernelResult: + advisory_id: int + impetus: str + rhcos_builds: list[str] = field(default_factory=list) + kernel_builds: list[KernelBuildInfo] = field(default_factory=list) + skipped: bool = False + error: Optional[str] = None + + @property + def passed(self) -> bool: + return not any(k.has_stop_ship for k in self.kernel_builds) and not self.error + + @property + def failed(self) -> bool: + return any(k.has_stop_ship for k in self.kernel_builds) or bool(self.error) + + +@dataclass +class VerifyKernelTagResult: + advisories: list[AdvisoryKernelResult] = field(default_factory=list) + stop_ship_tag: str = "" + + @property + def passed(self) -> bool: + return bool(self.advisories) and all(a.passed or a.skipped for a in self.advisories) + + @property + def failed(self) -> bool: + return any(a.failed for a in self.advisories) + + +def get_rpm_deliveries_config(runtime) -> list: + rpm_deliveries = runtime.group_config.get("rpm_deliveries") + if not rpm_deliveries: + return [] + return rpm_deliveries.primitive() if hasattr(rpm_deliveries, "primitive") else rpm_deliveries + + +def get_kernel_packages_and_tag(rpm_deliveries: list) -> tuple[set[str], str]: + packages = set() + tags = set() + for entry in rpm_deliveries: + tag = entry.get("stop_ship_tag", "") + if tag: + packages.update(entry.get("packages", [])) + tags.add(tag) + if len(tags) > 1: + raise ValueError(f"Multiple different stop_ship_tag values in rpm_deliveries: {tags}") + return packages, tags.pop() if tags else "" + + +def find_rhcos_nvrs(build_nvrs: set[str]) -> list[str]: + return sorted(nvr for nvr in build_nvrs if nvr.startswith("rhcos-")) + + +def nvr_to_brewroot_metadata_url(rhcos_nvr: str) -> str: + match = re.fullmatch(r"([A-Za-z0-9_.+-]+?)-([\d.]+)-(\d+)", rhcos_nvr) + if not match: + raise ValueError(f"Cannot parse RHCOS NVR: {rhcos_nvr}") + name, version, release = match.groups() + return f"{BREW_DOWNLOAD_URL}/packages/{name}/{version}/{release}/metadata.json" + + +def get_kernel_nvrs_from_metadata(metadata: dict, kernel_packages: set[str]) -> list[str]: + kernel_nvrs = [] + for entry in metadata.get("output", []): + for comp in entry.get("components", []) or []: + if comp.get("name") in kernel_packages: + nvr = f"{comp['name']}-{comp['version']}-{comp['release']}" + if nvr not in kernel_nvrs: + kernel_nvrs.append(nvr) + return kernel_nvrs + + +def get_kernel_rpms_from_rhcos(rhcos_nvr: str, kernel_packages: set[str]) -> list[str]: + url = nvr_to_brewroot_metadata_url(rhcos_nvr) + LOGGER.debug("Downloading RHCOS metadata from %s", url) + resp = requests.get(url, timeout=30) + resp.raise_for_status() + return get_kernel_nvrs_from_metadata(resp.json(), kernel_packages) + + +def check_kernel_tags(koji_api, kernel_nvrs: list[str], stop_ship_tag: str) -> list[KernelBuildInfo]: + results = [] + if not kernel_nvrs: + return results + build_tags = brew.get_builds_tags(kernel_nvrs, koji_api) + for nvr, tags in zip(kernel_nvrs, build_tags): + tag_names = {t["name"] for t in tags} + results.append(KernelBuildInfo(nvr=nvr, has_stop_ship=stop_ship_tag in tag_names)) + return results + + +async def check_advisory_kernel_tag( + api: AsyncErrataAPI, + advisory_id: int, + impetus: str, + koji_api, + kernel_packages: set[str], + stop_ship_tag: str, +) -> AdvisoryKernelResult: + result = AdvisoryKernelResult(advisory_id=advisory_id, impetus=impetus) + try: + all_builds = await api.get_builds_flattened(advisory_id) + rhcos_nvrs = find_rhcos_nvrs(all_builds) + result.rhcos_builds = rhcos_nvrs + + if not rhcos_nvrs: + LOGGER.info("Advisory %s (%s): no RHCOS builds found, skipping", advisory_id, impetus) + result.skipped = True + return result + + def _check(): + seen_nvrs = set() + all_kernel_nvrs = [] + for rhcos_nvr in rhcos_nvrs: + LOGGER.debug("Advisory %s (%s): checking RHCOS build %s", advisory_id, impetus, rhcos_nvr) + kernel_nvrs = get_kernel_rpms_from_rhcos(rhcos_nvr, kernel_packages) + for nvr in kernel_nvrs: + if nvr not in seen_nvrs: + seen_nvrs.add(nvr) + all_kernel_nvrs.append(nvr) + return check_kernel_tags(koji_api, all_kernel_nvrs, stop_ship_tag) + + result.kernel_builds = await asyncio.to_thread(_check) + + if not result.kernel_builds: + result.error = "no kernel RPMs found in RHCOS builds" + LOGGER.error("Advisory %s (%s): %s", advisory_id, impetus, result.error) + return result + + for kb in result.kernel_builds: + if kb.has_stop_ship: + LOGGER.error( + "Advisory %s (%s): kernel %s tagged %s!", + advisory_id, + impetus, + kb.nvr, + stop_ship_tag, + ) + else: + LOGGER.info("Advisory %s (%s): kernel %s OK", advisory_id, impetus, kb.nvr) + + except Exception as e: + LOGGER.error("Advisory %s (%s): error checking kernel tag: %s", advisory_id, impetus, e) + result.error = str(e) + + return result + + +async def verify_kernel_tag( + advisories: dict[str, int], + koji_api, + kernel_packages: set[str], + stop_ship_tag: str, +) -> VerifyKernelTagResult: + result = VerifyKernelTagResult(stop_ship_tag=stop_ship_tag) + + async with AsyncErrataAPI() as api: + for impetus, advisory_id in advisories.items(): + ar = await check_advisory_kernel_tag(api, advisory_id, impetus, koji_api, kernel_packages, stop_ship_tag) + result.advisories.append(ar) + + return result + + +def render_result(result: VerifyKernelTagResult, output: str) -> str: + if output == "json": + return json.dumps( + { + "passed": result.passed, + "failed": result.failed, + "stop_ship_tag": result.stop_ship_tag, + "advisories": [ + { + "advisory_id": a.advisory_id, + "impetus": a.impetus, + "rhcos_builds": a.rhcos_builds, + "kernel_builds": [{"nvr": k.nvr, "has_stop_ship": k.has_stop_ship} for k in a.kernel_builds], + "skipped": a.skipped, + "error": a.error, + } + for a in result.advisories + ], + }, + indent=2, + ) + + lines = [f"Kernel stop-ship tag check (tag: {result.stop_ship_tag})", ""] + for a in result.advisories: + if a.skipped: + lines.append(f" Advisory {a.advisory_id} ({a.impetus}): SKIPPED (no RHCOS/kernel)") + elif a.failed: + status = "STOP-SHIP" if any(k.has_stop_ship for k in a.kernel_builds) else "ERROR" + lines.append(f" Advisory {a.advisory_id} ({a.impetus}): {status}") + else: + lines.append(f" Advisory {a.advisory_id} ({a.impetus}): OK") + + if a.error: + lines.append(f" Error: {a.error}") + for k in a.kernel_builds: + tag_status = "STOP-SHIP" if k.has_stop_ship else "ok" + lines.append(f" {k.nvr}: {tag_status}") + lines.append("") + + overall = "PASS" if result.passed else "FAIL" + lines.append(f"Overall: {overall}") + return "\n".join(lines) + + +def get_advisory_ids(runtime) -> dict[str, int]: + releases_config = runtime.get_releases_config() + group_config = assembly_config_struct(releases_config, runtime.assembly, "group", {}) + advisories = group_config.get("advisories", {}) + result = {} + for impetus in KERNEL_TAG_ADVISORY_TYPES: + ad_id = advisories.get(impetus) + if ad_id: + result[impetus] = int(ad_id) + return result + + +@cli.command("verify-kernel-tag", short_help="Check RHCOS kernel builds for stop-ship tags") +@click.option( + "-o", + "--output", + type=click.Choice(["text", "json"]), + default="text", + show_default=True, + help="Output format.", +) +@click.pass_obj +@click_coroutine +async def verify_kernel_tag_cli(runtime, output): + """Check RHCOS kernel builds for early-kernel-stop-ship tags. + + Inspects image and RHCOS advisories from the assembly config. For each + advisory, finds RHCOS builds, extracts kernel RPMs, and checks whether + any kernel build is tagged with the stop-ship tag defined in + rpm_deliveries group config. + + Exits with code 1 if any kernel build has the stop-ship tag. + + Requires --group and --assembly global options. + + Example: + elliott --group openshift-4.18 --assembly 4.18.51 verify-kernel-tag + """ + runtime.initialize() + + rpm_deliveries = get_rpm_deliveries_config(runtime) + if not rpm_deliveries: + raise click.UsageError("No rpm_deliveries found in group config.") + + kernel_packages, stop_ship_tag = get_kernel_packages_and_tag(rpm_deliveries) + if not kernel_packages or not stop_ship_tag: + raise click.UsageError("No kernel packages or stop_ship_tag found in rpm_deliveries config.") + + advisories = get_advisory_ids(runtime) + if not advisories: + raise click.UsageError(f"No advisory IDs found for {KERNEL_TAG_ADVISORY_TYPES} in assembly config.") + + LOGGER.info( + "Checking kernel stop-ship tag '%s' for packages %s in advisories: %s", + stop_ship_tag, + kernel_packages, + advisories, + ) + + koji_api = koji.ClientSession(BREW_HUB) + + result = await verify_kernel_tag( + advisories=advisories, + koji_api=koji_api, + kernel_packages=kernel_packages, + stop_ship_tag=stop_ship_tag, + ) + click.echo(render_result(result, output)) + if not result.passed: + raise SystemExit(1) diff --git a/elliott/tests/test_verify_kernel_tag_cli.py b/elliott/tests/test_verify_kernel_tag_cli.py new file mode 100644 index 0000000000..4c6fd8f759 --- /dev/null +++ b/elliott/tests/test_verify_kernel_tag_cli.py @@ -0,0 +1,488 @@ +import json +from unittest import IsolatedAsyncioTestCase +from unittest.mock import AsyncMock, MagicMock, patch + +from elliottlib.cli.verify_kernel_tag_cli import ( + AdvisoryKernelResult, + KernelBuildInfo, + VerifyKernelTagResult, + check_advisory_kernel_tag, + check_kernel_tags, + find_rhcos_nvrs, + get_kernel_nvrs_from_metadata, + get_kernel_packages_and_tag, + get_kernel_rpms_from_rhcos, + get_rpm_deliveries_config, + nvr_to_brewroot_metadata_url, + render_result, + verify_kernel_tag, +) + + +class TestKernelBuildInfo(IsolatedAsyncioTestCase): + def test_no_stop_ship(self): + kb = KernelBuildInfo(nvr="kernel-5.14.0-284.14.1.el9_2") + self.assertFalse(kb.has_stop_ship) + + def test_has_stop_ship(self): + kb = KernelBuildInfo(nvr="kernel-5.14.0-284.14.1.el9_2", has_stop_ship=True) + self.assertTrue(kb.has_stop_ship) + + +class TestAdvisoryKernelResult(IsolatedAsyncioTestCase): + def test_passed_no_kernels(self): + r = AdvisoryKernelResult(advisory_id=1, impetus="image") + self.assertTrue(r.passed) + self.assertFalse(r.failed) + + def test_passed_with_ok_kernels(self): + r = AdvisoryKernelResult( + advisory_id=1, + impetus="image", + kernel_builds=[KernelBuildInfo(nvr="kernel-5.14.0-1.el9")], + ) + self.assertTrue(r.passed) + self.assertFalse(r.failed) + + def test_failed_stop_ship(self): + r = AdvisoryKernelResult( + advisory_id=1, + impetus="image", + kernel_builds=[KernelBuildInfo(nvr="kernel-5.14.0-1.el9", has_stop_ship=True)], + ) + self.assertFalse(r.passed) + self.assertTrue(r.failed) + + def test_failed_error(self): + r = AdvisoryKernelResult(advisory_id=1, impetus="image", error="some error") + self.assertFalse(r.passed) + self.assertTrue(r.failed) + + def test_mixed_kernels(self): + r = AdvisoryKernelResult( + advisory_id=1, + impetus="image", + kernel_builds=[ + KernelBuildInfo(nvr="kernel-5.14.0-1.el9"), + KernelBuildInfo(nvr="kernel-rt-5.14.0-1.el9", has_stop_ship=True), + ], + ) + self.assertFalse(r.passed) + self.assertTrue(r.failed) + + +class TestVerifyKernelTagResult(IsolatedAsyncioTestCase): + def test_passed(self): + r = VerifyKernelTagResult( + advisories=[ + AdvisoryKernelResult( + advisory_id=1, + impetus="image", + kernel_builds=[KernelBuildInfo(nvr="kernel-5.14.0-1.el9")], + ) + ], + stop_ship_tag="early-kernel-stop-ship", + ) + self.assertTrue(r.passed) + self.assertFalse(r.failed) + + def test_failed(self): + r = VerifyKernelTagResult( + advisories=[ + AdvisoryKernelResult( + advisory_id=1, + impetus="image", + kernel_builds=[KernelBuildInfo(nvr="kernel-5.14.0-1.el9", has_stop_ship=True)], + ) + ], + ) + self.assertFalse(r.passed) + self.assertTrue(r.failed) + + def test_empty(self): + r = VerifyKernelTagResult() + self.assertFalse(r.passed) + self.assertFalse(r.failed) + + def test_passed_with_skipped(self): + r = VerifyKernelTagResult( + advisories=[ + AdvisoryKernelResult(advisory_id=1, impetus="image", skipped=True), + AdvisoryKernelResult( + advisory_id=2, + impetus="rhcos", + kernel_builds=[KernelBuildInfo(nvr="kernel-5.14.0-1.el9")], + ), + ], + ) + self.assertTrue(r.passed) + + +class TestGetRpmDeliveriesConfig(IsolatedAsyncioTestCase): + def test_with_config(self): + runtime = MagicMock() + mock_deliveries = MagicMock() + mock_deliveries.primitive.return_value = [ + { + "packages": ["kernel"], + "stop_ship_tag": "early-kernel-stop-ship", + "integration_tag": "early-kernel-candidate", + } + ] + runtime.group_config.get.return_value = mock_deliveries + result = get_rpm_deliveries_config(runtime) + self.assertEqual(len(result), 1) + + def test_without_config(self): + runtime = MagicMock() + runtime.group_config.get.return_value = None + result = get_rpm_deliveries_config(runtime) + self.assertEqual(result, []) + + +class TestGetKernelPackagesAndTag(IsolatedAsyncioTestCase): + def test_standard_config(self): + config = [ + { + "packages": ["kernel", "kernel-rt"], + "stop_ship_tag": "early-kernel-stop-ship", + "integration_tag": "early-kernel-candidate", + } + ] + packages, tag = get_kernel_packages_and_tag(config) + self.assertEqual(packages, {"kernel", "kernel-rt"}) + self.assertEqual(tag, "early-kernel-stop-ship") + + def test_no_stop_ship_tag(self): + config = [{"packages": ["foo"], "integration_tag": "bar"}] + packages, tag = get_kernel_packages_and_tag(config) + self.assertEqual(packages, set()) + self.assertEqual(tag, "") + + def test_multiple_entries_same_tag(self): + config = [ + {"packages": ["kernel"], "stop_ship_tag": "early-kernel-stop-ship", "integration_tag": "ic1"}, + {"packages": ["kernel-rt"], "stop_ship_tag": "early-kernel-stop-ship", "integration_tag": "ic2"}, + ] + packages, tag = get_kernel_packages_and_tag(config) + self.assertEqual(packages, {"kernel", "kernel-rt"}) + self.assertEqual(tag, "early-kernel-stop-ship") + + def test_multiple_entries_different_tags_raises(self): + config = [ + {"packages": ["kernel"], "stop_ship_tag": "tag1", "integration_tag": "ic1"}, + {"packages": ["kernel-rt"], "stop_ship_tag": "tag2", "integration_tag": "ic2"}, + ] + with self.assertRaises(ValueError): + get_kernel_packages_and_tag(config) + + +class TestFindRhcosNvrs(IsolatedAsyncioTestCase): + def test_finds_rhcos(self): + builds = {"ose-installer-4.18.0-1.el9", "rhcos-x86_64-418.92.1-1", "rhcos-aarch64-418.92.1-1"} + result = find_rhcos_nvrs(builds) + self.assertEqual(result, ["rhcos-aarch64-418.92.1-1", "rhcos-x86_64-418.92.1-1"]) + + def test_no_rhcos(self): + builds = {"ose-installer-4.18.0-1.el9"} + result = find_rhcos_nvrs(builds) + self.assertEqual(result, []) + + +class TestNvrToBrewrootMetadataUrl(IsolatedAsyncioTestCase): + def test_standard_nvr(self): + url = nvr_to_brewroot_metadata_url("rhcos-x86_64-418.92.202407091253-1") + self.assertIn("/packages/rhcos-x86_64/418.92.202407091253/1/metadata.json", url) + + def test_rhcos_421(self): + url = nvr_to_brewroot_metadata_url("rhcos-x86_64-4.21.9.6.202608051529-0") + self.assertIn("/packages/rhcos-x86_64/4.21.9.6.202608051529/0/metadata.json", url) + + def test_malformed_nvr_raises(self): + with self.assertRaises(ValueError): + nvr_to_brewroot_metadata_url("not-a-valid-nvr") + + +class TestGetKernelNvrsFromMetadata(IsolatedAsyncioTestCase): + def test_finds_kernel(self): + metadata = { + "output": [ + { + "components": [ + {"name": "kernel", "version": "5.14.0", "release": "570.132.1.el9_6"}, + {"name": "bash", "version": "5.2.26", "release": "1.el9"}, + {"name": "kernel-core", "version": "5.14.0", "release": "570.132.1.el9_6"}, + ] + } + ] + } + result = get_kernel_nvrs_from_metadata(metadata, {"kernel", "kernel-rt"}) + self.assertEqual(result, ["kernel-5.14.0-570.132.1.el9_6"]) + + def test_no_kernel(self): + metadata = {"output": [{"components": [{"name": "bash", "version": "5.2.26", "release": "1.el9"}]}]} + result = get_kernel_nvrs_from_metadata(metadata, {"kernel"}) + self.assertEqual(result, []) + + def test_empty_output(self): + result = get_kernel_nvrs_from_metadata({"output": []}, {"kernel"}) + self.assertEqual(result, []) + + def test_null_components(self): + metadata = {"output": [{"components": None}]} + result = get_kernel_nvrs_from_metadata(metadata, {"kernel"}) + self.assertEqual(result, []) + + def test_dedup_across_outputs(self): + metadata = { + "output": [ + {"components": [{"name": "kernel", "version": "5.14.0", "release": "1.el9"}]}, + {"components": [{"name": "kernel", "version": "5.14.0", "release": "1.el9"}]}, + ] + } + result = get_kernel_nvrs_from_metadata(metadata, {"kernel"}) + self.assertEqual(len(result), 1) + + +class TestGetKernelRpmsFromRhcos(IsolatedAsyncioTestCase): + @patch("elliottlib.cli.verify_kernel_tag_cli.requests.get") + def test_success(self, mock_get): + mock_resp = MagicMock() + mock_resp.json.return_value = { + "output": [ + { + "components": [ + {"name": "kernel", "version": "5.14.0", "release": "1.el9"}, + ] + } + ] + } + mock_get.return_value = mock_resp + result = get_kernel_rpms_from_rhcos("rhcos-x86_64-418.92.1-1", {"kernel"}) + self.assertEqual(result, ["kernel-5.14.0-1.el9"]) + mock_resp.raise_for_status.assert_called_once() + + @patch("elliottlib.cli.verify_kernel_tag_cli.requests.get") + def test_http_error(self, mock_get): + from requests.exceptions import HTTPError + + mock_resp = MagicMock() + mock_resp.raise_for_status.side_effect = HTTPError("404 Not Found") + mock_get.return_value = mock_resp + with self.assertRaises(HTTPError) as cm: + get_kernel_rpms_from_rhcos("rhcos-x86_64-418.92.1-1", {"kernel"}) + self.assertIn("404 Not Found", str(cm.exception)) + + +class TestCheckKernelTags(IsolatedAsyncioTestCase): + @patch("elliottlib.cli.verify_kernel_tag_cli.brew.get_builds_tags") + def test_no_stop_ship(self, mock_get_tags): + mock_get_tags.return_value = [ + [{"name": "rhaos-4.18-rhel-9-candidate"}, {"name": "early-kernel-candidate"}], + ] + result = check_kernel_tags(MagicMock(), ["kernel-5.14.0-1.el9"], "early-kernel-stop-ship") + self.assertEqual(len(result), 1) + self.assertFalse(result[0].has_stop_ship) + + @patch("elliottlib.cli.verify_kernel_tag_cli.brew.get_builds_tags") + def test_has_stop_ship(self, mock_get_tags): + mock_get_tags.return_value = [ + [{"name": "early-kernel-stop-ship"}, {"name": "early-kernel-candidate"}], + ] + result = check_kernel_tags(MagicMock(), ["kernel-5.14.0-1.el9"], "early-kernel-stop-ship") + self.assertEqual(len(result), 1) + self.assertTrue(result[0].has_stop_ship) + + @patch("elliottlib.cli.verify_kernel_tag_cli.brew.get_builds_tags") + def test_empty_nvrs(self, mock_get_tags): + result = check_kernel_tags(MagicMock(), [], "early-kernel-stop-ship") + self.assertEqual(result, []) + mock_get_tags.assert_not_called() + + +class TestCheckAdvisoryKernelTag(IsolatedAsyncioTestCase): + @patch("elliottlib.cli.verify_kernel_tag_cli.check_kernel_tags") + @patch("elliottlib.cli.verify_kernel_tag_cli.get_kernel_rpms_from_rhcos") + async def test_no_rhcos_builds(self, mock_get_rpms, mock_check_tags): + api = AsyncMock() + api.get_builds_flattened.return_value = {"ose-installer-4.18.0-1.el9"} + result = await check_advisory_kernel_tag(api, 12345, "image", MagicMock(), {"kernel"}, "early-kernel-stop-ship") + self.assertTrue(result.skipped) + mock_get_rpms.assert_not_called() + + @patch("elliottlib.cli.verify_kernel_tag_cli.check_kernel_tags") + @patch("elliottlib.cli.verify_kernel_tag_cli.get_kernel_rpms_from_rhcos") + async def test_kernel_ok(self, mock_get_rpms, mock_check_tags): + api = AsyncMock() + api.get_builds_flattened.return_value = {"rhcos-x86_64-418.92.1-1"} + mock_get_rpms.return_value = ["kernel-5.14.0-1.el9"] + mock_check_tags.return_value = [KernelBuildInfo(nvr="kernel-5.14.0-1.el9")] + + result = await check_advisory_kernel_tag(api, 12345, "image", MagicMock(), {"kernel"}, "early-kernel-stop-ship") + self.assertTrue(result.passed) + self.assertFalse(result.skipped) + + @patch("elliottlib.cli.verify_kernel_tag_cli.check_kernel_tags") + @patch("elliottlib.cli.verify_kernel_tag_cli.get_kernel_rpms_from_rhcos") + async def test_kernel_stop_ship(self, mock_get_rpms, mock_check_tags): + api = AsyncMock() + api.get_builds_flattened.return_value = {"rhcos-x86_64-418.92.1-1"} + mock_get_rpms.return_value = ["kernel-5.14.0-1.el9"] + mock_check_tags.return_value = [KernelBuildInfo(nvr="kernel-5.14.0-1.el9", has_stop_ship=True)] + + result = await check_advisory_kernel_tag(api, 12345, "image", MagicMock(), {"kernel"}, "early-kernel-stop-ship") + self.assertTrue(result.failed) + + @patch("elliottlib.cli.verify_kernel_tag_cli.check_kernel_tags") + @patch("elliottlib.cli.verify_kernel_tag_cli.get_kernel_rpms_from_rhcos") + async def test_no_kernel_rpms_found(self, mock_get_rpms, mock_check_tags): + api = AsyncMock() + api.get_builds_flattened.return_value = {"rhcos-x86_64-418.92.1-1"} + mock_get_rpms.return_value = [] + mock_check_tags.return_value = [] + + result = await check_advisory_kernel_tag(api, 12345, "image", MagicMock(), {"kernel"}, "early-kernel-stop-ship") + self.assertTrue(result.failed) + self.assertIn("no kernel RPMs found", result.error) + + @patch("elliottlib.cli.verify_kernel_tag_cli.check_kernel_tags") + @patch("elliottlib.cli.verify_kernel_tag_cli.get_kernel_rpms_from_rhcos") + async def test_api_error(self, mock_get_rpms, mock_check_tags): + api = AsyncMock() + api.get_builds_flattened.side_effect = Exception("connection failed") + + result = await check_advisory_kernel_tag(api, 12345, "image", MagicMock(), {"kernel"}, "early-kernel-stop-ship") + self.assertEqual(result.error, "connection failed") + self.assertTrue(result.failed) + + @patch("elliottlib.cli.verify_kernel_tag_cli.check_kernel_tags") + @patch("elliottlib.cli.verify_kernel_tag_cli.get_kernel_rpms_from_rhcos") + async def test_multiple_rhcos_dedup(self, mock_get_rpms, mock_check_tags): + api = AsyncMock() + api.get_builds_flattened.return_value = {"rhcos-x86_64-418.92.1-1", "rhcos-aarch64-418.92.1-1"} + mock_get_rpms.side_effect = [ + ["kernel-5.14.0-1.el9"], + ["kernel-5.14.0-1.el9"], + ] + mock_check_tags.return_value = [KernelBuildInfo(nvr="kernel-5.14.0-1.el9")] + + await check_advisory_kernel_tag(api, 12345, "image", MagicMock(), {"kernel"}, "early-kernel-stop-ship") + mock_check_tags.assert_called_once() + args = mock_check_tags.call_args[0] + self.assertEqual(args[1], ["kernel-5.14.0-1.el9"]) + + +class TestVerifyKernelTag(IsolatedAsyncioTestCase): + @patch("elliottlib.cli.verify_kernel_tag_cli.AsyncErrataAPI") + @patch("elliottlib.cli.verify_kernel_tag_cli.check_advisory_kernel_tag") + async def test_checks_all_advisories(self, mock_check, mock_api_cls): + mock_api = AsyncMock() + mock_api_cls.return_value.__aenter__ = AsyncMock(return_value=mock_api) + mock_api_cls.return_value.__aexit__ = AsyncMock(return_value=False) + + mock_check.side_effect = [ + AdvisoryKernelResult( + advisory_id=100, + impetus="image", + kernel_builds=[KernelBuildInfo(nvr="kernel-5.14.0-1.el9")], + ), + AdvisoryKernelResult(advisory_id=200, impetus="rhcos", skipped=True), + ] + + koji_api = MagicMock() + result = await verify_kernel_tag({"image": 100, "rhcos": 200}, koji_api, {"kernel"}, "early-kernel-stop-ship") + self.assertEqual(len(result.advisories), 2) + self.assertTrue(result.passed) + self.assertEqual(mock_check.call_count, 2) + + +class TestRenderResult(IsolatedAsyncioTestCase): + def test_text_passed(self): + result = VerifyKernelTagResult( + advisories=[ + AdvisoryKernelResult( + advisory_id=100, + impetus="image", + kernel_builds=[KernelBuildInfo(nvr="kernel-5.14.0-1.el9")], + ) + ], + stop_ship_tag="early-kernel-stop-ship", + ) + text = render_result(result, "text") + self.assertIn("OK", text) + self.assertIn("PASS", text) + self.assertIn("kernel-5.14.0-1.el9", text) + + def test_text_failed(self): + result = VerifyKernelTagResult( + advisories=[ + AdvisoryKernelResult( + advisory_id=100, + impetus="image", + kernel_builds=[KernelBuildInfo(nvr="kernel-5.14.0-1.el9", has_stop_ship=True)], + ) + ], + stop_ship_tag="early-kernel-stop-ship", + ) + text = render_result(result, "text") + self.assertIn("STOP-SHIP", text) + self.assertIn("FAIL", text) + + def test_text_skipped(self): + result = VerifyKernelTagResult( + advisories=[ + AdvisoryKernelResult(advisory_id=100, impetus="rpm", skipped=True), + ], + stop_ship_tag="early-kernel-stop-ship", + ) + text = render_result(result, "text") + self.assertIn("SKIPPED", text) + + def test_text_error(self): + result = VerifyKernelTagResult( + advisories=[ + AdvisoryKernelResult(advisory_id=100, impetus="image", error="connection failed"), + ], + stop_ship_tag="early-kernel-stop-ship", + ) + text = render_result(result, "text") + self.assertIn("ERROR", text) + self.assertIn("connection failed", text) + + def test_json_output(self): + result = VerifyKernelTagResult( + advisories=[ + AdvisoryKernelResult( + advisory_id=100, + impetus="image", + rhcos_builds=["rhcos-x86_64-418.92.1-1"], + kernel_builds=[KernelBuildInfo(nvr="kernel-5.14.0-1.el9")], + ) + ], + stop_ship_tag="early-kernel-stop-ship", + ) + output = render_result(result, "json") + data = json.loads(output) + self.assertTrue(data["passed"]) + self.assertFalse(data["failed"]) + self.assertEqual(data["stop_ship_tag"], "early-kernel-stop-ship") + self.assertEqual(len(data["advisories"]), 1) + self.assertEqual(data["advisories"][0]["advisory_id"], 100) + self.assertEqual(len(data["advisories"][0]["kernel_builds"]), 1) + + def test_json_stop_ship(self): + result = VerifyKernelTagResult( + advisories=[ + AdvisoryKernelResult( + advisory_id=100, + impetus="image", + kernel_builds=[KernelBuildInfo(nvr="kernel-5.14.0-1.el9", has_stop_ship=True)], + ) + ], + stop_ship_tag="early-kernel-stop-ship", + ) + output = render_result(result, "json") + data = json.loads(output) + self.assertFalse(data["passed"]) + self.assertTrue(data["failed"]) + self.assertTrue(data["advisories"][0]["kernel_builds"][0]["has_stop_ship"])