Add Python scripts to consume internal APIs (SMASH, SMELT, Bugzilla)

jenkins_pipelines/scripts/bsc_list_generator/bsc_finder.py

@@ -0,0 +1,124 @@
+import argparse
+import json
+import logging
+from typing import Any
+
+from bugzilla_client import BugzillaClient
+
+_FORMATS_DEFAULT_FILE_NAMES: dict[str, str] = {
+    "json": "bsc_list.json",
+    "txt": "bsc_list.txt"
+}
+
+_PRODUCT_VERSIONS: list[str] = ["4.3", "5.0", "5.1"]
+
+# version -> (project, package, filename)
+_IBS_RELEASE_NOTES_FOR_SUMA_VERSION: dict[str, tuple[tuple[str, str, str]]] = {
+    "4.3": (
+        ("Devel:Galaxy:Manager:4.3:ToSLE", "release-notes-susemanager", "release-notes-susemanager.changes"),
+        ("Devel:Galaxy:Manager:4.3:ToSLE", "release-notes-susemanager-proxy", "release-notes-susemanager-proxy.changes")
+    ),
+    "5.0": (
+        ("Devel:Galaxy:Manager:5.0", "release-notes-susemanager", "release-notes-susemanager.changes"),
+        ("Devel:Galaxy:Manager:5.0", "release-notes-susemanager-proxy", "release-notes-susemanager-proxy.changes"),
+    ),
+    "5.1": {
+        ("Devel:Galaxy:Manager:5.1", "release-notes-multi-linux-manager", "release-notes-multi-linux-manager.changes"),
+        ("Devel:Galaxy:Manager:5.1", "release-notes-multi-linux-manager-proxy", "release-notes-multi-linux-manager-proxy.changes"),
+    }
+}
+
+def setup_logging():
+    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+
+def parse_cli_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="This script retrieves a list of relevant open BSCs by querying the SUSE Bugzilla REST API and returns either a JSON output or a check-list of BSC links and summaries"
+    )
+    # required API key
+    parser.add_argument("-k", "--api-key", dest="api_key", help="Bugzilla API key", action='store', required=True)
+    # filters
+    parser.add_argument("-a", "--all", dest="all", default=False, help="Returns results for all supported products (overrides 'version' and 'cloud' flags)", action="store_true")
+    parser.add_argument("-p", "--product-version", dest="product_version", help="Product version of SUMA you want to run this script for, the options are 4.3, 5.0 and 5.1. The default is 4.3 for now",
+        choices=_PRODUCT_VERSIONS, default="4.3", action='store'
+    )
+    parser.add_argument("-n", "--release-notes", dest="use_release", default=False, help="Obtain the bugs list from the release notes for the specified SUMA version", action="store_true")
+    parser.add_argument("-c", "--cloud", dest="cloud", default=False, help="Return BSCs for SUMA/MLM in Public Clouds", action="store_true")
+    parser.add_argument("-s", "--status", dest="status", help="Status to filter BSCs by", action="store",
+        choices=["NEW", "CONFIRMED", "IN_PROGRESS", "RESOLVED"]
+    )
+    parser.add_argument("-r", "--resolution", dest="resolution", help="Resolution to filter issues for (default empty string means open bugs)", action="store")
+    parser.add_argument("-t", "--reporter", dest="reporter", help="Email address of the person who opened the bug to filter issues for", action="store")
+    # output related flags
+    parser.add_argument("-o", "--output", dest="output_file", help="File in which the results will be saved", action="store")
+    parser.add_argument("-f", "--format", dest="output_format", default="txt", help="Output file format (txt default)", action="store",
+        choices=["json", "txt"]
+    )
+
+    args: argparse.Namespace = parser.parse_args()
+    return args
+
+def get_suma_product_name(product_version: str, cloud: bool) -> str:
+    product_name: str = "SUSE Multi-Linux Manager" if product_version == '5.1' else "SUSE Manager"
+    return f"{product_name} {product_version}{' in Public Clouds' if cloud else ''}"
+
+def get_suma_bugzilla_products(all, product_version: str, cloud: bool) -> list[str]:
+    if not all:
+        return [get_suma_product_name(product_version, cloud)]
+
+    bugzilla_products: list[str] = []
+    for version in _PRODUCT_VERSIONS:
+        # get both "standard" and cloud product versions
+        bugzilla_products.append(get_suma_product_name(version, False))
+        bugzilla_products.append(get_suma_product_name(version, True))
+
+    return bugzilla_products
+
+# return a txt file formatted according to .md syntax, so that it can be used in GitHub cards and the likes
+def bugs_to_links_list(products_bugs: dict[str, list[dict]], bugzilla_url: str) -> list[str]:
+    lines: list[str] = []
+
+    for product, bugs_list in products_bugs.items():
+        lines.append(f"## {product}\n\n")
+        for bug in bugs_list:
+            bug_id: str = bug['id']
+            bug_url: str = f"{bugzilla_url}?id={bug_id}"
+            lines.append(f"- [ ] [Bug {bug_id}]({bug_url}) - {bug['priority']} - ({bug['component']}) {bug['summary']}\n")
+        lines.append("\n")
+
+    return lines
+
+def store_results(products_bugs: dict[str, list[dict]], output_file: str, output_format: str, bugzilla_url: str = ""):
+    logging.info(f"Storing results at {output_file} ({output_format} file)")
+
+    with open(output_file, 'w', encoding='utf-8') as f:
+        if output_format == "json":
+            json.dump(products_bugs, f, indent=2, sort_keys=True)
+        elif output_format == "txt":
+            issues_links: list[str] = bugs_to_links_list(products_bugs, bugzilla_url)
+            f.writelines(issues_links)
+        else:
+            raise ValueError(f"Invalid output format: {output_format} - supported formats {_FORMATS_DEFAULT_FILE_NAMES.keys()}")
+
+def main():
+    setup_logging()
+    args: argparse.Namespace = parse_cli_args()
+    bugzilla_client: BugzillaClient = BugzillaClient(args.api_key)
+
+    product_bugs: dict[str, list[dict[str, Any]]] = {}
+    if args.use_release:
+        product_versions: list[str] = _PRODUCT_VERSIONS if args.all else [args.product_version]
+        for version in product_versions:
+            release_notes_paths: tuple[tuple[str, str, str]] = _IBS_RELEASE_NOTES_FOR_SUMA_VERSION[version]
+            product_bugs[f"SUSE Manager/MLM {version}"] = bugzilla_client.bscs_from_release_notes(release_notes_paths, reporter = args.reporter, status = args.status, resolution = args.resolution)
+    else:
+        bugzilla_products: list[str] = get_suma_bugzilla_products(args.all, args.product_version, args.cloud)
+        product_bugs = bugzilla_client.find_suma_bscs(bugzilla_products, reporter = args.reporter, status = args.status, resolution = args.resolution)
+
+    output_format: str = args.output_format
+    output_file: str = args.output_file if args.output_file else _FORMATS_DEFAULT_FILE_NAMES[output_format]
+
+    store_results(product_bugs, output_file, output_format, bugzilla_client.show_bug_url)
+
+if __name__ == '__main__':
+    main()

jenkins_pipelines/scripts/bsc_list_generator/bsc_finder_README.md

@@ -0,0 +1,141 @@
+# bsc_finder.py
+
+## Table of Contents
+
+- [Overview](#overview)
+- [Features](#features)
+- [Requirements](#requirements)
+- [Installation](#installation)
+- [Usage](#usage)
+- [Output](#output)
+- [Logging](#logging)
+- [Error Handling](#error-handling)
+- [Dependencies](#dependencies)
+- [License](#license)
+- [Notes](#notes)
+
+---
+
+## Overview
+
+This Python script automates the process of gathering, processing and storing bug reports
+consuming Bugzilla REST API.
+It requires a valid API key to function.
+
+The script allows users to filter bug reports for SUMA 4.3. SUMA 5.0 and MLM 5.1 products.
+
+## Features
+
+- Fetches and stores bug reports from Bugzilla REST API through a CLI
+- Can retrieve the bugs mentioned in SUMA 4.3, SUMA 5.0 and MLM 5.1 release notes
+
+## Requirements
+
+- A valid Bugzilla API key
+- osc CLI if you intend to retrieve bug reports from release notes
+- Python 3.6 or higher
+- `requests` library
+- `bugzilla_client` library: Ensure you have the `bugzilla_client` module available in
+  your environment.
+
+## Installation
+
+To install the required dependencies, ensure you have `requests` installed:
+
+```bash
+pip install requests
+```
+
+## Usage
+
+Command-Line Arguments
+
+The script accepts several command-line arguments to control its behavior.
+
+```bash
+python bsc_finder.py [options]
+```
+
+Options:
+
+1) Required Argument:
+    - API Key (-k or --api-key):
+        Description: Bugzilla API key (required).
+        Usage: -k YOUR_API_KEY or --api-key YOUR_API_KEY
+
+2) Filter Options:
+    - All Products (-a or --all):
+        Description: Returns results for all supported products, overriding version and cloud flags.
+        Usage: -a or --all (flag)
+    - Product Version (-p or --product-version):
+        Description: Specify the product version of SUMA/MLM to run the script for. Options: 4.3, 5.0 or 5.1. Default is 4.3.
+        Usage: -p 5.0 or --product-version 5.0
+    - Release notes (-n or --release-notes):
+        Description: Retrieves the bug reports mentioned in the latest release notes for the selected SUMA/MLM version(s)
+        Usage: -n or --release-notes
+    - Cloud (-c or --cloud):
+        Description: Returns BSCs for SUMA/MLM in Public Clouds.
+        Usage: -c or --cloud (flag)
+    - Status (-s or --status):
+        Description: Filters BSCs by status. Options: NEW, CONFIRMED, IN_PROGRESS, RESOLVED.
+        Usage: -s NEW or --status NEW
+    - Resolution (-r or --resolution):
+        Description: Filters issues by resolution. Leave empty for open bugs.
+        Usage: -r FIXED or --resolution FIXED
+
+3) Output Options:
+    - Output File (-o or --output):
+        Description: Specifies the file in which the results will be saved.
+        Usage: -o results.txt or --output results.txt
+    - Output Format (-f or --format):
+        Description: Format for the output file. Options: json, txt. Default is txt.
+        Usage: -f json or --format json
+
+Example:
+
+```bash
+python bsc_finder.py -k YOUR_API_KEY -p 5.0 -s NEW -c -o results.txt -f txt
+```
+
+This command will:
+
+1) Instantiate a new Bugzilla REST API client using your API key
+2) Query the API for all the bug reports related to 'SUSE Manager 5.0 in Public Clouds', in status NEW
+3) Save the results to a file called 'results.txt', as a .md formatted list of link-summary elements
+
+## Output
+
+The produced output can be one of:
+1) a JSON file, containing all the bug reports info
+2) a txt file formatted in .md syntax, containing links and a summary for each report.
+
+## Logging
+
+The script includes basic logging for informational messages. To enable logging,
+ensure the setup_logging function is called at the beginning of the script. Log
+messages will display timestamped INFO-level messages.
+
+## Error Handling
+
+- If no, or an invalid, API key is provided via CLI the script will print an error
+message and halt execution.
+- Invalid flags values in appropriate error messages.
+
+## Dependencies
+
+`requests`: A popular Python library for making HTTP requests. It is used here to
+handle communication with the SMASH API.
+
+## License
+
+This script is licensed under the [MIT License](https://opensource.org/licenses/MIT).
+
+## Notes
+
+Ensure that the requests library is installed in your environment.
+This script relies on the Bugzilla REST API being available and responsive and having a valid API key for it.
+
+Handle possible exceptions appropriately in production environments.
+Caching helps in reducing the load on the API and speeds up access to the
+embargoed IDs, but make sure to handle cache invalidation if the data can change
+frequently.

jenkins_pipelines/scripts/bsc_list_generator/bugzilla_client.py

@@ -0,0 +1,114 @@
+import logging
+from os import remove
+import subprocess
+import requests
+from typing import Any
+
+_SUSE_BUGZILLA_BASE_URL = "https://bugzilla.suse.com"
+_IBS_API_URL: str = "https://api.suse.de"
+
+class BugzillaClient:
+
+    def __init__(self, api_key: str, base_url: str = _SUSE_BUGZILLA_BASE_URL, api_type: str = "rest"):
+        # api_key is needed for actual API calls so we may as well fail here
+        if not api_key:
+            raise ValueError("api_key is None or empty")
+        # private
+        self._api_key: str = api_key
+        self._base_url: str = base_url
+        self._api_url: str = f"{base_url}/{api_type}"
+        self._bugs_endpoint = f"{base_url}/{api_type}/bug"
+        self._params: dict[str, Any] = { 'Bugzilla_api_key': self._api_key }
+        # public 
+        self.show_bug_url: str = f"{base_url}/show_bug.cgi"
+
+    def find_suma_bscs(self, bugzilla_products: list[str], **kwargs) -> dict[str, list[dict[str, Any]]]:
+        product_bugs: dict[str, list[dict[str, Any]]] = {}
+
+        for bugzilla_product in bugzilla_products:
+            logging.info(f"Retrieving BSCs for product '{bugzilla_product}'...")
+            product_bugs[bugzilla_product] = self._get_bugs(product = bugzilla_product, **kwargs)
+            logging.info("Done")
+
+        return product_bugs
+
+    def bscs_from_release_notes(self, release_note_paths: tuple[tuple[str, str, str]], **kwargs) -> list[dict[str, Any]]:
+        bsc_ids: list[str] = []
+
+        for rn_path in release_note_paths:
+            logging.info(f"Parsing release notes: {rn_path[0]} - {rn_path[1]}")
+            rn_ids: list[str] = self._get_mentioned_bscs(*rn_path)
+            # avoid duplicating a BSC between Proxy and Server
+            for bug_id in rn_ids:
+                if bug_id not in bsc_ids:
+                    bsc_ids.append(bug_id)
+
+        return self._get_bugs(id=','.join(bsc_ids), **kwargs)
+
+    def _bug_under_embargo(self, bsc: dict[str, Any]) -> bool:
+        summary: str = bsc['summary']
+        if "embargo" in summary.lower():
+            logging.info(f"BSC#{bsc['id']} is under embargo and will not be displayed in the results.")
+            return True
+
+        return False
+
+    def _get_mentioned_bscs(self, project: str, package:str, filename: str) -> list[str]:
+        # check=True -> raise subprocess.CalledProcessError if the return code is != 0
+        subprocess.run(
+            ["osc", "--apiurl", _IBS_API_URL, "co", project, package, filename],
+            check=True,
+            stdout=subprocess.PIPE, stderr=subprocess.PIPE
+        )
+        bugs_ids: list[str] = self._parse_release_notes(filename)
+        # cleanup
+        remove(filename)
+
+        return bugs_ids
+
+    def _parse_release_notes(self, notes_filename: str) -> list[str]:
+        bsc_ids: list[str] = []
+        # retrieve BSC IDs only for the latest release notes block
+        with open(notes_filename) as nf:
+            # first line should delimit a block, better bail out if not
+            firstline: str = nf.readline().strip()
+            if not len(firstline) or not all(char == '-' for char in firstline):
+                raise ValueError("Irregular or missing release notes block: first line should only be composed by '-'")
+
+            bsc_block: bool = False
+            while(True):
+                cur_line: str = nf.readline().strip()
+
+                if cur_line.startswith("bsc#"):
+                    bsc_block = True
+                    bsc_entries: str = cur_line.split(", ")
+                    bsc_ids.extend([entry.replace("bsc#", "") for entry in bsc_entries])
+                    continue
+
+                # this is True only only if we have ended parsing a previous bsc block
+                if bsc_block:
+                    break
+
+        return bsc_ids
+
+    def _get_bugs(self, **kwargs) -> list[dict[str, Any]]:
+        # drops CLI args that have not been used and have no default
+        additional_params: dict[str, Any] = { k: v for k, v in kwargs.items() if v is not None }
+        response: requests.Response = requests.get(self._bugs_endpoint, params={**self._params, **additional_params})
+        if not response.ok:
+            response.raise_for_status()
+
+        json_res: dict = response.json()
+        bugs: list[dict[str, Any]] = json_res['bugs']
+        filtered_bugs: list[dict[str, Any]] = [ bug for bug in bugs if not self._bug_under_embargo(bug) ]
+
+        return filtered_bugs
+
+    def _get_bug_comments(self, bug_id: str) -> list[dict[str, Any]]:
+        response: requests.Response = requests.get(f"{self._bugs_endpoint}/{bug_id}/comment", params={**self._params})
+        if not response.ok:
+            response.raise_for_status()
+
+        json_res: dict = response.json()
+        comments: list[dict[str, Any]] = json_res['bugs'][str(bug_id)]['comments']
+        return comments

jenkins_pipelines/scripts/json_generator/README.md

@@ -6,7 +6,7 @@ README files for each script:
 ## Scripts
 
 - [maintenance_json_generator script README](./maintenance_json_generator_README.md) - Automates gathering and
-processing of open QAM SLE requests that affect SUMA 4.3 and 5.0 and generates JSON output for BV testsuite
+processing of open QAM SLE requests that affect SUMA 4.3, 5.0 and 5.1 generates JSON output for BV testsuite
 - [ibs_osc_client_README](./ibs_osc_client_README.md) - checks embargo status and processes repository information
 - [smash_client_README](./smash_client_README.md) - interacts with SUSE Manager's SMASH API to retrieve and manage embargoed bug IDs and CVEs.