Add multi-OCP FBC support for LP assembly pipelines

- Extract shared FBC validation utilities into pyartcd/fbc_util.py
  (extract_fbc_labels, validate_fbc_related_images, extract_ocp_version_from_nvr)
- Refactor release_from_fbc.py to delegate to shared fbc_util functions
- gen_assembly_lp: validate FBC consistency when multiple pullspecs are
  provided, and store basis.fbc_pullspecs as a dict keyed by OCP version
- prepare_release_lp: loop over OCP_TARGET_VERSIONS from group config,
  build FBC per version with --major-minor, and validate consistency
  before creating shipments
- Add comprehensive tests for fbc_util, gen_assembly_lp, and
  prepare_release_lp multi-FBC scenarios

Co-authored-by: Cursor <cursoragent@cursor.com>

rh-pre-commit.version: 2.3.2
rh-pre-commit.check-secrets: ENABLED

Author: ashwindasr

pyartcd/pyartcd/fbc_util.py

@@ -0,0 +1,191 @@
+"""
+Shared utilities for FBC (File-Based Catalog) image validation.
+
+Used by release_from_fbc, gen_assembly_lp, and prepare_release_lp pipelines
+to validate consistency of related images across FBC builds targeting
+different OCP versions.
+"""
+
+import asyncio
+import json
+import logging
+import os
+import re
+from collections import defaultdict
+from typing import Dict, List, Optional
+
+from artcommonlib import exectools
+from elliottlib.util import extract_nvrs_from_fbc
+from tenacity import retry, stop_after_attempt
+
+logger = logging.getLogger(__name__)
+
+_OCP_VERSION_RE = re.compile(r'\.ocp(\d+\.\d+)')
+
+
+def extract_ocp_version_from_nvr(nvr: str) -> Optional[str]:
+    """Extract the OCP target version from an LP FBC NVR.
+
+    LP FBC NVRs encode the target OCP version in the release field as
+    ``.ocp{major}.{minor}`` (e.g. ``operator-fbc-6.4.1-1234.ocp4.18``).
+    Returns the version string (e.g. ``"4.18"``) or None if not found.
+    """
+    match = _OCP_VERSION_RE.search(nvr)
+    return match.group(1) if match else None
+
+
+async def extract_fbc_labels(fbc_pullspec: str) -> Dict[str, Optional[str]]:
+    """
+    Extract the NVR and __doozer_key labels from an FBC image.
+
+    Returns a dict with 'nvr' and 'doozer_key' keys. On failure, returns
+    None values rather than raising.
+    """
+    logger.info("Extracting FBC labels from FBC pullspec: %s", fbc_pullspec)
+
+    @retry(reraise=True, stop=stop_after_attempt(3))
+    async def _get_image_info():
+        oc_cmd = ['oc', 'image', 'info', fbc_pullspec, '--filter-by-os', 'amd64', '-o', 'json']
+
+        quay_auth_file = os.getenv("QUAY_AUTH_FILE")
+        if quay_auth_file:
+            oc_cmd.extend(['--registry-config', quay_auth_file])
+
+        rc, stdout, stderr = await exectools.cmd_gather_async(oc_cmd, check=False)
+        if rc != 0:
+            raise RuntimeError(f"Failed to get image info for FBC {fbc_pullspec}: {stderr}")
+        return stdout
+
+    try:
+        image_info_output = await _get_image_info()
+        image_info = json.loads(image_info_output)
+
+        labels = image_info.get('config', {}).get('config', {}).get('Labels', {})
+        nvr = labels.get('com.redhat.art.nvr')
+        doozer_key = labels.get('__doozer_key')
+
+        logger.info("Extracted FBC labels - nvr: %s, doozer_key: %s", nvr, doozer_key)
+        return {'nvr': nvr, 'doozer_key': doozer_key}
+
+    except Exception as e:
+        logger.exception("Failed to extract labels from FBC %s: %s", fbc_pullspec, e)
+        return {'nvr': None, 'doozer_key': None}
+
+
+async def validate_fbc_related_images(fbc_pullspecs: List[str], product: str) -> List[str]:
+    """
+    Validate that FBC pullspecs from the same operator have the same related images.
+
+    Different operators are allowed to have different related images, but
+    multiple FBC builds for the *same* operator (targeting different OCP versions)
+    must contain identical related-image NVR sets.
+
+    Returns the combined list of all unique related images across all operators.
+    Raises RuntimeError if any same-operator FBCs have mismatched related images.
+    """
+    logger.info("Validating related images for %d FBC builds (grouped by operator)...", len(fbc_pullspecs))
+
+    # Step 1: Extract operator identifiers from all FBCs in parallel
+    logger.info("Extracting operator identifiers from FBC images...")
+    label_tasks = [extract_fbc_labels(fbc) for fbc in fbc_pullspecs]
+    fbc_labels_list = await asyncio.gather(*label_tasks)
+
+    # Step 2: Map FBCs to operators
+    fbc_to_operator = {}
+    for fbc_pullspec, labels in zip(fbc_pullspecs, fbc_labels_list):
+        operator = labels.get('doozer_key')
+
+        if not operator:
+            try:
+                tag = fbc_pullspec.split(':')[-1]
+                if '-fbc-' in tag:
+                    operator = tag.split('-fbc-')[0]
+                    logger.error(
+                        "Missing __doozer_key label for %s, using extracted operator name: %s",
+                        fbc_pullspec, operator,
+                    )
+                else:
+                    operator = f"UNKNOWN-{hash(fbc_pullspec) & 0xFFFFFFFF:08x}"
+                    logger.error(
+                        "Missing __doozer_key label for %s, using generated operator identifier: %s",
+                        fbc_pullspec, operator,
+                    )
+            except Exception as e:
+                logger.error("Failed to extract operator from %s: %s", fbc_pullspec, e)
+                operator = f"UNKNOWN-{hash(fbc_pullspec) & 0xFFFFFFFF:08x}"
+
+        fbc_to_operator[fbc_pullspec] = operator
+
+    # Step 3: Group FBCs by operator
+    operator_groups = defaultdict(list)
+    for fbc, operator in fbc_to_operator.items():
+        operator_groups[operator].append(fbc)
+
+    logger.info("Found %d operator group(s):", len(operator_groups))
+    for operator, fbcs in operator_groups.items():
+        logger.info("  %s: %d FBC(s)", operator, len(fbcs))
+
+    # Step 4: Extract related images from each FBC sequentially to avoid temp directory conflicts
+    logger.info("Extracting related images from all FBCs...")
+    fbc_related_images = {}
+    for fbc_pullspec in fbc_pullspecs:
+        related_nvrs = await extract_nvrs_from_fbc(fbc_pullspec, product)
+        fbc_related_images[fbc_pullspec] = sorted(related_nvrs)
+
+    # Step 5: Validate related images within each operator group
+    all_related_images = set()
+    validation_errors = []
+
+    for operator, fbcs_in_group in operator_groups.items():
+        if len(fbcs_in_group) == 1:
+            logger.info("Operator '%s': single FBC, no validation needed", operator)
+            all_related_images.update(fbc_related_images[fbcs_in_group[0]])
+            continue
+
+        logger.info("Validating %d FBCs for operator '%s'...", len(fbcs_in_group), operator)
+        reference_fbc = fbcs_in_group[0]
+        reference_images = fbc_related_images[reference_fbc]
+
+        mismatches = []
+        for fbc_pullspec in fbcs_in_group[1:]:
+            current_images = fbc_related_images[fbc_pullspec]
+            if current_images != reference_images:
+                only_in_reference = set(reference_images) - set(current_images)
+                only_in_current = set(current_images) - set(reference_images)
+                mismatches.append(
+                    {
+                        'fbc': fbc_pullspec,
+                        'only_in_reference': sorted(only_in_reference),
+                        'only_in_current': sorted(only_in_current),
+                    }
+                )
+
+        if mismatches:
+            error_msg = f"Operator '{operator}': FBC builds have mismatched related images.\n"
+            error_msg += f"Reference FBC: {reference_fbc}\n"
+            for mismatch in mismatches:
+                error_msg += f"\nMismatch with: {mismatch['fbc']}\n"
+                if mismatch['only_in_reference']:
+                    error_msg += f"  Only in reference: {mismatch['only_in_reference']}\n"
+                if mismatch['only_in_current']:
+                    error_msg += f"  Only in current: {mismatch['only_in_current']}\n"
+            validation_errors.append(error_msg)
+        else:
+            logger.info(
+                "Operator '%s': all %d FBCs have matching related images (%d images)",
+                operator, len(fbcs_in_group), len(reference_images),
+            )
+            all_related_images.update(reference_images)
+
+    # Step 6: Report validation results
+    if validation_errors:
+        full_error = "\n".join(validation_errors)
+        logger.error("Validation failed:\n%s", full_error)
+        raise RuntimeError(full_error)
+
+    unique_images = sorted(all_related_images)
+    logger.info(
+        "Validation passed: %d total unique related images across %d operator(s)",
+        len(unique_images), len(operator_groups),
+    )
+    return unique_images

pyartcd/pyartcd/pipelines/gen_assembly_lp.py

@@ -8,10 +8,14 @@
 from artcommonlib.github_auth import get_github_client_for_org
 from artcommonlib.rpm_utils import parse_nvr
 from artcommonlib.util import merge_objects, new_roundtrip_yaml_handler, split_git_url
-from elliottlib.util import extract_nvrs_from_fbc
 
 from pyartcd import constants
 from pyartcd.cli import cli, click_coroutine, pass_runtime
+from pyartcd.fbc_util import (
+    extract_fbc_labels,
+    extract_ocp_version_from_nvr,
+    validate_fbc_related_images,
+)
 from pyartcd.git import GitRepository
 from pyartcd.runtime import Runtime
 
@@ -108,11 +112,18 @@ def _categorize_nvrs(self, nvrs: List[str]) -> Dict[str, List[str]]:
     async def _extract_nvrs_from_fbc(self) -> Dict[str, str]:
         """
         Extract operand NVRs from FBC pullspecs and return a dict of distgit_key -> NVR.
+
+        When multiple FBC pullspecs are provided (one per OCP target version),
+        validates that all FBCs for the same operator contain identical related images.
         """
-        all_related_nvrs: List[str] = []
-        for fbc_pullspec in self.fbc_pullspecs:
-            related_nvrs = await extract_nvrs_from_fbc(fbc_pullspec, self.product)
-            all_related_nvrs.extend(related_nvrs)
+        if len(self.fbc_pullspecs) > 1:
+            all_related_nvrs = await validate_fbc_related_images(self.fbc_pullspecs, self.product)
+        else:
+            from elliottlib.util import extract_nvrs_from_fbc as _extract
+            all_related_nvrs = []
+            for fbc_pullspec in self.fbc_pullspecs:
+                related_nvrs = await _extract(fbc_pullspec, self.product)
+                all_related_nvrs.extend(related_nvrs)
 
         all_related_nvrs = sorted(set(all_related_nvrs))
         categorized = self._categorize_nvrs(all_related_nvrs)
@@ -203,7 +214,6 @@ async def _resolve_parent_operands(self) -> Dict[str, str]:
             )
 
         releases_config = yaml.load(releases_yaml_path)
-        releases_model = Model(dict_to_model=releases_config)
 
         parent_assembly = releases_config.get('releases', {}).get(self.basis_assembly)
         if parent_assembly is None:
@@ -226,18 +236,46 @@ async def _resolve_parent_operands(self) -> Dict[str, str]:
         )
         return operand_map
 
+    async def _build_fbc_pullspecs_map(self) -> Dict[str, str]:
+        """Map each FBC pullspec to its OCP target version.
+
+        Extracts the OCP version from the NVR embedded in the FBC image labels.
+        Returns a dict of ``{"4.18": "quay.io/...", "4.19": "quay.io/...", ...}``.
+        If only a single pullspec is provided or the OCP version cannot be determined,
+        falls back to a flat list representation.
+        """
+        fbc_map: Dict[str, str] = {}
+        for pullspec in self.fbc_pullspecs:
+            labels = await extract_fbc_labels(pullspec)
+            nvr = labels.get('nvr')
+            ocp_version = extract_ocp_version_from_nvr(nvr) if nvr else None
+            if ocp_version:
+                fbc_map[ocp_version] = pullspec
+            else:
+                self._logger.warning(
+                    "Could not determine OCP target version from FBC %s (nvr=%s); "
+                    "falling back to flat list",
+                    pullspec, nvr,
+                )
+                return {}
+        return fbc_map
+
     def _generate_assembly_definition(
         self,
         operand_map: Dict[str, str],
         parent_operand_map: Optional[Dict[str, str]] = None,
+        fbc_pullspecs_map: Optional[Dict[str, str]] = None,
     ) -> OrderedDict:
         """Generate the assembly YAML definition."""
 
         basis: Dict = {}
         if self.basis_assembly:
             basis['assembly'] = self.basis_assembly
         if self.fbc_pullspecs:
-            basis['fbc_pullspecs'] = list(self.fbc_pullspecs)
+            if fbc_pullspecs_map:
+                basis['fbc_pullspecs'] = dict(fbc_pullspecs_map)
+            else:
+                basis['fbc_pullspecs'] = list(self.fbc_pullspecs)
 
         group_info: Dict = {
             'shipment': {
@@ -375,7 +413,13 @@ async def run(self) -> OrderedDict:
         for key, nvr in sorted(operand_map.items()):
             self._logger.info("  %s = %s", key, nvr)
 
-        assembly_definition = self._generate_assembly_definition(operand_map, parent_operand_map)
+        fbc_pullspecs_map: Optional[Dict[str, str]] = None
+        if self.fbc_pullspecs:
+            fbc_pullspecs_map = await self._build_fbc_pullspecs_map()
+
+        assembly_definition = self._generate_assembly_definition(
+            operand_map, parent_operand_map, fbc_pullspecs_map,
+        )
 
         if self.runtime.dry_run:
             from io import StringIO
@@ -408,7 +452,9 @@ async def run(self) -> OrderedDict:
     metavar="FBC_PULLSPECS",
     default="",
     help="Comma-separated FBC image pullspecs to extract operand NVRs from. "
-         "Required unless --basis-assembly is provided.",
+         "Provide one pullspec per OCP target version for multi-OCP products. "
+         "When multiple are provided, related images are validated for consistency "
+         "across OCP versions. Required unless --basis-assembly is provided.",
 )
 @click.option(
     "--basis-assembly",