diff --git a/elliott/elliottlib/cli/verify_cdn_push_cli.py b/elliott/elliottlib/cli/verify_cdn_push_cli.py index 5734380573..dae5258b1f 100644 --- a/elliott/elliottlib/cli/verify_cdn_push_cli.py +++ b/elliott/elliottlib/cli/verify_cdn_push_cli.py @@ -1,13 +1,17 @@ -import json import logging from dataclasses import dataclass, field from typing import Optional import click -from artcommonlib.assembly import assembly_config_struct from elliottlib.cli.common import cli, click_coroutine from elliottlib.errata_async import AsyncErrataAPI +from elliottlib.verify_common import ( + VerifyResultBase, + get_assembly_advisory_ids, + handle_verify_result, + verify_output_option, +) LOGGER = logging.getLogger(__name__) @@ -54,17 +58,51 @@ def pending(self) -> bool: @dataclass -class VerifyCdnPushResult: +class VerifyCdnPushResult(VerifyResultBase): advisories: list[AdvisoryPushResult] = field(default_factory=list) @property - def complete(self) -> bool: + def passed(self) -> bool: return bool(self.advisories) and all(a.complete for a in self.advisories) @property def failed(self) -> bool: return any(a.failed for a in self.advisories) + def to_dict(self) -> dict: + return { + "passed": self.passed, + "failed": self.failed, + "advisories": [ + { + "advisory_id": a.advisory_id, + "impetus": a.impetus, + "complete": a.complete, + "failed": a.failed, + "push_triggered": a.push_triggered, + "error": a.error, + "push_jobs": [{"target": j.target, "job_id": j.job_id, "status": j.status} for j in a.push_jobs], + } + for a in self.advisories + ], + } + + def render_text(self) -> str: + lines = ["CDN staging push status", ""] + for a in self.advisories: + status = "COMPLETE" if a.complete else ("FAIL" if a.failed else "PENDING") + lines.append(f" Advisory {a.advisory_id} ({a.impetus}): {status}") + if a.push_triggered: + lines.append(" Push re-triggered") + if a.error: + lines.append(f" Error: {a.error}") + for j in a.push_jobs: + lines.append(f" {j.target}: {j.status} (job {j.job_id})") + lines.append("") + overall = "COMPLETE" if self.passed else ("FAIL" if self.failed else "PENDING") + lines.append(f"Overall: {overall}") + return "\n".join(lines) + def parse_push_jobs(raw_jobs: list) -> list[PushJobInfo]: latest_by_target: dict[str, PushJobInfo] = {} @@ -166,59 +204,6 @@ async def verify_cdn_push(advisories: dict[str, int], do_push: bool) -> VerifyCd return result -def render_result(result: VerifyCdnPushResult, output: str) -> str: - if output == "json": - return json.dumps( - { - "complete": result.complete, - "failed": result.failed, - "advisories": [ - { - "advisory_id": a.advisory_id, - "impetus": a.impetus, - "complete": a.complete, - "failed": a.failed, - "push_triggered": a.push_triggered, - "error": a.error, - "push_jobs": [ - {"target": j.target, "job_id": j.job_id, "status": j.status} for j in a.push_jobs - ], - } - for a in result.advisories - ], - }, - indent=2, - ) - - lines = ["CDN staging push status", ""] - for a in result.advisories: - status = "COMPLETE" if a.complete else ("FAIL" if a.failed else "PENDING") - lines.append(f" Advisory {a.advisory_id} ({a.impetus}): {status}") - if a.push_triggered: - lines.append(" Push re-triggered") - if a.error: - lines.append(f" Error: {a.error}") - for j in a.push_jobs: - lines.append(f" {j.target}: {j.status} (job {j.job_id})") - lines.append("") - - overall = "COMPLETE" if result.complete else ("FAIL" if result.failed else "PENDING") - 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 CDN_PUSH_ADVISORY_TYPES: - ad_id = advisories.get(impetus) - if ad_id: - result[impetus] = int(ad_id) - return result - - @cli.command("verify-cdn-push", short_help="Verify CDN staging push jobs for advisories") @click.option( "--push/--no-push", @@ -226,14 +211,7 @@ def get_advisory_ids(runtime) -> dict[str, int]: show_default=True, help="Re-trigger CDN push for advisories with failed or missing push jobs.", ) -@click.option( - "-o", - "--output", - type=click.Choice(["text", "json"]), - default="text", - show_default=True, - help="Output format.", -) +@verify_output_option @click.pass_obj @click_coroutine async def verify_cdn_push_cli(runtime, push, output): @@ -250,12 +228,10 @@ async def verify_cdn_push_cli(runtime, push, output): elliott --group openshift-4.18 --assembly 4.18.51 verify-cdn-push """ runtime.initialize() - advisories = get_advisory_ids(runtime) + advisories = get_assembly_advisory_ids(runtime, include_types=CDN_PUSH_ADVISORY_TYPES) if not advisories: raise click.UsageError(f"No advisory IDs found for {CDN_PUSH_ADVISORY_TYPES} in assembly config.") LOGGER.info("Checking CDN push status for advisories: %s", advisories) result = await verify_cdn_push(advisories=advisories, do_push=push) - click.echo(render_result(result, output)) - if not result.complete: - raise SystemExit(1) + handle_verify_result(result, output) diff --git a/elliott/elliottlib/cli/verify_kernel_tag_cli.py b/elliott/elliottlib/cli/verify_kernel_tag_cli.py index 6fb156a83d..3d16c6609b 100644 --- a/elliott/elliottlib/cli/verify_kernel_tag_cli.py +++ b/elliott/elliottlib/cli/verify_kernel_tag_cli.py @@ -1,5 +1,4 @@ import asyncio -import json import logging import re from dataclasses import dataclass, field @@ -8,12 +7,17 @@ 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 +from elliottlib.verify_common import ( + VerifyResultBase, + get_assembly_advisory_ids, + handle_verify_result, + verify_output_option, +) LOGGER = logging.getLogger(__name__) @@ -45,7 +49,7 @@ def failed(self) -> bool: @dataclass -class VerifyKernelTagResult: +class VerifyKernelTagResult(VerifyResultBase): advisories: list[AdvisoryKernelResult] = field(default_factory=list) stop_ship_tag: str = "" @@ -57,6 +61,45 @@ def passed(self) -> bool: def failed(self) -> bool: return any(a.failed for a in self.advisories) + def to_dict(self) -> dict: + return { + "passed": self.passed, + "failed": self.failed, + "stop_ship_tag": self.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 self.advisories + ], + } + + def render_text(self) -> str: + lines = [f"Kernel stop-ship tag check (tag: {self.stop_ship_tag})", ""] + for a in self.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 self.passed else "FAIL" + lines.append(f"Overall: {overall}") + return "\n".join(lines) + def get_rpm_deliveries_config(runtime) -> list: rpm_deliveries = runtime.group_config.get("rpm_deliveries") @@ -196,71 +239,8 @@ async def verify_kernel_tag( 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.", -) +@verify_output_option @click.pass_obj @click_coroutine async def verify_kernel_tag_cli(runtime, output): @@ -288,7 +268,7 @@ async def verify_kernel_tag_cli(runtime, output): 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) + advisories = get_assembly_advisory_ids(runtime, include_types=KERNEL_TAG_ADVISORY_TYPES) if not advisories: raise click.UsageError(f"No advisory IDs found for {KERNEL_TAG_ADVISORY_TYPES} in assembly config.") @@ -307,6 +287,4 @@ async def verify_kernel_tag_cli(runtime, output): kernel_packages=kernel_packages, stop_ship_tag=stop_ship_tag, ) - click.echo(render_result(result, output)) - if not result.passed: - raise SystemExit(1) + handle_verify_result(result, output) diff --git a/elliott/elliottlib/cli/verify_metadata_url_cli.py b/elliott/elliottlib/cli/verify_metadata_url_cli.py index 2241e17cfb..ef09c373eb 100644 --- a/elliott/elliottlib/cli/verify_metadata_url_cli.py +++ b/elliott/elliottlib/cli/verify_metadata_url_cli.py @@ -9,6 +9,7 @@ from artcommonlib import exectools from elliottlib.cli.common import cli, click_coroutine +from elliottlib.verify_common import VerifyResultBase, handle_verify_result, verify_output_option LOGGER = logging.getLogger(__name__) @@ -16,8 +17,8 @@ @dataclass -class VerifyMetadataUrlResult: - release: str +class VerifyMetadataUrlResult(VerifyResultBase): + release: str = "" pullspec: str = "" metadata_url: str = "" accessible: bool = False @@ -27,9 +28,31 @@ class VerifyMetadataUrlResult: def passed(self) -> bool: return self.accessible and not self.error - @property - def failed(self) -> bool: - return not self.accessible or bool(self.error) + def to_dict(self) -> dict: + return { + "passed": self.passed, + "failed": self.failed, + "release": self.release, + "pullspec": self.pullspec, + "metadata_url": self.metadata_url, + "accessible": self.accessible, + "error": self.error, + } + + def render_text(self) -> str: + lines = ["Metadata URL check", ""] + lines.append(f" Release: {self.release}") + if self.pullspec: + lines.append(f" Pullspec: {self.pullspec}") + if self.metadata_url: + lines.append(f" Metadata URL: {self.metadata_url}") + lines.append(f" Accessible: {'yes' if self.accessible else 'no'}") + if self.error: + lines.append(f" Error: {self.error}") + lines.append("") + overall = "PASS" if self.passed else "FAIL" + lines.append(f"Overall: {overall}") + return "\n".join(lines) def _release_stream_name(release: str) -> str: @@ -123,46 +146,8 @@ async def verify_metadata_url(release: str) -> VerifyMetadataUrlResult: return result -def render_result(result: VerifyMetadataUrlResult, output: str) -> str: - if output == "json": - return json.dumps( - { - "passed": result.passed, - "failed": result.failed, - "release": result.release, - "pullspec": result.pullspec, - "metadata_url": result.metadata_url, - "accessible": result.accessible, - "error": result.error, - }, - indent=2, - ) - - lines = ["Metadata URL check", ""] - lines.append(f" Release: {result.release}") - if result.pullspec: - lines.append(f" Pullspec: {result.pullspec}") - if result.metadata_url: - lines.append(f" Metadata URL: {result.metadata_url}") - lines.append(f" Accessible: {'yes' if result.accessible else 'no'}") - if result.error: - lines.append(f" Error: {result.error}") - lines.append("") - - overall = "PASS" if result.passed else "FAIL" - lines.append(f"Overall: {overall}") - return "\n".join(lines) - - @cli.command("verify-metadata-url", short_help="Check release payload metadata URL accessibility") -@click.option( - "-o", - "--output", - type=click.Choice(["text", "json"]), - default="text", - show_default=True, - help="Output format.", -) +@verify_output_option @click.pass_obj @click_coroutine async def verify_metadata_url_cli(runtime, output): @@ -184,6 +169,4 @@ async def verify_metadata_url_cli(runtime, output): LOGGER.info("Verifying metadata URL for release %s", release) result = await verify_metadata_url(release=release) - click.echo(render_result(result, output)) - if not result.passed: - raise SystemExit(1) + handle_verify_result(result, output) diff --git a/elliott/elliottlib/cli/verify_qe_qualifier_cli.py b/elliott/elliottlib/cli/verify_qe_qualifier_cli.py index c9dedc8eba..bf20e3eb4c 100644 --- a/elliott/elliottlib/cli/verify_qe_qualifier_cli.py +++ b/elliott/elliottlib/cli/verify_qe_qualifier_cli.py @@ -1,5 +1,4 @@ import asyncio -import json import logging from dataclasses import dataclass, field from typing import Optional @@ -10,6 +9,7 @@ from artcommonlib.assembly import assembly_basis from elliottlib.cli.common import cli, click_coroutine +from elliottlib.verify_common import VerifyResultBase, handle_verify_result, verify_output_option LOGGER = logging.getLogger(__name__) @@ -30,8 +30,8 @@ def passed(self) -> bool: @dataclass -class VerifyQeQualifierResult: - assembly: str +class VerifyQeQualifierResult(VerifyResultBase): + assembly: str = "" stable_results: list[QualifierCheckResult] = field(default_factory=list) nightly_results: list[QualifierCheckResult] = field(default_factory=list) @@ -40,6 +40,59 @@ def passed(self) -> bool: all_results = self.stable_results + self.nightly_results return bool(all_results) and all(r.passed for r in all_results) + def to_dict(self) -> dict: + return { + "assembly": self.assembly, + "passed": self.passed, + "stable": [ + { + "release_tag": r.release_tag, + "arch": r.arch, + "badge_earned": r.badge_earned, + "passed": r.passed, + "error": r.error, + } + for r in self.stable_results + ], + "nightly": [ + { + "release_tag": r.release_tag, + "arch": r.arch, + "badge_earned": r.badge_earned, + "passed": r.passed, + "error": r.error, + } + for r in self.nightly_results + ], + } + + def render_text(self) -> str: + lines = [f"Assembly: {self.assembly}", ""] + + if self.stable_results: + lines.append("Stable:") + for r in self.stable_results: + status = "PASS" if r.passed else "FAIL" + if r.error: + lines.append(f" {r.arch}: ERROR - {r.error}") + else: + lines.append(f" {r.arch}: {status} (tag: {r.release_tag})") + lines.append("") + + if self.nightly_results: + lines.append("Nightly:") + for r in self.nightly_results: + status = "PASS" if r.passed else "FAIL" + if r.error: + lines.append(f" {r.arch}: ERROR - {r.error}") + else: + lines.append(f" {r.arch}: {status} (tag: {r.release_tag})") + lines.append("") + + overall = "PASS" if self.passed else "FAIL" + lines.append(f"Overall: {overall}") + return "\n".join(lines) + async def check_qe_qualifier(release_tag: str, go_arch: str, session: aiohttp.ClientSession) -> QualifierCheckResult: url = RELEASE_CONTROLLER_URL.format(go_arch=go_arch) @@ -94,65 +147,8 @@ async def verify_qe_qualifier( return result -def render_result(result: VerifyQeQualifierResult, output: str) -> str: - if output == "json": - data = { - "assembly": result.assembly, - "passed": result.passed, - "stable": [ - { - "release_tag": r.release_tag, - "arch": r.arch, - "badge_earned": r.badge_earned, - "passed": r.passed, - "error": r.error, - } - for r in result.stable_results - ], - "nightly": [ - { - "release_tag": r.release_tag, - "arch": r.arch, - "badge_earned": r.badge_earned, - "passed": r.passed, - "error": r.error, - } - for r in result.nightly_results - ], - } - return json.dumps(data, indent=2) - - lines = [f"Assembly: {result.assembly}", ""] - - if result.stable_results: - lines.append("Stable:") - for r in result.stable_results: - status = "PASS" if r.passed else "FAIL" - if r.error: - lines.append(f" {r.arch}: ERROR - {r.error}") - else: - lines.append(f" {r.arch}: {status} (tag: {r.release_tag})") - lines.append("") - - if result.nightly_results: - lines.append("Nightly:") - for r in result.nightly_results: - status = "PASS" if r.passed else "FAIL" - if r.error: - lines.append(f" {r.arch}: ERROR - {r.error}") - else: - lines.append(f" {r.arch}: {status} (tag: {r.release_tag})") - lines.append("") - - overall = "PASS" if result.passed else "FAIL" - lines.append(f"Overall: {overall}") - return "\n".join(lines) - - @cli.command("verify-qe-qualifier", short_help="Check release controller QE qualifier for stable and nightly builds") -@click.option( - "-o", "--output", type=click.Choice(["text", "json"]), default="text", show_default=True, help="Output format." -) +@verify_output_option @click.option("--stable/--no-stable", default=True, show_default=True, help="Check stable build QE qualifier.") @click.option("--nightly/--no-nightly", default=True, show_default=True, help="Check nightly build QE qualifier.") @click.pass_obj @@ -199,6 +195,4 @@ async def verify_qe_qualifier_cli(runtime, output, stable, nightly): check_stable=stable, check_nightly=nightly, ) - click.echo(render_result(result, output)) - if not result.passed: - raise SystemExit(1) + handle_verify_result(result, output) diff --git a/elliott/elliottlib/cli/verify_security_alerts_cli.py b/elliott/elliottlib/cli/verify_security_alerts_cli.py index c7c7b517fb..0a242ce415 100644 --- a/elliott/elliottlib/cli/verify_security_alerts_cli.py +++ b/elliott/elliottlib/cli/verify_security_alerts_cli.py @@ -1,14 +1,18 @@ import asyncio -import json import logging from dataclasses import dataclass, field from typing import Optional import click -from artcommonlib.assembly import assembly_config_struct from elliottlib.cli.common import cli, click_coroutine from elliottlib.errata_async import AsyncErrataAPI +from elliottlib.verify_common import ( + VerifyResultBase, + get_assembly_advisory_ids, + handle_verify_result, + verify_output_option, +) LOGGER = logging.getLogger(__name__) @@ -32,16 +36,46 @@ def failed(self) -> bool: @dataclass -class VerifySecurityAlertsResult: +class VerifySecurityAlertsResult(VerifyResultBase): advisories: list[AdvisoryAlertResult] = field(default_factory=list) @property - def ok(self) -> bool: + def passed(self) -> bool: return all(a.ok for a in self.advisories) - @property - def failed(self) -> bool: - return any(a.failed for a in self.advisories) + def to_dict(self) -> dict: + return { + "passed": self.passed, + "failed": self.failed, + "advisories": [ + { + "advisory_id": a.advisory_id, + "impetus": a.impetus, + "errata_type": a.errata_type, + "blocking": a.blocking, + "skipped": a.skipped, + "error": a.error, + } + for a in self.advisories + ], + } + + def render_text(self) -> str: + lines = ["Security alerts check", ""] + for a in self.advisories: + if a.skipped: + lines.append(f" Advisory {a.advisory_id} ({a.impetus}): SKIPPED ({a.errata_type.upper()})") + elif a.blocking: + lines.append(f" Advisory {a.advisory_id} ({a.impetus}): BLOCKING") + elif a.error: + lines.append(f" Advisory {a.advisory_id} ({a.impetus}): ERROR") + lines.append(f" {a.error}") + else: + lines.append(f" Advisory {a.advisory_id} ({a.impetus}): OK") + lines.append("") + overall = "OK" if self.passed else "FAIL" + lines.append(f"Overall: {overall}") + return "\n".join(lines) def get_errata_type(advisory_data: dict) -> str: @@ -106,69 +140,12 @@ async def verify_security_alerts(advisories: dict[str, int]) -> VerifySecurityAl return result -def render_result(result: VerifySecurityAlertsResult, output: str) -> str: - if output == "json": - return json.dumps( - { - "ok": result.ok, - "failed": result.failed, - "advisories": [ - { - "advisory_id": a.advisory_id, - "impetus": a.impetus, - "errata_type": a.errata_type, - "blocking": a.blocking, - "skipped": a.skipped, - "error": a.error, - } - for a in result.advisories - ], - }, - indent=2, - ) - - lines = ["Security alerts check", ""] - for a in result.advisories: - if a.skipped: - lines.append(f" Advisory {a.advisory_id} ({a.impetus}): SKIPPED ({a.errata_type.upper()})") - elif a.blocking: - lines.append(f" Advisory {a.advisory_id} ({a.impetus}): BLOCKING") - elif a.error: - lines.append(f" Advisory {a.advisory_id} ({a.impetus}): ERROR") - lines.append(f" {a.error}") - else: - lines.append(f" Advisory {a.advisory_id} ({a.impetus}): OK") - lines.append("") - - overall = "OK" if result.ok else "FAIL" - lines.append(f"Overall: {overall}") - return "\n".join(lines) - - # microshift advisories are managed separately and don't go through ProdSec alert flow SKIPPED_IMPETUS = ("microshift",) -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, ad_id in advisories.items(): - if ad_id and impetus not in SKIPPED_IMPETUS: - result[impetus] = int(ad_id) - return result - - @cli.command("verify-security-alerts", short_help="Check RHSA advisories for blocking security alerts") -@click.option( - "-o", - "--output", - type=click.Choice(["text", "json"]), - default="text", - show_default=True, - help="Output format.", -) +@verify_output_option @click.pass_obj @click_coroutine async def verify_security_alerts_cli(runtime, output): @@ -186,12 +163,10 @@ async def verify_security_alerts_cli(runtime, output): elliott --group openshift-4.18 --assembly 4.18.51 verify-security-alerts """ runtime.initialize() - advisories = get_advisory_ids(runtime) + advisories = get_assembly_advisory_ids(runtime, exclude_types=SKIPPED_IMPETUS) if not advisories: raise click.UsageError("No advisory IDs found in assembly config.") LOGGER.info("Checking security alerts for advisories: %s", advisories) result = await verify_security_alerts(advisories=advisories) - click.echo(render_result(result, output)) - if not result.ok: - raise SystemExit(1) + handle_verify_result(result, output) diff --git a/elliott/elliottlib/verify_common.py b/elliott/elliottlib/verify_common.py new file mode 100644 index 0000000000..b17ba91833 --- /dev/null +++ b/elliott/elliottlib/verify_common.py @@ -0,0 +1,125 @@ +"""Shared utilities for verify-* elliott subcommands. + +Provides: +- get_assembly_advisory_ids(): unified advisory ID lookup from assembly config +- VerifyResultBase: abstract base for verify result dataclasses +- render_verify_result(): generic JSON/text rendering +- verify_output_option: shared --output click option +- handle_verify_result(): echo result and exit on failure +""" + +import json +from abc import ABC, abstractmethod +from dataclasses import dataclass + +import click +from artcommonlib.assembly import assembly_config_struct + + +def get_assembly_advisory_ids( + runtime, + include_types: tuple[str, ...] | None = None, + exclude_types: tuple[str, ...] = (), +) -> dict[str, int]: + """Get advisory IDs from assembly config, filtered by impetus type. + + Reads the assembly's group config from releases.yml and returns + advisory IDs filtered by impetus type. + + Args: + runtime: Elliott runtime (must be initialized). + include_types: If set, only include these impetus types. + exclude_types: Impetus types to skip. + + Returns: + dict mapping impetus name to advisory ID. + """ + releases_config = runtime.get_releases_config() + group_config = assembly_config_struct(releases_config, runtime.assembly, "group", {}) + advisories = group_config.get("advisories", {}) + result = {} + for impetus, ad_id in advisories.items(): + if not ad_id: + continue + if include_types is not None and impetus not in include_types: + continue + if impetus in exclude_types: + continue + result[impetus] = int(ad_id) + return result + + +@dataclass +class VerifyResultBase(ABC): + """Abstract base for verify-* command top-level result dataclasses. + + Subclasses must implement: + - ``passed``: whether the verification succeeded + - ``to_dict()``: JSON-serializable dict representation + - ``render_text()``: human-readable text representation + + The ``failed`` property defaults to ``not self.passed`` but can be + overridden for tri-state results (e.g. complete / pending / failed). + """ + + @property + @abstractmethod + def passed(self) -> bool: + """Whether the verification succeeded.""" + ... + + @property + def failed(self) -> bool: + """Whether the verification failed. + + Override for tri-state results where ``not passed`` does not + imply ``failed`` (e.g. a pending state). + """ + return not self.passed + + @abstractmethod + def to_dict(self) -> dict: + """Return a JSON-serializable dict representation.""" + ... + + @abstractmethod + def render_text(self) -> str: + """Return a human-readable text representation.""" + ... + + +def render_verify_result(result: VerifyResultBase, output: str) -> str: + """Render a verify result in the requested format. + + Args: + result: A VerifyResultBase subclass instance. + output: ``"json"`` or ``"text"``. + + Returns: + Formatted string. + """ + if output == "json": + return json.dumps(result.to_dict(), indent=2) + return result.render_text() + + +verify_output_option = click.option( + "-o", + "--output", + type=click.Choice(["text", "json"]), + default="text", + show_default=True, + help="Output format.", +) + + +def handle_verify_result(result: VerifyResultBase, output: str) -> None: + """Render and echo a verify result, exiting with code 1 on failure. + + Args: + result: A VerifyResultBase subclass instance. + output: ``"json"`` or ``"text"``. + """ + click.echo(render_verify_result(result, output)) + if not result.passed: + raise SystemExit(1) diff --git a/elliott/tests/test_verify_cdn_push_cli.py b/elliott/tests/test_verify_cdn_push_cli.py index 0be614f310..44886eb549 100644 --- a/elliott/tests/test_verify_cdn_push_cli.py +++ b/elliott/tests/test_verify_cdn_push_cli.py @@ -8,9 +8,9 @@ VerifyCdnPushResult, check_advisory_push, parse_push_jobs, - render_result, verify_cdn_push, ) +from elliottlib.verify_common import render_verify_result class TestPushJobInfo(TestCase): @@ -96,7 +96,7 @@ def test_all_complete(self): ), ] ) - self.assertTrue(r.complete) + self.assertTrue(r.passed) self.assertFalse(r.failed) def test_one_failed(self): @@ -114,7 +114,7 @@ def test_one_failed(self): ), ] ) - self.assertFalse(r.complete) + self.assertFalse(r.passed) self.assertTrue(r.failed) @@ -221,7 +221,7 @@ async def test_all_complete(self, mock_api_cls): ] result = await verify_cdn_push({"rpm": 111}, do_push=True) - self.assertTrue(result.complete) + self.assertTrue(result.passed) @patch("elliottlib.cli.verify_cdn_push_cli.AsyncErrataAPI") async def test_with_blocking_advisory(self, mock_api_cls): @@ -235,7 +235,7 @@ async def test_with_blocking_advisory(self, mock_api_cls): ] result = await verify_cdn_push({"rpm": 111}, do_push=True) - self.assertTrue(result.complete) + self.assertTrue(result.passed) self.assertEqual(len(result.advisories), 2) @patch("elliottlib.cli.verify_cdn_push_cli.AsyncErrataAPI") @@ -250,7 +250,7 @@ async def test_blocking_incomplete_skips_main(self, mock_api_cls): ] result = await verify_cdn_push({"rpm": 111}, do_push=True) - self.assertFalse(result.complete) + self.assertFalse(result.passed) @patch("elliottlib.cli.verify_cdn_push_cli.AsyncErrataAPI") async def test_blocking_lookup_failure(self, mock_api_cls): @@ -261,7 +261,7 @@ async def test_blocking_lookup_failure(self, mock_api_cls): api.get_advisory.side_effect = RuntimeError("connection failed") result = await verify_cdn_push({"rpm": 111}, do_push=True) - self.assertFalse(result.complete) + self.assertFalse(result.passed) self.assertTrue(result.failed) @@ -276,7 +276,7 @@ def test_text_complete(self): ), ] ) - text = render_result(r, "text") + text = render_verify_result(r, "text") self.assertIn("COMPLETE", text) self.assertIn("12345", text) @@ -291,7 +291,7 @@ def test_text_with_push_triggered(self): ), ] ) - text = render_result(r, "text") + text = render_verify_result(r, "text") self.assertIn("re-triggered", text.lower()) def test_json_output(self): @@ -304,8 +304,8 @@ def test_json_output(self): ), ] ) - data = json.loads(render_result(r, "json")) - self.assertTrue(data["complete"]) + data = json.loads(render_verify_result(r, "json")) + self.assertTrue(data["passed"]) self.assertEqual(len(data["advisories"]), 1) self.assertEqual(data["advisories"][0]["advisory_id"], 12345) @@ -315,6 +315,6 @@ def test_text_fail(self): AdvisoryPushResult(advisory_id=12345, impetus="rpm", error="boom"), ] ) - text = render_result(r, "text") + text = render_verify_result(r, "text") self.assertIn("FAIL", text) self.assertIn("boom", text) diff --git a/elliott/tests/test_verify_common.py b/elliott/tests/test_verify_common.py new file mode 100644 index 0000000000..dd6f2102e7 --- /dev/null +++ b/elliott/tests/test_verify_common.py @@ -0,0 +1,150 @@ +import json +from dataclasses import dataclass +from typing import Optional +from unittest import TestCase +from unittest.mock import MagicMock + +from elliottlib.verify_common import ( + VerifyResultBase, + get_assembly_advisory_ids, + handle_verify_result, + render_verify_result, +) + + +class TestGetAssemblyAdvisoryIds(TestCase): + def _make_runtime(self, advisories: dict): + runtime = MagicMock() + runtime.assembly = "4.18.51" + mock_config = MagicMock() + mock_config.get.return_value = advisories + runtime.get_releases_config.return_value = MagicMock() + + import artcommonlib.assembly as asm + + original = asm.assembly_config_struct + + def fake_config_struct(releases_config, assembly, key, default): + if key == "group": + return mock_config + return original(releases_config, assembly, key, default) + + self._patcher = MagicMock() + import unittest.mock + + self._patcher = unittest.mock.patch( + "elliottlib.verify_common.assembly_config_struct", + side_effect=fake_config_struct, + ) + self._patcher.start() + return runtime + + def tearDown(self): + if hasattr(self, "_patcher") and hasattr(self._patcher, "stop"): + self._patcher.stop() + + def test_no_filter(self): + runtime = self._make_runtime({"rpm": 111, "image": 222, "rhcos": 333}) + result = get_assembly_advisory_ids(runtime) + self.assertEqual(result, {"rpm": 111, "image": 222, "rhcos": 333}) + + def test_include_types(self): + runtime = self._make_runtime({"rpm": 111, "image": 222, "rhcos": 333}) + result = get_assembly_advisory_ids(runtime, include_types=("rpm", "rhcos")) + self.assertEqual(result, {"rpm": 111, "rhcos": 333}) + + def test_exclude_types(self): + runtime = self._make_runtime({"rpm": 111, "image": 222, "microshift": 444}) + result = get_assembly_advisory_ids(runtime, exclude_types=("microshift",)) + self.assertEqual(result, {"rpm": 111, "image": 222}) + + def test_include_and_exclude(self): + runtime = self._make_runtime({"rpm": 111, "image": 222, "rhcos": 333}) + result = get_assembly_advisory_ids(runtime, include_types=("rpm", "image", "rhcos"), exclude_types=("image",)) + self.assertEqual(result, {"rpm": 111, "rhcos": 333}) + + def test_skips_empty_ids(self): + runtime = self._make_runtime({"rpm": 111, "image": 0, "rhcos": None}) + result = get_assembly_advisory_ids(runtime) + self.assertEqual(result, {"rpm": 111}) + + def test_converts_to_int(self): + runtime = self._make_runtime({"rpm": "111"}) + result = get_assembly_advisory_ids(runtime) + self.assertEqual(result, {"rpm": 111}) + self.assertIsInstance(result["rpm"], int) + + +# Concrete subclass for testing VerifyResultBase +@dataclass +class _TestResult(VerifyResultBase): + success: bool = True + error: Optional[str] = None + + @property + def passed(self) -> bool: + return self.success and not self.error + + def to_dict(self) -> dict: + return {"passed": self.passed, "failed": self.failed, "error": self.error} + + def render_text(self) -> str: + status = "PASS" if self.passed else "FAIL" + lines = [f"Test result: {status}"] + if self.error: + lines.append(f" Error: {self.error}") + lines.append(f"Overall: {status}") + return "\n".join(lines) + + +class TestVerifyResultBase(TestCase): + def test_passed(self): + r = _TestResult(success=True) + self.assertTrue(r.passed) + self.assertFalse(r.failed) + + def test_failed(self): + r = _TestResult(success=False) + self.assertFalse(r.passed) + self.assertTrue(r.failed) + + def test_error_implies_failed(self): + r = _TestResult(success=True, error="boom") + self.assertFalse(r.passed) + self.assertTrue(r.failed) + + +class TestRenderVerifyResult(TestCase): + def test_json(self): + r = _TestResult(success=True) + output = render_verify_result(r, "json") + data = json.loads(output) + self.assertTrue(data["passed"]) + self.assertFalse(data["failed"]) + + def test_text(self): + r = _TestResult(success=True) + output = render_verify_result(r, "text") + self.assertIn("PASS", output) + + def test_text_failed(self): + r = _TestResult(success=False, error="something broke") + output = render_verify_result(r, "text") + self.assertIn("FAIL", output) + self.assertIn("something broke", output) + + +class TestHandleVerifyResult(TestCase): + def test_passed_no_exit(self): + r = _TestResult(success=True) + # Should not raise + try: + handle_verify_result(r, "text") + except SystemExit: + self.fail("handle_verify_result raised SystemExit on passing result") + + def test_failed_exits(self): + r = _TestResult(success=False) + with self.assertRaises(SystemExit) as ctx: + handle_verify_result(r, "text") + self.assertEqual(ctx.exception.code, 1) diff --git a/elliott/tests/test_verify_kernel_tag_cli.py b/elliott/tests/test_verify_kernel_tag_cli.py index 019e7f32ff..1414b006dd 100644 --- a/elliott/tests/test_verify_kernel_tag_cli.py +++ b/elliott/tests/test_verify_kernel_tag_cli.py @@ -14,9 +14,9 @@ get_kernel_rpms_from_rhcos, get_rpm_deliveries_config, nvr_to_brewroot_metadata_url, - render_result, verify_kernel_tag, ) +from elliottlib.verify_common import render_verify_result from requests.exceptions import HTTPError @@ -407,7 +407,7 @@ def test_text_passed(self): ], stop_ship_tag="early-kernel-stop-ship", ) - text = render_result(result, "text") + text = render_verify_result(result, "text") self.assertIn("OK", text) self.assertIn("PASS", text) self.assertIn("kernel-5.14.0-1.el9", text) @@ -423,7 +423,7 @@ def test_text_failed(self): ], stop_ship_tag="early-kernel-stop-ship", ) - text = render_result(result, "text") + text = render_verify_result(result, "text") self.assertIn("STOP-SHIP", text) self.assertIn("FAIL", text) @@ -434,7 +434,7 @@ def test_text_skipped(self): ], stop_ship_tag="early-kernel-stop-ship", ) - text = render_result(result, "text") + text = render_verify_result(result, "text") self.assertIn("SKIPPED", text) def test_text_error(self): @@ -444,7 +444,7 @@ def test_text_error(self): ], stop_ship_tag="early-kernel-stop-ship", ) - text = render_result(result, "text") + text = render_verify_result(result, "text") self.assertIn("ERROR", text) self.assertIn("connection failed", text) @@ -460,7 +460,7 @@ def test_json_output(self): ], stop_ship_tag="early-kernel-stop-ship", ) - output = render_result(result, "json") + output = render_verify_result(result, "json") data = json.loads(output) self.assertTrue(data["passed"]) self.assertFalse(data["failed"]) @@ -480,7 +480,7 @@ def test_json_stop_ship(self): ], stop_ship_tag="early-kernel-stop-ship", ) - output = render_result(result, "json") + output = render_verify_result(result, "json") data = json.loads(output) self.assertFalse(data["passed"]) self.assertTrue(data["failed"]) diff --git a/elliott/tests/test_verify_metadata_url_cli.py b/elliott/tests/test_verify_metadata_url_cli.py index 4ef03a9407..aa5db301a7 100644 --- a/elliott/tests/test_verify_metadata_url_cli.py +++ b/elliott/tests/test_verify_metadata_url_cli.py @@ -8,10 +8,10 @@ check_url_accessible, extract_metadata_url, get_release_pullspec, - render_result, validate_metadata_url, verify_metadata_url, ) +from elliottlib.verify_common import render_verify_result class TestVerifyMetadataUrlResult(TestCase): @@ -302,7 +302,7 @@ def test_text_pass(self): metadata_url="https://access.redhat.com/errata/RHBA-2025:1234", accessible=True, ) - text = render_result(r, "text") + text = render_verify_result(r, "text") self.assertIn("PASS", text) self.assertIn("4.18", text) self.assertIn("access.redhat.com", text) @@ -314,13 +314,13 @@ def test_text_fail(self): metadata_url="https://access.redhat.com/errata/RHBA-2025:1234", accessible=False, ) - text = render_result(r, "text") + text = render_verify_result(r, "text") self.assertIn("FAIL", text) self.assertIn("no", text) def test_text_error(self): r = VerifyMetadataUrlResult(release="4.18", error="API unreachable") - text = render_result(r, "text") + text = render_verify_result(r, "text") self.assertIn("FAIL", text) self.assertIn("API unreachable", text) @@ -331,7 +331,7 @@ def test_json_pass(self): metadata_url="https://access.redhat.com/errata/RHBA-2025:1234", accessible=True, ) - data = json.loads(render_result(r, "json")) + data = json.loads(render_verify_result(r, "json")) self.assertTrue(data["passed"]) self.assertFalse(data["failed"]) self.assertEqual(data["release"], "4.18") @@ -339,7 +339,7 @@ def test_json_pass(self): def test_json_fail(self): r = VerifyMetadataUrlResult(release="4.18", error="boom") - data = json.loads(render_result(r, "json")) + data = json.loads(render_verify_result(r, "json")) self.assertFalse(data["passed"]) self.assertTrue(data["failed"]) self.assertEqual(data["error"], "boom") diff --git a/elliott/tests/test_verify_qe_qualifier_cli.py b/elliott/tests/test_verify_qe_qualifier_cli.py index 07ac3d9dc1..fe1a7c0790 100644 --- a/elliott/tests/test_verify_qe_qualifier_cli.py +++ b/elliott/tests/test_verify_qe_qualifier_cli.py @@ -6,9 +6,9 @@ QualifierCheckResult, VerifyQeQualifierResult, check_qe_qualifier, - render_result, verify_qe_qualifier, ) +from elliottlib.verify_common import render_verify_result class TestQualifierCheckResult(TestCase): @@ -314,7 +314,7 @@ def test_text_all_pass(self): QualifierCheckResult(release_tag="4.22.0-0.nightly-2026-08-05-104816", arch="amd64", badge_earned=True) ], ) - text = render_result(result, "text") + text = render_verify_result(result, "text") self.assertIn("Assembly: 4.22.9", text) self.assertIn("Overall: PASS", text) self.assertIn("Stable:", text) @@ -325,7 +325,7 @@ def test_text_fail(self): assembly="4.22.9", stable_results=[QualifierCheckResult(release_tag="4.22.9", arch="amd64", badge_earned=False)], ) - text = render_result(result, "text") + text = render_verify_result(result, "text") self.assertIn("Overall: FAIL", text) def test_text_error(self): @@ -333,7 +333,7 @@ def test_text_error(self): assembly="4.22.9", stable_results=[QualifierCheckResult(release_tag="4.22.9", arch="amd64", error="release not found")], ) - text = render_result(result, "text") + text = render_verify_result(result, "text") self.assertIn("ERROR", text) self.assertIn("release not found", text) @@ -345,7 +345,7 @@ def test_json_output(self): QualifierCheckResult(release_tag="4.22.0-0.nightly-2026-08-05-104816", arch="amd64", badge_earned=True) ], ) - text = render_result(result, "json") + text = render_verify_result(result, "json") data = json.loads(text) self.assertEqual(data["assembly"], "4.22.9") self.assertTrue(data["passed"]) @@ -358,7 +358,7 @@ def test_json_with_error(self): assembly="4.22.9", stable_results=[QualifierCheckResult(release_tag="4.22.9", arch="amd64", error="not found")], ) - text = render_result(result, "json") + text = render_verify_result(result, "json") data = json.loads(text) self.assertFalse(data["passed"]) self.assertEqual(data["stable"][0]["error"], "not found") diff --git a/elliott/tests/test_verify_security_alerts_cli.py b/elliott/tests/test_verify_security_alerts_cli.py index d1b69088ef..3c6f90ded6 100644 --- a/elliott/tests/test_verify_security_alerts_cli.py +++ b/elliott/tests/test_verify_security_alerts_cli.py @@ -7,9 +7,9 @@ VerifySecurityAlertsResult, check_advisory_security_alerts, get_errata_type, - render_result, verify_security_alerts, ) +from elliottlib.verify_common import render_verify_result class TestAdvisoryAlertResult(TestCase): @@ -33,6 +33,9 @@ def test_skipped_is_ok(self): self.assertTrue(r.ok) self.assertFalse(r.failed) + # AdvisoryAlertResult keeps its domain-specific .ok property; + # VerifySecurityAlertsResult uses the standardized .passed property. + class TestVerifySecurityAlertsResult(TestCase): def test_all_ok(self): @@ -42,7 +45,7 @@ def test_all_ok(self): AdvisoryAlertResult(advisory_id=2, impetus="rhcos", errata_type="rhba", skipped=True), ] ) - self.assertTrue(r.ok) + self.assertTrue(r.passed) self.assertFalse(r.failed) def test_one_blocking(self): @@ -52,7 +55,7 @@ def test_one_blocking(self): AdvisoryAlertResult(advisory_id=2, impetus="rhcos", errata_type="rhba", skipped=True), ] ) - self.assertFalse(r.ok) + self.assertFalse(r.passed) self.assertTrue(r.failed) @@ -137,7 +140,7 @@ async def test_all_ok(self, mock_api_cls): api.get_advisory.return_value = {"errata": {"rhba": {}}} result = await verify_security_alerts({"rpm": 111, "rhcos": 222}) - self.assertTrue(result.ok) + self.assertTrue(result.passed) self.assertEqual(len(result.advisories), 2) @patch("elliottlib.cli.verify_security_alerts_cli.AsyncErrataAPI") @@ -150,7 +153,7 @@ async def test_rhsa_blocking(self, mock_api_cls): api.refresh_security_alerts.return_value = {"alerts": {"blocking": True, "alerts": [{"id": 1}]}} result = await verify_security_alerts({"rpm": 111}) - self.assertFalse(result.ok) + self.assertFalse(result.passed) self.assertTrue(result.failed) @patch("elliottlib.cli.verify_security_alerts_cli.AsyncErrataAPI") @@ -168,7 +171,7 @@ def get_advisory_side_effect(advisory_id): api.refresh_security_alerts.return_value = {"alerts": {"blocking": False, "alerts": []}} result = await verify_security_alerts({"rpm": 111, "rhcos": 222}) - self.assertTrue(result.ok) + self.assertTrue(result.passed) rhsa_result = next(a for a in result.advisories if a.impetus == "rpm") rhba_result = next(a for a in result.advisories if a.impetus == "rhcos") self.assertFalse(rhsa_result.skipped) @@ -182,7 +185,7 @@ def test_text_ok(self): AdvisoryAlertResult(advisory_id=12345, impetus="rpm", errata_type="rhsa"), ] ) - text = render_result(r, "text") + text = render_verify_result(r, "text") self.assertIn("OK", text) self.assertIn("12345", text) @@ -192,7 +195,7 @@ def test_text_blocking(self): AdvisoryAlertResult(advisory_id=12345, impetus="rpm", errata_type="rhsa", blocking=True), ] ) - text = render_result(r, "text") + text = render_verify_result(r, "text") self.assertIn("BLOCKING", text) self.assertIn("FAIL", text) @@ -202,7 +205,7 @@ def test_text_skipped(self): AdvisoryAlertResult(advisory_id=12345, impetus="rpm", errata_type="rhba", skipped=True), ] ) - text = render_result(r, "text") + text = render_verify_result(r, "text") self.assertIn("SKIPPED", text) self.assertIn("RHBA", text) @@ -212,8 +215,8 @@ def test_json_output(self): AdvisoryAlertResult(advisory_id=12345, impetus="rpm", errata_type="rhsa", blocking=True), ] ) - data = json.loads(render_result(r, "json")) - self.assertFalse(data["ok"]) + data = json.loads(render_verify_result(r, "json")) + self.assertFalse(data["passed"]) self.assertTrue(data["failed"]) self.assertTrue(data["advisories"][0]["blocking"]) @@ -223,6 +226,6 @@ def test_text_error(self): AdvisoryAlertResult(advisory_id=12345, impetus="rpm", error="boom"), ] ) - text = render_result(r, "text") + text = render_verify_result(r, "text") self.assertIn("ERROR", text) self.assertIn("boom", text)