fix: correctness, cross-backend parity & enforced quality gate (audit Sprints 2-4)

Audit-driven fixes across the dual backend, the premium workflow, and the
engineering layer. 318 tests, ruff-clean.

Quality gate (the moat):
- drawing_finalize now runs the premium ISO-128 critique in addition to the
  8-step validator; leftover construction, non-ISO lineweights, untrimmed
  corners, duplicates and dim overlap block the gate (strict_critique fails on
  warnings too).
- dim_overlap critique no longer a no-op (dim reference point + bbox fallback);
  dim/construction layers resolve from the active layer set (iso13567 too).

Stability / correctness:
- view_screenshot & drawing_export_pdf render headless via Agg (no more
  worker-thread SIGSEGV).
- gear tooth profile no longer self-overlaps at high tooth counts; section view
  no longer duplicates bore lines; validator keyway heuristic fixed.
- ezdxf aligned/angular dim API, full-circle polar array duplicate already
  covered by Sprint 1; this bundle adds save-format-from-extension,
  entity_offset side selection, ARC length, properties parity (bounding_box /
  rotation / char_height), mtext rotation.

COM robustness (mocked-COM tested):
- CoUninitialize on teardown; transaction commit/rollback try/finally + UNDO via
  the CMDACTIVE guard; system_run_lisp guard + USERS1 value capture;
  set_variable type coercion; HWND via Application.HWND; BlockInfo.description.

Security:
- HTTP remote-bind guard fires on every launch path (incl. `fastmcp run`);
  StaticTokenVerifier wired into FastMCP; AutoLISP allowlist bypass-vector tests.

Hygiene: linetype-on-demand in _apply_attrs, honest view_zoom returns, snap-on-arc
sweep handling, point_tangent acos clamp, dynamic tool_groups / tool count,
word-boundary SPUR/HELICAL, undo-stack locking.

Deferred: R22 (undo/transaction stack isolation) lands with the I1 refiner (1.2),
which depends on it.
This commit is contained in:
Umutcan Edizsalan
2026-06-19 08:36:58 +03:00
parent 00e14423df
commit 32a68be919
18 changed files with 2016 additions and 166 deletions
+132 -25
View File
@@ -201,7 +201,7 @@ class AutoCADBackend(ABC):
@abstractmethod
async def entity_create_mtext(
self, text: str, x: float, y: float,
width: float = 100.0, height: float = 2.5,
width: float = 100.0, height: float = 2.5, rotation: float = 0.0,
layer: str | None = None, color: int | None = None,
) -> EntityInfo: ...
@@ -545,7 +545,10 @@ class AutoCADBackend(ABC):
notes: list[str] | None = None,
) -> PlanSpec:
"""Commit a drawing intent before any geometry is created.
Returns a PlanSpec held on the backend for `drawing_critique`."""
Returns a PlanSpec and stores it on the backend; read it back with
`get_plan_spec()`. (N8: the previous claim that `drawing_critique`
replays this plan was untrue — critique runs geometric checks, not a
plan diff.)"""
from engineering.plan_spec import PlanSpec
plan = PlanSpec(
intent=str(intent),
@@ -562,11 +565,19 @@ class AutoCADBackend(ABC):
async def drawing_critique(
self, focus: list[CritiqueFocus] | None = None,
) -> list[Issue]:
"""Run premium-quality checks; return zero issues before
`drawing_finalize`. `focus=None` runs all checks."""
"""Run premium-quality geometric checks; return zero issues before
`drawing_finalize`. `focus=None` runs all checks. (N8: this does NOT
replay the committed PlanSpec — it inspects the live drawing geometry.)"""
from engineering.critique import run_critique
return await run_critique(self, focus)
def get_plan_spec(self) -> dict | None:
"""Return the PlanSpec committed by `drawing_plan` as a dict, or None
if no plan has been committed. (N8: makes the stored `_plan_spec`
actually readable instead of being write-only dead state.)"""
plan = getattr(self, "_plan_spec", None)
return plan.to_dict() if plan is not None else None
async def point_from_snap(
self, handle: str, snap: SnapType,
ref_x: float | None = None, ref_y: float | None = None,
@@ -601,6 +612,19 @@ class AutoCADBackend(ABC):
)
return float(c[0]), float(c[1]), float(r)
def _arc_angles():
# N7: use .get() + descriptive RuntimeError (matching _line_pts /
# _circle_geom) instead of bracket indexing that raises an opaque
# KeyError when start_angle/end_angle are missing.
sa = p.get("start_angle")
ea = p.get("end_angle")
if sa is None or ea is None:
raise RuntimeError(
f"point_from_snap: entity {handle} ({et}) has no "
"start_angle/end_angle."
)
return math.radians(float(sa)), math.radians(float(ea))
if snap == "end":
if et == "LINE":
s, e = _line_pts()
@@ -611,8 +635,7 @@ class AutoCADBackend(ABC):
return s if ds <= de else e
if et == "ARC":
cx, cy, r = _circle_geom()
sa = math.radians(float(p["start_angle"]))
ea = math.radians(float(p["end_angle"]))
sa, ea = _arc_angles()
sp = (cx + r * math.cos(sa), cy + r * math.sin(sa))
ep = (cx + r * math.cos(ea), cy + r * math.sin(ea))
if ref is None:
@@ -628,8 +651,7 @@ class AutoCADBackend(ABC):
return ((s[0] + e[0]) / 2.0, (s[1] + e[1]) / 2.0)
if et == "ARC":
cx, cy, r = _circle_geom()
sa = math.radians(float(p["start_angle"]))
ea = math.radians(float(p["end_angle"]))
sa, ea = _arc_angles()
if ea < sa:
ea += 2 * math.pi
ma = (sa + ea) / 2.0
@@ -646,13 +668,49 @@ class AutoCADBackend(ABC):
return (float(c[0]), float(c[1]))
raise RuntimeError(f"point_from_snap: 'center' snap not supported on {et}")
# NEW-snap-quad-arc / NEW-snap-near-arc: an ARC only covers part of the
# full circle, so quad/near candidates derived from the circle may fall
# outside the [start_angle, end_angle] sweep. Restrict (quad) / clamp
# (near) the projected angle to the sweep, with 2π wraparound.
def _arc_sweep():
sa, ea = _arc_angles()
if ea < sa:
ea += 2 * math.pi
return sa, ea
def _angle_in_sweep(theta: float, sa: float, ea: float) -> bool:
# Normalise theta into [sa, sa+2π) and test against [sa, ea].
t = sa + ((theta - sa) % (2 * math.pi))
return t <= ea + 1e-9
def _arc_endpoints(cx: float, cy: float, r: float, sa: float, ea: float):
return (
(cx + r * math.cos(sa), cy + r * math.sin(sa)),
(cx + r * math.cos(ea), cy + r * math.sin(ea)),
)
if snap == "quad":
if et in ("CIRCLE", "ARC"):
if et == "CIRCLE":
cx, cy, r = _circle_geom()
pts = [(cx + r, cy), (cx, cy + r), (cx - r, cy), (cx, cy - r)]
if ref is None:
return pts[0]
return min(pts, key=lambda q: (q[0] - ref[0]) ** 2 + (q[1] - ref[1]) ** 2)
if et == "ARC":
cx, cy, r = _circle_geom()
sa, ea = _arc_sweep()
quad_angles = (0.0, math.pi / 2.0, math.pi, 3 * math.pi / 2.0)
pts = [
(cx + r * math.cos(a), cy + r * math.sin(a))
for a in quad_angles
if _angle_in_sweep(a, sa, ea)
]
if not pts:
# No quadrant lies on the sweep — fall back to arc endpoints.
pts = list(_arc_endpoints(cx, cy, r, sa, ea))
if ref is None:
return pts[0]
return min(pts, key=lambda q: (q[0] - ref[0]) ** 2 + (q[1] - ref[1]) ** 2)
raise RuntimeError(f"point_from_snap: 'quad' snap not supported on {et}")
if snap == "perp":
@@ -682,13 +740,29 @@ class AutoCADBackend(ABC):
t = ((ref[0] - s[0]) * dx + (ref[1] - s[1]) * dy) / L2
t = max(0.0, min(1.0, t))
return (s[0] + t * dx, s[1] + t * dy)
if et in ("CIRCLE", "ARC"):
if et == "CIRCLE":
cx, cy, r = _circle_geom()
vx, vy = ref[0] - cx, ref[1] - cy
L = math.hypot(vx, vy)
if L < 1e-18:
return (cx + r, cy)
return (cx + r * vx / L, cy + r * vy / L)
if et == "ARC":
cx, cy, r = _circle_geom()
sa, ea = _arc_sweep()
vx, vy = ref[0] - cx, ref[1] - cy
L = math.hypot(vx, vy)
if L < 1e-18:
# Degenerate ref at center: return the start endpoint.
return (cx + r * math.cos(sa), cy + r * math.sin(sa))
theta = math.atan2(vy, vx)
if _angle_in_sweep(theta, sa, ea):
return (cx + r * math.cos(theta), cy + r * math.sin(theta))
# Projected angle is off the sweep — fall back to the nearer endpoint.
sp, ep = _arc_endpoints(cx, cy, r, sa, ea)
ds = (sp[0] - ref[0]) ** 2 + (sp[1] - ref[1]) ** 2
de = (ep[0] - ref[0]) ** 2 + (ep[1] - ref[1]) ** 2
return sp if ds <= de else ep
raise RuntimeError(f"point_from_snap: 'near' snap not supported on {et}")
if snap == "int":
@@ -831,7 +905,13 @@ class AutoCADBackend(ABC):
# Solving cos(θ - α) = -r/d where α = atan2(C-F) gives the two
# circle angles θ at which the tangent touches the perimeter.
alpha = math.atan2(dy, dx) # direction from F to C
acos_neg = math.acos(-r / d) # in (π/2, π) for external point
# NEW-base-1: clamp the acos argument to [-1, 1] so a from-point in the
# narrow band [r-1e-9, r) (past the internal-point guard but with
# |-r/d| marginally > 1 from float error) degenerates gracefully to the
# tangent-at-perimeter case instead of raising a raw ValueError
# ("math domain error"). Genuine internal points (d < r) are already
# rejected by the RuntimeError guard above.
acos_neg = math.acos(max(-1.0, min(1.0, -r / d))) # (π/2, π] for external pt
candidates: list[tuple[float, float]] = [
(cx + r * math.cos(alpha + acos_neg), cy + r * math.sin(alpha + acos_neg)),
(cx + r * math.cos(alpha - acos_neg), cy + r * math.sin(alpha - acos_neg)),
@@ -864,12 +944,31 @@ class AutoCADBackend(ABC):
except Exception:
pass # already created / racy create — harmless
def _active_layer_set_id(self) -> str:
"""Best-effort active layer set: an explicit `drawing_apply_iso_layers`
wins, else the committed PlanSpec, else 'mech' (the default bootstrap)."""
explicit = getattr(self, "_active_layer_set", None)
if explicit:
return explicit
plan = getattr(self, "_plan_spec", None)
if plan is not None:
return getattr(plan, "layer_set_id", "mech")
return "mech"
def _role_layer(self, role: str) -> str:
"""Resolve the layer name for a role ('dim' | 'construction') in the
active layer set, so dims/scaffolding land on the right layer for
iso13567 (M-DIMEN-T-N / M-CONST-E-N) not just mech/pid (DIM/CONSTRUCTION)."""
from engineering.layers import resolve_role_layer
return resolve_role_layer(self._active_layer_set_id(), role)
async def construction_xline(
self, x: float, y: float, angle_deg: float,
layer: str = "CONSTRUCTION",
layer: str | None = None,
) -> EntityInfo:
"""Create an infinite construction line (XLINE) on a CONSTRUCTION
layer (auto-created with color 250 = lightest if missing)."""
"""Create an infinite construction line (XLINE) on the active layer set's
construction layer (auto-created with color 250 = lightest if missing)."""
layer = layer or self._role_layer("construction")
await self._ensure_layer(layer, color=250)
ang = math.radians(float(angle_deg))
return await self._create_xline(
@@ -877,10 +976,11 @@ class AutoCADBackend(ABC):
)
async def construction_clear(
self, layer: str = "CONSTRUCTION",
self, layer: str | None = None,
) -> dict:
"""Delete every entity on the CONSTRUCTION layer.
"""Delete every entity on the active layer set's construction layer.
Must be called before `drawing_finalize`."""
layer = layer or self._role_layer("construction")
try:
ents = await self.entity_list(layer_filter=layer, limit=5000)
except Exception:
@@ -901,17 +1001,20 @@ class AutoCADBackend(ABC):
and lineweights. Idempotent."""
from engineering.layers import apply_layer_set
result = await apply_layer_set(self, standard)
self._active_layer_set = standard
return {"ok": True, "standard": standard, "layers": result}
async def dimension_auto(
self, handles: list[str], style: DimStyle = "chain",
offset: float = 10.0,
offset: float = 10.0, layer: str | None = None,
) -> list[EntityInfo]:
"""Generate ISO 129 dimensions across `handles` in the chosen style
(chain | baseline | ordinate). `offset` is the dimension-line
distance from the reference geometry (mm)."""
distance from the reference geometry (mm). `layer` defaults to the
active layer set's dimension layer (DIM, or M-DIMEN-T-N for iso13567)."""
if not handles:
return []
dim_layer = layer or self._role_layer("dim")
if style not in ("chain", "baseline", "ordinate"):
raise RuntimeError(
f"dimension_auto: unknown style '{style}'. "
@@ -945,7 +1048,7 @@ class AutoCADBackend(ABC):
nx, ny = -dy / L, dx / L
dim = await self.dimension_linear(
sx, sy, ex, ey, mx + nx * off, my + ny * off,
rotation=math.degrees(math.atan2(dy, dx)), layer="DIM",
rotation=math.degrees(math.atan2(dy, dx)), layer=dim_layer,
)
results.append(dim)
@@ -954,27 +1057,31 @@ class AutoCADBackend(ABC):
dx, dy = ex0 - sx0, ey0 - sy0
L0 = math.hypot(dx, dy) or 1.0
ux, uy = dx / L0, dy / L0
base_rot = math.degrees(math.atan2(dy, dx))
for i, (_sx, _sy, ex, ey) in enumerate(segs):
dim = await self.dimension_linear(
sx0, sy0, ex, ey,
sx0 - uy * (off + i * off), sy0 + ux * (off + i * off),
layer="DIM",
# Dimension line must run parallel to the (possibly rotated)
# baseline, not horizontal.
rotation=base_rot, layer=dim_layer,
)
results.append(dim)
else: # ordinate — X then Y distance of each segment's end from start
for sx, sy, ex, ey in segs:
else: # ordinate — stacked X/Y distances; each feature offset so dims don't overlap
for i, (sx, sy, ex, ey) in enumerate(segs):
rung = off + i * off # stagger per feature to avoid coincident dims
if abs(ex - sx) > 1e-9:
dim = await self.dimension_linear(
sx, sy, ex, sy,
sx + (ex - sx) / 2.0, max(sy, ey) + off, layer="DIM",
sx + (ex - sx) / 2.0, max(sy, ey) + rung, layer=dim_layer,
)
results.append(dim)
if abs(ey - sy) > 1e-9:
dim = await self.dimension_linear(
ex, sy, ex, ey,
max(sx, ex) + off, (sy + ey) / 2.0,
rotation=90.0, layer="DIM",
max(sx, ex) + rung, (sy + ey) / 2.0,
rotation=90.0, layer=dim_layer,
)
results.append(dim)
+150 -20
View File
@@ -72,6 +72,21 @@ def _com_init():
pythoncom.CoInitialize()
def _com_teardown():
"""COM thread finalizer — drop the cached app and release the STA apartment.
Submitted to the executor (R5) so the CoInitialize done in _com_init is
paired with a CoUninitialize when the worker is torn down or a stuck
SendCommand eventually unblocks, instead of leaking the apartment forever.
"""
_COM_STATE.pop("app", None)
if _COM_IMPORTS_OK:
try:
pythoncom.CoUninitialize()
except Exception:
pass
def _apoint(x: float, y: float, z: float = 0.0):
"""Create a VARIANT double-array point for AutoCAD COM."""
return win32com.client.VARIANT(
@@ -220,6 +235,9 @@ def _entity_info(entity) -> EntityInfo:
props["radius"] = entity.Radius
props["start_angle"] = rad2deg(entity.StartAngle)
props["end_angle"] = rad2deg(entity.EndAngle)
# Arc length (N3) — parity with ezdxf so length_range selects ARCs.
_sweep = (entity.EndAngle - entity.StartAngle) % (2.0 * math.pi)
props["length"] = entity.Radius * _sweep
elif obj_name in ("AcDbLWPolyline", "AcDb2dPolyline"):
coords = list(entity.Coordinates)
pts = [[coords[i], coords[i+1]] for i in range(0, len(coords), 2)]
@@ -231,10 +249,14 @@ def _entity_info(entity) -> EntityInfo:
ins = entity.InsertionPoint
props["insertion"] = [ins[0], ins[1]]
props["height"] = entity.Height
props["rotation"] = rad2deg(entity.Rotation) # N5 parity with ezdxf
elif obj_name == "AcDbMText":
props["text"] = entity.Contents
ins = entity.InsertionPoint
props["insertion"] = [ins[0], ins[1]]
# N5 parity with ezdxf: MTEXT char_height + rotation.
props["char_height"] = entity.Height
props["rotation"] = rad2deg(entity.Rotation)
elif obj_name == "AcDbBlockReference":
ins = entity.InsertionPoint
props["insertion"] = [ins[0], ins[1]]
@@ -285,6 +307,13 @@ def _find_autocad_hwnd() -> int | None:
"""Find the main AutoCAD window handle."""
if not _WIN32_AVAILABLE:
return None
# R28: prefer the COM application's real main-frame handle. The EnumWindows
# title-substring scan can match a palette / About dialog / Start tab, so it
# is only a fallback for when the HWND read fails.
try:
return int(_acad_app().HWND)
except Exception as exc:
log.debug("Application.HWND read failed, falling back to EnumWindows: %s", exc)
result = {"hwnd": None}
def _enum_cb(hwnd, _):
@@ -385,6 +414,10 @@ class ComBackend(AutoCADBackend):
async def disconnect(self) -> None:
if self._executor:
try:
self._executor.submit(_com_teardown) # release COM apartment (R5)
except Exception:
pass
self._executor.shutdown(wait=False)
self._executor = None
self._connected = False
@@ -422,6 +455,12 @@ class ComBackend(AutoCADBackend):
stuck = self._executor
self._executor = ThreadPoolExecutor(max_workers=1, initializer=_com_init)
if stuck is not None:
# When the stuck SendCommand eventually unblocks, the queued
# teardown releases its COM apartment instead of leaking it (R5).
try:
stuck.submit(_com_teardown)
except Exception:
pass
stuck.shutdown(wait=False)
_COM_STATE.pop("app", None)
self._connected = False
@@ -637,12 +676,14 @@ class ComBackend(AutoCADBackend):
return await self._run(_sync)
async def entity_create_mtext(
self, text, x, y, width=100.0, height=2.5, layer=None, color=None,
self, text, x, y, width=100.0, height=2.5, rotation=0.0, layer=None, color=None,
) -> EntityInfo:
def _sync():
mspace = _msp()
mt = mspace.AddMText(_apoint(x, y), float(width), text)
mt.Height = float(height)
if rotation:
mt.Rotation = deg2rad(rotation) # NEW-mtext-1: honor caller rotation
_apply_entity_attrs(mt, layer, color, None)
_regen()
return _entity_info(mt)
@@ -863,14 +904,49 @@ class ComBackend(AutoCADBackend):
def _sync():
doc = _acad_doc()
ent = doc.HandleToObject(handle)
offset_ent = ent.Offset(float(distance))
# Offset returns a variant array of entities
try:
first = offset_ent[0]
except Exception as exc:
log.debug("Offset result indexing failed, using raw result: %s", exc)
first = offset_ent
return _entity_info(first)
def _as_list(res):
# Offset returns a variant array of entities (1 for line/arc/circle).
try:
return [res[i] for i in range(len(res))]
except TypeError:
return [res]
def _center(e):
bb_min, bb_max = e.GetBoundingBox()
return ((bb_min[0] + bb_max[0]) / 2.0, (bb_min[1] + bb_max[1]) / 2.0)
d = float(distance)
# R13: honor side_x/side_y so positive distance means the same side as
# ezdxf. Offset both ways, keep the copy nearest the side point, delete
# the rest (also fixes the orphan-extras leak, NEW-com-offset-orphan).
if side_x is not None and side_y is not None and abs(d) > 1e-12:
sx, sy = float(side_x), float(side_y)
pos = _as_list(ent.Offset(abs(d)))
neg = _as_list(ent.Offset(-abs(d)))
def _sq_dist(e):
cx, cy = _center(e)
return (cx - sx) ** 2 + (cy - sy) ** 2
if _sq_dist(pos[0]) <= _sq_dist(neg[0]):
keep, drop = pos, neg
else:
keep, drop = neg, pos
for extra in drop + keep[1:]:
try:
extra.Delete()
except Exception:
pass
return _entity_info(keep[0])
result = _as_list(ent.Offset(d))
for extra in result[1:]:
try:
extra.Delete()
except Exception:
pass
return _entity_info(result[0])
return await self._run(_sync)
async def entity_delete(self, handle) -> dict:
@@ -1178,12 +1254,18 @@ class ComBackend(AutoCADBackend):
1 for j in range(blk.Count)
if blk.Item(j).ObjectName == "AcDbAttributeDefinition"
)
# S3: populate description from the block definition's .Comments.
try:
description = blk.Comments or ""
except Exception:
description = ""
blocks.append(BlockInfo(
name=blk.Name,
origin=(blk.Origin[0], blk.Origin[1]),
attribute_count=attr_count,
entity_count=blk.Count,
is_xref=bool(blk.IsXRef),
description=description,
))
return blocks
return await self._run(_sync)
@@ -1403,19 +1485,25 @@ class ComBackend(AutoCADBackend):
doc = _acad_doc()
doc.EndUndoMark()
return {"ok": True, "message": "Transaction committed"}
result = await self._run(_sync)
self._transaction_active = False
return result
# R16: clear the flag even if _run raises, so a failed commit can't leave
# _transaction_active permanently stale.
try:
return await self._run(_sync)
finally:
self._transaction_active = False
async def transaction_rollback(self) -> dict:
def _sync():
doc = _acad_doc()
doc.EndUndoMark()
doc.SendCommand("_UNDO B\n") # Back to last undo mark
# R16: route UNDO B through the CMDACTIVE-aware sender instead of a raw
# SendCommand that could deadlock if a command/prompt is active.
self._safe_send_command(doc, "_UNDO B")
return {"ok": True, "message": "Transaction rolled back"}
result = await self._run(_sync)
self._transaction_active = False
return result
try:
return await self._run(_sync)
finally:
self._transaction_active = False
# ── system ──────────────────────────────────────────────────────────────
@@ -1453,8 +1541,22 @@ class ComBackend(AutoCADBackend):
async def system_set_variable(self, name, value) -> dict:
def _sync():
app = _acad_app()
app.SetVariable(name, value)
return {"ok": True, "variable": name, "value": value}
# Coerce to the sysvar's actual type (NEW-com-set-variable-coercion):
# AutoCAD rejects e.g. a string "0" for the integer OSMODE. Probe the
# current value's type and convert to match; pass through on failure.
coerced = value
try:
current = app.GetVariable(name)
if isinstance(current, bool):
coerced = bool(int(value)) if isinstance(value, str) else bool(value)
elif isinstance(current, int):
coerced = int(float(value)) if isinstance(value, str) else int(value)
elif isinstance(current, float):
coerced = float(value)
except Exception as exc:
log.debug("set_variable type probe for %s failed: %s", name, exc)
app.SetVariable(name, coerced)
return {"ok": True, "variable": name, "value": coerced}
return await self._run(_sync)
async def system_run_command(self, command) -> dict:
@@ -1482,9 +1584,37 @@ class ComBackend(AutoCADBackend):
async def system_run_lisp(self, expression) -> dict:
def _sync():
app = _acad_app()
doc = _acad_doc()
result = doc.SendCommand(f"(progn {expression})\n")
return {"ok": True, "expression": expression, "result": str(result) if result else "nil"}
# R23: refuse to send while a command/prompt is active (raw SendCommand
# of a prompting form would deadlock the single STA thread).
try:
if int(app.GetVariable("CMDACTIVE")):
raise RuntimeError(
"AutoCAD has an active command/prompt (CMDACTIVE). "
"Press ESC in AutoCAD and retry."
)
except RuntimeError:
raise
except Exception as exc:
log.debug("CMDACTIVE read failed, proceeding: %s", exc)
# N6: SendCommand is void-returning, so the value can't be read from its
# return. Stash it in USERS1 and read it back after the form completes
# (CMDACTIVE-polled by _safe_send_command).
wrapped = (
'(vl-load-com)'
f'(setvar "USERS1" (vl-princ-to-string (progn {expression})))'
)
self._safe_send_command(doc, wrapped)
try:
result = app.GetVariable("USERS1")
except Exception:
result = None
return {
"ok": True,
"expression": expression,
"result": str(result) if result not in (None, "") else "nil",
}
return await self._run(_sync)
# ── corner ops ──────────────────────────────────────────────────────────
+97 -23
View File
@@ -30,6 +30,25 @@ from .base import (
log = logging.getLogger(__name__)
def _new_agg_figure(*, figsize=None, dpi=100):
"""Create a matplotlib Figure bound to the headless Agg canvas.
Rendering runs inside an ``asyncio.to_thread`` worker. ``pyplot.figure``
selects a *GUI* backend from global state (Qt/Tk), and instantiating a GUI
canvas off the main thread can hard-crash the interpreter (SIGSEGV). Using
``Figure`` + ``FigureCanvasAgg`` directly bypasses pyplot's global backend
selection entirely and is thread-safe; ``savefig`` still dispatches to the
PDF backend by file extension when needed.
"""
from matplotlib.backends.backend_agg import FigureCanvasAgg
from matplotlib.figure import Figure
fig = Figure(figsize=figsize, dpi=dpi)
FigureCanvasAgg(fig) # binds fig.canvas to Agg (no pyplot, no GUI)
return fig
try:
import ezdxf
from ezdxf import colors, units # noqa: F401
@@ -132,6 +151,9 @@ def _entity_info_dxf(ent) -> EntityInfo:
props["radius"] = ent.dxf.radius
props["start_angle"] = ent.dxf.start_angle
props["end_angle"] = ent.dxf.end_angle
# Arc length (N3) so entity_select_smart length_range works on ARCs.
_sweep = (ent.dxf.end_angle - ent.dxf.start_angle) % 360.0
props["length"] = ent.dxf.radius * math.radians(_sweep)
elif ent_type == "LWPOLYLINE":
props["points"] = [[float(pt[0]), float(pt[1])] for pt in ent.get_points()]
props["closed"] = ent.closed
@@ -145,6 +167,7 @@ def _entity_info_dxf(ent) -> EntityInfo:
props["text"] = ent.text
props["insertion"] = _v2(ent.dxf.insert)
props["char_height"] = ent.dxf.char_height
props["rotation"] = ent.dxf.get("rotation", 0.0)
elif ent_type == "INSERT":
props["block_name"] = ent.dxf.name
props["insertion"] = _v2(ent.dxf.insert)
@@ -159,12 +182,37 @@ def _entity_info_dxf(ent) -> EntityInfo:
if ent.dxf.hasattr("fit_points"):
props["fit_point_count"] = len(list(ent.fit_points))
props["degree"] = ent.dxf.degree
elif ent_type in ("DIMLINEAR", "DIMALIGNED", "DIMANGULAR",
"DIMRADIUS", "DIMDIAMETER"):
elif ent_type in ("DIMENSION", "DIMLINEAR", "DIMALIGNED", "DIMANGULAR",
"DIMRADIUS", "DIMDIAMETER", "DIMORDINATE"):
props["dim_type"] = ent_type
# Reference points for dim_overlap critique. ezdxf dimensions carry
# a definition point and (usually) the text midpoint.
try:
props["defpoint"] = _v2(ent.dxf.defpoint)
except Exception:
pass
try:
tm = ent.dxf.get("text_midpoint", None)
if tm is not None:
props["text_position"] = _v2(tm)
except Exception:
pass
except Exception as exc:
log.debug("extracting entity properties for %s: %s", ent_type, exc)
# bounding_box parity with the COM backend (same {min,max} shape) so clients
# reading properties["bounding_box"] work on both engines (N5).
try:
from ezdxf import bbox as _bbox
_bb = _bbox.extents([ent])
if _bb.has_data:
props["bounding_box"] = {
"min": [float(_bb.extmin.x), float(_bb.extmin.y)],
"max": [float(_bb.extmax.x), float(_bb.extmax.y)],
}
except Exception as exc:
log.debug("bbox extents failed for %s: %s", ent_type, exc)
return EntityInfo(
handle=str(handle),
type=ent_type,
@@ -361,17 +409,15 @@ class EzdxfBackend(AutoCADBackend):
doc = self._require_doc()
msp = self._msp()
try:
import matplotlib.pyplot as plt
from ezdxf.addons.drawing import Frontend, RenderContext
from ezdxf.addons.drawing.matplotlib import MatplotlibBackend
fig = plt.figure()
fig = _new_agg_figure()
ax = fig.add_axes([0, 0, 1, 1])
ctx = RenderContext(doc)
out = MatplotlibBackend(ax)
Frontend(ctx, out).draw_layout(msp, finalize=True)
fig.savefig(path, dpi=150)
plt.close(fig)
return {"ok": True, "path": path}
except ImportError:
raise RuntimeError(
@@ -469,9 +515,11 @@ class EzdxfBackend(AutoCADBackend):
return await self._async(_sync)
async def drawing_undo(self) -> dict:
if not self._undo_stack:
return {"ok": False, "error": "Nothing to undo (no undo history in ezdxf backend)"}
# NEW-undo-1: read+mutate _undo_stack under self._lock (via _async) so
# the empty-check and the pop are serialized with all other sync work.
def _sync():
if not self._undo_stack:
return {"ok": False, "error": "Nothing to undo (no undo history in ezdxf backend)"}
p = self._undo_stack.pop()
try:
self._doc = ezdxf.readfile(str(p))
@@ -499,6 +547,10 @@ class EzdxfBackend(AutoCADBackend):
if color is not None:
entity.dxf.color = int(color)
if linetype is not None:
# R24: load HIDDEN/CENTER/etc. on demand before assignment so the
# linetype actually renders instead of silently falling back to
# Continuous. Same loader used in entity_set_properties/layer_*.
_ensure_linetype_loaded(self._require_doc(), linetype)
entity.dxf.linetype = linetype
def _mark_dirty(self):
@@ -577,7 +629,7 @@ class EzdxfBackend(AutoCADBackend):
return await self._async(_sync)
async def entity_create_mtext(
self, text, x, y, width=100.0, height=2.5, layer=None, color=None,
self, text, x, y, width=100.0, height=2.5, rotation=0.0, layer=None, color=None,
) -> EntityInfo:
def _sync():
msp = self._msp()
@@ -586,6 +638,8 @@ class EzdxfBackend(AutoCADBackend):
dxfattribs={"char_height": float(height), "width": float(width)},
)
ent.dxf.insert = (float(x), float(y), 0.0)
if rotation:
ent.dxf.rotation = float(rotation) # NEW-mtext-1: honor caller rotation
self._apply_attrs(ent, layer, color)
self._mark_dirty()
return _entity_info_dxf(ent)
@@ -1212,12 +1266,15 @@ class EzdxfBackend(AutoCADBackend):
)
origin = blk.block.dxf.get("base_point", (0.0, 0.0, 0.0))
is_xref = bool(blk.block.dxf.get("xref_path", ""))
# S3: surface the block definition's description instead of always "".
description = blk.block.dxf.get("description", "")
blocks.append(BlockInfo(
name=blk.name,
origin=(origin[0], origin[1]),
attribute_count=attr_count,
entity_count=len(list(blk)),
is_xref=is_xref,
description=description,
))
return blocks
return await self._async(_sync)
@@ -1412,10 +1469,23 @@ class EzdxfBackend(AutoCADBackend):
# ── view / screenshot ──────────────────────────────────────────────────────
async def view_zoom_extents(self) -> dict:
return {"ok": True, "message": "Zoom extents not applicable for ezdxf backend (no display)"}
# R26/N9: ezdxf has no viewport, so framing is never actually applied.
# Return applied=False with a consistent shape so a client cannot mistake
# the no-op for real framing.
return {
"ok": True,
"applied": False,
"message": "Zoom extents not applicable for ezdxf backend (no display/viewport)",
}
async def view_zoom_window(self, x1, y1, x2, y2) -> dict:
return {"ok": True, "message": "Zoom window not applicable for ezdxf backend (no display)"}
# R26/N9: window region is ignored (no viewport). Consistent shape with
# view_zoom_extents; applied=False signals the framing was not honored.
return {
"ok": True,
"applied": False,
"message": "Zoom window not applicable for ezdxf backend (no display/viewport)",
}
async def view_screenshot(self) -> bytes | None:
"""Render drawing to PNG using matplotlib."""
@@ -1423,11 +1493,10 @@ class EzdxfBackend(AutoCADBackend):
try:
doc = self._require_doc()
msp = self._msp()
import matplotlib.pyplot as plt
from ezdxf.addons.drawing import Frontend, RenderContext
from ezdxf.addons.drawing.matplotlib import MatplotlibBackend
fig = plt.figure(figsize=(16, 9), dpi=100)
fig = _new_agg_figure(figsize=(16, 9), dpi=100)
ax = fig.add_axes([0, 0, 1, 1])
ctx = RenderContext(doc)
out = MatplotlibBackend(ax)
@@ -1435,7 +1504,6 @@ class EzdxfBackend(AutoCADBackend):
buf = io.BytesIO()
fig.savefig(buf, format="png", bbox_inches="tight", dpi=100)
plt.close(fig)
buf.seek(0)
return buf.read()
except ImportError:
@@ -1470,19 +1538,25 @@ class EzdxfBackend(AutoCADBackend):
return await self._async(_sync)
async def transaction_commit(self) -> dict:
if not self._undo_stack:
return {"ok": False, "error": "No active transaction"}
p = self._undo_stack.pop()
try:
p.unlink()
except OSError:
pass
return {"ok": True, "message": "Transaction committed (snapshot discarded)"}
# NEW-undo-1: guard the _undo_stack read+pop with self._lock (via _async)
# instead of touching it outside any lock.
def _sync():
if not self._undo_stack:
return {"ok": False, "error": "No active transaction"}
p = self._undo_stack.pop()
try:
p.unlink()
except OSError:
pass
return {"ok": True, "message": "Transaction committed (snapshot discarded)"}
return await self._async(_sync)
async def transaction_rollback(self) -> dict:
if not self._undo_stack:
return {"ok": False, "error": "No active transaction to rollback"}
# NEW-undo-1: read+mutate _undo_stack under self._lock (via _async) so the
# empty-check and the pop/readfile are serialized with other sync work.
def _sync():
if not self._undo_stack:
return {"ok": False, "error": "No active transaction to rollback"}
p = self._undo_stack.pop()
try:
self._doc = ezdxf.readfile(str(p))
+44 -13
View File
@@ -94,22 +94,29 @@ async def _check_layer_color(backend: AutoCADBackend) -> list[Issue]:
async def _check_construction_left(backend: AutoCADBackend) -> list[Issue]:
"""CONSTRUCTION layer must be empty before drawing_finalize."""
"""No construction/scratch geometry may remain before drawing_finalize.
Matches any layer whose name contains 'CONST' (CONSTRUCTION, M-CONST-E-N, …)
so the check is independent of the active layer set — a leftover iso13567
scaffold on M-CONST-E-N is caught just like a mech one on CONSTRUCTION.
"""
try:
ents = await backend.entity_list(layer_filter="CONSTRUCTION", limit=5000)
ents = await backend.entity_list(limit=5000)
except Exception:
return []
if not ents:
leftover = [e for e in ents if "CONST" in (e.layer or "").upper()]
if not leftover:
return []
return [Issue(
severity="error",
focus="construction_left",
message=(
f"{len(ents)} entity/entities still on CONSTRUCTION layer. "
f"{len(leftover)} entity/entities still on a construction layer. "
"Call construction_clear() before drawing_finalize."
),
handles=[e.handle for e in ents[:50]],
detail={"count": len(ents)},
handles=[e.handle for e in leftover[:50]],
detail={"count": len(leftover),
"layers": sorted({e.layer for e in leftover if e.layer})},
)]
@@ -185,22 +192,46 @@ async def _check_duplicate_entities(backend: AutoCADBackend) -> list[Issue]:
return issues
def _dim_ref_point(e) -> tuple[float, float] | None:
"""Best-effort reference point for a dimension entity, tolerant of the
per-backend property shape: explicit text/def points first, then the
bounding-box centre (COM seeds `bounding_box` for every entity)."""
for key in ("text_position", "defpoint", "insertion_point", "insertion"):
ip = e.properties.get(key)
if ip and len(ip) >= 2:
return (float(ip[0]), float(ip[1]))
bb = e.properties.get("bounding_box")
try:
if isinstance(bb, dict):
mn, mx = bb.get("min"), bb.get("max")
return ((mn[0] + mx[0]) / 2.0, (mn[1] + mx[1]) / 2.0)
if bb and len(bb) == 2 and hasattr(bb[0], "__len__"):
return ((bb[0][0] + bb[1][0]) / 2.0, (bb[0][1] + bb[1][1]) / 2.0)
if bb and len(bb) >= 4:
return ((bb[0] + bb[2]) / 2.0, (bb[1] + bb[3]) / 2.0)
except (TypeError, IndexError, KeyError):
return None
return None
async def _check_dim_overlap(backend: AutoCADBackend) -> list[Issue]:
"""Two dimensions whose insertion points are within `tol` likely overlap."""
"""Two dimensions whose reference points are within `tol` likely overlap.
Filters by entity TYPE (not layer) so dims land-anywhere — including the
iso13567 M-DIMEN-T-N layer — are still considered.
"""
tol = 5.0 # mm
try:
ents = await backend.entity_list(layer_filter="DIM", limit=5000)
ents = await backend.entity_list(limit=5000)
except Exception:
return []
dim_points: list[tuple[str, tuple[float, float]]] = []
for e in ents:
if e.type not in _DIM_TYPES and "DIM" not in e.type:
continue
ip = (e.properties.get("text_position")
or e.properties.get("defpoint")
or e.properties.get("insertion_point"))
if ip and len(ip) >= 2:
dim_points.append((e.handle, (float(ip[0]), float(ip[1]))))
ip = _dim_ref_point(e)
if ip is not None:
dim_points.append((e.handle, ip))
issues: list[Issue] = []
seen: set[frozenset[str]] = set()
for i, (h1, p1) in enumerate(dim_points):
+39 -15
View File
@@ -33,19 +33,36 @@ def generate_involute_flank(
outer_r: float,
n_points: int = 40,
) -> list[tuple[float, float]]:
"""One tooth flank from root through base circle out to addendum tip."""
"""One tooth flank from the root circle out to the addendum tip.
The first returned point ALWAYS sits exactly on the root circle (radius
`root_r`) so the caller's root arc connects seamlessly:
* root_r < base_r — the involute is undefined below the base circle, so the
flank runs radially from the root circle up to (base_r, 0) and then follows
the involute outward (a straight radial fillet approximation).
* root_r >= base_r — (high tooth counts, z>=~42 @20deg) the active profile
starts where the involute crosses the root circle; sampling begins at that
parameter so the flank never dips inside the root circle (which previously
produced a self-overlapping outline).
"""
if n_points < 30:
n_points = 30
points: list[tuple[float, float]] = []
if root_r < base_r:
points.append((root_r, 0.0))
ratio = outer_r / base_r if base_r > 0 else 1.0
if ratio <= 1.0:
points.append((base_r, 0.0))
return points
# Degenerate (outer circle at/below the base circle) — unreachable for
# real gears (outer_r = pitch_r + module > base_r). Anchor at the root.
return [(max(base_r, root_r), 0.0)]
t_max = math.sqrt(ratio * ratio - 1.0)
if root_r > base_r:
# Involute radius r(t) = base_r * sqrt(1 + t^2); solve r(t)=root_r.
t_min = math.sqrt((root_r / base_r) ** 2 - 1.0)
else:
t_min = 0.0
points.append((root_r, 0.0)) # radial fillet up to the base circle
for i in range(n_points):
t = t_max * i / (n_points - 1)
t = t_min + (t_max - t_min) * i / (n_points - 1)
points.append(involute_xy(base_r, t))
return points
@@ -208,7 +225,10 @@ async def draw_helical_gear_front_view(
outer_circ = await backend.entity_create_circle(cx, cy, outer_r, layer="GEOMETRY")
pitch_circ = await backend.entity_create_circle(cx, cy, pitch_r, layer="CENTER")
base_circ = await backend.entity_create_circle(cx, cy, base_r, layer="CONSTRUCTION")
# Base circle is a persistent involute reference on PHANTOM — NOT CONSTRUCTION,
# which the mandated construction_clear() before finalize would delete, leaving
# the returned base_circle handle dangling.
base_circ = await backend.entity_create_circle(cx, cy, base_r, layer="PHANTOM")
root_circ = await backend.entity_create_circle(cx, cy, root_r, layer="GEOMETRY")
pts = generate_full_gear_outline(
@@ -350,15 +370,10 @@ async def draw_gear_section_aa(
keyway_dict: dict | None = None
if bore_d is not None:
bore_r = float(bore_d) / 2.0
bore_top = await backend.entity_create_line(
x_left, cy + bore_r, x_right, cy + bore_r, layer="GEOMETRY",
)
bore_bot = await backend.entity_create_line(
x_left, cy - bore_r, x_right, cy - bore_r, layer="GEOMETRY",
)
bore_handles = [bore_top.handle, bore_bot.handle]
if keyway_w is not None or keyway_h is not None:
# draw_keyway_section draws BOTH bore lines (same span/y) AND the
# notch — drawing the bore lines here too would duplicate them and
# trip the duplicate_entities critique. Let it own the bore lines.
keyway_dict = await draw_keyway_section(
backend,
center=(section_cx, cy),
@@ -367,6 +382,15 @@ async def draw_gear_section_aa(
keyway_width=keyway_w,
keyway_depth=keyway_h,
)
bore_handles = [keyway_dict["bore_top"], keyway_dict["bore_bottom"]]
else:
bore_top = await backend.entity_create_line(
x_left, cy + bore_r, x_right, cy + bore_r, layer="GEOMETRY",
)
bore_bot = await backend.entity_create_line(
x_left, cy - bore_r, x_right, cy - bore_r, layer="GEOMETRY",
)
bore_handles = [bore_top.handle, bore_bot.handle]
return {
"top": top.handle,
+17
View File
@@ -68,6 +68,23 @@ LAYER_SET_REGISTRY: dict[str, list[tuple[str, int, str, float, str]]] = {
"iso13567": ISO13567_LAYERS,
}
# Role → layer name, per layer set. Lets dimension_auto / construction ops
# resolve the correct layer for the active standard instead of hardcoding
# "DIM"/"CONSTRUCTION" (which only exist in the mech/pid sets — iso13567 uses
# M-DIMEN-T-N / M-CONST-E-N).
LAYER_ROLES: dict[str, dict[str, str]] = {
"mech": {"dim": "DIM", "construction": "CONSTRUCTION"},
"pid": {"dim": "DIM", "construction": "CONSTRUCTION"},
"iso13567": {"dim": "M-DIMEN-T-N", "construction": "M-CONST-E-N"},
}
def resolve_role_layer(layer_set_id: str | None, role: str) -> str:
"""Resolve the layer name for `role` ('dim' | 'construction') in the active
layer set, falling back to the mech/default name when unknown."""
roles = LAYER_ROLES.get(layer_set_id or "mech", LAYER_ROLES["mech"])
return roles.get(role, LAYER_ROLES["mech"][role])
STANDARD_LINETYPES: list[str] = ["CENTER", "HIDDEN", "PHANTOM", "DASHED", "DASHDOT"]
+33 -9
View File
@@ -22,6 +22,14 @@ DIMENSION_LIKE_TEXT = re.compile(
re.IGNORECASE,
)
# R27 — match SPUR / HELICAL as whole words (not substrings of e.g.
# "SPURIOUS") when checking title/helix consistency.
_SPUR_RE = re.compile(r"\bSPUR\b", re.IGNORECASE)
_HELICAL_RE = re.compile(r"\bHELICAL\b", re.IGNORECASE)
# Layers that hold title-block text (scoped scan target for step 5).
_TITLE_LAYERS = {"TITLE", "TITLEBLOCK"}
@dataclass
class ValidationFinding:
@@ -141,38 +149,54 @@ class DrawingValidator:
"bore_missing",
"Expected bore but no circle on GEOMETRY layer found",
))
# draw_keyed_bore emits the keyway notch as an OPEN 4-point polyline
# on GEOMETRY — requiring `closed` here made this a permanent
# false-negative. Match a small (3-6 point) polyline, open or closed.
keyway_polys = [
e for e in ents
if e.type in ("LWPOLYLINE", "POLYLINE")
and e.layer == "GEOMETRY"
and len(e.properties.get("points", [])) <= 6
and e.properties.get("closed")
and 3 <= len(e.properties.get("points", [])) <= 6
]
if expected.get("must_have_keyway") and len(keyway_polys) < 1:
findings.append(ValidationFinding(
"warning",
"keyway_section_unverified",
"Could not confirm keyway in section view via heuristic "
"(small closed polyline on GEOMETRY)",
"Could not confirm keyway via heuristic "
"(small 3-6 point polyline on GEOMETRY)",
))
# 5 — title consistency
helix = expected.get("helix_angle")
if helix:
title_texts = [
(e.properties.get("text", "") or "").upper()
# R27 — scope the scan to title-block text where possible (TITLE /
# TITLEBLOCK layer); fall back to all TEXT-layer text if no title
# layer carries any text.
title_block_texts = [
e.properties.get("text", "") or ""
for e in ents
if e.type in ("TEXT", "MTEXT") and e.layer == "TEXT"
if e.type in ("TEXT", "MTEXT") and e.layer in _TITLE_LAYERS
]
if title_block_texts:
title_texts = title_block_texts
else:
title_texts = [
e.properties.get("text", "") or ""
for e in ents
if e.type in ("TEXT", "MTEXT") and e.layer == "TEXT"
]
joined = " | ".join(title_texts)
if "HELICAL SPUR" in joined:
# R27 — whole-word matching so "SPURIOUS" no longer trips SPUR.
has_spur = bool(_SPUR_RE.search(joined))
has_helical = bool(_HELICAL_RE.search(joined))
if has_spur and has_helical:
findings.append(ValidationFinding(
"error",
"title_helical_spur",
"Title contains 'HELICAL SPUR GEAR' — invalid; use "
"'HELICAL GEAR' or 'SPUR GEAR'",
))
elif "SPUR" in joined and "HELICAL" not in joined:
elif has_spur:
findings.append(ValidationFinding(
"error",
"title_says_spur_for_helical",
+196 -61
View File
@@ -241,16 +241,111 @@ def _dc(obj) -> dict:
return obj
def _registered_tool_count() -> int:
"""Number of @mcp.tool registrations currently on the server.
async def _registered_tools() -> list:
"""Return the FunctionTool objects currently registered on the server.
Used by system_status / system_about so the count never drifts from reality.
R20: prefer the public FastMCP 3.0.1 accessor (``mcp._list_tools()``,
also exposed as ``mcp.list_tools()``) which returns a ``Sequence[Tool]``.
Only fall back to the private ``_local_provider._components`` registry if
the public path is unavailable. Never raises; returns [] when unknown so
callers can label/omit rather than surface a bogus count.
"""
try:
tools = await mcp._list_tools()
return list(tools)
except Exception as exc: # pragma: no cover - public path is the norm
log.debug("public _list_tools() failed, falling back: %s", exc)
try:
components = mcp._local_provider._components
return sum(1 for k in components if k.startswith("tool:"))
except Exception:
return -1
return [v for k, v in components.items() if k.startswith("tool:")]
except Exception as exc:
log.debug("private tool registry unavailable: %s", exc)
return []
async def _registered_tool_count() -> int | None:
"""Number of @mcp.tool registrations currently on the server.
R20: backed by the public accessor (see ``_registered_tools``). Returns
``None`` (never -1) when the count is genuinely unknown so system_status /
system_about can omit/label it rather than report a fake value.
"""
tools = await _registered_tools()
return len(tools) if tools else None
# Priority order used to bucket each registered tool into exactly one group
# for the system_about breakdown (R15). The first matching tag wins, so more
# specific tags (engineering/premium/corner) take precedence over the broad
# entity/drawing tags. Tools whose tags match none fall into "other".
_GROUP_TAG_PRIORITY = (
"engineering",
"premium",
"corner",
"batch",
"template",
"dimension",
"transaction",
"view",
"validation",
"analysis",
"block",
"layer",
"linetype",
"create",
"modify",
"query",
"drawing",
"system",
)
# Map the winning tag to a human-readable group label for the breakdown.
_GROUP_TAG_LABELS = {
"engineering": "engineering",
"premium": "premium",
"corner": "corner_ops",
"batch": "batch",
"template": "templates",
"dimension": "dimensions",
"transaction": "transactions",
"view": "view",
"analysis": "analysis",
"validation": "validation",
"block": "blocks",
"layer": "layers",
"linetype": "layers",
"create": "entity_creation",
"modify": "entity_modification",
"query": "entity_query",
"drawing": "drawing",
"system": "system",
}
async def _tool_groups() -> dict:
"""Derive the system_about per-group tool breakdown dynamically from each
registered tool's tags (R15).
Replaces the hand-maintained static dict that omitted ~22 tools (all
engineering/premium/corner-ops + drawing_close) and misfiled
entity_delete_many under entity_creation. Because this reads the live
``tags={...}`` on each @mcp.tool, the breakdown can never drift again.
"""
groups: dict[str, list[str]] = {}
for tool in await _registered_tools():
name = getattr(tool, "name", None)
if not name:
continue
tags = getattr(tool, "tags", None) or set()
label = "other"
for tag in _GROUP_TAG_PRIORITY:
if tag in tags:
label = _GROUP_TAG_LABELS[tag]
break
groups.setdefault(label, []).append(name)
for names in groups.values():
names.sort()
return dict(sorted(groups.items()))
# ---------------------------------------------------------------------------
@@ -353,13 +448,22 @@ async def drawing_save(
)
async def drawing_save_as(
path: Annotated[str, "Full destination path including extension"],
format: Annotated[str, "Output format: dwg, dxf, dwt"] = "dwg",
format: Annotated[str, "Output format override: dwg, dxf, dwt. Default: derive "
"from the path extension (the extension is authoritative)."] = "",
ctx: Context = None,
) -> dict:
"""Save current drawing to a new path/format (DWG, DXF, or DWT template)."""
"""Save current drawing to a new path/format (DWG, DXF, or DWT template).
The on-disk format is derived from the file extension so the bytes always
match the name (N2) — e.g. 'part.dxf' writes DXF, not DWG. `format` overrides
only when the path has no recognised extension.
"""
validated = validate_path(path, allow_write=True)
await ctx.info(f"Saving as {format}: {validated}")
return await _backend(ctx).drawing_save_as(str(validated), format)
from pathlib import Path as _P
ext = _P(str(validated)).suffix.lstrip(".").lower()
fmt = ext if ext in ("dxf", "dwg", "dwt") else (format.lower() or "dxf")
await ctx.info(f"Saving as {fmt}: {validated}")
return await _backend(ctx).drawing_save_as(str(validated), fmt)
@mcp.tool(
@@ -583,13 +687,14 @@ async def entity_create_mtext(
y: Annotated[float, "Insertion point Y"],
width: Annotated[float, "Text box width in drawing units"] = 100.0,
height: Annotated[float, "Character height in drawing units"] = 2.5,
rotation: Annotated[float, "Rotation in degrees, CCW from +X"] = 0.0,
layer: Annotated[str | None, "Layer name"] = None,
color: Annotated[int | None, "ACI color code"] = None,
ctx: Context = None,
) -> dict:
"""Create a multi-line text entity (MTEXT) with word-wrap at the specified width."""
await ctx.debug(f"Creating mtext at ({x},{y}) w={width}")
result = await _backend(ctx).entity_create_mtext(text, x, y, width, height, layer, color)
result = await _backend(ctx).entity_create_mtext(text, x, y, width, height, rotation, layer, color)
return _dc(result)
@@ -1986,21 +2091,26 @@ async def system_status(ctx: Context = None) -> dict:
Returns backend name, connection status, capabilities, document info.
"""
b = ctx.lifespan_context.get("backend")
tool_count = _registered_tool_count()
# R20: tool_count may be None when the registry is unavailable; omit the
# key rather than surface a bogus value (e.g. the old -1).
tool_count = await _registered_tool_count()
unsafe = config.settings.dangerous_commands_enabled
if b is None:
return {
out = {
"server": "AutoCAD MCP Pro",
"backend": "none",
"connected": False,
"tool_count": tool_count,
"unsafe_mode": unsafe,
"error": ctx.lifespan_context.get("init_error"),
"hint": "Set AUTOCAD_MCP_BACKEND=ezdxf to use headless mode, or start AutoCAD for COM mode.",
}
if tool_count is not None:
out["tool_count"] = tool_count
return out
status = await b.system_status()
status["server"] = "AutoCAD MCP Pro"
status["tool_count"] = tool_count
if tool_count is not None:
status["tool_count"] = tool_count
status["unsafe_mode"] = unsafe
if unsafe:
status["unsafe_mode_warning"] = (
@@ -2082,51 +2192,23 @@ async def system_about(ctx: Context = None) -> dict:
"""Get detailed information about AutoCAD MCP Pro capabilities and available tools."""
b = ctx.lifespan_context.get("backend")
backend_name = b.name if b else "none"
return {
# R15: derive the per-group breakdown dynamically from each tool's tags so
# it can never drift from the registered surface (the old hand-maintained
# dict omitted all engineering/premium/corner-ops tools + drawing_close and
# misfiled entity_delete_many under entity_creation).
tool_count = await _registered_tool_count()
out = {
"name": "AutoCAD MCP Pro",
"version": "1.0.0",
"description": "Production-grade AutoCAD MCP server with dual COM+ezdxf engine",
"active_backend": backend_name,
"tool_groups": {
"drawing": ["drawing_info", "drawing_new", "drawing_open", "drawing_save",
"drawing_save_as", "drawing_export_dxf", "drawing_export_pdf",
"drawing_purge", "drawing_audit", "drawing_undo", "drawing_redo"],
"entity_creation": ["entity_create_line", "entity_create_circle", "entity_create_arc",
"entity_create_polyline", "entity_create_rectangle",
"entity_create_text", "entity_create_mtext", "entity_create_hatch",
"entity_create_spline", "entity_create_ellipse",
"entity_create_point", "entity_create_block_ref",
"entity_delete_many"],
"dimensions": ["dimension_linear", "dimension_aligned", "dimension_angular",
"dimension_radius", "dimension_diameter"],
"entity_modification": ["entity_move", "entity_copy", "entity_rotate", "entity_scale",
"entity_mirror", "entity_offset", "entity_delete",
"entity_array_rectangular", "entity_array_polar",
"entity_set_properties"],
"entity_query": ["entity_get", "entity_list"],
"batch": ["entity_batch_create", "entity_batch_modify"],
"templates": ["template_apply_layers", "template_list"],
"layers": ["layer_list", "layer_create", "layer_delete", "layer_set_current",
"layer_modify", "layer_freeze", "layer_thaw", "layer_lock",
"layer_unlock", "layer_hide", "layer_show", "layer_isolate",
"linetype_list", "linetype_load"],
"blocks": ["block_list", "block_insert", "block_explode",
"block_get_attributes", "block_set_attributes",
"block_create_from_entities", "block_find_references"],
"analysis": ["analysis_entity_stats", "analysis_find_in_region",
"analysis_measure_distance", "analysis_measure_area",
"analysis_bounding_box", "analysis_select_by_layer",
"analysis_select_by_type", "analysis_layer_stats"],
"validation": ["validation_check"],
"view": ["view_zoom_extents", "view_zoom_window", "view_screenshot",
"view_zoom_and_screenshot"],
"transactions": ["transaction_begin", "transaction_commit", "transaction_rollback"],
"system": ["system_status", "system_get_variable", "system_set_variable",
"system_run_command", "system_run_lisp", "system_about"],
},
"total_tools": _registered_tool_count(),
"tool_groups": await _tool_groups(),
"unsafe_mode": config.settings.dangerous_commands_enabled,
}
# R20: omit total_tools when unknown rather than reporting a fake -1.
if tool_count is not None:
out["total_tools"] = tool_count
return out
# ---------------------------------------------------------------------------
@@ -2305,11 +2387,19 @@ async def drawing_finalize(
description="If given, write PNG screenshot to this path.")] = None,
expected: Annotated[dict | None, Field(default=None,
description="Optional contract: {'part_type', 'helix_angle', 'must_have_bore', 'must_have_keyway'}")] = None,
strict_critique: Annotated[bool, Field(default=False,
description="If true, ANY critique issue (including warnings) fails the gate — the full "
"premium discipline. Default false: only critique 'error' issues (e.g. leftover "
"construction geometry) fail; warnings are surfaced in the payload.")] = False,
ctx: Context = None,
) -> dict:
"""8-step pre-completion validator: real dimensions, no hidden orphans, linetypes loaded,
bore+keyway in both views (if expected), title consistency, saved to disk, screenshot exported,
DWG path returned. Raises ToolError if any 'error' severity finding is present.
"""Premium completion gate: runs BOTH the 8-step validator AND the premium critique focuses
(iso128, layer_color, dim_overlap, untrimmed_corner, duplicate_entities, construction_left),
then saves to disk, exports a screenshot, and returns the DWG path.
Raises ToolError if any validator 'error' finding is present, or if critique reports an
'error' (or, with strict_critique=True, any critique issue). Critique warnings are surfaced
under payload['critique'] without failing the gate by default.
"""
from engineering import DrawingValidator
backend = _backend(ctx)
@@ -2328,17 +2418,44 @@ async def drawing_finalize(
from pathlib import Path as _P
_P(str(validated_shot)).write_bytes(shot)
except Exception as exc:
await ctx.warning(f"Screenshot export failed: {exc}")
if ctx is not None:
await ctx.warning(f"Screenshot export failed: {exc}")
result = await DrawingValidator().run(backend, expected=expected or {})
payload = result.to_dict()
# I15 — the premium critique focuses are part of the finalize gate, not merely advisory.
critique_issues = await backend.drawing_critique(focus=None)
crit_summary = {"error": 0, "warning": 0, "info": 0}
for issue in critique_issues:
crit_summary[issue.severity] = crit_summary.get(issue.severity, 0) + 1
payload["critique"] = [issue.to_dict() for issue in critique_issues]
payload["critique_summary"] = crit_summary
info = await backend.drawing_info()
payload["dwg_path"] = getattr(info, "full_path", "")
if ctx is not None:
for issue in critique_issues:
if issue.severity == "warning":
await ctx.warning(f"critique[{issue.focus}]: {issue.message}")
if not result.ok:
raise ToolError(
f"drawing_finalize: validation failed with {result.summary['error']} error(s). "
f"First: {result.findings[0].code}: {result.findings[0].message}"
)
gate_failures = [
i for i in critique_issues
if i.severity == "error" or (strict_critique and i.severity != "info")
]
if gate_failures:
first = gate_failures[0]
raise ToolError(
f"drawing_finalize: critique gate failed with {len(gate_failures)} blocking issue(s). "
f"First: {first.focus}: {first.message}"
)
return payload
@@ -2370,9 +2487,9 @@ async def drawing_plan(
) -> dict:
"""Commit a PlanSpec before any geometry is created.
The PlanSpec is held by the backend and replayed by `drawing_critique`
at finalize time to verify the drawing matches the original intent.
Always call this FIRST in a premium workflow.
The PlanSpec is stored on the backend and surfaced for reference during
the workflow (it is not replayed as a critique). Always call this FIRST
in a premium workflow.
"""
plan = await _backend(ctx).drawing_plan(
intent, sheet_size, scale, layer_set_id, view_count, dim_style, notes,
@@ -2918,6 +3035,24 @@ def _validate_http_bind(host: str) -> None:
)
# NEW-AUTH-1 — enforce the bind guard on EVERY launch path, not just __main__.
# The documented `fastmcp run server.py:mcp --transport http` imports this module
# and calls `mcp.run_async(...)` directly, bypassing the __main__ block. Both
# `mcp.run()` (via anyio.run(self.run_async, ...)) and the CLI funnel through
# run_async, so wrapping it on the instance closes the anonymous-remote-bind hole.
_HTTP_TRANSPORTS = {"http", "sse", "streamable-http"}
_orig_run_async = mcp.run_async
async def _guarded_run_async(transport=None, *args, **kwargs):
if (transport in _HTTP_TRANSPORTS) or (kwargs.get("transport") in _HTTP_TRANSPORTS):
_validate_http_bind(kwargs.get("host", "127.0.0.1"))
return await _orig_run_async(transport, *args, **kwargs)
mcp.run_async = _guarded_run_async
if __name__ == "__main__":
transport = "stdio"
host = "127.0.0.1"
+4
View File
@@ -39,6 +39,10 @@ NL_TABLE: list[tuple[Any, Any]] = [
(50, 50),
(100, 100),
(200, 200),
# Ambiguous bare ints (<= 2.05) — the P0 truncation hazard: treated as mm,
# NOT as hundredths (NEW-test-com-1). 2 -> 2.00mm -> 200, 1 -> 1.00mm -> 100.
(1, 100),
(2, 200),
# Sentinels: -1 = ByLayer, -2 = ByBlock, -3 = Default
(-1, -1),
(-2, -2),
+158
View File
@@ -0,0 +1,158 @@
"""Mocked-COM regression tests for the Sprint-3 COM fixes.
These exercise the COM `_sync` bodies without a live AutoCAD by:
* bypassing the STA executor (replace ``_run`` with a direct caller), and
* patching the module-level ``_acad_app`` / ``_acad_doc`` / ``_entity_info``.
Covers: N6/R23 (run_lisp guard + USERS1 capture), set_variable coercion,
R16 (transaction flag cleared on error), R13 (offset honors side_x/side_y).
"""
from __future__ import annotations
from unittest.mock import MagicMock
import pytest
import backends.com_backend as cb
from backends.com_backend import ComBackend
pytestmark = pytest.mark.asyncio
def _backend_no_executor():
b = ComBackend()
async def _run(func, *args, **kwargs): # bypass the ThreadPoolExecutor
return func(*args, **kwargs)
b._run = _run
return b
class _Doc:
"""Idle doc: GetVariable returns 0 (CMDACTIVE clear) and ModelSpace is empty."""
def __init__(self):
self.sent: list[str] = []
self._vars: dict = {}
self.ModelSpace: list = []
def GetVariable(self, name):
return self._vars.get(name, 0)
def SetVariable(self, name, val):
self._vars[name] = val
def SendCommand(self, cmd):
self.sent.append(cmd)
def EndUndoMark(self):
pass
# ── NEW-com-set-variable-coercion ───────────────────────────────────────────
async def test_set_variable_coerces_string_to_sysvar_int_type(monkeypatch):
app = MagicMock()
app.GetVariable.return_value = 4159 # OSMODE current value is an int
monkeypatch.setattr(cb, "_acad_app", lambda: app)
b = _backend_no_executor()
out = await b.system_set_variable("OSMODE", "0")
name, val = app.SetVariable.call_args[0]
assert name == "OSMODE"
assert val == 0 and isinstance(val, int) # coerced from "0" to int 0
assert out["value"] == 0
# ── R23 / N6 — run_lisp guard + value capture ───────────────────────────────
async def test_run_lisp_refuses_when_cmdactive(monkeypatch):
app = MagicMock()
app.GetVariable.side_effect = lambda n: 1 if n == "CMDACTIVE" else 0
monkeypatch.setattr(cb, "_acad_app", lambda: app)
monkeypatch.setattr(cb, "_acad_doc", lambda: _Doc())
b = _backend_no_executor()
with pytest.raises(RuntimeError):
await b.system_run_lisp("(+ 1 2)")
async def test_run_lisp_captures_value_from_users1(monkeypatch):
app = MagicMock()
app.GetVariable.side_effect = lambda n: {"CMDACTIVE": 0, "USERS1": "3"}.get(n, 0)
doc = _Doc()
monkeypatch.setattr(cb, "_acad_app", lambda: app)
monkeypatch.setattr(cb, "_acad_doc", lambda: doc)
b = _backend_no_executor()
out = await b.system_run_lisp("(+ 1 2)")
assert out["result"] == "3" # captured, not the bogus "nil"
assert any("USERS1" in c for c in doc.sent) # value stashed out-of-band
# ── R16 — transaction flag cleared even on error ────────────────────────────
async def test_transaction_begin_sets_flag_and_marks_undo(monkeypatch):
doc = _Doc()
doc.StartUndoMark = MagicMock()
monkeypatch.setattr(cb, "_acad_doc", lambda: doc)
b = _backend_no_executor()
out = await b.transaction_begin()
assert out["ok"] is True
assert b._transaction_active is True
doc.StartUndoMark.assert_called_once()
async def test_transaction_commit_clears_flag_on_error():
b = _backend_no_executor()
async def boom(func, *a, **k):
raise RuntimeError("EndUndoMark failed")
b._run = boom
b._transaction_active = True
with pytest.raises(RuntimeError):
await b.transaction_commit()
assert b._transaction_active is False # not left stale
async def test_transaction_rollback_uses_safe_send_command(monkeypatch):
doc = _Doc() # idle: CMDACTIVE=0 so _safe_send_command returns promptly
monkeypatch.setattr(cb, "_acad_doc", lambda: doc)
b = _backend_no_executor()
b._transaction_active = True
out = await b.transaction_rollback()
assert out["ok"] is True
assert b._transaction_active is False
# UNDO B went through _safe_send_command (which appends a trailing newline).
assert any("_UNDO B" in c for c in doc.sent)
# ── R13 — offset honors side_x/side_y and deletes the unused copy ───────────
async def test_offset_picks_copy_nearest_side_point(monkeypatch):
# ent.Offset(+d) -> copy centred at (0, +10); ent.Offset(-d) -> (0, -10).
pos_copy = MagicMock(name="pos")
pos_copy.GetBoundingBox.return_value = ((-1, 9), (1, 11)) # centre (0, 10)
neg_copy = MagicMock(name="neg")
neg_copy.GetBoundingBox.return_value = ((-1, -11), (1, -9)) # centre (0, -10)
ent = MagicMock()
ent.Offset.side_effect = lambda d: [pos_copy] if d > 0 else [neg_copy]
doc = MagicMock()
doc.HandleToObject.return_value = ent
monkeypatch.setattr(cb, "_acad_doc", lambda: doc)
monkeypatch.setattr(cb, "_entity_info", lambda e: e) # return the entity itself
b = _backend_no_executor()
# Side point near the negative copy -> keep neg, delete pos.
kept = await b.entity_offset("AA", 5.0, side_x=0.0, side_y=-10.0)
assert kept is neg_copy
pos_copy.Delete.assert_called_once()
neg_copy.Delete.assert_not_called()
+43
View File
@@ -229,3 +229,46 @@ class TestSanitizeLisp:
monkeypatch.delenv("DANGEROUS_COMMANDS_ENABLED", raising=False)
config.settings = Settings()
class TestSanitizerBypassVectors:
"""R8 — adversarial coverage: the deny-list must hold against common evasion
tricks. Each asserts the guard REJECTS the payload (defensive verification),
plus that legitimate input is NOT over-blocked."""
def test_command_newline_injected_second_command(self):
# A safe head command with a dangerous one smuggled after a newline.
with pytest.raises(ToolError, match="restricted command"):
sanitize_command("_ZOOM E\n_ERASE")
def test_command_case_insensitive(self):
with pytest.raises(ToolError, match="restricted command"):
sanitize_command("eRaSe all")
def test_command_legitimate_not_overblocked(self):
assert sanitize_command("_LINE 0,0 10,10") == "_LINE 0,0 10,10"
def test_lisp_symbol_aliasing_blocked(self):
# (setq f command) references `command` as a value to dodge head-only
# matching — the bare-token scan must still catch it.
with pytest.raises(ToolError, match="restricted function"):
sanitize_lisp("(setq f command)(f)")
def test_lisp_activex_family_blocked(self):
with pytest.raises(ToolError, match="restricted function"):
sanitize_lisp('(vlax-invoke obj "SendCommand" "x")')
def test_lisp_express_tools_shell_blocked(self):
with pytest.raises(ToolError, match="restricted function"):
sanitize_lisp('(acet-sys-shell "cmd")')
def test_lisp_custom_command_invocation_blocked(self):
with pytest.raises(ToolError, match="restricted function"):
sanitize_lisp("(c:mycmd)")
def test_lisp_dangerous_token_inside_wrapper_blocked(self):
with pytest.raises(ToolError, match="restricted function"):
sanitize_lisp('(startapp (strcat "c" "md"))')
def test_lisp_legitimate_not_overblocked(self):
assert sanitize_lisp("(setq x (* 2 3))") == "(setq x (* 2 3))"
+186
View File
@@ -0,0 +1,186 @@
"""Regression tests for the Sprint-2 audit fixes.
Covers, one block per audit finding:
* NEW-screenshot-1 — headless Agg render returns a valid PNG (no GUI-thread crash)
* I15 — drawing_finalize runs the premium critique as part of the gate
* R4 — dim_overlap critique actually fires (was a guaranteed no-op)
* N4 — dim / construction layers resolve per the active layer set
* NEW-AUTH-1 — the remote-bind guard runs on the run_async launch path
* NEW-gear-roottooth— gear outline never dips inside the root circle (high tooth counts)
* NEW-section-dupbore / NEW-validator-keyway / NEW-gear-baseconstr
"""
from __future__ import annotations
import math
import pytest
pytestmark = pytest.mark.asyncio
class _FakeCtx:
"""Minimal Context stand-in: drawing_finalize only needs lifespan_context + warning."""
def __init__(self, backend):
self.lifespan_context = {"backend": backend}
self.warnings: list[str] = []
async def warning(self, message):
self.warnings.append(message)
# ── NEW-screenshot-1 ────────────────────────────────────────────────────────
async def test_screenshot_renders_png_without_gui_thread_crash(backend):
"""view_screenshot runs in a worker thread; the Agg figure must not crash and
must return valid PNG bytes."""
await backend.entity_create_line(0, 0, 100, 100)
png = await backend.view_screenshot()
assert png is not None
assert png[:4] == b"\x89PNG" # PNG magic
# ── I15 — critique is part of the finalize gate ─────────────────────────────
async def test_finalize_runs_critique_and_blocks_on_construction(backend, tmp_path):
from fastmcp.exceptions import ToolError
import server
await backend.drawing_apply_iso_layers("mech")
await backend.entity_create_line(0, 0, 100, 0, layer="GEOMETRY")
await backend.construction_xline(0, 0, 90) # leftover scaffold -> construction_left ERROR
ctx = _FakeCtx(backend)
save = str(tmp_path / "part.dxf")
with pytest.raises(ToolError) as exc:
await server.drawing_finalize(save_path=save, ctx=ctx)
assert "critique" in str(exc.value).lower()
# Clear the scaffold; finalize now passes and the payload carries the critique.
await backend.construction_clear()
payload = await server.drawing_finalize(save_path=save, ctx=ctx)
assert payload["ok"] is True
assert "critique" in payload
assert payload["critique_summary"]["error"] == 0
# ── R4 — dim_overlap critique fires ─────────────────────────────────────────
async def test_dim_overlap_critique_is_not_a_noop(backend):
await backend.drawing_apply_iso_layers("mech")
# Two dimensions measuring the same segment overlap exactly.
await backend.dimension_linear(0, 0, 100, 0, 50, 20, layer="DIM")
await backend.dimension_linear(0, 0, 100, 0, 50, 20, layer="DIM")
issues = await backend.drawing_critique(focus=["dim_overlap"])
assert len(issues) >= 1
assert all(i.focus == "dim_overlap" for i in issues)
# ── N4 — dim / construction layers follow the active layer set ──────────────
async def test_iso13567_dim_and_construction_layers_resolve(backend):
await backend.drawing_apply_iso_layers("iso13567")
assert backend._active_layer_set_id() == "iso13567"
assert backend._role_layer("dim") == "M-DIMEN-T-N"
assert backend._role_layer("construction") == "M-CONST-E-N"
ln = await backend.entity_create_line(0, 0, 80, 0, layer="M-GEOMET-E-N")
dims = await backend.dimension_auto([ln.handle], style="chain")
assert dims[0].layer == "M-DIMEN-T-N"
xl = await backend.construction_xline(0, 0, 90)
assert xl.layer == "M-CONST-E-N"
# construction_left must catch iso13567 scaffold (layer name independent).
left = await backend.drawing_critique(focus=["construction_left"])
assert len(left) == 1 and left[0].severity == "error"
cleared = await backend.construction_clear()
assert cleared["deleted"] == 1
assert await backend.drawing_critique(focus=["construction_left"]) == []
# ── NEW-AUTH-1 — bind guard on the run_async launch path ────────────────────
async def test_run_async_invokes_bind_guard_for_http(monkeypatch):
import server
seen = {}
def fake_guard(host):
seen["host"] = host
raise SystemExit("blocked before bind")
monkeypatch.setattr(server, "_validate_http_bind", fake_guard)
with pytest.raises(SystemExit):
await server.mcp.run_async(transport="http", host="0.0.0.0")
assert seen["host"] == "0.0.0.0"
async def test_run_async_skips_guard_for_stdio(monkeypatch):
import server
seen = {}
monkeypatch.setattr(server, "_validate_http_bind",
lambda h: seen.setdefault("host", h))
async def fake_orig(*args, **kwargs):
return "stdio-ran"
monkeypatch.setattr(server, "_orig_run_async", fake_orig)
out = await server.mcp.run_async(transport="stdio")
assert out == "stdio-ran"
assert seen == {} # guard not invoked for stdio
# ── NEW-gear-roottooth — outline never dips inside the root circle ──────────
async def test_gear_outline_stays_outside_root_circle():
from engineering.gear import generate_full_gear_outline
for module, teeth, pa in [(2, 20, 20.0), (2, 50, 20.0), (3, 60, 20.0), (1, 6, 20.0)]:
pitch_r = module * teeth / 2.0
root_r = pitch_r - 1.25 * module
pts = generate_full_gear_outline(module, teeth, pa)
radii = [math.hypot(x, y) for x, y in pts]
# No point may sit inside the root circle (the self-overlap bug for z>=~42).
assert min(radii) >= root_r - 1e-6, f"m={module} z={teeth}: min_r {min(radii)} < root_r {root_r}"
# ── NEW-section-dupbore / NEW-validator-keyway / NEW-gear-baseconstr ─────────
async def test_gear_front_view_base_circle_persists_and_keyway_validates(backend, tmp_path):
from engineering.gear import draw_helical_gear_front_view
from engineering.validator import DrawingValidator
await backend.drawing_apply_iso_layers("mech")
res = await draw_helical_gear_front_view(
backend, module=2, teeth=20, helix_angle=0, center=(0, 0),
bore_diameter=20, keyway_width=6, keyway_depth=2.8,
)
# NEW-gear-baseconstr: base circle is a persistent PHANTOM reference, not CONSTRUCTION.
base = await backend.entity_get(res["base_circle"])
assert base.layer == "PHANTOM"
await backend.construction_clear()
assert await backend.entity_get(res["base_circle"]) # survives, handle still valid
# NEW-validator-keyway: the open 4-point keyway polyline is now recognised.
await backend.drawing_save_as(str(tmp_path / "gear.dxf"))
vr = await DrawingValidator().run(
backend, expected={"must_have_bore": True, "must_have_keyway": True},
)
assert "keyway_section_unverified" not in [f.code for f in vr.findings]
async def test_gear_section_does_not_duplicate_bore_lines(backend):
from engineering.gear import draw_gear_section_aa
await backend.drawing_apply_iso_layers("mech")
await draw_gear_section_aa(
backend,
gear_metadata={"outer_radius": 22, "center": [0, 0], "bore_diameter": 20,
"keyway_width": 6, "keyway_depth": 2.8},
x_offset=100, face_width=30,
)
dups = await backend.drawing_critique(focus=["duplicate_entities"])
assert dups == []
+107
View File
@@ -0,0 +1,107 @@
"""Regression tests for the CI-verifiable (ezdxf / shared) Sprint-3 fixes.
* N2 — drawing_save_as derives the on-disk format from the path extension
* N3 — ARC carries a `length` property; entity_select_smart length_range selects it
* N5 — properties parity: ezdxf sets bounding_box; MTEXT carries char_height + rotation
* NEW-dimauto-baseline-rotation — baseline dims run parallel to a rotated baseline
* NEW-dimauto-ordinate-misnomer — ordinate dims are staggered, not stacked at one offset
"""
from __future__ import annotations
import os
import tempfile
import pytest
pytestmark = pytest.mark.asyncio
class _Ctx:
def __init__(self, backend):
self.lifespan_context = {"backend": backend}
async def info(self, message):
pass
# ── N2 — format follows the file extension ──────────────────────────────────
async def test_save_as_derives_format_from_extension(backend):
import server
ctx = _Ctx(backend)
d = tempfile.mkdtemp()
res = await server.drawing_save_as(path=os.path.join(d, "part.dxf"), ctx=ctx)
assert res.get("format") == "dxf" # not the old "dwg" default
# ezdxf cannot write real DWG — must refuse rather than mislabel a .dwg file.
with pytest.raises(RuntimeError):
await server.drawing_save_as(path=os.path.join(d, "part.dwg"), ctx=ctx)
# ── N3 — ARC length + length_range selection ────────────────────────────────
async def test_arc_has_length_and_is_selectable(backend):
arc = await backend.entity_create_arc(0, 0, 10, 0, 90) # r=10, 90deg -> len ~15.708
info = await backend.entity_get(arc.handle)
assert info.properties.get("length") == pytest.approx(15.70796, rel=1e-4)
sel = await backend.entity_select_smart({"type": "ARC", "length_range": [15, 16]})
assert len(sel) == 1
# Out-of-range excludes it (was: all arcs silently rejected for lack of length).
assert await backend.entity_select_smart({"type": "ARC", "length_range": [1, 2]}) == []
# ── N5 — cross-backend property parity ──────────────────────────────────────
async def test_ezdxf_populates_bounding_box(backend):
line = await backend.entity_create_line(0, 0, 30, 40)
info = await backend.entity_get(line.handle)
bb = info.properties.get("bounding_box")
assert bb is not None and set(bb) == {"min", "max"} # same shape as COM
async def test_mtext_has_char_height_and_rotation(backend):
mt = await backend.entity_create_mtext("hello", 5, 5, height=2.5)
info = await backend.entity_get(mt.handle)
assert "char_height" in info.properties
assert "rotation" in info.properties
async def test_mtext_honors_caller_rotation(backend):
# NEW-mtext-1: entity_create_mtext now accepts a rotation that round-trips.
mt = await backend.entity_create_mtext("rot", 0, 0, width=50, height=2.5, rotation=30.0)
info = await backend.entity_get(mt.handle)
assert float(info.properties.get("rotation", 0.0)) == pytest.approx(30.0, abs=1e-6)
# ── NEW-dimauto-baseline-rotation ───────────────────────────────────────────
async def test_baseline_dim_runs_parallel_to_rotated_baseline(backend):
ln = await backend.entity_create_line(0, 0, 40, 40) # 45-degree baseline
dims = await backend.dimension_auto([ln.handle], style="baseline")
doc = backend._require_doc()
ent = doc.entitydb[dims[0].handle]
# Dimension line angle must match the baseline (45), not the old hardcoded 0.
assert ent.dxf.get("angle", 0.0) == pytest.approx(45.0, abs=1e-6)
assert ent.get_measurement() == pytest.approx(56.5685, rel=1e-3) # true len, not projection
# ── NEW-dimauto-ordinate-misnomer ───────────────────────────────────────────
async def test_ordinate_dims_are_staggered_not_overlapping(backend):
l1 = await backend.entity_create_line(0, 0, 20, 10)
l2 = await backend.entity_create_line(0, 0, 40, 30)
dims = await backend.dimension_auto([l1.handle, l2.handle], style="ordinate")
# Each segment emits an X and a Y dim; with per-feature staggering the
# X-dim reference Y-offsets differ between the two features.
doc = backend._require_doc()
y_offsets = []
for d in dims:
ent = doc.entitydb[d.handle]
dp = ent.dxf.get("defpoint", None)
if dp is not None:
y_offsets.append(round(float(dp[1]), 3))
# Not every dim collapses onto a single offset value.
assert len(set(y_offsets)) > 1
+223
View File
@@ -0,0 +1,223 @@
"""Sprint-4 regression tests for backends/base.py concrete helpers.
Covers:
- N7 — point_from_snap ARC end/mid raises a descriptive RuntimeError (not an
opaque KeyError) when start_angle/end_angle are absent.
- NEW-base-1 — point_tangent does not raise a raw ValueError ("math domain
error") for a from-point exactly on / marginally inside the perimeter band.
- NEW-snap-quad-arc / NEW-snap-near-arc — quad/near snaps on an ARC return a
point that lies on the arc sweep (not an off-sweep full-circle point).
- N8 — get_plan_spec() makes the stored PlanSpec readable.
"""
from __future__ import annotations
import math
import pytest
from backends.base import EntityInfo
pytestmark = pytest.mark.asyncio
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _on_arc(pt, cx, cy, r, sa_deg, ea_deg, tol=1e-6):
"""True if pt lies on the circle of radius r AND within the [sa, ea] sweep
(degrees, CCW, with wraparound)."""
if abs(math.hypot(pt[0] - cx, pt[1] - cy) - r) > tol:
return False
theta = math.degrees(math.atan2(pt[1] - cy, pt[0] - cx)) % 360.0
sa = sa_deg % 360.0
sweep = (ea_deg - sa_deg) % 360.0
rel = (theta - sa) % 360.0
return rel <= sweep + 1e-6
# ---------------------------------------------------------------------------
# N7 — ARC snap angle-missing raises RuntimeError, not KeyError
# ---------------------------------------------------------------------------
class TestN7ArcAngleMissing:
async def _patch_entity_get(self, backend, info: EntityInfo):
async def _fake(handle):
return info
backend.entity_get = _fake # type: ignore[method-assign]
async def test_end_snap_missing_angles_raises_runtimeerror(self, backend):
"""An ARC entity whose properties lack start_angle/end_angle must raise
a descriptive RuntimeError, not a KeyError."""
bad = EntityInfo(
handle="DEAD",
type="ARC",
layer="0",
color=256,
linetype="ByLayer",
visible=True,
properties={"center": [0.0, 0.0], "radius": 10.0}, # no angles
)
await self._patch_entity_get(backend, bad)
with pytest.raises(RuntimeError, match="start_angle/end_angle"):
await backend.point_from_snap("DEAD", "end")
async def test_mid_snap_missing_angles_raises_runtimeerror(self, backend):
bad = EntityInfo(
handle="DEAD",
type="ARC",
layer="0",
color=256,
linetype="ByLayer",
visible=True,
properties={"center": [0.0, 0.0], "radius": 10.0},
)
await self._patch_entity_get(backend, bad)
with pytest.raises(RuntimeError, match="start_angle/end_angle"):
await backend.point_from_snap("DEAD", "mid")
async def test_end_snap_missing_angles_not_keyerror(self, backend):
"""Explicitly assert the failure is NOT the old opaque KeyError."""
bad = EntityInfo(
handle="DEAD",
type="ARC",
layer="0",
color=256,
linetype="ByLayer",
visible=True,
properties={"center": [0.0, 0.0], "radius": 10.0},
)
await self._patch_entity_get(backend, bad)
with pytest.raises(Exception) as excinfo:
await backend.point_from_snap("DEAD", "end")
assert not isinstance(excinfo.value, KeyError)
assert isinstance(excinfo.value, RuntimeError)
async def test_real_arc_end_snap_still_works(self, backend):
"""Sanity: a well-formed ARC still resolves end-snap correctly."""
arc = await backend.entity_create_arc(0, 0, 10, 0, 90)
sp = await backend.point_from_snap(arc.handle, "end", ref_x=11, ref_y=0)
# start angle 0 -> (10, 0)
assert abs(sp[0] - 10.0) < 1e-6
assert abs(sp[1] - 0.0) < 1e-6
# ---------------------------------------------------------------------------
# NEW-base-1 — point_tangent on/near the perimeter must not raise ValueError
# ---------------------------------------------------------------------------
class TestNewBase1TangentClamp:
async def test_from_point_exactly_on_circle_no_valueerror(self, backend):
"""A from-point exactly on the perimeter (d == r) degenerates to the
tangent-at-that-point; -r/d == -1 must not blow up math.acos."""
circ = await backend.entity_create_circle(0, 0, 10)
# Point exactly on perimeter at (10, 0).
pt = await backend.point_tangent(circ.handle, 10.0, 0.0)
# Tangent point coincides with the from-point (on the circle).
assert abs(math.hypot(pt[0], pt[1]) - 10.0) < 1e-6
assert abs(pt[0] - 10.0) < 1e-6
assert abs(pt[1] - 0.0) < 1e-6
async def test_from_point_in_narrow_inside_band_no_valueerror(self, backend):
"""A from-point a hair inside the perimeter (within 1e-9 of r, so past
the d < r-1e-9 guard) used to raise a raw 'math domain error'
ValueError. After clamping it must degenerate without ValueError."""
circ = await backend.entity_create_circle(0, 0, 10)
# d = 10 - 5e-10 -> inside r but NOT below r-1e-9, so the guard passes
# and -r/d is marginally < -1.
fx = 10.0 - 5e-10
try:
pt = await backend.point_tangent(circ.handle, fx, 0.0)
except ValueError as exc: # pragma: no cover - this is the bug we fixed
pytest.fail(f"point_tangent raised ValueError (NEW-base-1): {exc}")
# Result still lies on the circle.
assert abs(math.hypot(pt[0], pt[1]) - 10.0) < 1e-4
async def test_genuine_internal_point_still_raises_runtimeerror(self, backend):
"""The internal-point guard (d < r) must still raise RuntimeError."""
circ = await backend.entity_create_circle(0, 0, 10)
with pytest.raises(RuntimeError, match="inside"):
await backend.point_tangent(circ.handle, 1.0, 1.0)
# ---------------------------------------------------------------------------
# NEW-snap-quad-arc / NEW-snap-near-arc — snaps stay on the arc sweep
# ---------------------------------------------------------------------------
class TestNewSnapArcSweep:
async def test_quad_on_quarter_arc_stays_on_sweep(self, backend):
"""Quarter arc 0..90 deg, r=10 at origin. Only the 0-deg and 90-deg
quadrants lie on the sweep; quad snap must return one of those, never
the 180/270 quadrant points that are off the arc."""
arc = await backend.entity_create_arc(0, 0, 10, 0, 90)
# ref near the (10,0) quadrant
q = await backend.point_from_snap(arc.handle, "quad", ref_x=12, ref_y=0)
assert _on_arc(q, 0, 0, 10, 0, 90), f"quad point {q} off sweep"
# ref near the (0,10) quadrant
q2 = await backend.point_from_snap(arc.handle, "quad", ref_x=0, ref_y=12)
assert _on_arc(q2, 0, 0, 10, 0, 90), f"quad point {q2} off sweep"
async def test_quad_ref_toward_offsweep_quadrant_falls_back_to_sweep(self, backend):
"""Even when ref points toward the (-10,0) quadrant (off the 0..90
sweep), the returned quad point must remain on the sweep."""
arc = await backend.entity_create_arc(0, 0, 10, 0, 90)
q = await backend.point_from_snap(arc.handle, "quad", ref_x=-50, ref_y=-50)
assert _on_arc(q, 0, 0, 10, 0, 90), f"quad point {q} off sweep"
async def test_quad_no_quadrant_on_sweep_falls_back_to_endpoint(self, backend):
"""A narrow arc (10..80 deg) contains no axis quadrant; quad must fall
back to an arc endpoint, which is on the sweep."""
arc = await backend.entity_create_arc(0, 0, 10, 10, 80)
q = await backend.point_from_snap(arc.handle, "quad", ref_x=20, ref_y=0)
assert _on_arc(q, 0, 0, 10, 10, 80), f"quad point {q} off sweep"
async def test_near_within_sweep_projects_onto_arc(self, backend):
"""A ref inside the 0..90 sweep direction projects onto the arc."""
arc = await backend.entity_create_arc(0, 0, 10, 0, 90)
# 45-degree direction is inside the sweep.
n = await backend.point_from_snap(arc.handle, "near", ref_x=5, ref_y=5)
assert _on_arc(n, 0, 0, 10, 0, 90), f"near point {n} off sweep"
# Should be ~ (10/sqrt2, 10/sqrt2)
assert abs(n[0] - 10 / math.sqrt(2)) < 1e-6
assert abs(n[1] - 10 / math.sqrt(2)) < 1e-6
async def test_near_outside_sweep_clamps_to_endpoint(self, backend):
"""A ref in the (-x,-y) direction is outside the 0..90 sweep; near must
clamp to the nearer arc endpoint (on the sweep), not return an off-sweep
full-circle projection."""
arc = await backend.entity_create_arc(0, 0, 10, 0, 90)
# Direction ~ -27deg (just below the x-axis) is off the 0..90 sweep.
# Of the two endpoints (10,0) and (0,10), the start (10,0) is clearly
# nearer to (10, -5), so near must clamp there.
n = await backend.point_from_snap(arc.handle, "near", ref_x=10, ref_y=-5)
assert _on_arc(n, 0, 0, 10, 0, 90), f"near point {n} off sweep"
# Endpoints are (10,0) [start] and (0,10) [end]; (10,0) is nearer.
assert abs(n[0] - 10.0) < 1e-6
assert abs(n[1] - 0.0) < 1e-6
async def test_circle_quad_unchanged(self, backend):
"""CIRCLE quad behaviour is preserved (all four quadrants available)."""
circ = await backend.entity_create_circle(0, 0, 10)
q = await backend.point_from_snap(circ.handle, "quad", ref_x=-12, ref_y=0)
assert abs(q[0] - (-10.0)) < 1e-6
assert abs(q[1] - 0.0) < 1e-6
# ---------------------------------------------------------------------------
# N8 — get_plan_spec accessor exposes the stored PlanSpec
# ---------------------------------------------------------------------------
class TestN8GetPlanSpec:
async def test_none_before_plan(self, backend):
assert backend.get_plan_spec() is None
async def test_returns_dict_after_plan(self, backend):
await backend.drawing_plan("test shaft", sheet_size="A3", scale=2.0)
spec = backend.get_plan_spec()
assert isinstance(spec, dict)
assert spec["intent"] == "test shaft"
assert spec["sheet_size"] == "A3"
assert spec["scale"] == 2.0
# Returned object is a plain dict snapshot (not the live dataclass).
assert "layer_set_id" in spec
+180
View File
@@ -0,0 +1,180 @@
"""Mocked-COM regression tests for the Sprint-4 COM fixes.
These run without a live AutoCAD (and OS-agnostically) by:
* patching the module-level ``_acad_app`` / ``_acad_doc`` with MagicMocks, and
* forcing ``_WIN32_AVAILABLE`` True so the win32-guarded paths execute
(no real win32 calls are made — every win32 attribute is mocked).
* bypassing the STA executor (replace ``_run`` with a direct caller).
Covers: R28 (_find_autocad_hwnd prefers Application.HWND over EnumWindows),
S3 (block_list populates BlockInfo.description from the definition .Comments).
"""
from __future__ import annotations
from unittest.mock import MagicMock
import pytest
import backends.com_backend as cb
from backends.com_backend import ComBackend
pytestmark = pytest.mark.asyncio
def _backend_no_executor() -> ComBackend:
b = ComBackend()
async def _run(func, *args, **kwargs): # bypass the ThreadPoolExecutor
return func(*args, **kwargs)
b._run = _run
return b
# ── R28 — _find_autocad_hwnd prefers Application.HWND over EnumWindows ───────
async def test_find_hwnd_prefers_application_hwnd(monkeypatch):
"""The real main-frame handle from the COM app wins; EnumWindows is not
consulted when the HWND read succeeds."""
app = MagicMock()
app.HWND = 0xABCD # known int handle
monkeypatch.setattr(cb, "_acad_app", lambda: app)
monkeypatch.setattr(cb, "_WIN32_AVAILABLE", True)
enum = MagicMock(side_effect=AssertionError("EnumWindows must not be called"))
monkeypatch.setattr(cb, "win32gui", MagicMock(EnumWindows=enum))
assert cb._find_autocad_hwnd() == 0xABCD
enum.assert_not_called()
async def test_find_hwnd_coerces_hwnd_to_int(monkeypatch):
"""A COM HWND that arrives as a numeric string is coerced to int."""
app = MagicMock()
app.HWND = "43981" # COM can hand back a stringy handle
monkeypatch.setattr(cb, "_acad_app", lambda: app)
monkeypatch.setattr(cb, "_WIN32_AVAILABLE", True)
monkeypatch.setattr(cb, "win32gui", MagicMock())
result = cb._find_autocad_hwnd()
assert result == 43981
assert isinstance(result, int)
async def test_find_hwnd_falls_back_to_enumwindows(monkeypatch):
"""When the HWND read raises, the EnumWindows title scan is used as a
fallback and its discovered handle is returned."""
app = MagicMock()
type(app).HWND = property( # reading .HWND raises
lambda self: (_ for _ in ()).throw(RuntimeError("no HWND"))
)
monkeypatch.setattr(cb, "_acad_app", lambda: app)
monkeypatch.setattr(cb, "_WIN32_AVAILABLE", True)
def _enum_windows(cb_fn, _extra):
# Simulate one visible AutoCAD top-level window.
cb_fn(0x1234, None)
win32gui = MagicMock()
win32gui.EnumWindows.side_effect = _enum_windows
win32gui.GetWindowText.return_value = "AutoCAD 2025 - [Drawing1.dwg]"
win32gui.IsWindowVisible.return_value = True
monkeypatch.setattr(cb, "win32gui", win32gui)
assert cb._find_autocad_hwnd() == 0x1234
win32gui.EnumWindows.assert_called_once()
async def test_find_hwnd_returns_none_when_win32_unavailable(monkeypatch):
"""Short-circuits to None on non-Windows without touching the COM app."""
sentinel = MagicMock(side_effect=AssertionError("_acad_app must not be called"))
monkeypatch.setattr(cb, "_acad_app", sentinel)
monkeypatch.setattr(cb, "_WIN32_AVAILABLE", False)
assert cb._find_autocad_hwnd() is None
sentinel.assert_not_called()
# ── S3 — block_list populates BlockInfo.description from .Comments ───────────
class _Block:
"""Minimal stand-in for an AutoCAD block definition."""
def __init__(self, name, comments="", count=0, is_xref=False, raise_comments=False):
self.Name = name
self._comments = comments
self.Count = count
self.IsXRef = is_xref
self.Origin = (0.0, 0.0, 0.0)
self._raise_comments = raise_comments
@property
def Comments(self):
if self._raise_comments:
raise RuntimeError("Comments not supported")
return self._comments
def Item(self, _j): # no attribute definitions in these fixtures
return MagicMock(ObjectName="AcDbEntity")
class _Blocks:
def __init__(self, items):
self._items = items
self.Count = len(items)
def Item(self, i):
return self._items[i]
def _doc_with_blocks(items):
doc = MagicMock()
doc.Blocks = _Blocks(items)
return doc
async def test_block_list_populates_description_from_comments(monkeypatch):
doc = _doc_with_blocks([_Block("BOLT", comments="M8 hex bolt")])
monkeypatch.setattr(cb, "_acad_doc", lambda: doc)
b = _backend_no_executor()
blocks = await b.block_list()
assert len(blocks) == 1
assert blocks[0].name == "BOLT"
assert blocks[0].description == "M8 hex bolt"
async def test_block_list_description_defaults_empty_when_comments_raise(monkeypatch):
doc = _doc_with_blocks([_Block("NUT", raise_comments=True)])
monkeypatch.setattr(cb, "_acad_doc", lambda: doc)
b = _backend_no_executor()
blocks = await b.block_list()
assert blocks[0].description == "" # try/except default, not a crash
async def test_block_list_description_empty_when_comments_none(monkeypatch):
doc = _doc_with_blocks([_Block("WASHER", comments=None)])
monkeypatch.setattr(cb, "_acad_doc", lambda: doc)
b = _backend_no_executor()
blocks = await b.block_list()
assert blocks[0].description == "" # `Comments or ""` normalises None
async def test_block_list_skips_anonymous_layout_blocks(monkeypatch):
doc = _doc_with_blocks([
_Block("*Model_Space", comments="ignored"),
_Block("GEAR", comments="spur gear"),
])
monkeypatch.setattr(cb, "_acad_doc", lambda: doc)
b = _backend_no_executor()
blocks = await b.block_list()
assert [blk.name for blk in blocks] == ["GEAR"]
assert blocks[0].description == "spur gear"
+136
View File
@@ -0,0 +1,136 @@
"""Regression tests for the Sprint-4 ezdxf-backend fixes.
* R24 — _apply_attrs loads CENTER/HIDDEN on demand so entity_create_line(
linetype=...) actually renders the dashed linetype instead of
silently falling back to Continuous.
* S3 — block_list surfaces the block definition's description instead of "".
* R26/N9 — view_zoom_extents / view_zoom_window return a consistent
{"ok": True, "applied": False, ...} shape (no fake framing success).
* NEW-undo-1 — drawing_undo / transaction_* serialize _undo_stack access
(no read/mutate outside self._lock); behaviour still correct.
All tests are backend-agnostic ezdxf and use the shared `backend` fixture.
"""
from __future__ import annotations
import pytest
pytestmark = pytest.mark.asyncio
# ── R24 — CENTER/HIDDEN linetype actually loaded + applied ──────────────────
async def test_create_line_loads_center_linetype(backend):
# Fresh doc: CENTER is not loaded yet.
assert "CENTER" not in backend._doc.linetypes
line = await backend.entity_create_line(0, 0, 100, 0, linetype="CENTER")
# The linetype must now be defined in the document (a real dashed pattern),
# not silently dropped — otherwise it renders as Continuous.
assert "CENTER" in backend._doc.linetypes
# And it must be assigned to the entity.
info = await backend.entity_get(line.handle)
assert info.linetype == "CENTER"
async def test_create_line_loads_hidden_linetype(backend):
assert "HIDDEN" not in backend._doc.linetypes
line = await backend.entity_create_line(0, 0, 50, 50, linetype="HIDDEN")
assert "HIDDEN" in backend._doc.linetypes
info = await backend.entity_get(line.handle)
assert info.linetype == "HIDDEN"
async def test_create_line_continuous_does_not_error(backend):
# Built-in linetypes are a no-op for the loader — must not raise.
line = await backend.entity_create_line(0, 0, 10, 0, linetype="Continuous")
info = await backend.entity_get(line.handle)
assert info.linetype == "Continuous"
# ── S3 — block description surfaced from the definition ─────────────────────
async def test_block_list_surfaces_description(backend):
# Create a block from an entity, then stamp a description on its definition.
seg = await backend.entity_create_line(0, 0, 10, 0)
await backend.block_create_from_entities("WIDGET", [seg.handle], 0.0, 0.0)
backend._doc.blocks.get("WIDGET").block.dxf.description = "Mechanical widget"
blocks = await backend.block_list()
widget = next(b for b in blocks if b.name == "WIDGET")
assert widget.description == "Mechanical widget"
async def test_block_list_description_defaults_empty(backend):
# A block with no description must still surface "" (no crash, default).
seg = await backend.entity_create_line(0, 0, 5, 0)
await backend.block_create_from_entities("PLAIN", [seg.handle], 0.0, 0.0)
blocks = await backend.block_list()
plain = next(b for b in blocks if b.name == "PLAIN")
assert plain.description == ""
# ── R26 / N9 — honest zoom payload shape ────────────────────────────────────
async def test_zoom_extents_reports_not_applied(backend):
res = await backend.view_zoom_extents()
assert res["ok"] is True
assert res["applied"] is False
assert "message" in res
async def test_zoom_window_reports_not_applied(backend):
res = await backend.view_zoom_window(0, 0, 100, 100)
assert res["ok"] is True
assert res["applied"] is False
assert "message" in res
# ── NEW-undo-1 — undo/transaction stack behaviour under lock ────────────────
async def test_undo_empty_stack_returns_error(backend):
res = await backend.drawing_undo()
assert res["ok"] is False
assert "undo" in res["error"].lower()
async def test_commit_without_transaction_returns_error(backend):
res = await backend.transaction_commit()
assert res["ok"] is False
assert "transaction" in res["error"].lower()
async def test_rollback_without_transaction_returns_error(backend):
res = await backend.transaction_rollback()
assert res["ok"] is False
assert "transaction" in res["error"].lower()
async def test_transaction_begin_commit_roundtrip(backend):
begin = await backend.transaction_begin()
assert begin["ok"] is True
assert len(backend._undo_stack) == 1
commit = await backend.transaction_commit()
assert commit["ok"] is True
assert len(backend._undo_stack) == 0
async def test_transaction_rollback_restores_snapshot(backend):
# Snapshot an empty doc, add an entity, then roll back to the snapshot.
await backend.transaction_begin()
line = await backend.entity_create_line(0, 0, 10, 0)
before = await backend.entity_get(line.handle)
assert before.handle # entity exists
res = await backend.transaction_rollback()
assert res["ok"] is True
assert len(backend._undo_stack) == 0
# The line is gone after rollback.
listed = await backend.entity_list(type_filter="LINE")
assert listed == []
+143
View File
@@ -0,0 +1,143 @@
"""Regression tests for the Sprint-4 server.py fixes.
* R20 — _registered_tool_count() uses the public FastMCP accessor
(mcp._list_tools) and returns a real positive count, never -1/None.
* R15 — system_about's tool_groups is derived dynamically from each tool's
tags: it now includes the engineering/premium groups and no longer
misfiles entity_delete_many under entity_creation.
* N8 — drawing_plan docstring no longer claims the PlanSpec is "replayed by
drawing_critique".
These tests drive server.py's underlying async functions directly with a
minimal fake ctx (lifespan_context + async info/warning), the same pattern the
other sprint test files use. They are ezdxf-backend-agnostic.
"""
from __future__ import annotations
import pytest
pytestmark = pytest.mark.asyncio
class _Ctx:
"""Minimal async context mimicking fastmcp.Context for direct tool calls."""
def __init__(self, backend):
self.lifespan_context = {"backend": backend}
async def info(self, message):
pass
async def warning(self, message):
pass
# ── R20 — tool count comes from the public accessor and is a real number ─────
async def test_registered_tool_count_is_positive_never_minus_one():
import server
count = await server._registered_tool_count()
assert count is not None # never the old sentinel state
assert isinstance(count, int)
assert count != -1 # the old silent-failure value must never surface
assert count > 50 # the server registers dozens of tools
async def test_registered_tools_returns_objects_with_name_and_tags():
import server
tools = await server._registered_tools()
assert len(tools) > 50
# Every entry is a real tool object usable for tag-based grouping.
for t in tools:
assert getattr(t, "name", None)
# tags is a set (possibly empty) on FunctionTool
assert hasattr(t, "tags")
# ── R15 — system_about tool_groups is dynamic + correct ──────────────────────
async def test_system_about_includes_engineering_group(backend):
import server
about = await server.system_about(ctx=_Ctx(backend))
groups = about["tool_groups"]
# An engineering tool must appear somewhere in the breakdown (the old
# static dict omitted all engineering tools entirely).
all_listed = {name for names in groups.values() for name in names}
assert "gear_draw_section_aa" in all_listed
# The engineering group itself is present and non-empty.
assert groups.get("engineering")
assert "gear_draw_section_aa" in groups["engineering"]
async def test_system_about_includes_premium_group(backend):
import server
about = await server.system_about(ctx=_Ctx(backend))
groups = about["tool_groups"]
all_listed = {name for names in groups.values() for name in names}
# Premium meta-tools (e.g. drawing_plan) used to be missing from the dict.
assert "drawing_plan" in all_listed
assert groups.get("premium")
async def test_entity_delete_many_not_under_entity_creation(backend):
import server
about = await server.system_about(ctx=_Ctx(backend))
groups = about["tool_groups"]
# The bug: entity_delete_many was hardcoded under entity_creation.
assert "entity_delete_many" not in groups.get("entity_creation", [])
# It is tagged {"entity", "modify"} so it belongs with modification.
assert "entity_delete_many" in groups.get("entity_modification", [])
async def test_system_about_total_tools_matches_count(backend):
import server
about = await server.system_about(ctx=_Ctx(backend))
count = await server._registered_tool_count()
# total_tools is reported dynamically (never hardcoded) and agrees with
# the count helper.
assert about.get("total_tools") == count
assert about["total_tools"] > 50
# The sum of all grouped tools equals the total (every tool buckets once).
grouped = sum(len(v) for v in about["tool_groups"].values())
assert grouped == about["total_tools"]
async def test_drawing_close_appears_in_groups(backend):
import server
about = await server.system_about(ctx=_Ctx(backend))
all_listed = {name for names in about["tool_groups"].values() for name in names}
# drawing_close was omitted by the old static dict.
assert "drawing_close" in all_listed
# ── R20 — system_status reports a real tool_count, never -1 ──────────────────
async def test_system_status_reports_real_tool_count(backend):
import server
status = await server.system_status(ctx=_Ctx(backend))
# tool_count is present and a real positive number (omitted only if unknown).
assert status.get("tool_count") is not None
assert status["tool_count"] != -1
assert status["tool_count"] > 50
# ── N8 — drawing_plan docstring corrected ────────────────────────────────────
async def test_drawing_plan_docstring_not_misleading():
import server
doc = server.drawing_plan.__doc__ or ""
# The corrected docstring must not claim the plan is replayed as a critique.
assert "replayed by `drawing_critique`" not in doc
assert "stored on the backend" in doc
+128
View File
@@ -0,0 +1,128 @@
"""Sprint-4 R27 regression tests for engineering.validator step 5.
R27: title/helix consistency must match SPUR / HELICAL as whole words (not
substrings) and scope the scan to title-block text where possible.
"""
import pytest
from engineering.layers import ensure_engineering_layers, ensure_standard_linetypes
from engineering.validator import DrawingValidator
pytestmark = pytest.mark.asyncio
def _codes(result) -> set[str]:
return {f.code for f in result.findings}
async def _ensure_hidden_linetype(backend) -> None:
"""ezdxf doesn't ship a HIDDEN pattern; add a synthetic one."""
doc = backend._require_doc()
if "HIDDEN" not in doc.linetypes:
doc.linetypes.add(
"HIDDEN",
pattern=[6.35, 5.0, -1.27],
description="Hidden _ _ _ _ _ _",
)
async def _bootstrap(backend):
await ensure_standard_linetypes(backend)
await _ensure_hidden_linetype(backend)
await ensure_engineering_layers(backend)
async def test_spur_as_substring_does_not_false_match(backend, tmp_path):
"""R27: 'SPURIOUS' contains 'spur' but must NOT trip a SPUR mismatch."""
await _bootstrap(backend)
await backend.entity_create_text(
text="SPURIOUS HELICAL GEAR", x=0, y=0, height=5, layer="TEXT",
)
await backend.drawing_save_as(str(tmp_path / "spurious.dxf"))
result = await DrawingValidator().run(backend, expected={"helix_angle": 15})
codes = _codes(result)
assert "title_says_spur_for_helical" not in codes
assert "title_helical_spur" not in codes
async def test_genuine_spur_title_flags_when_helical_expected(backend, tmp_path):
"""R27: a real 'SPUR GEAR' title still flags the mismatch when helix declared."""
await _bootstrap(backend)
await backend.entity_create_text(
text="SPUR GEAR", x=0, y=0, height=5, layer="TEXT",
)
await backend.drawing_save_as(str(tmp_path / "spur.dxf"))
result = await DrawingValidator().run(backend, expected={"helix_angle": 15})
codes = _codes(result)
assert "title_says_spur_for_helical" in codes
assert result.ok is False
async def test_spur_word_in_titleblock_layer_flags(backend, tmp_path):
"""R27: title-block-scoped scan — SPUR on TITLEBLOCK layer still flags."""
await _bootstrap(backend)
await backend.entity_create_text(
text="SPUR GEAR", x=0, y=0, height=5, layer="TITLEBLOCK",
)
await backend.drawing_save_as(str(tmp_path / "spur_tb.dxf"))
result = await DrawingValidator().run(backend, expected={"helix_angle": 15})
assert "title_says_spur_for_helical" in _codes(result)
async def test_substring_in_titleblock_layer_does_not_false_match(backend, tmp_path):
"""R27: 'SPURIOUS' on TITLEBLOCK layer must not trip a SPUR match either."""
await _bootstrap(backend)
await backend.entity_create_text(
text="SPURIOUS NOTE", x=0, y=0, height=5, layer="TITLEBLOCK",
)
await backend.drawing_save_as(str(tmp_path / "spurious_tb.dxf"))
result = await DrawingValidator().run(backend, expected={"helix_angle": 15})
codes = _codes(result)
assert "title_says_spur_for_helical" not in codes
assert "title_helical_spur" not in codes
async def test_helical_spur_combo_still_flags_invalid(backend, tmp_path):
"""R27: both whole words present -> invalid combo error preserved."""
await _bootstrap(backend)
await backend.entity_create_text(
text="HELICAL SPUR GEAR", x=0, y=0, height=5, layer="TEXT",
)
await backend.drawing_save_as(str(tmp_path / "combo.dxf"))
result = await DrawingValidator().run(backend, expected={"helix_angle": 15})
assert "title_helical_spur" in _codes(result)
assert result.ok is False
async def test_no_helix_expected_skips_title_check(backend, tmp_path):
"""Gating preserved: without helix_angle, a SPUR title raises no title finding."""
await _bootstrap(backend)
await backend.entity_create_text(
text="SPUR GEAR", x=0, y=0, height=5, layer="TEXT",
)
await backend.drawing_save_as(str(tmp_path / "nohelix.dxf"))
result = await DrawingValidator().run(backend)
codes = _codes(result)
assert "title_says_spur_for_helical" not in codes
assert "title_helical_spur" not in codes
async def test_titleblock_layer_takes_precedence_over_text_layer(backend, tmp_path):
"""R27 scoping: when title-block text exists, the TEXT-layer scan is not used.
SPUR sits on the generic TEXT layer (a body note), while the title block on
TITLEBLOCK is clean — the scoped scan should ignore the body note.
"""
await _bootstrap(backend)
await backend.entity_create_text(
text="HELICAL GEAR", x=0, y=0, height=5, layer="TITLEBLOCK",
)
await backend.entity_create_text(
text="SPUR TOOTH DETAIL NOTE", x=0, y=-20, height=3.5, layer="TEXT",
)
await backend.drawing_save_as(str(tmp_path / "scope.dxf"))
result = await DrawingValidator().run(backend, expected={"helix_angle": 15})
codes = _codes(result)
assert "title_says_spur_for_helical" not in codes
assert "title_helical_spur" not in codes