fix(swig): detect and recover from BOARD proxy dehydration (#173)
KiCAD nightly builds occasionally return a SwigPyObject from pcbnew.LoadBoard, and SaveBoard can leave self.board with no method dispatch as a side-effect of certain sequences (delete_trace + auto-save is the one users have hit in the wild). Before this fix, open_project would keep reporting "Opened project: foo.kicad_pcb" while every subsequent board operation failed with AttributeError on GetDesignSettings / GetBoardEdgesBoundingBox / GetCurrentViaSize, with no path to recovery short of restarting the MCP server. Add two helpers in KiCADInterface: * _is_board_healthy(board=None) probes for stable BOARD methods (GetDesignSettings, GetBoardEdgesBoundingBox, GetFileName) — these are missing on a dehydrated SwigPyObject, so hasattr() catches the state without segfaulting. * _safe_load_board(path) wraps pcbnew.LoadBoard, checks health, and on dehydration reloads the pcbnew module via importlib.reload and retries once. Returns None when recovery is impossible so callers surface real failure rather than fake success. Wire the helpers in: * handle_command's open_project / create_project path validates the loaded board and either recovers (with a warnings[] entry) or returns success=False with an explicit "restart the MCP server" errorDetails — never claims success when the board is unusable. * _auto_save_board now detects dehydration introduced by SaveBoard itself and reloads from disk so the next command sees a usable proxy. This is the post-delete_trace failure mode users hit. * _handle_place_component, _handle_sync_schematic_to_board, _handle_import_svg_logo and _handle_refill_zones all go through _safe_load_board instead of bare LoadBoard, surfacing real errors consistently. Also fix two adjacent issues observed in the same incident: * _handle_check_kicad_ui used to call manager.is_running() and manager.get_process_info() separately, with different detection methods. They could disagree, producing the confusing running=True, processes=[] state users hit after manually quitting KiCAD. processes is now the single source of truth and running is derived from len(). * run_drc accepts a timeoutSec param (default 600s, clamped to [10, 1800]) so callers with smaller MCP transport budgets can bound the kicad-cli subprocess. Same timeout is applied to the optional report-generation subprocess. Error message names the actual timeout that fired. Tests: tests/test_swig_dehydration.py adds 17 unit tests covering detection, recovery, the open_project surfacing path, the auto-save post-recovery path, the check_kicad_ui consistency, and the run_drc timeout clamping. Full suite: same 12 pre-existing failures both before and after this change, +17 new tests passing. Note: the SWIG dehydration is fundamentally a pcbnew memory bug exposed by repeated LoadBoard calls in a single Python process; this PR is a defensive recovery layer, not a fix to the underlying binding. Complementary to PR #151 (auto-save-guard), which expands the LoadBoard call rate by refusing saves on external file change and forcing re-open_project cycles.
This commit is contained in:
@@ -187,6 +187,13 @@ class DesignRuleCommands:
|
||||
}
|
||||
|
||||
report_path = params.get("reportPath")
|
||||
# Caller-overridable timeout (seconds). Defaults to 600s for big boards
|
||||
# but smaller MCP transport budgets (e.g. 120s) can lower it explicitly.
|
||||
try:
|
||||
timeout_sec = int(params.get("timeoutSec", 600))
|
||||
except (TypeError, ValueError):
|
||||
timeout_sec = 600
|
||||
timeout_sec = max(10, min(timeout_sec, 1800)) # clamp to [10, 1800]
|
||||
|
||||
# Get the board file path
|
||||
board_file = self.board.GetFileName()
|
||||
@@ -225,14 +232,14 @@ class DesignRuleCommands:
|
||||
board_file,
|
||||
]
|
||||
|
||||
logger.info(f"Running DRC command: {' '.join(cmd)}")
|
||||
logger.info(f"Running DRC command (timeout={timeout_sec}s): {' '.join(cmd)}")
|
||||
|
||||
# Run DRC
|
||||
# Run DRC. subprocess.run kills the child on TimeoutExpired.
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=600, # 10 minute timeout for large boards (21MB PCB needs time)
|
||||
timeout=timeout_sec,
|
||||
)
|
||||
|
||||
if result.returncode != 0:
|
||||
@@ -318,7 +325,7 @@ class DesignRuleCommands:
|
||||
"mm",
|
||||
board_file,
|
||||
]
|
||||
subprocess.run(cmd_report, capture_output=True, timeout=600)
|
||||
subprocess.run(cmd_report, capture_output=True, timeout=timeout_sec)
|
||||
|
||||
# Return summary only (not full violations list)
|
||||
return {
|
||||
@@ -339,11 +346,14 @@ class DesignRuleCommands:
|
||||
os.unlink(json_output)
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.error("DRC command timed out")
|
||||
logger.error(f"DRC command timed out after {timeout_sec}s")
|
||||
return {
|
||||
"success": False,
|
||||
"message": "DRC command timed out",
|
||||
"errorDetails": "Command took longer than 600 seconds (10 minutes)",
|
||||
"errorDetails": (
|
||||
f"Command took longer than {timeout_sec} seconds; "
|
||||
"raise timeoutSec param for very large boards"
|
||||
),
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error running DRC: {str(e)}")
|
||||
|
||||
@@ -567,6 +567,47 @@ class KiCADInterface:
|
||||
logger.info("Updating board reference...")
|
||||
# Get board from the project commands handler
|
||||
self.board = self.project_commands.board
|
||||
|
||||
# Detect SWIG dehydration before claiming success.
|
||||
# Without this, every later board op sees a raw
|
||||
# SwigPyObject and raises AttributeError, while the
|
||||
# MCP keeps reporting "Opened project" — the exact
|
||||
# symptom users hit on KiCAD nightlies.
|
||||
if not self._is_board_healthy():
|
||||
board_path = (result.get("project") or {}).get("boardPath")
|
||||
recovered = None
|
||||
if board_path:
|
||||
logger.warning(
|
||||
"Board after %s is SWIG-dehydrated; attempting recovery",
|
||||
command,
|
||||
)
|
||||
recovered = self._safe_load_board(board_path)
|
||||
if recovered is not None:
|
||||
self.board = recovered
|
||||
self.project_commands.board = recovered
|
||||
result.setdefault("warnings", []).append(
|
||||
"SWIG board proxy was dehydrated on load; "
|
||||
"recovered via pcbnew module reload"
|
||||
)
|
||||
else:
|
||||
# Surface the truth — never claim success when
|
||||
# the board is unusable.
|
||||
return {
|
||||
"success": False,
|
||||
"message": (
|
||||
f"{command} loaded the board but the SWIG "
|
||||
"proxy is dehydrated and recovery failed"
|
||||
),
|
||||
"errorDetails": (
|
||||
"pcbnew.LoadBoard returned a BOARD whose "
|
||||
"method dispatch is missing (raw SwigPyObject). "
|
||||
"This indicates SWIG state corruption in the "
|
||||
"current Python process — restart the MCP "
|
||||
"server to recover."
|
||||
),
|
||||
"_backend": "swig",
|
||||
"_realtime": False,
|
||||
}
|
||||
self._update_command_handlers()
|
||||
# Record the file's signature so subsequent auto-saves
|
||||
# can detect external modifications and refuse to
|
||||
@@ -694,6 +735,9 @@ class KiCADInterface:
|
||||
(rotating, keeps the most recent `_auto_save_backup_keep`),
|
||||
then call pcbnew.SaveBoard().
|
||||
* Update the recorded signature on success.
|
||||
* If SaveBoard leaves the in-memory BOARD dehydrated (observed on
|
||||
KiCAD nightlies after delete_trace + auto-save), reload from disk
|
||||
so the next command sees a usable proxy instead of a SwigPyObject.
|
||||
|
||||
Returns a status dict that handle_command merges into the caller's
|
||||
response so warnings about refused saves are visible:
|
||||
@@ -772,11 +816,30 @@ class KiCADInterface:
|
||||
pcbnew.SaveBoard(board_path, self.board)
|
||||
logger.debug(f"Auto-saved board to: {board_path}")
|
||||
self._board_disk_signature = self._disk_signature(board_path)
|
||||
return {"saved": True, "boardPath": board_path, "backup": backup_path}
|
||||
except Exception as e:
|
||||
logger.warning(f"Auto-save failed: {e}")
|
||||
return {"saved": False, "error": str(e), "backup": backup_path}
|
||||
|
||||
# Post-save dehydration check. If the BOARD lost its bindings during
|
||||
# save, reload from disk while we still know the path. board_path is
|
||||
# guaranteed non-empty here (we returned early above otherwise).
|
||||
if not self._is_board_healthy():
|
||||
logger.warning(
|
||||
"Board became dehydrated during auto-save; reloading from %s",
|
||||
board_path,
|
||||
)
|
||||
recovered = self._safe_load_board(board_path)
|
||||
if recovered is not None:
|
||||
self.board = recovered
|
||||
self._update_command_handlers()
|
||||
else:
|
||||
logger.error(
|
||||
"Board dehydration after auto-save is unrecoverable — "
|
||||
"subsequent commands will fail until MCP restart"
|
||||
)
|
||||
|
||||
return {"saved": True, "boardPath": board_path, "backup": backup_path}
|
||||
|
||||
def _update_command_handlers(self) -> None:
|
||||
"""Update board reference in all command handlers"""
|
||||
logger.debug("Updating board reference in command handlers")
|
||||
@@ -788,6 +851,70 @@ class KiCADInterface:
|
||||
self.export_commands.board = self.board
|
||||
self.freerouting_commands.board = self.board
|
||||
|
||||
# Stable BOARD methods used to detect SWIG dehydration. Newer KiCAD nightly
|
||||
# builds occasionally return a raw SwigPyObject from pcbnew.LoadBoard after
|
||||
# certain mutating sequences (delete_trace, refill_zones, …) — the proxy
|
||||
# type-checks but every method access raises AttributeError. Probing for
|
||||
# these methods catches that state without segfaulting.
|
||||
_BOARD_HEALTH_METHODS = (
|
||||
"GetDesignSettings",
|
||||
"GetBoardEdgesBoundingBox",
|
||||
"GetFileName",
|
||||
)
|
||||
|
||||
def _is_board_healthy(self, board: Optional[Any] = None) -> bool:
|
||||
"""Return True if the board (default self.board) has live SWIG dispatch."""
|
||||
target = board if board is not None else self.board
|
||||
if target is None:
|
||||
return False
|
||||
return all(hasattr(target, m) for m in self._BOARD_HEALTH_METHODS)
|
||||
|
||||
def _safe_load_board(self, path: str) -> Optional[Any]:
|
||||
"""Load a board from disk, recovering from SWIG dehydration if pcbnew is broken.
|
||||
|
||||
If pcbnew.LoadBoard returns a dehydrated proxy, reload the pcbnew
|
||||
module once and retry. Returns the new board, or None if recovery
|
||||
is impossible (caller must surface a real failure rather than fake
|
||||
success).
|
||||
"""
|
||||
global pcbnew
|
||||
try:
|
||||
board = pcbnew.LoadBoard(path)
|
||||
except Exception as e:
|
||||
logger.error(f"LoadBoard({path!r}) raised: {e}")
|
||||
return None
|
||||
|
||||
if self._is_board_healthy(board):
|
||||
return board
|
||||
|
||||
logger.warning(
|
||||
f"LoadBoard({path!r}) returned a dehydrated SWIG proxy; "
|
||||
"reloading pcbnew module and retrying"
|
||||
)
|
||||
try:
|
||||
import importlib
|
||||
|
||||
pcbnew = importlib.reload(pcbnew)
|
||||
except Exception as e:
|
||||
logger.error(f"pcbnew module reload failed: {e}")
|
||||
return None
|
||||
|
||||
try:
|
||||
board = pcbnew.LoadBoard(path)
|
||||
except Exception as e:
|
||||
logger.error(f"LoadBoard retry after pcbnew reload failed: {e}")
|
||||
return None
|
||||
|
||||
if not self._is_board_healthy(board):
|
||||
logger.error(
|
||||
"Board still dehydrated after pcbnew reload; SWIG state is "
|
||||
"unrecoverable in this process — restart the MCP server"
|
||||
)
|
||||
return None
|
||||
|
||||
logger.info("Recovered from SWIG dehydration via pcbnew reload")
|
||||
return board
|
||||
|
||||
# Schematic command handlers
|
||||
def _handle_create_schematic(self, params: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Create a new schematic"""
|
||||
@@ -921,17 +1048,19 @@ class KiCADInterface:
|
||||
current_board_file = str(Path(self.board.GetFileName()).resolve()) if self.board else ""
|
||||
if board_path_norm != current_board_file:
|
||||
logger.info(f"boardPath differs from current board — reloading: {board_path}")
|
||||
try:
|
||||
self.board = pcbnew.LoadBoard(board_path)
|
||||
self._update_command_handlers()
|
||||
logger.info("Board reloaded from boardPath")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to reload board from boardPath: {e}")
|
||||
reloaded = self._safe_load_board(board_path)
|
||||
if reloaded is None:
|
||||
return {
|
||||
"success": False,
|
||||
"message": f"Could not load board from boardPath: {board_path}",
|
||||
"errorDetails": str(e),
|
||||
"errorDetails": (
|
||||
"pcbnew.LoadBoard failed or returned a dehydrated "
|
||||
"SWIG proxy that could not be recovered"
|
||||
),
|
||||
}
|
||||
self.board = reloaded
|
||||
self._update_command_handlers()
|
||||
logger.info("Board reloaded from boardPath")
|
||||
|
||||
project_path = Path(board_path).parent
|
||||
if project_path != getattr(self, "_current_project_path", None):
|
||||
@@ -4197,7 +4326,16 @@ class KiCADInterface:
|
||||
# Determine board to work with
|
||||
board = None
|
||||
if board_path:
|
||||
board = pcbnew.LoadBoard(board_path)
|
||||
board = self._safe_load_board(board_path)
|
||||
if board is None:
|
||||
return {
|
||||
"success": False,
|
||||
"message": f"Could not load board from {board_path}",
|
||||
"errorDetails": (
|
||||
"pcbnew.LoadBoard failed or returned a dehydrated "
|
||||
"SWIG proxy that could not be recovered"
|
||||
),
|
||||
}
|
||||
elif self.board:
|
||||
board = self.board
|
||||
board_path = board.GetFileName() if not board_path else board_path
|
||||
@@ -4764,14 +4902,15 @@ class KiCADInterface:
|
||||
# call would overwrite the file with the stale in-memory state, erasing the
|
||||
# logo. Reload the board from disk so pcbnew's memory matches the file.
|
||||
if result.get("success") and self.board:
|
||||
try:
|
||||
self.board = pcbnew.LoadBoard(pcb_path)
|
||||
# Propagate updated board reference to all command handlers
|
||||
reloaded = self._safe_load_board(pcb_path)
|
||||
if reloaded is not None:
|
||||
self.board = reloaded
|
||||
self._update_command_handlers()
|
||||
logger.info("Reloaded board into pcbnew after SVG logo import")
|
||||
except Exception as reload_err:
|
||||
else:
|
||||
logger.warning(
|
||||
f"Board reload after SVG import failed (non-fatal): {reload_err}"
|
||||
"Board reload after SVG import failed (non-fatal); "
|
||||
"next mutation may operate on stale in-memory state"
|
||||
)
|
||||
|
||||
return result
|
||||
@@ -4868,12 +5007,19 @@ class KiCADInterface:
|
||||
return {"success": False, "message": str(e)}
|
||||
|
||||
def _handle_check_kicad_ui(self, params: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Check if KiCAD UI is running"""
|
||||
"""Check if KiCAD UI is running.
|
||||
|
||||
`processes` is the single source of truth — `running` is derived from
|
||||
its length so the two fields cannot disagree. Previously they came
|
||||
from separate detection methods (pgrep regex vs. ps-aux substring) and
|
||||
could race or use different filters, producing the confusing
|
||||
`running=True, processes=[]` state users hit after quitting KiCAD.
|
||||
"""
|
||||
logger.info("Checking if KiCAD UI is running")
|
||||
try:
|
||||
manager = KiCADProcessManager()
|
||||
is_running = manager.is_running()
|
||||
processes = manager.get_process_info() if is_running else []
|
||||
processes = manager.get_process_info()
|
||||
is_running = len(processes) > 0
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
@@ -4956,7 +5102,18 @@ print("ok")
|
||||
)
|
||||
if result.returncode == 0 and "ok" in result.stdout:
|
||||
# Reload board after subprocess modified it
|
||||
self.board = pcbnew.LoadBoard(board_path)
|
||||
reloaded = self._safe_load_board(board_path)
|
||||
if reloaded is None:
|
||||
return {
|
||||
"success": False,
|
||||
"message": (
|
||||
"Zone fill subprocess succeeded but the board "
|
||||
"could not be reloaded into pcbnew (SWIG state "
|
||||
"is corrupt — restart the MCP server)"
|
||||
),
|
||||
"zoneCount": zone_count,
|
||||
}
|
||||
self.board = reloaded
|
||||
self._update_command_handlers()
|
||||
logger.info("Zone fill subprocess succeeded")
|
||||
return {
|
||||
|
||||
376
tests/test_swig_dehydration.py
Normal file
376
tests/test_swig_dehydration.py
Normal file
@@ -0,0 +1,376 @@
|
||||
"""
|
||||
Tests for SWIG board-dehydration detection and recovery in KiCADInterface.
|
||||
|
||||
Reproduces the failure mode users hit on KiCAD nightly builds: after sequences
|
||||
like delete_trace + add_via, pcbnew.LoadBoard returns a "dehydrated" BOARD —
|
||||
a SWIG proxy whose method dispatch table is missing, so every method access
|
||||
raises AttributeError. Without recovery, every subsequent call fails and
|
||||
open_project keeps reporting fake success because LoadBoard didn't raise.
|
||||
|
||||
Also covers the check_kicad_ui consistency fix where running and processes
|
||||
could disagree.
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent / "python"))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures: stand-ins for SWIG BOARD proxies
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_dehydrated_board() -> Any:
|
||||
"""A bare object simulating the post-dehydration SwigPyObject — no methods."""
|
||||
|
||||
class _Dehydrated:
|
||||
# Intentionally empty: hasattr(...) for the health-check methods
|
||||
# returns False, matching the symptom users hit.
|
||||
pass
|
||||
|
||||
return _Dehydrated()
|
||||
|
||||
|
||||
def _make_healthy_board() -> Any:
|
||||
"""A MagicMock that satisfies _is_board_healthy."""
|
||||
board = MagicMock(name="HealthyBoard")
|
||||
board.GetDesignSettings = MagicMock(return_value=MagicMock())
|
||||
board.GetBoardEdgesBoundingBox = MagicMock(return_value=MagicMock())
|
||||
board.GetFileName = MagicMock(return_value="/tmp/test.kicad_pcb")
|
||||
return board
|
||||
|
||||
|
||||
def _make_iface() -> Any:
|
||||
"""Build a KiCADInterface skipping __init__ — same pattern as test_erc_handler.
|
||||
|
||||
handle_command reads `use_ipc` and `ipc_board_api` early; set them so a
|
||||
test exercising the routing layer doesn't crash on the IPC short-circuit.
|
||||
"""
|
||||
with patch("kicad_interface.USE_IPC_BACKEND", False):
|
||||
from kicad_interface import KiCADInterface
|
||||
|
||||
iface = KiCADInterface.__new__(KiCADInterface)
|
||||
iface.use_ipc = False
|
||||
iface.ipc_board_api = None
|
||||
return iface
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _is_board_healthy
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_is_board_healthy_returns_false_for_none():
|
||||
iface = _make_iface()
|
||||
iface.board = None
|
||||
assert iface._is_board_healthy() is False
|
||||
|
||||
|
||||
def test_is_board_healthy_returns_false_for_dehydrated_proxy():
|
||||
iface = _make_iface()
|
||||
iface.board = _make_dehydrated_board()
|
||||
assert iface._is_board_healthy() is False
|
||||
|
||||
|
||||
def test_is_board_healthy_returns_true_for_real_board():
|
||||
iface = _make_iface()
|
||||
iface.board = _make_healthy_board()
|
||||
assert iface._is_board_healthy() is True
|
||||
|
||||
|
||||
def test_is_board_healthy_accepts_explicit_target():
|
||||
iface = _make_iface()
|
||||
iface.board = None # self.board is broken
|
||||
healthy = _make_healthy_board()
|
||||
# Explicit argument bypasses self.board
|
||||
assert iface._is_board_healthy(healthy) is True
|
||||
assert iface._is_board_healthy(_make_dehydrated_board()) is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _safe_load_board
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_safe_load_board_returns_loaded_board_when_healthy():
|
||||
iface = _make_iface()
|
||||
healthy = _make_healthy_board()
|
||||
with patch("kicad_interface.pcbnew") as mock_pcbnew:
|
||||
mock_pcbnew.LoadBoard = MagicMock(return_value=healthy)
|
||||
result = iface._safe_load_board("/tmp/x.kicad_pcb")
|
||||
assert result is healthy
|
||||
|
||||
|
||||
def test_safe_load_board_retries_after_pcbnew_reload_when_dehydrated():
|
||||
iface = _make_iface()
|
||||
dehydrated = _make_dehydrated_board()
|
||||
healthy = _make_healthy_board()
|
||||
|
||||
# First LoadBoard returns dehydrated → reload pcbnew → second LoadBoard returns healthy
|
||||
with patch("kicad_interface.pcbnew") as mock_pcbnew, patch("importlib.reload") as mock_reload:
|
||||
mock_pcbnew.LoadBoard = MagicMock(side_effect=[dehydrated, healthy])
|
||||
mock_reload.return_value = mock_pcbnew # reload returns the (mock) module
|
||||
result = iface._safe_load_board("/tmp/x.kicad_pcb")
|
||||
|
||||
assert result is healthy
|
||||
assert mock_pcbnew.LoadBoard.call_count == 2
|
||||
mock_reload.assert_called_once()
|
||||
|
||||
|
||||
def test_safe_load_board_returns_none_when_recovery_fails():
|
||||
iface = _make_iface()
|
||||
dehydrated = _make_dehydrated_board()
|
||||
|
||||
with patch("kicad_interface.pcbnew") as mock_pcbnew, patch("importlib.reload") as mock_reload:
|
||||
mock_pcbnew.LoadBoard = MagicMock(return_value=dehydrated) # always dehydrated
|
||||
mock_reload.return_value = mock_pcbnew
|
||||
result = iface._safe_load_board("/tmp/x.kicad_pcb")
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_safe_load_board_returns_none_when_loadboard_raises():
|
||||
iface = _make_iface()
|
||||
with patch("kicad_interface.pcbnew") as mock_pcbnew:
|
||||
mock_pcbnew.LoadBoard = MagicMock(side_effect=RuntimeError("io error"))
|
||||
result = iface._safe_load_board("/tmp/missing.kicad_pcb")
|
||||
assert result is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# handle_command("open_project") — surfacing dehydration vs. recovery
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _wire_open_project(iface: Any, board_to_assign: Any, board_path: str) -> None:
|
||||
"""Make iface.project_commands.open_project return success and assign a board."""
|
||||
iface.project_commands = MagicMock()
|
||||
|
||||
def _fake(params):
|
||||
iface.project_commands.board = board_to_assign
|
||||
return {
|
||||
"success": True,
|
||||
"message": f"Opened project: {Path(board_path).name}",
|
||||
"project": {
|
||||
"name": Path(board_path).stem,
|
||||
"path": board_path,
|
||||
"boardPath": board_path,
|
||||
},
|
||||
}
|
||||
|
||||
iface.project_commands.open_project = _fake
|
||||
# Stub the rest of _update_command_handlers' targets
|
||||
for attr in (
|
||||
"board_commands",
|
||||
"component_commands",
|
||||
"routing_commands",
|
||||
"design_rule_commands",
|
||||
"export_commands",
|
||||
"freerouting_commands",
|
||||
):
|
||||
setattr(iface, attr, MagicMock())
|
||||
# Provide minimal command_routes so handle_command can route
|
||||
iface.command_routes = {"open_project": iface.project_commands.open_project}
|
||||
|
||||
|
||||
def test_handle_open_project_surfaces_dehydration_when_recovery_fails():
|
||||
"""If LoadBoard returns a dehydrated proxy and recovery fails, the MCP must
|
||||
return success=False — never claim "Opened project" while the board is
|
||||
unusable. This is the fix for the silent-success bug users hit."""
|
||||
iface = _make_iface()
|
||||
dehydrated = _make_dehydrated_board()
|
||||
_wire_open_project(iface, dehydrated, "/tmp/test.kicad_pcb")
|
||||
iface._safe_load_board = MagicMock(return_value=None) # recovery impossible
|
||||
|
||||
result = iface.handle_command("open_project", {"filename": "/tmp/test.kicad_pro"})
|
||||
|
||||
assert result["success"] is False, f"Expected failure, got: {result}"
|
||||
combined = (result.get("errorDetails", "") + result.get("message", "")).lower()
|
||||
assert "dehydrated" in combined or "swigpyobject" in combined
|
||||
|
||||
|
||||
def test_handle_open_project_recovers_via_safe_load_board():
|
||||
"""When _safe_load_board succeeds, open_project should report success
|
||||
with a warning and the recovered board should be installed."""
|
||||
iface = _make_iface()
|
||||
dehydrated = _make_dehydrated_board()
|
||||
healthy = _make_healthy_board()
|
||||
_wire_open_project(iface, dehydrated, "/tmp/test.kicad_pcb")
|
||||
iface._safe_load_board = MagicMock(return_value=healthy)
|
||||
|
||||
result = iface.handle_command("open_project", {"filename": "/tmp/test.kicad_pro"})
|
||||
|
||||
assert result["success"] is True, f"Expected recovery success, got: {result}"
|
||||
assert iface.board is healthy
|
||||
iface._safe_load_board.assert_called_once_with("/tmp/test.kicad_pcb")
|
||||
warnings = result.get("warnings", [])
|
||||
assert any("dehydrated" in w.lower() for w in warnings)
|
||||
|
||||
|
||||
def test_handle_open_project_passes_through_when_already_healthy():
|
||||
"""The fast path: LoadBoard returned a healthy board, no recovery needed."""
|
||||
iface = _make_iface()
|
||||
healthy = _make_healthy_board()
|
||||
_wire_open_project(iface, healthy, "/tmp/test.kicad_pcb")
|
||||
iface._safe_load_board = MagicMock(
|
||||
side_effect=AssertionError("should not be called when board is healthy")
|
||||
)
|
||||
|
||||
result = iface.handle_command("open_project", {"filename": "/tmp/test.kicad_pro"})
|
||||
|
||||
assert result["success"] is True
|
||||
assert iface.board is healthy
|
||||
assert "warnings" not in result # no recovery happened
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _auto_save_board — recover if save dehydrates the board in-memory
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_auto_save_recovers_when_save_leaves_board_dehydrated():
|
||||
"""If pcbnew.SaveBoard somehow corrupts the in-memory BOARD (observed on
|
||||
nightlies after delete_trace + auto-save), _auto_save_board reloads from
|
||||
disk so the next command sees a usable proxy."""
|
||||
iface = _make_iface()
|
||||
|
||||
# Start with a healthy board that becomes dehydrated after SaveBoard runs
|
||||
healthy = _make_healthy_board()
|
||||
healthy.GetFileName = MagicMock(return_value="/tmp/test.kicad_pcb")
|
||||
iface.board = healthy
|
||||
|
||||
recovered = _make_healthy_board()
|
||||
iface._safe_load_board = MagicMock(return_value=recovered)
|
||||
# Stub _update_command_handlers' downstream targets
|
||||
for attr in (
|
||||
"project_commands",
|
||||
"board_commands",
|
||||
"component_commands",
|
||||
"routing_commands",
|
||||
"design_rule_commands",
|
||||
"export_commands",
|
||||
"freerouting_commands",
|
||||
):
|
||||
setattr(iface, attr, MagicMock())
|
||||
|
||||
def _save_then_dehydrate(path, board):
|
||||
# SaveBoard succeeds but leaves iface.board unusable
|
||||
iface.board = _make_dehydrated_board()
|
||||
iface.board.GetFileName = MagicMock(return_value="/tmp/test.kicad_pcb")
|
||||
|
||||
with patch("kicad_interface.pcbnew") as mock_pcbnew:
|
||||
mock_pcbnew.SaveBoard = MagicMock(side_effect=_save_then_dehydrate)
|
||||
iface._auto_save_board()
|
||||
|
||||
assert iface.board is recovered
|
||||
iface._safe_load_board.assert_called_once_with("/tmp/test.kicad_pcb")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# check_kicad_ui consistency fix
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_check_kicad_ui_running_false_when_no_processes():
|
||||
iface = _make_iface()
|
||||
with patch("kicad_interface.KiCADProcessManager") as MockMgr:
|
||||
MockMgr.return_value.get_process_info.return_value = []
|
||||
result = iface._handle_check_kicad_ui({})
|
||||
assert result["success"] is True
|
||||
assert result["running"] is False
|
||||
assert result["processes"] == []
|
||||
|
||||
|
||||
def test_check_kicad_ui_running_true_when_processes_present():
|
||||
iface = _make_iface()
|
||||
with patch("kicad_interface.KiCADProcessManager") as MockMgr:
|
||||
MockMgr.return_value.get_process_info.return_value = [
|
||||
{"pid": "1234", "name": "kicad", "command": "/Applications/KiCad/.../kicad"}
|
||||
]
|
||||
result = iface._handle_check_kicad_ui({})
|
||||
assert result["success"] is True
|
||||
assert result["running"] is True
|
||||
assert len(result["processes"]) == 1
|
||||
|
||||
|
||||
def test_check_kicad_ui_running_and_processes_never_disagree():
|
||||
"""Regression test for the old race where running=True coexisted with
|
||||
processes=[] because is_running() and get_process_info() used different
|
||||
detection methods."""
|
||||
iface = _make_iface()
|
||||
with patch("kicad_interface.KiCADProcessManager") as MockMgr:
|
||||
# Even if some hypothetical separate is_running() returned True,
|
||||
# only get_process_info matters now.
|
||||
MockMgr.return_value.is_running.return_value = True
|
||||
MockMgr.return_value.get_process_info.return_value = []
|
||||
result = iface._handle_check_kicad_ui({})
|
||||
assert result["running"] is False
|
||||
assert result["processes"] == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# run_drc — caller-overridable timeout
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_run_drc_honors_timeout_sec_param():
|
||||
"""timeoutSec param must be passed to the subprocess.run timeout argument."""
|
||||
import subprocess as _sp
|
||||
|
||||
from commands.design_rules import DesignRuleCommands
|
||||
|
||||
cmds = DesignRuleCommands(board=_make_healthy_board())
|
||||
cmds.board.GetFileName = MagicMock(return_value="/tmp/exists.kicad_pcb")
|
||||
|
||||
captured: dict = {}
|
||||
|
||||
def _fake_run(cmd, **kwargs):
|
||||
captured["timeout"] = kwargs.get("timeout")
|
||||
# Raise TimeoutExpired so we don't need to fake JSON output
|
||||
raise _sp.TimeoutExpired(cmd, kwargs.get("timeout", 0))
|
||||
|
||||
with (
|
||||
patch("os.path.exists", return_value=True),
|
||||
patch("subprocess.run", side_effect=_fake_run),
|
||||
patch.object(cmds, "_find_kicad_cli", return_value="/usr/bin/kicad-cli"),
|
||||
):
|
||||
result = cmds.run_drc({"timeoutSec": 45})
|
||||
|
||||
assert captured["timeout"] == 45
|
||||
assert result["success"] is False
|
||||
assert "45" in result["errorDetails"]
|
||||
|
||||
|
||||
def test_run_drc_clamps_extreme_timeout_values():
|
||||
"""Bad timeoutSec values must be clamped, not crash."""
|
||||
import subprocess as _sp
|
||||
|
||||
from commands.design_rules import DesignRuleCommands
|
||||
|
||||
cmds = DesignRuleCommands(board=_make_healthy_board())
|
||||
cmds.board.GetFileName = MagicMock(return_value="/tmp/exists.kicad_pcb")
|
||||
|
||||
captured: dict = {}
|
||||
|
||||
def _fake_run(cmd, **kwargs):
|
||||
captured.setdefault("timeouts", []).append(kwargs.get("timeout"))
|
||||
raise _sp.TimeoutExpired(cmd, kwargs.get("timeout", 0))
|
||||
|
||||
with (
|
||||
patch("os.path.exists", return_value=True),
|
||||
patch("subprocess.run", side_effect=_fake_run),
|
||||
patch.object(cmds, "_find_kicad_cli", return_value="/usr/bin/kicad-cli"),
|
||||
):
|
||||
cmds.run_drc({"timeoutSec": -100}) # below clamp
|
||||
cmds.run_drc({"timeoutSec": 99999}) # above clamp
|
||||
cmds.run_drc({"timeoutSec": "garbage"}) # unparseable → default 600
|
||||
|
||||
assert captured["timeouts"][0] == 10 # clamped to min
|
||||
assert captured["timeouts"][1] == 1800 # clamped to max
|
||||
assert captured["timeouts"][2] == 600 # fallback default
|
||||
Reference in New Issue
Block a user