fix: resolve all 28 mypy type errors + re-enable coverage in CI
All checks were successful
CI / lint (push) Successful in 11s
CI / security (push) Successful in 7s
CI / test (push) Successful in 13s
CI / build (push) Successful in 1s

This commit is contained in:
Bruno Charest 2026-05-28 12:57:30 -04:00
parent 7096050da5
commit 1a14927f36
8 changed files with 29 additions and 27 deletions

View File

@ -50,14 +50,10 @@ jobs:
pip install -r backend/requirements.txt
- name: Run tests
run: pytest tests/ --cov=backend --cov-report=xml --cov-report=term -q
- name: Upload coverage
if: false # Disabled — needs external GitHub actions (not available on self-hosted runner)
uses: actions/upload-artifact@v4
with:
name: coverage-report
path: coverage.xml
run: pytest tests/ --cov=backend --cov-report=term -q
# Coverage report is printed in console (--cov-report=term).
# XML artifact upload (actions/upload-artifact@v4) not available
# on self-hosted Gitea runner without GitHub internet access.
# ── Security scan ─────────────────────────────────────────────────
security:

View File

@ -1,7 +1,7 @@
import asyncio
import logging
from pathlib import Path
from typing import Dict, List, Optional
from typing import Any, Dict, List, Optional
import threading
logger = logging.getLogger("obsigate.attachment_indexer")
@ -34,7 +34,7 @@ def clear_resolution_cache(vault_name: Optional[str] = None) -> None:
del _resolution_cache[key]
def _scan_vault_attachments(vault_name: str, vault_path: str, vault_cfg: dict = None) -> Dict[str, List[Path]]:
def _scan_vault_attachments(vault_name: str, vault_path: str, vault_cfg: dict | None = None) -> Dict[str, List[Path]]:
"""Synchronously scan a vault directory for image attachments.
Walks the vault tree and builds a filename -> absolute path mapping
@ -84,7 +84,7 @@ def _scan_vault_attachments(vault_name: str, vault_path: str, vault_cfg: dict =
return index
async def build_attachment_index(vault_config: Dict[str, Dict[str, any]]) -> None:
async def build_attachment_index(vault_config: Dict[str, Dict[str, Any]]) -> None:
"""Build the attachment index for all configured vaults.
Runs vault scans concurrently in a thread pool, then performs
@ -103,7 +103,7 @@ async def build_attachment_index(vault_config: Dict[str, Dict[str, any]]) -> Non
loop = asyncio.get_event_loop()
new_index: Dict[str, Dict[str, List[Path]]] = {}
tasks = []
tasks: list[tuple[str, asyncio.Future[Dict[str, List[Path]]]]] = []
for name, config in vault_config.items():
vault_path = config.get("path")
if not vault_path:
@ -132,7 +132,7 @@ async def build_attachment_index(vault_config: Dict[str, Dict[str, any]]) -> Non
logger.info(f"Attachment index built: {len(attachment_index)} vaults, {total_attachments} total attachments")
async def rescan_vault_attachments(vault_name: str, vault_path: str, vault_cfg: dict = None) -> int:
async def rescan_vault_attachments(vault_name: str, vault_path: str, vault_cfg: dict | None = None) -> int:
"""Rescan attachments for a single vault.
Args:

View File

@ -250,6 +250,7 @@ async def change_password(
):
"""Change own password."""
user = get_user(current_user["username"])
assert user is not None, f"User {current_user['username']} not found"
if not verify_password(req.current_password, user["password_hash"]):
raise HTTPException(400, "Mot de passe actuel incorrect")
update_user(current_user["username"], {"password": req.new_password})

View File

@ -5,7 +5,7 @@ import re
import threading
from pathlib import Path
from datetime import datetime, timezone
from typing import Dict, List, Optional, Any
from typing import Any, Callable, Dict, List, Optional
import frontmatter
@ -21,14 +21,14 @@ vault_config: Dict[str, Dict[str, Any]] = {}
_index_lock = threading.Lock()
# Async lock for partial index updates (coexists with threading lock)
_async_index_lock: asyncio.Lock = None # initialized lazily
_async_index_lock: asyncio.Lock | None = None # initialized lazily
# Generation counter — incremented on each index rebuild so consumers
# (e.g. the inverted index in search.py) can detect staleness.
_index_generation: int = 0
# Hook for incremental inverted index updates: called as (action, vault, path, file_info)
_on_index_change: callable = None
_on_index_change: Callable[..., None] | None = None
def set_index_change_hook(hook):
@ -615,7 +615,7 @@ def _remove_file_from_structures(vault_name: str, rel_path: str) -> Optional[Dic
# Notify inverted index for incremental update
if _on_index_change:
_on_index_change('remove', vault_name, rel_path, removed)
_on_index_change('remove', vault_name, rel_path, removed) # type: ignore[misc]
return removed
@ -686,7 +686,7 @@ def _add_file_to_structures(vault_name: str, file_info: Dict[str, Any]):
# Notify inverted index for incremental update
if _on_index_change:
_on_index_change('add', vault_name, file_info["path"], file_info)
_on_index_change('add', vault_name, file_info["path"], file_info) # type: ignore[misc]
async def update_single_file(vault_name: str, abs_file_path: str) -> Optional[Dict[str, Any]]:

View File

@ -583,8 +583,8 @@ from backend.audit import log_file_save, log_file_delete # noqa: E402
try:
from backend.pdf_export import generate_pdf, build_pdf_html # noqa: E402
except OSError:
generate_pdf = None
build_pdf_html = None
generate_pdf = None # type: ignore[assignment]
build_pdf_html = None # type: ignore[assignment]
import logging
logging.getLogger("obsigate").warning("PDF export unavailable (WeasyPrint/GTK not found)")
from backend.share import create_share, get_share_by_token, record_access, revoke_share, list_shares # noqa: E402
@ -1026,7 +1026,7 @@ async def api_toggle_bookmark(req: BookmarkToggleRequest, current_user=Depends(r
if not check_vault_access(req.vault, current_user):
raise HTTPException(status_code=403, detail="Access denied to vault")
is_now_bookmarked = toggle_bookmark(username, req.vault, req.path, req.title)
is_now_bookmarked = toggle_bookmark(username, req.vault, req.path, req.title or "")
# Update the file's YAML frontmatter: favoris: true/false
vault_data = get_vault_data(req.vault)
@ -3063,6 +3063,10 @@ async def api_conflict_resolve(body: dict = Body(...), current_user=Depends(requ
conflict_path = body.get("conflict_path")
original_path = body.get("original_path")
action = body.get("action") # "keep_local" or "keep_conflict"
# mypy: narrow down from dict values
assert isinstance(vault_name, str), "'vault' is required and must be a string"
assert isinstance(conflict_path, str), "'conflict_path' is required and must be a string"
assert isinstance(original_path, str), "'original_path' is required and must be a string"
if not check_vault_access(vault_name, current_user):
raise HTTPException(403, f"Accès refusé à la vault '{vault_name}'")
vault_data = get_vault_data(vault_name)

View File

@ -73,7 +73,7 @@ def is_rate_limited(ip: str) -> bool:
return failures >= MAX_ATTEMPTS
def get_status(ip: str = None) -> dict:
def get_status(ip: str | None = None) -> dict:
"""Get rate limit status for an IP (for diagnostics)."""
_cleanup_expired()
if ip:

View File

@ -63,7 +63,7 @@ def redact(text: str) -> tuple:
if callable(replacement):
new_result, n = pattern.subn(replacement, result)
else:
new_result, n = pattern.subn(replacement, result)
new_result, n = pattern.subn(str(replacement), result)
count += n
result = new_result
if count > 0:

View File

@ -54,7 +54,7 @@ class VaultEventHandler(FileSystemEventHandler):
basename_lower = p.name.lower()
return suffix in SUPPORTED_EXTENSIONS or basename_lower in ("dockerfile", "makefile", "cmakelists.txt")
def _enqueue(self, event_type: str, src: str, dest: str = None):
def _enqueue(self, event_type: str, src: str, dest: str | None = None):
"""Thread-safe : envoyer l'événement vers l'event loop asyncio."""
event = {
'type': event_type,
@ -103,7 +103,7 @@ class VaultWatcher:
self.debounce_seconds = debounce_seconds
self.use_polling = use_polling
self.polling_interval = polling_interval
self.observers: Dict[str, Observer] = {}
self.observers: dict[str, object] = {}
self.event_queue: asyncio.Queue = asyncio.Queue()
self._processor_task: Optional[asyncio.Task] = None
self._running = False
@ -171,9 +171,10 @@ class VaultWatcher:
async def remove_vault(self, vault_name: str):
"""Arrêter la surveillance d'une vault."""
if vault_name in self.observers:
observer = self.observers[vault_name]
try:
self.observers[vault_name].stop()
self.observers[vault_name].join(timeout=5)
observer.stop() # type: ignore[attr-defined]
observer.join(timeout=5) # type: ignore[attr-defined]
except Exception as e:
logger.warning(f"Error stopping observer for '{vault_name}': {e}")
del self.observers[vault_name]