Merge pull request #6 from RedHatQE/main

[pull] main from RedHatQE:main

Author: dbasunag

.pre-commit-config.yaml

@@ -10,7 +10,7 @@ ci:
 
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: "v6.0.0"
+    rev: v6.0.0
     hooks:
       - id: check-merge-conflict
       - id: debug-statements
@@ -23,7 +23,7 @@ repos:
       - id: check-toml
 
   - repo: https://github.com/PyCQA/flake8
-    rev: "7.3.0"
+    rev: 7.3.0
     hooks:
       - id: flake8
         args: [--config=.flake8]
@@ -45,18 +45,18 @@ repos:
           ]
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.13.0
+    rev: v0.14.5
     hooks:
       - id: ruff
       - id: ruff-format
 
   - repo: https://github.com/gitleaks/gitleaks
-    rev: v8.28.0
+    rev: v8.29.0
     hooks:
       - id: gitleaks
 
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.18.1
+    rev: v1.18.2
     hooks:
       - id: mypy
         exclude: ^(tests/|examples/|docs/)

VERSION

@@ -1,14 +1,14 @@
 """Command-line interface for the class generator."""
 
 import fnmatch
+import logging
+import os
 import shutil
 import sys
 from datetime import datetime
 from pathlib import Path
 from typing import Any
 
-import logging
-
 import cloup
 from cloup.constraints import If, IsSet, accept_none, require_one
 from simple_logger.logger import get_logger
@@ -17,7 +17,7 @@
 from class_generator.core.coverage import analyze_coverage, generate_report
 from class_generator.core.discovery import discover_generated_resources
 from class_generator.core.generator import class_generator
-from class_generator.core.schema import update_kind_schema, ClusterVersionError
+from class_generator.core.schema import ClusterVersionError, update_kind_schema
 from class_generator.tests.test_generation import generate_class_generator_tests
 from class_generator.utils import execute_parallel_tasks
 from ocp_resources.utils.utils import convert_camel_case_to_snake_case
@@ -57,7 +57,7 @@ def handle_schema_update(update_schema: bool, generate_missing: bool) -> bool:
         LOGGER.info("Updating resource schema...")
         try:
             update_kind_schema()
-        except (RuntimeError, IOError, ClusterVersionError) as e:
+        except (OSError, RuntimeError, ClusterVersionError) as e:
             LOGGER.exception(f"Failed to update schema: {e}")
             sys.exit(1)
 
@@ -251,9 +251,9 @@ def regenerate_single_resource(resource: dict[str, Any]) -> tuple[str, bool, str
             return resource_kind, False, str(e)
 
     # Process results from parallel execution
-    def process_regeneration_result(resource: dict[str, Any], result: tuple[str, bool, str | None]) -> None:
+    def process_regeneration_result(_resource: dict[str, Any], result: tuple[str, bool, str | None]) -> None:
         nonlocal success_count, error_count
-        resource_kind, success, error = result
+        _resource_kind, success, _error = result
         if success:
             success_count += 1
         else:
@@ -391,7 +391,7 @@ def generate_with_backup(kind_to_generate: str) -> tuple[str, bool, str | None]:
                 return kind_to_generate, False, str(e)
 
         # Process results from parallel execution
-        def process_generation_result(kind_to_generate: str, result: tuple[str, bool, str | None]) -> None:
+        def process_generation_result(_kind_to_generate: str, result: tuple[str, bool, str | None]) -> None:
             nonlocal success_count, error_count, failed_kinds
             kind_name, success, error = result
             if success:
@@ -447,8 +447,6 @@ def handle_test_generation(add_tests: bool) -> None:
 
         # Run the generated test file
         LOGGER.info("Running generated tests...")
-        import os
-
         test_file = "class_generator/tests/test_class_generator.py"
         exit_code = os.system(f"uv run pytest {test_file}")
 

class_generator/cli.py

@@ -68,7 +68,7 @@ def analyze_coverage(
             continue
 
         try:
-            with open(filepath, "r") as f:
+            with open(filepath) as f:
                 content = f.read()
 
             # Check if file is auto-generated

class_generator/core/coverage.py

@@ -1,8 +1,8 @@
 """Discovery functions for finding cluster resources and generated files."""
 
+from concurrent.futures import Future, ThreadPoolExecutor, as_completed
 from pathlib import Path
 from typing import Any
-from concurrent.futures import ThreadPoolExecutor, as_completed, Future
 
 from kubernetes.dynamic import DynamicClient
 from simple_logger.logger import get_logger
@@ -221,7 +221,7 @@ def discover_generated_resources() -> list[dict[str, Any]]:
     for info in resource_infos:
         # Read file to check for user code
         try:
-            with open(info.file_path, "r", encoding="utf-8") as f:
+            with open(info.file_path, encoding="utf-8") as f:
                 content = f.read()
         except FileNotFoundError:
             LOGGER.warning(f"File not found: {info.file_path}, skipping...")

class_generator/core/discovery.py

@@ -155,7 +155,7 @@ def class_generator(
             LOGGER.info("Updating schema")
             try:
                 update_kind_schema()
-            except (RuntimeError, IOError) as e:
+            except (OSError, RuntimeError) as e:
                 error_msg = f"Failed to update schema: {e}"
                 LOGGER.error(error_msg)
                 raise RuntimeError(error_msg) from e

class_generator/core/generator.py

@@ -12,7 +12,7 @@
 from simple_logger.logger import get_logger
 
 from class_generator.constants import DEFINITIONS_FILE, RESOURCES_MAPPING_FILE, SCHEMA_DIR
-from class_generator.utils import execute_parallel_with_mapping, execute_parallel_tasks
+from class_generator.utils import execute_parallel_tasks, execute_parallel_with_mapping
 from ocp_resources.utils.archive_utils import save_json_archive
 from ocp_resources.utils.schema_validator import SchemaValidator
 
@@ -136,9 +136,9 @@ def check_and_update_cluster_version(client: str) -> bool:
     last_cluster_version_generated: str = ""
 
     try:
-        with open(cluster_version_file, "r") as fd:
+        with open(cluster_version_file) as fd:
             last_cluster_version_generated = fd.read().strip()
-    except (FileNotFoundError, IOError):
+    except (OSError, FileNotFoundError):
         # Treat missing file as first run - use baseline version that allows updates
         last_cluster_version_generated = "v0.0.0"
         LOGGER.info("Cluster version file not found - treating as first run with baseline version v0.0.0")
@@ -530,7 +530,7 @@ def process_schema_definitions(
     definitions = {}
     if not allow_updates:
         try:
-            with open(DEFINITIONS_FILE, "r") as f:
+            with open(DEFINITIONS_FILE) as f:
                 existing_definitions_data = json.load(f)
                 definitions = existing_definitions_data.get("definitions", {})
                 LOGGER.info(f"Loaded {len(definitions)} existing definitions to preserve")
@@ -1064,12 +1064,12 @@ def process_explain_result(spec_tuple: tuple[str, str], result: Any) -> None:
             LOGGER.debug(f"Failed to obtain explain data for {ref_name} from {explain_path}")
 
     def handle_explain_error(spec_tuple: tuple[str, str], exc: Exception) -> None:
-        ref_name, explain_path = spec_tuple
-        LOGGER.debug(f"Exception occurred while explaining {ref_name} from {explain_path}: {exc}")
+        ref_name, _explain_path = spec_tuple
+        LOGGER.debug(f"Exception occurred while explaining {ref_name}: {exc}")
         explain_results[ref_name] = None
 
     def create_explain_task(spec_tuple: tuple[str, str]) -> Any:
-        ref_name, explain_path = spec_tuple
+        _ref_name, explain_path = spec_tuple
         return _run_explain_and_parse(client, explain_path)
 
     execute_parallel_tasks(
@@ -1244,7 +1244,7 @@ def process_required_field_result(task_tuple: tuple[str, str], result: Any) -> N
                 definitions[schema_key] = updated_schema
 
         def handle_required_field_error(task_tuple: tuple[str, str], exc: Exception) -> None:
-            schema_key, explain_path = task_tuple
+            schema_key, _explain_path = task_tuple
             LOGGER.debug(f"Failed to process required fields for {schema_key}: {exc}")
             # Set empty list if explain fails
             current_schema = definitions[schema_key]
@@ -1307,14 +1307,14 @@ def _get_missing_core_definitions(
     if refs_to_fetch:
 
         def process_missing_definition_result(task_tuple: tuple[str, str], result: Any) -> None:
-            ref_name, oc_path = task_tuple
+            _ref_name, _oc_path = task_tuple
             if result:
                 fetched_ref_name, schema = result
                 missing_definitions[fetched_ref_name] = schema
                 LOGGER.debug(f"Successfully fetched definition for {fetched_ref_name}")
 
         def handle_missing_definition_error(task_tuple: tuple[str, str], exc: Exception) -> None:
-            ref_name, oc_path = task_tuple
+            ref_name, _oc_path = task_tuple
             LOGGER.debug(f"Failed to fetch definition for {ref_name}: {exc}")
 
         def create_missing_definition_task(task_tuple: tuple[str, str]) -> Any:
@@ -1360,10 +1360,10 @@ def write_schema_files(
     # Ensure schema directory exists
     try:
         Path(SCHEMA_DIR).mkdir(parents=True, exist_ok=True)
-    except (OSError, IOError) as e:
+    except OSError as e:
         error_msg = f"Failed to create schema directory {SCHEMA_DIR}: {e}"
         LOGGER.error(error_msg)
-        raise IOError(error_msg) from e
+        raise OSError(error_msg) from e
 
     # Fetch missing core definitions if schemas are available
     if schemas:
@@ -1400,18 +1400,18 @@ def write_schema_files(
             with open(definitions_file, "w") as fd:
                 json.dump(definitions_data, fd, indent=2, sort_keys=True)
             LOGGER.info(f"Written {len(definitions)} definitions to {definitions_file}")
-        except (OSError, IOError, TypeError) as e:
+        except (OSError, TypeError) as e:
             error_msg = f"Failed to write definitions file {definitions_file}: {e}"
             LOGGER.error(error_msg)
-            raise IOError(error_msg) from e
+            raise OSError(error_msg) from e
 
     # Write and archive resources mapping
     try:
         save_json_archive(resources_mapping, RESOURCES_MAPPING_FILE)
-    except (OSError, IOError, TypeError) as e:
+    except (OSError, TypeError) as e:
         error_msg = f"Failed to save and archive resources mapping file {RESOURCES_MAPPING_FILE}: {e}"
         LOGGER.error(error_msg)
-        raise IOError(error_msg) from e
+        raise OSError(error_msg) from e
 
 
 @dataclasses.dataclass
@@ -1523,11 +1523,11 @@ def _handle_no_schemas_case() -> None:
     """
     LOGGER.info("No schemas fetched. Preserving existing data to avoid overwriting with empty definitions.")
     try:
-        with open(DEFINITIONS_FILE, "r") as fd:
+        with open(DEFINITIONS_FILE) as fd:
             existing_definitions_data = json.load(fd)
             definitions = existing_definitions_data.get("definitions", {})
             LOGGER.info(f"Found {len(definitions)} existing definitions that will be preserved")
-    except (FileNotFoundError, IOError, json.JSONDecodeError):
+    except (OSError, FileNotFoundError, json.JSONDecodeError):
         LOGGER.debug("Could not load existing definitions file. No existing definitions to preserve.")
 
 

class_generator/core/schema.py

@@ -1,7 +1,7 @@
 """Jinja template rendering for resource generation."""
 
-from typing import Any
 from pathlib import Path
+from typing import Any
 
 from jinja2 import DebugUndefined, Environment, FileSystemLoader, meta
 from simple_logger.logger import get_logger
@@ -38,7 +38,7 @@ def render_jinja_template(template_dict: dict[Any, Any], template_dir: str, temp
     except AttributeError:
         # Fallback: read the template file directly
         template_path = Path(template_dir) / template_name
-        with open(template_path, "r", encoding="utf-8") as f:
+        with open(template_path, encoding="utf-8") as f:
             template_source = f.read()
 
     ast = env.parse(source=template_source)

class_generator/formatters/template_renderer.py

@@ -55,7 +55,7 @@ def parse_explain(kind: str) -> list[dict[str, Any]]:
 
     # For each API group, select the latest version
     filtered_schemas = []
-    for group, group_schemas in schemas_by_group.items():
+    for _group, group_schemas in schemas_by_group.items():
         if len(group_schemas) > 1:
             # Multiple versions in same group - pick latest
             versions = []

class_generator/parsers/explain_parser.py

@@ -6,7 +6,7 @@
 
 from simple_logger.logger import get_logger
 
-from class_generator.constants import MISSING_DESCRIPTION_STR, SPEC_STR, DEFINITIONS_FILE
+from class_generator.constants import DEFINITIONS_FILE, MISSING_DESCRIPTION_STR, SPEC_STR
 from class_generator.utils import sanitize_python_name
 from ocp_resources.utils.utils import convert_camel_case_to_snake_case