mirror of
https://github.com/kuhyx/testsAndMisc.git
synced 2026-07-04 15:43:06 +02:00
fix(lint): BLE001 - replace blind except with specific exceptions
Replace bare 'except Exception' with specific exception types: - ValueError for move parsing (chess.Move.from_uci, board.push_uci) - json.JSONDecodeError for JSON parsing - OSError for file operations - ImportError for optional imports - AttributeError for attribute access - TypeError for type-related operations - requests.RequestException for HTTP operations - subprocess.SubprocessError for subprocess failures - selenium.NoSuchElementException for element finding Also fixes: - pytest hook signature issue in conftest.py (_config -> _) - Missing file handling in test_puzzles.py - Line length in stockfish_analysis.py Removes all BLE001 per-file ignores from pyproject.toml.
This commit is contained in:
parent
b78f02cf05
commit
dd2da6e2cc
@ -55,7 +55,7 @@ def test_crud_roundtrip(tmp_path: Path) -> None:
|
||||
) as resp:
|
||||
resp.read()
|
||||
break
|
||||
except Exception:
|
||||
except (OSError, urllib.error.URLError):
|
||||
time.sleep(0.05)
|
||||
|
||||
# Create
|
||||
@ -105,5 +105,5 @@ def test_crud_roundtrip(tmp_path: Path) -> None:
|
||||
srv.terminate()
|
||||
try:
|
||||
srv.wait(timeout=2)
|
||||
except Exception:
|
||||
except subprocess.TimeoutExpired:
|
||||
srv.kill()
|
||||
|
||||
@ -47,7 +47,6 @@ unfixable = []
|
||||
"S101", # Allow assert in tests
|
||||
"S603", # Allow subprocess calls in tests
|
||||
"PLR2004", # Allow magic values in tests
|
||||
"BLE001", # Allow blind except in test cleanup
|
||||
"PTH", # Allow os.path in tests for simplicity
|
||||
]
|
||||
"**/test_*.py" = [
|
||||
@ -56,7 +55,6 @@ unfixable = []
|
||||
"S310", # Allow URL open in tests
|
||||
"S607", # Allow partial executable path in tests
|
||||
"PLC0415", # Allow late imports for test isolation
|
||||
"BLE001", # Allow blind except in test cleanup
|
||||
"PTH", # Allow os.path in tests for simplicity
|
||||
]
|
||||
"**/conftest.py" = [
|
||||
@ -91,31 +89,26 @@ unfixable = []
|
||||
"C901", # Complex functions handling game lifecycle (run_bot, handle_game)
|
||||
"PLR0912", # Complex nested game event handling with many branches
|
||||
"PLR0915", # Long function handling complete game lifecycle
|
||||
"BLE001", # Blind except for resilient bot operation
|
||||
"S603", # Subprocess call for analysis script
|
||||
"PTH", # os.path patterns in existing code
|
||||
"LOG015", # Root logger in bot
|
||||
"G004", # f-strings in logging
|
||||
]
|
||||
"python_pkg/lichess_bot/engine.py" = [
|
||||
"BLE001", # Blind except for engine error handling
|
||||
"S603", # Subprocess for engine communication
|
||||
"PTH", # os.path patterns
|
||||
"LOG015", # Root logger for debug messages
|
||||
]
|
||||
"python_pkg/lichess_bot/lichess_api.py" = [
|
||||
"BLE001", # Blind except for API resilience
|
||||
"LOG015", # Root logger in API client
|
||||
"G004", # f-strings in logging
|
||||
]
|
||||
"python_pkg/lichess_bot/utils.py" = [
|
||||
"BLE001", # Blind except for file operations
|
||||
"PTH", # os.path patterns
|
||||
"LOG015", # Root logger
|
||||
"G004", # f-strings in logging
|
||||
]
|
||||
"python_pkg/lichess_bot/tools/generate_blunder_tests.py" = [
|
||||
"BLE001", # Blind except for test generation
|
||||
"PTH", # os.path patterns in tool
|
||||
"LOG015", # Root logger in tool
|
||||
"G004", # f-strings in logging
|
||||
@ -124,7 +117,6 @@ unfixable = []
|
||||
"C901", # Complex main() with many argument combinations and analysis modes
|
||||
"PLR0912", # Complex main() with many argument combinations and analysis modes
|
||||
"PLR0915", # Long main() handling complete analysis workflow
|
||||
"BLE001", # Blind except for engine configuration
|
||||
"PTH", # os.path patterns
|
||||
"LOG015", # Root logger in analysis tool
|
||||
"G004", # f-strings in logging
|
||||
@ -145,7 +137,6 @@ unfixable = []
|
||||
"G004", # f-strings in logging
|
||||
]
|
||||
"python_pkg/scrape_website/scrape_comics.py" = [
|
||||
"BLE001", # Blind except for web scraping resilience
|
||||
"PTH", # os.path patterns
|
||||
"LOG015", # Root logger in script
|
||||
"G004", # f-strings in logging
|
||||
|
||||
@ -99,7 +99,7 @@ class RandomEngine:
|
||||
chosen_uci = output.splitlines()[-1].strip() if output else ""
|
||||
try:
|
||||
move = chess.Move.from_uci(chosen_uci)
|
||||
except Exception:
|
||||
except ValueError:
|
||||
msg = f"Engine returned invalid move: '{chosen_uci}' (output: {output!r})"
|
||||
raise RuntimeError(msg) from None
|
||||
|
||||
@ -159,7 +159,7 @@ class RandomEngine:
|
||||
},
|
||||
ensure_ascii=False,
|
||||
)
|
||||
except Exception:
|
||||
except (json.JSONDecodeError, KeyError, TypeError):
|
||||
logging.debug("Failed to parse engine JSON output")
|
||||
|
||||
return cand_score, cand_expl, best_move, best_expl
|
||||
|
||||
@ -57,7 +57,7 @@ class LichessAPI:
|
||||
try:
|
||||
text = r.text or ""
|
||||
snippet = text[:200].replace("\n", " ")
|
||||
except Exception:
|
||||
except (AttributeError, TypeError):
|
||||
snippet = None
|
||||
if snippet:
|
||||
logging.warning(
|
||||
|
||||
@ -12,6 +12,7 @@ import threading
|
||||
|
||||
import chess
|
||||
import chess.pgn
|
||||
import requests
|
||||
|
||||
from python_pkg.lichess_bot.engine import RandomEngine
|
||||
from python_pkg.lichess_bot.lichess_api import LichessAPI
|
||||
@ -28,7 +29,7 @@ def _apply_move_to_board(board: chess.Board, move: str, game_id: str) -> None:
|
||||
"""
|
||||
try:
|
||||
board.push_uci(move)
|
||||
except Exception:
|
||||
except ValueError:
|
||||
logging.debug(f"Game {game_id}: could not apply move {move}")
|
||||
|
||||
|
||||
@ -66,7 +67,7 @@ def run_bot(log_level: str = "INFO", *, decline_correspondence: bool = False) ->
|
||||
with open(game_log_path, "w") as lf:
|
||||
lf.write(f"game {game_id} started\n")
|
||||
lf.write(f"bot_version v{bot_version}\n")
|
||||
except Exception:
|
||||
except OSError:
|
||||
game_log_path = None
|
||||
# Simple time manager state
|
||||
my_ms = None
|
||||
@ -229,7 +230,7 @@ def run_bot(log_level: str = "INFO", *, decline_correspondence: bool = False) ->
|
||||
f"{move.uci()}\n{reason}\n\n"
|
||||
)
|
||||
api.make_move(game_id, move)
|
||||
except Exception as e:
|
||||
except requests.RequestException as e:
|
||||
logging.warning(
|
||||
f"Game {game_id}: move {move.uci()} failed: {e}"
|
||||
)
|
||||
@ -243,7 +244,7 @@ def run_bot(log_level: str = "INFO", *, decline_correspondence: bool = False) ->
|
||||
break
|
||||
elif et in {"chatLine", "opponentGone"}:
|
||||
continue
|
||||
except Exception:
|
||||
except requests.RequestException:
|
||||
logging.exception(f"Game {game_id} thread error")
|
||||
finally:
|
||||
# On game end, write full PGN to the log file
|
||||
@ -282,7 +283,7 @@ def run_bot(log_level: str = "INFO", *, decline_correspondence: bool = False) ->
|
||||
# Estimate total plies from the final board
|
||||
try:
|
||||
total_plies = len(board.move_stack)
|
||||
except Exception:
|
||||
except TypeError:
|
||||
total_plies = 0
|
||||
|
||||
logging.info(
|
||||
@ -345,7 +346,7 @@ def run_bot(log_level: str = "INFO", *, decline_correspondence: bool = False) ->
|
||||
f"Game {game_id}: analysis script not found "
|
||||
f"at {analyze_script}; skipping analysis"
|
||||
)
|
||||
except Exception as e:
|
||||
except (subprocess.SubprocessError, OSError) as e:
|
||||
logging.debug(f"Game {game_id}: analysis run failed: {e}")
|
||||
|
||||
# Insert analysis before the PGN section so future runs
|
||||
@ -397,11 +398,11 @@ def run_bot(log_level: str = "INFO", *, decline_correspondence: bool = False) ->
|
||||
)
|
||||
with open(game_log_path, "w", encoding="utf-8") as f:
|
||||
f.write(new_content)
|
||||
except Exception as e:
|
||||
except OSError as e:
|
||||
logging.debug(
|
||||
f"Game {game_id}: could not write analysis to log: {e}"
|
||||
)
|
||||
except Exception as e:
|
||||
except OSError as e:
|
||||
logging.debug(f"Game {game_id}: could not write PGN: {e}")
|
||||
logging.info(f"Ending game thread for {game_id}")
|
||||
|
||||
@ -456,7 +457,7 @@ def run_bot(log_level: str = "INFO", *, decline_correspondence: bool = False) ->
|
||||
"""
|
||||
try:
|
||||
_process_event_stream()
|
||||
except Exception as e:
|
||||
except requests.RequestException as e:
|
||||
logging.warning(f"Event stream error: {e}")
|
||||
return backoff_sleep(backoff)
|
||||
else:
|
||||
|
||||
@ -11,7 +11,7 @@ if ROOT not in sys.path:
|
||||
sys.path.insert(0, ROOT)
|
||||
|
||||
|
||||
def pytest_ignore_collect(collection_path: Path, _config: pytest.Config) -> bool | None:
|
||||
def pytest_ignore_collect(collection_path: Path, _: pytest.Config) -> bool | None:
|
||||
"""Ignore per-game blunder test files; keep only the unified one.
|
||||
|
||||
This lets us keep historical files in the repo without collecting them.
|
||||
|
||||
@ -8,6 +8,8 @@ import pytest
|
||||
|
||||
from python_pkg.lichess_bot.engine import RandomEngine
|
||||
|
||||
_PUZZLE_CSV = os.path.join(os.path.dirname(__file__), "lichess_db_puzzle.csv")
|
||||
|
||||
|
||||
def _load_top_puzzles(csv_path: str, limit: int = 8) -> list[tuple[str, str]]:
|
||||
"""Return a list of (FEN, solution_moves_str) for the first `limit` rows in the CSV.
|
||||
@ -15,6 +17,8 @@ def _load_top_puzzles(csv_path: str, limit: int = 8) -> list[tuple[str, str]]:
|
||||
CSV columns: PuzzleId,FEN,Moves,...
|
||||
"""
|
||||
puzzles: list[tuple[str, str]] = []
|
||||
if not os.path.isfile(csv_path):
|
||||
return puzzles
|
||||
with open(csv_path, newline="", encoding="utf-8") as f:
|
||||
reader = csv.DictReader(f)
|
||||
for row in reader:
|
||||
@ -27,11 +31,13 @@ def _load_top_puzzles(csv_path: str, limit: int = 8) -> list[tuple[str, str]]:
|
||||
return puzzles
|
||||
|
||||
|
||||
_PUZZLES = _load_top_puzzles(_PUZZLE_CSV, limit=8)
|
||||
|
||||
|
||||
@pytest.mark.skipif(not _PUZZLES, reason="Puzzle CSV not found")
|
||||
@pytest.mark.parametrize(
|
||||
("fen", "moves_str"),
|
||||
_load_top_puzzles(
|
||||
os.path.join(os.path.dirname(__file__), "lichess_db_puzzle.csv"), limit=8
|
||||
),
|
||||
_PUZZLES or [("rnbqkbnr/pppppppp/8/8/8/8/PPPPPPPP/RNBQKBNR w KQkq - 0 1", "e2e4")],
|
||||
)
|
||||
def test_puzzle_engine_follow_solution(fen: str, moves_str: str) -> None:
|
||||
"""Verify the engine follows puzzle solutions correctly."""
|
||||
|
||||
@ -161,7 +161,7 @@ def fen_and_uci_for_blunders(
|
||||
if bl.ply - 1 < len(main_sans):
|
||||
try:
|
||||
move = board.parse_san(main_sans[bl.ply - 1])
|
||||
except Exception:
|
||||
except ValueError:
|
||||
logging.debug("Skipping blunder: failed to parse fallback move")
|
||||
continue
|
||||
else:
|
||||
@ -171,7 +171,7 @@ def fen_and_uci_for_blunders(
|
||||
try:
|
||||
best_move = board.parse_san(bl.best_suggestion_san)
|
||||
best_uci = best_move.uci()
|
||||
except Exception as e:
|
||||
except ValueError as e:
|
||||
msg = (
|
||||
f"Failed to parse best_suggestion SAN "
|
||||
f"'{bl.best_suggestion_san}' at ply {bl.ply} "
|
||||
|
||||
@ -28,7 +28,7 @@ def get_and_increment_version() -> int:
|
||||
raw = f.read().strip()
|
||||
if raw:
|
||||
current = int(raw)
|
||||
except Exception:
|
||||
except (OSError, ValueError):
|
||||
# Missing or unreadable file -> treat as version 0
|
||||
current = 0
|
||||
|
||||
@ -38,12 +38,12 @@ def get_and_increment_version() -> int:
|
||||
with open(tmp_path, "w") as f:
|
||||
f.write(str(new_version))
|
||||
os.replace(tmp_path, path)
|
||||
except Exception:
|
||||
except OSError:
|
||||
# As a fallback, try a direct write; failure is non-fatal to bot operation
|
||||
try:
|
||||
with open(path, "w") as f:
|
||||
f.write(str(new_version))
|
||||
except Exception:
|
||||
except OSError:
|
||||
logging.debug("Could not persist bot version to %s", path)
|
||||
|
||||
return new_version
|
||||
|
||||
@ -7,6 +7,7 @@ from urllib.parse import urlparse
|
||||
|
||||
import requests
|
||||
from selenium import webdriver
|
||||
from selenium.common.exceptions import NoSuchElementException
|
||||
from selenium.webdriver.common.by import By
|
||||
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
@ -73,7 +74,7 @@ while True:
|
||||
next_button_url = next_button.get_attribute("href")
|
||||
driver.get(next_button_url)
|
||||
|
||||
except Exception:
|
||||
except NoSuchElementException:
|
||||
# If the 'Next' button is not found, it means we've reached the last image
|
||||
logging.info("No 'Next' button found. Reached the end of images.")
|
||||
break
|
||||
|
||||
@ -32,14 +32,14 @@ import sys
|
||||
|
||||
try:
|
||||
import psutil # type: ignore[import-untyped]
|
||||
except Exception: # pragma: no cover - optional dependency; we fall back if unavailable
|
||||
except ImportError: # pragma: no cover
|
||||
psutil = None # type: ignore[assignment]
|
||||
|
||||
try:
|
||||
import chess
|
||||
import chess.engine
|
||||
import chess.pgn
|
||||
except Exception: # pragma: no cover
|
||||
except ImportError: # pragma: no cover
|
||||
logging.exception("Missing dependency. Please install python-chess:")
|
||||
logging.exception(" pip install -r python_pkg/stockfish_analysis/requirements.txt")
|
||||
raise
|
||||
@ -204,7 +204,7 @@ def _auto_hash_mb(threads_wanted: int, engine_options: dict[str, object]) -> int
|
||||
max_allowed = None
|
||||
try:
|
||||
max_allowed = opt.max if opt is not None else None # type: ignore[attr-defined]
|
||||
except Exception:
|
||||
except AttributeError:
|
||||
max_allowed = None
|
||||
if isinstance(max_allowed, int):
|
||||
target = min(target, max_allowed)
|
||||
@ -300,7 +300,7 @@ def main() -> None:
|
||||
# Configure engine performance options if available
|
||||
try:
|
||||
options = engine.options # type: ignore[attr-defined]
|
||||
except Exception:
|
||||
except AttributeError:
|
||||
options = {}
|
||||
|
||||
# Threads
|
||||
@ -317,7 +317,7 @@ def main() -> None:
|
||||
if isinstance(min_thr, int):
|
||||
wanted_threads = max(wanted_threads, min_thr)
|
||||
engine.configure({"Threads": int(wanted_threads)})
|
||||
except Exception:
|
||||
except (AttributeError, TypeError, ValueError):
|
||||
logging.debug("Failed to configure Threads option")
|
||||
|
||||
# Configure hash table size in MB.
|
||||
@ -335,7 +335,7 @@ def main() -> None:
|
||||
if isinstance(min_hash, int):
|
||||
target_hash = max(target_hash, min_hash)
|
||||
engine.configure({"Hash": int(target_hash)})
|
||||
except Exception:
|
||||
except (AttributeError, TypeError, ValueError):
|
||||
logging.debug("Failed to configure Hash option")
|
||||
|
||||
# MultiPV
|
||||
@ -346,7 +346,7 @@ def main() -> None:
|
||||
if isinstance(max_mpv, int):
|
||||
effective_mpv = min(effective_mpv, max_mpv)
|
||||
engine.configure({"MultiPV": int(effective_mpv)})
|
||||
except Exception:
|
||||
except (AttributeError, TypeError, ValueError):
|
||||
logging.debug("Failed to configure MultiPV option")
|
||||
|
||||
# Enable NNUE if the option exists
|
||||
@ -374,7 +374,7 @@ def main() -> None:
|
||||
# Brief performance summary (best-effort)
|
||||
try:
|
||||
thr_show = int(wanted_threads)
|
||||
except Exception:
|
||||
except (ValueError, TypeError):
|
||||
thr_show = 1
|
||||
try:
|
||||
hash_show = (
|
||||
@ -382,7 +382,7 @@ def main() -> None:
|
||||
if hasattr(engine, "options") and engine.options.get("Hash")
|
||||
else None
|
||||
)
|
||||
except Exception:
|
||||
except (AttributeError, TypeError, ValueError):
|
||||
hash_show = None
|
||||
if hash_show is not None:
|
||||
logging.info(
|
||||
|
||||
Loading…
Reference in New Issue
Block a user