diff --git a/README.md b/README.md index 9246681..149c1ae 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,12 @@ Set `TRUSTIFY_URL` to point to your Trustify instance: export TRUSTIFY_URL="https://trustify.example.com" ``` +When the URL has no API path, `/api/v2/` is appended. To use another API version, include the full path: + +```bash +export TRUSTIFY_URL="https://atlas.release.devshift.net/api/v3/" +``` + ### Authentication (Optional) If your Trustify instance requires authentication, also set `AUTH_ENDPOINT`: diff --git a/src/trustshell/__init__.py b/src/trustshell/__init__.py index d82cbaf..2169485 100644 --- a/src/trustshell/__init__.py +++ b/src/trustshell/__init__.py @@ -1,22 +1,21 @@ -import time import importlib.metadata import logging import os import sys -from urllib.parse import urlparse, urlunparse, quote, parse_qs, urlencode -from typing import Optional, Any +import time +import webbrowser +from http.server import BaseHTTPRequestHandler, HTTPServer +from typing import Any +from urllib.parse import parse_qs, quote, urlencode, urlparse, urlunparse import httpx import jwt +from anytree import Node, RenderTree from packageurl import PackageURL from rich.console import Console from rich.logging import RichHandler from rich.theme import Theme -import webbrowser from univers.versions import RpmVersion -from anytree import Node, RenderTree - -from http.server import BaseHTTPRequestHandler, HTTPServer from trustshell.oidc.oidc_pkce_authcode import ( LOCAL_SERVER_PORT, @@ -42,13 +41,12 @@ if "TRUSTIFY_URL" in os.environ: url_env = os.getenv("TRUSTIFY_URL", "") parsed_url = urlparse(url_env) - if not parsed_url.path or parsed_url.path != TRUSTIFY_URL_PATH: + if not parsed_url.path or parsed_url.path == "/": TRUSTIFY_URL = urlunparse( (parsed_url.scheme, parsed_url.netloc, TRUSTIFY_URL_PATH, "", "", "") ) else: TRUSTIFY_URL = url_env - # Only enable authentication if AUTH_ENDPOINT is also set AUTH_ENABLED = bool(os.getenv("AUTH_ENDPOINT")) else: TRUSTIFY_URL = "http://localhost:8080/api/v2/" @@ -105,9 +103,7 @@ def get_tag_from_purl(purl: PackageURL) -> str: return tag -def build_node_purl( - purls: list[str], show_versions: bool = False -) -> Optional[PackageURL]: +def build_node_purl(purls: list[str], show_versions: bool = False) -> PackageURL | None: """ Generate a base purl with a version or tag qualifier from a list of purls with homogenous type/namespace, and name @@ -220,7 +216,7 @@ def check_or_get_access_token() -> str: console.print( "Unable to authenticate to Atlas, please try again after authenticating in the browser." ) - exit(0) + sys.exit(0) return access_token @@ -369,30 +365,46 @@ def make_request_with_retry( first_response = make_request_with_retry(client, query_params, auth_header) first_result = first_response.json() - total_available = first_result.get("total", 0) - if total_available == 0: + all_items = first_result.get("items", []) + total_available = first_result.get("total") + total_known = total_available is not None + + if not all_items and (not total_known or total_available == 0): if component_name: console.print(f"No items found for {component_name}") return {"items": [], "total": 0} - all_items = first_result.get("items", []) - total_pages = (total_available + limit - 1) // limit - - if logger.isEnabledFor(logging.DEBUG): - logger.debug( - f"Paginated request: {total_available} total items, " - f"{total_pages} page(s), page 1/{total_pages} complete" - ) + if total_known: + total_pages = (total_available + limit - 1) // limit + if logger.isEnabledFor(logging.DEBUG): + logger.debug( + f"Paginated request: {total_available} total items, " + f"{total_pages} page(s), page 1/{total_pages} complete" + ) + else: + if logger.isEnabledFor(logging.DEBUG): + logger.debug( + f"Paginated request: total unknown, " + f"page 1 returned {len(all_items)} items" + ) - # Fetch remaining pages sequentially offset = limit page_num = 2 - while offset < total_available: + while True: + if total_known and offset >= total_available: + break + if not total_known and len(all_items) - (offset - limit) < limit: + break + page_params = {**base_params, "limit": limit, "offset": offset} if logger.isEnabledFor(logging.DEBUG): - logger.debug( - f"Fetching page {page_num}/{total_pages} (offset {offset})..." - ) + if total_known: + total_pages = (total_available + limit - 1) // limit + logger.debug( + f"Fetching page {page_num}/{total_pages} (offset {offset})..." + ) + else: + logger.debug(f"Fetching page {page_num} (offset {offset})...") try: response = make_request_with_retry(client, page_params, auth_header) result = response.json() @@ -400,21 +412,23 @@ def make_request_with_retry( all_items.extend(page_items) if logger.isEnabledFor(logging.DEBUG): logger.debug( - f"Page {page_num}/{total_pages} complete " - f"({len(all_items)}/{total_available} items)" + f"Page {page_num} complete ({len(all_items)} items so far)" ) + if not page_items or len(page_items) < limit: + break offset += limit page_num += 1 - except Exception as e: + except httpx.HTTPError as e: logger.error(f"Error fetching page at offset {offset}: {e}") break + total_count = total_available if total_known else len(all_items) if component_name: console.print( - f"Retrieved {len(all_items)} items out of {total_available} total for {component_name}" + f"Retrieved {len(all_items)} items out of {total_count} total for {component_name}" ) - return {"items": all_items, "total": total_available} + return {"items": all_items, "total": total_count} def render_tree_to_string(root: Node) -> str: diff --git a/src/trustshell/api.py b/src/trustshell/api.py index 2abd589..64b3a2a 100644 --- a/src/trustshell/api.py +++ b/src/trustshell/api.py @@ -1,9 +1,9 @@ -import click -import httpx import json import logging from urllib.parse import quote +import click +import httpx from rich.console import Console from rich.theme import Theme @@ -78,9 +78,7 @@ def api(endpoint: str, subpath: str, params: tuple[str], debug: bool) -> None: f"HTTP error {exc.response.status_code}: {exc.response.text}", style="error" ) except httpx.RequestError as exc: - console.print(f"Request error: {str(exc)}", style="error") + console.print(f"Request error: {exc!s}", style="error") except json.JSONDecodeError as exc: console.print("Response is not valid JSON:", style="warning") - console.print(f"JSON decode error: {str(exc)}") - except Exception as exc: - console.print(f"Unexpected error: {str(exc)}", style="error") + console.print(f"JSON decode error: {exc!s}") diff --git a/src/trustshell/models.py b/src/trustshell/models.py index 9853a86..56a21eb 100644 --- a/src/trustshell/models.py +++ b/src/trustshell/models.py @@ -1,7 +1,6 @@ """Data models for trust-products search results.""" from dataclasses import dataclass, field -from typing import Optional @dataclass(frozen=True) @@ -18,7 +17,7 @@ class ProductResultRow: cpe: str ps_update_stream: str - ps_module: Optional[str] + ps_module: str | None matched_component: str # PURL that matched (important for wildcard search) shipped_component: ( str # PURL for affects: image-index/arch-specific OCI, or SRPM/binary RPM diff --git a/src/trustshell/oidc/__init__.py b/src/trustshell/oidc/__init__.py index 3ad0d25..b640f59 100644 --- a/src/trustshell/oidc/__init__.py +++ b/src/trustshell/oidc/__init__.py @@ -4,23 +4,23 @@ """ from .oidc_pkce_authcode import ( - gen_things, + AUTH_ENDPOINT, + LOCAL_SERVER_PORT, + REDIRECT_URI, build_url, code_to_token, - get_fresh_token, + gen_things, get_client_credentials_token, - LOCAL_SERVER_PORT, - REDIRECT_URI, - AUTH_ENDPOINT, + get_fresh_token, ) __all__ = [ - "gen_things", + "AUTH_ENDPOINT", + "LOCAL_SERVER_PORT", + "REDIRECT_URI", "build_url", "code_to_token", - "get_fresh_token", + "gen_things", "get_client_credentials_token", - "LOCAL_SERVER_PORT", - "REDIRECT_URI", - "AUTH_ENDPOINT", + "get_fresh_token", ] diff --git a/src/trustshell/oidc/oidc_pkce_authcode.py b/src/trustshell/oidc/oidc_pkce_authcode.py index 8c4674c..b18788f 100644 --- a/src/trustshell/oidc/oidc_pkce_authcode.py +++ b/src/trustshell/oidc/oidc_pkce_authcode.py @@ -1,15 +1,12 @@ -#!/usr/bin/env python - import json import logging import os import secrets - import urllib.parse +import httpx import jwt import pkce -import httpx logger = logging.getLogger("trustshell") # Client ID - Keycloak uses "atlas-frontend"; Cognito TPA uses a dedicated CLI client @@ -61,7 +58,7 @@ def get_client_credentials_token() -> str | None: def gen_things() -> tuple[str, str, str]: - logging.debug("Generating verifier, challenge, state") + logger.debug("Generating verifier, challenge, state") code_verifier, code_challenge = pkce.generate_pkce_pair() state = secrets.token_urlsafe(16) logger.debug(f"Code Verifier: {code_verifier}") @@ -155,6 +152,5 @@ def get_fresh_token(refresh_token: str) -> tuple[str, str]: logger.debug("Each raw part of the response body:") for k, v in r2a_json.items(): logger.debug(f"{k}:{v}") - else: - logger.debug(f"Access Token: {access_token}") + logger.debug(f"Access Token: {access_token}") return access_token, refresh_token diff --git a/src/trustshell/oidc/oidc_pkce_server.py b/src/trustshell/oidc/oidc_pkce_server.py index 0b26216..aaf253f 100755 --- a/src/trustshell/oidc/oidc_pkce_server.py +++ b/src/trustshell/oidc/oidc_pkce_server.py @@ -2,8 +2,8 @@ import json import os -from http.server import SimpleHTTPRequestHandler import socketserver +from http.server import SimpleHTTPRequestHandler from typing import Any from urllib.parse import ( parse_qs, @@ -13,18 +13,18 @@ try: # When run as part of the trustshell package from .oidc_pkce_authcode import ( + AUTH_ENDPOINT, code_to_token, gen_things, get_fresh_token, - AUTH_ENDPOINT, ) except ImportError: # When run as a standalone script from oidc_pkce_authcode import ( # type: ignore[import-not-found,no-redef] + AUTH_ENDPOINT, code_to_token, gen_things, get_fresh_token, - AUTH_ENDPOINT, ) PORT: int = int(os.getenv("LISTEN_PORT", "8080")) diff --git a/src/trustshell/osidb.py b/src/trustshell/osidb.py index 5ea9950..19a0e55 100644 --- a/src/trustshell/osidb.py +++ b/src/trustshell/osidb.py @@ -1,18 +1,18 @@ -from collections import defaultdict import logging import os import subprocess import sys import tempfile -from typing import Any, Union +from collections import defaultdict +from typing import Any import click - -from trustshell.models import Affect -from requests import HTTPError -from trustshell import console import osidb_bindings from osidb_bindings.bindings.python_client.models import Flaw +from requests import HTTPError + +from trustshell import console +from trustshell.models import Affect logger = logging.getLogger(__name__) @@ -21,9 +21,7 @@ class OSIDB: def __init__(self) -> None: endpoint = os.getenv("OSIDB_ENDPOINT") if endpoint is None: - raise EnvironmentError( - "The environment variable 'OSIDB_ENDPOINT' is not set." - ) + raise OSError("The environment variable 'OSIDB_ENDPOINT' is not set.") self.session = osidb_bindings.new_session(osidb_server_uri=endpoint) # type: ignore[attr-defined] @staticmethod @@ -52,7 +50,7 @@ def parse_stream_purl_tuples(tuples_list: list[str]) -> set[tuple[str, str]]: @staticmethod def edit_tuples_in_editor( - current_tuples: Union[list[tuple[str, str]], set[tuple[str, str]]], + current_tuples: list[tuple[str, str]] | set[tuple[str, str]], ) -> list[tuple[str, str]]: """ Opens the default text editor for the user to modify the ps_update_stream/purl tuples. @@ -78,13 +76,13 @@ def edit_tuples_in_editor( f"Error: Editor '{editor}' not found. Please set your EDITOR environment variable.", style="error", ) - exit(1) + sys.exit(1) except subprocess.CalledProcessError: console.print( "Editor exited with an error. Changes might not be saved.", style="error", ) - exit(1) + sys.exit(1) with open(temp_filepath, "r") as file: modified_content = file.read() @@ -121,7 +119,7 @@ def add_affects(self, flaw: Flaw, affects_to_add: list[Affect]) -> None: except HTTPError as e: msg = e.response.text console.print(f"Failed to update flaw: {e}: {msg}") - exit(1) + sys.exit(1) console.print(f"Added {len(bulk_create_response.results)} new affects") def edit_flaw_affects( @@ -139,7 +137,7 @@ def edit_flaw_affects( try: flaw = self.session.flaws.retrieve(id=flaw_id) - except Exception as e: + except (OSError, RuntimeError, ValueError, TypeError, KeyError) as e: console.print(f"Could not retrieve flaw {flaw_id}: {e}") return @@ -251,7 +249,7 @@ def edit_flaw_affects( f"Failed to delete flaw affect {existing_key}: {e}: {msg}", style="error", ) - exit(1) + sys.exit(1) # Add any new affects not already on the flaw in NEW state ps_stream_purls_as_affects = [ diff --git a/src/trustshell/product_definitions.py b/src/trustshell/product_definitions.py index 5b6953b..7a2037d 100644 --- a/src/trustshell/product_definitions.py +++ b/src/trustshell/product_definitions.py @@ -1,28 +1,29 @@ -from collections import defaultdict import copy -from datetime import datetime import json import logging import os import re -from typing import Any, Optional +from collections import defaultdict +from datetime import UTC, date, datetime +from typing import Any + import httpx +from anytree import LevelOrderGroupIter, NodeMixin -from anytree import NodeMixin, LevelOrderGroupIter from trustshell import CONFIG_DIR, console from trustshell.rhel_releases import EnhancedProdDefs logger = logging.getLogger(__name__) -class ProductBase(object): +class ProductBase: def __init__(self, name: str) -> None: self.name = name def __hash__(self) -> int: return hash(self.name) - def __eq__(self, other: Any) -> bool: + def __eq__(self, other: object) -> bool: if isinstance(other, ProductBase) and type(self) is type(other): return self.name == other.name return False @@ -46,7 +47,11 @@ def match(self, cpe: str) -> bool: class ProductStream(ProductBase, NodeMixin): - def __init__(self, name: str, cpes: list[str] = [], active: bool = False) -> None: + def __init__( + self, name: str, cpes: list[str] | None = None, active: bool = False + ) -> None: + if cpes is None: + cpes = [] super().__init__(name) # In rhel 10 we don't use mainline CPEs, so we need to filter them out if name.startswith("rhel-"): @@ -69,7 +74,7 @@ class ProdDefs: PRODUCT_FILE = os.path.join(CONFIG_DIR, "products.json") @classmethod - def get_etag(cls, url: str) -> Optional[str]: + def get_etag(cls, url: str) -> str | None: response = httpx.head(url) etag = response.headers.get("etag") return str(etag) if etag is not None else None @@ -82,7 +87,7 @@ def persist_etag(cls, etag: str, file_path: str) -> None: # Assisted by watsonx Code Assistant @classmethod - def load_etag(cls, file_path: str) -> Optional[str]: + def load_etag(cls, file_path: str) -> str | None: if os.path.exists(file_path): with open(file_path, "r") as f: return f.read().strip() @@ -97,7 +102,7 @@ def load_product_definitions(cls, url: str, file_path: str) -> None: @classmethod def get_product_definitions_service(cls) -> dict[str, Any]: - proddefs_url: Optional[str] = None + proddefs_url: str | None = None if "PRODDEFS_URL" not in os.environ: console.print( "PRODDEFS_URL not set, not product mappings will be available", @@ -137,14 +142,14 @@ def __init__( self.product_trees: list[NodeMixin] = [] # Initialize enhanced RHEL release data - self.enhanced_proddefs: Optional[EnhancedProdDefs] = None + self.enhanced_proddefs: EnhancedProdDefs | None = None if rhel_releases_path: # Use local file for testing try: self.enhanced_proddefs = EnhancedProdDefs( git_branch=rhel_git_branch, rhel_releases_path=rhel_releases_path ) - except Exception as e: + except (OSError, ValueError, TypeError, httpx.HTTPError) as e: logger.warning( f"Could not initialize enhanced product definitions: {e}" ) @@ -178,10 +183,8 @@ def __init__( supported_from = lifecycle.get("supported_from") if supported_from: try: - supported_from_date = datetime.strptime( - supported_from, "%Y-%m-%d" - ).date() - current_date = datetime.now().date() + supported_from_date = date.fromisoformat(supported_from) + current_date = datetime.now(tz=UTC).date() # Module is only active if supported_from date is today or in the past module_is_active = supported_from_date <= current_date except ValueError as e: @@ -233,7 +236,7 @@ def match_module_pattern(self, cpe: str) -> list[ProductModule]: def _load_rhel_release_data( self, git_branch: str = "main", rhel_releases_path: str = "" - ) -> Optional[EnhancedProdDefs]: + ) -> EnhancedProdDefs | None: """ Load RHEL release data from GitLab repository or local file. @@ -257,7 +260,7 @@ def _load_rhel_release_data( f"Loaded RHEL release data from GitLab (branch: {git_branch})" ) return enhanced_proddefs - except Exception as e: + except (OSError, ValueError, TypeError, httpx.HTTPError) as e: logger.error(f"Could not load RHEL release data: {e}") return None @@ -270,7 +273,7 @@ def _clean_cpe(cpe: str) -> str: # Remove trailing ':' characters return cleaned_cpe.rstrip(":") - def get_product_mappings_for_cpe(self, cpe: str) -> list[tuple[str, Optional[str]]]: + def get_product_mappings_for_cpe(self, cpe: str) -> list[tuple[str, str | None]]: """Return (ps_update_stream, ps_module) for each product matching the CPE. Tries ps_update_stream direct CPE match first, then falls back to ps_module @@ -284,7 +287,7 @@ def get_product_mappings_for_cpe(self, cpe: str) -> list[tuple[str, Optional[str if re.search(r":redhat:enterprise_linux:\d:", cleaned_cpe): return [] - mappings: list[tuple[str, Optional[str]]] = [] + mappings: list[tuple[str, str | None]] = [] # Try stream matches first (direct CPE match or enhanced) enhanced_streams = self._check_enhanced_streams(cleaned_cpe) diff --git a/src/trustshell/products.py b/src/trustshell/products.py index ca35adf..c384336 100644 --- a/src/trustshell/products.py +++ b/src/trustshell/products.py @@ -1,24 +1,25 @@ -from collections import defaultdict -import click -import httpx import logging import sys +from collections import defaultdict from typing import Any, Optional +import click +import httpx from anytree import NodeMixin, PreOrderIter from anytree.walker import Walker, WalkError from packageurl import PackageURL from rich.console import Console from rich.theme import Theme + from trustshell import ( AUTH_ENABLED, TRUSTIFY_URL, build_node_purl, check_or_get_access_token, config_logging, + paginated_trustify_query, print_version, purl_sans_version, - paginated_trustify_query, ) from trustshell.models import Affect, ProductResultRow, ProductSearchResult from trustshell.osidb import OSIDB @@ -40,7 +41,7 @@ def __init__( self, name: str, parent: Optional["ComponentNode"] = None, - sbom_id: Optional[str] = None, + sbom_id: str | None = None, ) -> None: self.name = name self.parent = parent @@ -95,7 +96,7 @@ def prime_cache(check: bool, debug: bool) -> None: except httpx.HTTPStatusError as e: console.print(f"HTTP error occurred: {e}", style="error") sys.exit(1) - except Exception as e: + except httpx.RequestError as e: console.print(f"An error occurred: {e}", style="error") sys.exit(1) @@ -162,7 +163,7 @@ def prime_cache(check: bool, debug: bool) -> None: ) def search( purl: str, - flaw: Optional[str], + flaw: str | None, replace: bool, output: str, show_module: bool, @@ -599,9 +600,12 @@ def _trees_with_cpes( trees_with_cpes: list[ComponentNode] = [] for tree in first_children: # Remove this once https://issues.redhat.com/browse/TC-2659 is implemented - if tree.name.startswith("pkg:rpm/") and not include_rpm_containers: - if container_in_tree(tree): - continue + if ( + tree.name.startswith("pkg:rpm/") + and not include_rpm_containers + and container_in_tree(tree) + ): + continue _remove_non_cpe_branches(tree) if not _has_cpe_node(tree): for leaf in tree.leaves: diff --git a/src/trustshell/purl.py b/src/trustshell/purl.py index a805cf6..a071d07 100644 --- a/src/trustshell/purl.py +++ b/src/trustshell/purl.py @@ -1,29 +1,28 @@ -import click import logging from collections import defaultdict from typing import Any +import click import httpx from anytree import Node +from packageurl import PackageURL from rich.console import Console from rich.theme import Theme -from packageurl import PackageURL from trustshell import ( AUTH_ENABLED, TRUSTIFY_URL, build_node_purl, check_or_get_access_token, - print_version, config_logging, get_tag_from_purl, - render_tree, paginated_trustify_query, + print_version, + render_tree, urlencoded, ) from trustshell.products import ANALYSIS_ENDPOINT, LATEST_ENDPOINT - custom_theme = Theme({"warning": "magenta", "error": "bold red", "info": "cyan"}) console = Console(color_system="auto", theme=custom_theme) logger = logging.getLogger("trustshell") @@ -258,7 +257,7 @@ def _query_trustify_packages_base_purl( try: purl_obj = PackageURL.from_string(base_purl_str) results.append(purl_obj) - except Exception as e: + except ValueError as e: logger.debug(f"Failed to parse PURL '{base_purl_str}': {e}") else: # Enhanced behavior: lookup version details for each base PURL @@ -281,7 +280,7 @@ def _query_trustify_packages_base_purl( version_purl_str ) results.append(version_purl_obj) - except Exception as e: + except ValueError as e: logger.debug( f"Failed to parse version PURL '{version_purl_str}': {e}" ) @@ -296,12 +295,12 @@ def _query_trustify_packages_base_purl( individual_purl_str ) results.append(individual_purl_obj) - except Exception as e: + except ValueError as e: logger.debug( f"Failed to parse individual PURL '{individual_purl_str}': {e}" ) - except Exception as e: + except httpx.HTTPError as e: logger.debug( f"Failed to lookup base PURL '{base_purl_str}': {e}" ) @@ -309,7 +308,7 @@ def _query_trustify_packages_base_purl( try: purl_obj = PackageURL.from_string(base_purl_str) results.append(purl_obj) - except Exception as parse_e: + except ValueError as parse_e: logger.debug( f"Failed to parse base PURL '{base_purl_str}': {parse_e}" ) @@ -322,7 +321,7 @@ def _query_trustify_packages_base_purl( except httpx.HTTPStatusError as e: console.print(f"HTTP error querying base purls: {e}", style="error") return [] - except Exception as e: + except httpx.RequestError as e: console.print(f"Error querying base purls: {e}", style="error") return [] diff --git a/src/trustshell/rhel_releases.py b/src/trustshell/rhel_releases.py index ab9b771..a58e6dc 100644 --- a/src/trustshell/rhel_releases.py +++ b/src/trustshell/rhel_releases.py @@ -6,12 +6,14 @@ on release hierarchy and relationships. """ -from collections import defaultdict, deque import fnmatch +import json import logging import os import re -from typing import Any, Dict, List, Set, Optional +from collections import defaultdict, deque +from typing import Any + import httpx import yaml @@ -27,8 +29,8 @@ def __init__( self, name: str, node_type: str, - cpes: List[str], - ps_update_stream: Optional[str] = None, + cpes: list[str], + ps_update_stream: str | None = None, ): self.name = name self.node_type = node_type # main, eus, aus, e4s @@ -39,8 +41,8 @@ def __init__( self.ps_update_stream = ( ps_update_stream # The ps_update_stream associated with this node ) - self.children: Set[str] = set() - self.parents: Set[str] = set() + self.children: set[str] = set() + self.parents: set[str] = set() def __repr__(self) -> str: return f"RHELReleaseNode({self.name}, {self.node_type}, {len(self.cpes)} CPEs)" @@ -92,8 +94,8 @@ def __init__( self.git_branch = git_branch self.yaml_file_path = yaml_file_path # Only for testing - self.nodes: Dict[str, RHELReleaseNode] = {} - self.cpe_to_nodes: Dict[str, List[RHELReleaseNode]] = defaultdict(list) + self.nodes: dict[str, RHELReleaseNode] = {} + self.cpe_to_nodes: dict[str, list[RHELReleaseNode]] = defaultdict(list) # Ensure cache directory exists os.makedirs(self.CACHE_DIR, exist_ok=True) @@ -115,10 +117,10 @@ def _load_release_data(self) -> None: self._parse_yaml_data(data) - except Exception as e: + except (OSError, yaml.YAMLError, httpx.HTTPError, ValueError, TypeError) as e: logger.error(f"Error loading RHEL release data: {e}") - def _load_from_local_file(self) -> Optional[Dict[str, Any]]: + def _load_from_local_file(self) -> dict[str, Any] | None: """Load YAML data from local file (for testing).""" if not os.path.exists(self.yaml_file_path): logger.warning(f"RHEL release data file not found: {self.yaml_file_path}") @@ -138,7 +140,7 @@ def _load_from_local_file(self) -> Optional[Dict[str, Any]]: logger.error(f"Failed to parse YAML file {self.yaml_file_path}: {e}") return None - def _fetch_from_gitlab(self) -> Optional[Dict[str, Any]]: + def _fetch_from_gitlab(self) -> dict[str, Any] | None: """Fetch YAML data from GitLab repository with caching. Supports loading from multiple files matching the *-releases.yml glob pattern. @@ -187,11 +189,17 @@ def _fetch_from_gitlab(self) -> Optional[Dict[str, Any]]: logger.error(f"Network error fetching RHEL release data: {e}") # Try to use cached data as fallback return self._load_cached_data(cache_file) - except Exception as e: + except ( + json.JSONDecodeError, + KeyError, + TypeError, + ValueError, + yaml.YAMLError, + ) as e: logger.error(f"Error fetching RHEL release data: {e}") return self._load_cached_data(cache_file) - def _list_matching_files(self, pattern: str) -> List[str]: + def _list_matching_files(self, pattern: str) -> list[str]: """List files in the repository that match the given glob pattern.""" try: # Build GitLab API URL for listing repository tree @@ -234,11 +242,11 @@ def _list_matching_files(self, pattern: str) -> List[str]: except httpx.RequestError as e: logger.error(f"Network error listing repository files: {e}") return [] - except Exception as e: + except (json.JSONDecodeError, KeyError, TypeError, ValueError) as e: logger.error(f"Error listing repository files: {e}") return [] - def _get_latest_commit_hash(self) -> Optional[str]: + def _get_latest_commit_hash(self) -> str | None: """Get the latest commit hash for the branch to use as cache invalidation.""" try: # Build GitLab API URL for getting latest commit @@ -269,17 +277,15 @@ def _get_latest_commit_hash(self) -> Optional[str]: except httpx.RequestError as e: logger.error(f"Network error getting latest commit: {e}") - except Exception as e: + except (json.JSONDecodeError, KeyError, TypeError, ValueError) as e: logger.error(f"Error getting latest commit: {e}") return None - def _fetch_and_combine_files( - self, file_paths: List[str] - ) -> Optional[Dict[str, Any]]: + def _fetch_and_combine_files(self, file_paths: list[str]) -> dict[str, Any] | None: """Fetch multiple YAML files and combine their data.""" - combined_nodes: Dict[str, Any] = {} - combined_edges: Dict[str, Any] = {} + combined_nodes: dict[str, Any] = {} + combined_edges: dict[str, Any] = {} # SSL certificate path can be set via SSL_CERT_FILE environment variable ssl_cert_file = os.environ.get("SSL_CERT_FILE") @@ -353,7 +359,7 @@ def _fetch_and_combine_files( except httpx.RequestError as e: logger.error(f"Network error fetching {file_path}: {e}") continue - except Exception as e: + except (KeyError, TypeError, ValueError) as e: logger.error(f"Error processing {file_path}: {e}") continue @@ -374,13 +380,13 @@ def _fetch_and_combine_files( return result - def _load_cached_commit_hash(self, commit_hash_file: str) -> Optional[str]: + def _load_cached_commit_hash(self, commit_hash_file: str) -> str | None: """Load cached commit hash value.""" try: if os.path.exists(commit_hash_file): with open(commit_hash_file, "r") as f: return f.read().strip() - except Exception as e: + except OSError as e: logger.debug(f"Could not load cached commit hash: {e}") return None @@ -389,10 +395,10 @@ def _cache_commit_hash(self, commit_hash_file: str, commit_hash: str) -> None: try: with open(commit_hash_file, "w") as f: f.write(commit_hash) - except Exception as e: + except OSError as e: logger.debug(f"Could not cache commit hash: {e}") - def _load_cached_data(self, cache_file: str) -> Optional[Dict[str, Any]]: + def _load_cached_data(self, cache_file: str) -> dict[str, Any] | None: """Load cached YAML data.""" try: if os.path.exists(cache_file): @@ -402,7 +408,7 @@ def _load_cached_data(self, cache_file: str) -> Optional[Dict[str, Any]]: return data else: logger.debug(f"Cached data in {cache_file} is not a dictionary") - except Exception as e: + except OSError as e: logger.debug(f"Could not load cached data: {e}") return None @@ -411,10 +417,10 @@ def _cache_data(self, cache_file: str, content: str) -> None: try: with open(cache_file, "w") as f: f.write(content) - except Exception as e: + except OSError as e: logger.debug(f"Could not cache data: {e}") - def _parse_yaml_data(self, data: Dict[str, Any]) -> None: + def _parse_yaml_data(self, data: dict[str, Any]) -> None: """Parse YAML data and build node structures.""" # Load nodes if "nodes" in data: @@ -444,11 +450,11 @@ def _parse_yaml_data(self, data: Dict[str, Any]) -> None: logger.info(f"Loaded {len(self.nodes)} RHEL release nodes") - def get_leaf_nodes(self) -> List[RHELReleaseNode]: + def get_leaf_nodes(self) -> list[RHELReleaseNode]: """Get all leaf nodes (nodes with no children).""" return [node for node in self.nodes.values() if not node.children] - def get_descendants(self, node_name: str) -> Set[str]: + def get_descendants(self, node_name: str) -> set[str]: """Get all descendants (children, grandchildren, etc.) of a node.""" if node_name not in self.nodes: return set() @@ -466,7 +472,7 @@ def get_descendants(self, node_name: str) -> Set[str]: return descendants - def get_ancestors(self, node_name: str) -> Set[str]: + def get_ancestors(self, node_name: str) -> set[str]: """Get all ancestors (parents, grandparents, etc.) of a node.""" if node_name not in self.nodes: return set() @@ -484,13 +490,13 @@ def get_ancestors(self, node_name: str) -> Set[str]: return ancestors - def find_matching_nodes_for_cpe(self, cpe: str) -> List[RHELReleaseNode]: + def find_matching_nodes_for_cpe(self, cpe: str) -> list[RHELReleaseNode]: """Find all RHEL release nodes that contain the given CPE.""" return self.cpe_to_nodes.get(cpe, []) def find_active_streams_for_cpe( - self, cpe: str, active_streams: Set[str], stream_cpes: Dict[str, List[str]] - ) -> Set[str]: + self, cpe: str, active_streams: set[str], stream_cpes: dict[str, list[str]] + ) -> set[str]: """ Find active ps_update_streams that should be associated with a given CPE. @@ -521,9 +527,8 @@ def find_active_streams_for_cpe( if not has_active_rhel_streams: # For non-RHEL 9 streams, use direct matching first for stream_name in active_streams: - if stream_name in stream_cpes: - if cpe in stream_cpes[stream_name]: - result_streams.add(stream_name) + if stream_name in stream_cpes and cpe in stream_cpes[stream_name]: + result_streams.add(stream_name) # If we found direct matches for non-RHEL streams, return them if result_streams: @@ -546,15 +551,17 @@ def find_active_streams_for_cpe( # Check if this node's ps_update_stream matches any active stream # This avoids relying on CPE matching between product-definitions and rhel_releases - if candidate_node.ps_update_stream: - if candidate_node.ps_update_stream in active_streams: - result_streams.add(candidate_node.ps_update_stream) + if ( + candidate_node.ps_update_stream + and candidate_node.ps_update_stream in active_streams + ): + result_streams.add(candidate_node.ps_update_stream) return result_streams def get_all_cpes_for_stream( - self, stream_name: str, stream_cpes: Dict[str, List[str]] - ) -> Set[str]: + self, stream_name: str, stream_cpes: dict[str, list[str]] + ) -> set[str]: """ Get all CPEs that should be associated with a given RHEL stream by traversing the release graph to find related nodes using ps_update_stream attribute matching. @@ -630,7 +637,7 @@ def __init__( git_branch: Git branch to use for RHEL release data rhel_releases_path: Local file path (for testing only) """ - self.rhel_releases: Optional[RHELReleaseData] = None + self.rhel_releases: RHELReleaseData | None = None try: # Only try to load if we have a valid file path that exists @@ -644,13 +651,13 @@ def __init__( else: logger.warning(f"RHEL release file not found: {rhel_releases_path}") self.rhel_releases = None - except Exception as e: + except (OSError, ValueError, TypeError, httpx.HTTPError, yaml.YAMLError) as e: logger.warning(f"Could not load RHEL release data: {e}") self.rhel_releases = None def enhance_cpe_matching( - self, cpe: str, active_streams: Set[str], stream_cpes: Dict[str, List[str]] - ) -> Set[str]: + self, cpe: str, active_streams: set[str], stream_cpes: dict[str, list[str]] + ) -> set[str]: """ Enhance CPE matching using RHEL release hierarchy data. @@ -670,8 +677,8 @@ def enhance_cpe_matching( return result if result is not None else set() def get_all_cpes_for_stream( - self, stream_name: str, stream_cpes: Dict[str, List[str]] - ) -> Set[str]: + self, stream_name: str, stream_cpes: dict[str, list[str]] + ) -> set[str]: """ Get all CPEs that should be associated with a given RHEL stream. diff --git a/tests/test_pagination.py b/tests/test_pagination.py new file mode 100644 index 0000000..1d11a99 --- /dev/null +++ b/tests/test_pagination.py @@ -0,0 +1,111 @@ +from unittest.mock import MagicMock, patch + +from trustshell import paginated_trustify_query + + +def _mock_response(json_data: dict) -> MagicMock: + response = MagicMock() + response.json.return_value = json_data + response.raise_for_status = MagicMock() + return response + + +def _item(n: int) -> dict[str, int]: + return {"id": n} + + +@patch("trustshell.AUTH_ENABLED", False) +@patch("trustshell.httpx.Client") +class TestPaginatedTrustifyQuery: + endpoint = "http://localhost:8080/api/v2/analysis/latest/component" + + def test_does_not_add_total_by_default(self, mock_client_cls: MagicMock) -> None: + mock_client = MagicMock() + mock_client_cls.return_value.__enter__.return_value = mock_client + mock_client.get.return_value = _mock_response({"items": [_item(1)], "total": 1}) + + paginated_trustify_query(self.endpoint, {"q": "purl~foo"}, {}, limit=100) + + first_call_params = mock_client.get.call_args_list[0].kwargs["params"] + assert "total" not in first_call_params + assert first_call_params["limit"] == 100 + assert first_call_params["offset"] == 0 + + def test_known_total_fetches_all_pages(self, mock_client_cls: MagicMock) -> None: + mock_client = MagicMock() + mock_client_cls.return_value.__enter__.return_value = mock_client + mock_client.get.side_effect = [ + _mock_response({"items": [_item(i) for i in range(100)], "total": 150}), + _mock_response( + {"items": [_item(i) for i in range(100, 150)], "total": 150} + ), + ] + + result = paginated_trustify_query( + self.endpoint, {"q": "purl~foo"}, {}, limit=100 + ) + + assert len(result["items"]) == 150 + assert result["total"] == 150 + assert mock_client.get.call_count == 2 + + def test_null_total_fetches_until_short_page( + self, mock_client_cls: MagicMock + ) -> None: + mock_client = MagicMock() + mock_client_cls.return_value.__enter__.return_value = mock_client + mock_client.get.side_effect = [ + _mock_response({"items": [_item(i) for i in range(100)], "total": None}), + _mock_response( + {"items": [_item(i) for i in range(100, 125)], "total": None} + ), + ] + + result = paginated_trustify_query( + self.endpoint, {"q": "purl~foo"}, {}, limit=100 + ) + + assert len(result["items"]) == 125 + assert result["total"] == 125 + assert mock_client.get.call_count == 2 + + def test_empty_response_with_null_total(self, mock_client_cls: MagicMock) -> None: + mock_client = MagicMock() + mock_client_cls.return_value.__enter__.return_value = mock_client + mock_client.get.return_value = _mock_response({"items": [], "total": None}) + + result = paginated_trustify_query( + self.endpoint, {"q": "purl~missing"}, {}, limit=100 + ) + + assert result == {"items": [], "total": 0} + assert mock_client.get.call_count == 1 + + def test_respects_explicit_total_param(self, mock_client_cls: MagicMock) -> None: + mock_client = MagicMock() + mock_client_cls.return_value.__enter__.return_value = mock_client + mock_client.get.return_value = _mock_response({"items": [_item(1)], "total": 1}) + + paginated_trustify_query( + self.endpoint, + {"q": "purl~foo", "total": True}, + {}, + limit=100, + ) + + first_call_params = mock_client.get.call_args_list[0].kwargs["params"] + assert first_call_params["total"] is True + + def test_numeric_total_pagination(self, mock_client_cls: MagicMock) -> None: + mock_client = MagicMock() + mock_client_cls.return_value.__enter__.return_value = mock_client + mock_client.get.side_effect = [ + _mock_response({"items": [_item(0)], "total": 2}), + _mock_response({"items": [_item(1)], "total": 2}), + ] + + result = paginated_trustify_query(self.endpoint, {"q": "purl~foo"}, {}, limit=1) + + assert len(result["items"]) == 2 + assert result["total"] == 2 + assert mock_client.get.call_count == 2 diff --git a/tests/test_product_definitions.py b/tests/test_product_definitions.py index 5d59973..396c331 100644 --- a/tests/test_product_definitions.py +++ b/tests/test_product_definitions.py @@ -1,11 +1,12 @@ import json -import tempfile import os +import tempfile import unittest -from anytree import Node from unittest.mock import patch +from anytree import Node from test_products import _check_node_names_at_depth + from trustshell import render_tree from trustshell.product_definitions import ProdDefs from trustshell.products import build_product_search_result @@ -241,11 +242,12 @@ def _create_test_rhel_releases_yaml(self): - RHEL-9.2.0.Z.EUS - RHEL-9.3.0.GA """ - temp_file = tempfile.NamedTemporaryFile(mode="w", suffix=".yml", delete=False) - temp_file.write(test_data) - temp_file.flush() - temp_file.close() - return temp_file.name + with tempfile.NamedTemporaryFile( + mode="w", suffix=".yml", delete=False + ) as temp_file: + temp_file.write(test_data) + temp_file.flush() + return temp_file.name def _create_enhanced_product_definitions(self): """Create product definitions with enhanced RHEL streams for testing.""" diff --git a/tests/test_products.py b/tests/test_products.py index ba66111..8188275 100644 --- a/tests/test_products.py +++ b/tests/test_products.py @@ -1,22 +1,22 @@ import json import unittest -from parameterized import parameterized from unittest.mock import patch from anytree import Node +from parameterized import parameterized from trustshell import build_node_purl, render_tree +from trustshell.product_definitions import ProdDefs from trustshell.products import ( ComponentNode, _get_branch_signature, + _has_cpe_node, _remove_duplicate_parent_nodes, _remove_non_cpe_branches, _trees_with_cpes, - _has_cpe_node, - container_in_tree, build_product_search_result, + container_in_tree, ) -from trustshell.product_definitions import ProdDefs class TestProducts(unittest.TestCase): @@ -32,14 +32,14 @@ def setUp(self): ) def test_build_node_purl_rpm(self, show_versions, expected_purl): purls = [ - "pkg:rpm/redhat/webkit2gtk3@2.42.5-1.el9?arch=src&repository_id=rhel-9-for-aarch64-appstrea" - "pkg:rpm/redhat/webkit2gtk3@2.42.5-1.el9?arch=src&repository_id=rhel-9-for-ppc64le-appstrea" - "pkg:rpm/redhat/webkit2gtk3@2.42.5-1.el9?arch=src&repository_id=rhel-9-for-aarch64-appstrea" - "pkg:rpm/redhat/webkit2gtk3@2.42.5-1.el9?arch=src&repository_id=rhel-9-for-s390x-appstream-" - "pkg:rpm/redhat/webkit2gtk3@2.42.5-1.el9?arch=src&repository_id=rhel-9-for-s390x-appstream-" - "pkg:rpm/redhat/webkit2gtk3@2.42.5-1.el9?arch=src&repository_id=rhel-9-for-ppc64le-appstrea" - "pkg:rpm/redhat/webkit2gtk3@2.42.5-1.el9?arch=src&repository_id=rhel-9-for-x86_64-appstream" - "pkg:rpm/redhat/webkit2gtk3@2.42.5-1.el9?arch=src&repository_id=rhel-9-for-x86_64-appstream" + "pkg:rpm/redhat/webkit2gtk3@2.42.5-1.el9?arch=src&repository_id=rhel-9-for-aarch64-appstrea", + "pkg:rpm/redhat/webkit2gtk3@2.42.5-1.el9?arch=src&repository_id=rhel-9-for-ppc64le-appstrea", + "pkg:rpm/redhat/webkit2gtk3@2.42.5-1.el9?arch=src&repository_id=rhel-9-for-aarch64-appstrea", + "pkg:rpm/redhat/webkit2gtk3@2.42.5-1.el9?arch=src&repository_id=rhel-9-for-s390x-appstream-", + "pkg:rpm/redhat/webkit2gtk3@2.42.5-1.el9?arch=src&repository_id=rhel-9-for-s390x-appstream-", + "pkg:rpm/redhat/webkit2gtk3@2.42.5-1.el9?arch=src&repository_id=rhel-9-for-ppc64le-appstrea", + "pkg:rpm/redhat/webkit2gtk3@2.42.5-1.el9?arch=src&repository_id=rhel-9-for-x86_64-appstream", + "pkg:rpm/redhat/webkit2gtk3@2.42.5-1.el9?arch=src&repository_id=rhel-9-for-x86_64-appstream", ] result = build_node_purl(purls, show_versions=show_versions).to_string() assert result == expected_purl @@ -66,8 +66,10 @@ def test_build_node_purl_oci(self): ) def test_build_node_purl_maven_type(self, show_versions, expected_purl): purls = [ - "pkg:maven/io.agroal/agroal-api@1.3.0.redhat-00001?repository_url=https%3A%2F%2Fmaven.repository.redhat.com%2Fga%2F&type=jar" - "pkg:maven/io.agroal/agroal-api@1.3.0.redhat-00001?repository_url=https%3A%2F%2Fmaven.repository.redhat.com%2Fga%2F&type=jar&hash=sha256:1234567890" + ( + "pkg:maven/io.agroal/agroal-api@1.3.0.redhat-00001?repository_url=https%3A%2F%2Fmaven.repository.redhat.com%2Fga%2F&type=jar" + "pkg:maven/io.agroal/agroal-api@1.3.0.redhat-00001?repository_url=https%3A%2F%2Fmaven.repository.redhat.com%2Fga%2F&type=jar&hash=sha256:1234567890" + ) ] result = build_node_purl(purls, show_versions=show_versions).to_string() print(result) diff --git a/tests/test_purl.py b/tests/test_purl.py index ecaaac4..5d3ed07 100644 --- a/tests/test_purl.py +++ b/tests/test_purl.py @@ -1,4 +1,5 @@ from unittest.mock import Mock, patch + from packageurl import PackageURL from trustshell.purl import _query_trustify_packages_base_purl diff --git a/tests/test_rhel_releases.py b/tests/test_rhel_releases.py index c0ddf57..583a856 100644 --- a/tests/test_rhel_releases.py +++ b/tests/test_rhel_releases.py @@ -1,11 +1,11 @@ """Tests for RHEL release data parsing and CPE matching functionality.""" -import tempfile import os +import tempfile from unittest.mock import patch -from trustshell.rhel_releases import RHELReleaseData, EnhancedProdDefs from trustshell.product_definitions import ProdDefs +from trustshell.rhel_releases import EnhancedProdDefs, RHELReleaseData def create_test_rhel_data(): diff --git a/tests/test_rhel_releases_multifile.py b/tests/test_rhel_releases_multifile.py index 0473a2b..ed71e12 100644 --- a/tests/test_rhel_releases_multifile.py +++ b/tests/test_rhel_releases_multifile.py @@ -1,8 +1,9 @@ """Tests for RHEL release data parsing with multiple files.""" -import tempfile import os -from unittest.mock import patch, MagicMock +import tempfile +from unittest.mock import MagicMock, patch + import yaml from trustshell.rhel_releases import RHELReleaseData @@ -167,44 +168,45 @@ def mock_get_side_effect(url, **kwargs): try: # Set environment variable for GitLab URL - with patch.dict( - os.environ, - {"RHEL_RELEASE_GRAPH_URL": "https://example.com/api/v4/projects/123"}, - ): - # Create RHELReleaseData and force it to use the GitLab URL and temp cache dir - with patch.object( + with ( + patch.dict( + os.environ, + { + "RHEL_RELEASE_GRAPH_URL": "https://example.com/api/v4/projects/123" + }, + ), + patch.object( RHELReleaseData, "RHEL_RELEASE_GRAPH_BASE", "https://example.com/api/v4/projects/123", - ): - with patch.object(RHELReleaseData, "CACHE_DIR", temp_cache_dir): - rhel_data = RHELReleaseData() - - # Verify that all files were processed - assert ( - len(rhel_data.nodes) == 5 - ) # 2 from rhel8 + 2 from rhel9 + 1 from rhel10 - assert "RHEL-8.0.0.GA" in rhel_data.nodes - assert "RHEL-9.0.0.GA" in rhel_data.nodes - assert "RHEL-10.0.0.GA" in rhel_data.nodes - - # Verify the listing call was made - tree_calls = [ - call - for call in mock_get.call_args_list - if "/repository/tree" in str(call) - ] - assert len(tree_calls) == 1 - - # Verify file fetching calls were made - file_calls = [ - call - for call in mock_get.call_args_list - if "/repository/files/" in str(call) - ] - assert ( - len(file_calls) == 3 - ) # Should fetch 3 *-releases.yml files + ), + patch.object(RHELReleaseData, "CACHE_DIR", temp_cache_dir), + ): + rhel_data = RHELReleaseData() + + # Verify that all files were processed + assert ( + len(rhel_data.nodes) == 5 + ) # 2 from rhel8 + 2 from rhel9 + 1 from rhel10 + assert "RHEL-8.0.0.GA" in rhel_data.nodes + assert "RHEL-9.0.0.GA" in rhel_data.nodes + assert "RHEL-10.0.0.GA" in rhel_data.nodes + + # Verify the listing call was made + tree_calls = [ + call + for call in mock_get.call_args_list + if "/repository/tree" in str(call) + ] + assert len(tree_calls) == 1 + + # Verify file fetching calls were made + file_calls = [ + call + for call in mock_get.call_args_list + if "/repository/files/" in str(call) + ] + assert len(file_calls) == 3 # Should fetch 3 *-releases.yml files finally: # Clean up temp directory diff --git a/tests/test_trustify_url.py b/tests/test_trustify_url.py new file mode 100644 index 0000000..42a321e --- /dev/null +++ b/tests/test_trustify_url.py @@ -0,0 +1,42 @@ +import importlib +import os +from unittest.mock import patch + +import trustshell + + +def _reload_trustshell() -> None: + importlib.reload(trustshell) + + +class TestTrustifyUrl: + def test_default_appends_v2_when_no_path(self) -> None: + with patch.dict( + os.environ, + {"TRUSTIFY_URL": "https://atlas.example.com"}, + clear=False, + ): + _reload_trustshell() + assert trustshell.TRUSTIFY_URL == "https://atlas.example.com/api/v2/" + + def test_preserves_explicit_v3_path(self) -> None: + with patch.dict( + os.environ, + {"TRUSTIFY_URL": "https://atlas.example.com/api/v3/"}, + clear=False, + ): + _reload_trustshell() + assert trustshell.TRUSTIFY_URL == "https://atlas.example.com/api/v3/" + + def test_preserves_explicit_v2_path(self) -> None: + with patch.dict( + os.environ, + {"TRUSTIFY_URL": "https://atlas.example.com/api/v2/"}, + clear=False, + ): + _reload_trustshell() + assert trustshell.TRUSTIFY_URL == "https://atlas.example.com/api/v2/" + + def teardown_method(self) -> None: + os.environ.pop("TRUSTIFY_URL", None) + _reload_trustshell()