fix: Resolve all pre-commit ruff and mypy errors (92 total) (#2569)

* fix: Resolve all pre-commit ruff and mypy errors (92 total)

Fixed 90 ruff errors and 2 mypy errors across 242 files through proper code
improvements without using noqa suppressions.

Ruff errors resolved:
- ARG001/ARG002: Removed unused function/method arguments
- RUF059: Removed unused unpacked variables from tuples
- PLC0415: Moved imports to module top-level
- B904: Added proper exception chaining with 'from' clause
- B007: Prefixed unused loop variables with underscore
- B905: Added strict=True parameter to zip() calls
- B011: Replaced 'assert False' with 'raise AssertionError'
- B026: Moved star-arg unpacking before keyword arguments
- B028: Added stacklevel parameter to warnings.warn()
- UP031: Converted percent formatting to f-strings

Mypy errors resolved:
- Fixed type annotation issues in exception handling
- Corrected function signature type hints

Test fixes:
- Fixed 4 failing tests affected by argument removal
- Updated test assertions and mock calls

Impact:
- All pre-commit checks now pass
- Code quality improved through proper refactoring
- Test suite fully passing
- Zero technical debt from error suppression

* fix: organize imports in resource files

Fix import formatting in authentication and gateway resource classes:
- Add blank lines between import groups per PEP 8
- Alphabetize imports in data_science_pipelines_application.py

Files updated:
- authentication_config_openshift_io.py
- authentication_operator_openshift_io.py
- data_science_pipelines_application.py
- gateway_config.py

Author: myakove

class_generator/cli.py

@@ -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/core/coverage.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/discovery.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/generator.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/schema.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/formatters/template_renderer.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/parsers/explain_parser.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/type_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
 

class_generator/parsers/user_code_parser.py

@@ -26,20 +26,20 @@ def parse_user_code_from_file(file_path: str) -> tuple[str, str]:
     try:
         with open(file_path, encoding="utf-8") as fd:
             data = fd.read()
-    except FileNotFoundError:
-        raise FileNotFoundError(f"File not found: {file_path}. Please ensure the file exists.")
-    except PermissionError:
-        raise PermissionError(f"Permission denied when accessing file: {file_path}. Check file permissions.")
+    except FileNotFoundError as err:
+        raise FileNotFoundError(f"File not found: {file_path}. Please ensure the file exists.") from err
+    except PermissionError as err:
+        raise PermissionError(f"Permission denied when accessing file: {file_path}. Check file permissions.") from err
     except UnicodeDecodeError as e:
         raise UnicodeDecodeError(
             e.encoding,
             e.object,
             e.start,
             e.end,
             f"Failed to decode file {file_path} with UTF-8 encoding. The file may contain invalid characters.",
-        )
+        ) from e
     except Exception as e:
-        raise Exception(f"Unexpected error reading file {file_path}: {type(e).__name__}: {str(e)}")
+        raise Exception(f"Unexpected error reading file {file_path}: {type(e).__name__}: {str(e)}") from e
 
     end_of_generated_code_line = "    # End of generated code"
     user_code: str = ""

class_generator/tests/manifests/APIServer/api_server.py

@@ -2,6 +2,7 @@
 
 
 from typing import Any
+
 from ocp_resources.resource import Resource