From 32a68be91959df187d7da59ec2fcbf5549b8cb1d Mon Sep 17 00:00:00 2001 From: Umutcan Edizsalan Date: Fri, 19 Jun 2026 08:36:58 +0300 Subject: [PATCH] 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. --- backends/base.py | 157 +++++++++++++++---- backends/com_backend.py | 170 ++++++++++++++++++--- backends/ezdxf_backend.py | 120 ++++++++++++--- engineering/critique.py | 57 +++++-- engineering/gear.py | 54 +++++-- engineering/layers.py | 17 +++ engineering/validator.py | 42 ++++-- server.py | 257 ++++++++++++++++++++++++-------- tests/test_com_backend.py | 4 + tests/test_com_sprint3.py | 158 ++++++++++++++++++++ tests/test_security.py | 43 ++++++ tests/test_sprint2_fixes.py | 186 +++++++++++++++++++++++ tests/test_sprint3_fixes.py | 107 +++++++++++++ tests/test_sprint4_base.py | 223 +++++++++++++++++++++++++++ tests/test_sprint4_com.py | 180 ++++++++++++++++++++++ tests/test_sprint4_ezdxf.py | 136 +++++++++++++++++ tests/test_sprint4_server.py | 143 ++++++++++++++++++ tests/test_sprint4_validator.py | 128 ++++++++++++++++ 18 files changed, 2016 insertions(+), 166 deletions(-) create mode 100644 tests/test_com_sprint3.py create mode 100644 tests/test_sprint2_fixes.py create mode 100644 tests/test_sprint3_fixes.py create mode 100644 tests/test_sprint4_base.py create mode 100644 tests/test_sprint4_com.py create mode 100644 tests/test_sprint4_ezdxf.py create mode 100644 tests/test_sprint4_server.py create mode 100644 tests/test_sprint4_validator.py diff --git a/backends/base.py b/backends/base.py index fb9afa2..a2fb030 100644 --- a/backends/base.py +++ b/backends/base.py @@ -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) diff --git a/backends/com_backend.py b/backends/com_backend.py index 0d9467e..e986c0a 100644 --- a/backends/com_backend.py +++ b/backends/com_backend.py @@ -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 ────────────────────────────────────────────────────────── diff --git a/backends/ezdxf_backend.py b/backends/ezdxf_backend.py index edeb6b2..bbf0fd0 100644 --- a/backends/ezdxf_backend.py +++ b/backends/ezdxf_backend.py @@ -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)) diff --git a/engineering/critique.py b/engineering/critique.py index 06227a3..ddc96a8 100644 --- a/engineering/critique.py +++ b/engineering/critique.py @@ -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): diff --git a/engineering/gear.py b/engineering/gear.py index 0c115db..5e2a21c 100644 --- a/engineering/gear.py +++ b/engineering/gear.py @@ -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, diff --git a/engineering/layers.py b/engineering/layers.py index 4436350..ee28b65 100644 --- a/engineering/layers.py +++ b/engineering/layers.py @@ -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"] diff --git a/engineering/validator.py b/engineering/validator.py index 5e22c99..98ef988 100644 --- a/engineering/validator.py +++ b/engineering/validator.py @@ -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", diff --git a/server.py b/server.py index 5e49a78..d2bbf6f 100644 --- a/server.py +++ b/server.py @@ -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" diff --git a/tests/test_com_backend.py b/tests/test_com_backend.py index 7a236ac..916bbc9 100644 --- a/tests/test_com_backend.py +++ b/tests/test_com_backend.py @@ -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), diff --git a/tests/test_com_sprint3.py b/tests/test_com_sprint3.py new file mode 100644 index 0000000..7489c02 --- /dev/null +++ b/tests/test_com_sprint3.py @@ -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() diff --git a/tests/test_security.py b/tests/test_security.py index 9cf8f9f..b4fe15f 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -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))" diff --git a/tests/test_sprint2_fixes.py b/tests/test_sprint2_fixes.py new file mode 100644 index 0000000..0ed25f0 --- /dev/null +++ b/tests/test_sprint2_fixes.py @@ -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 == [] diff --git a/tests/test_sprint3_fixes.py b/tests/test_sprint3_fixes.py new file mode 100644 index 0000000..ded4f4c --- /dev/null +++ b/tests/test_sprint3_fixes.py @@ -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 diff --git a/tests/test_sprint4_base.py b/tests/test_sprint4_base.py new file mode 100644 index 0000000..3ba8829 --- /dev/null +++ b/tests/test_sprint4_base.py @@ -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 diff --git a/tests/test_sprint4_com.py b/tests/test_sprint4_com.py new file mode 100644 index 0000000..56ef536 --- /dev/null +++ b/tests/test_sprint4_com.py @@ -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" diff --git a/tests/test_sprint4_ezdxf.py b/tests/test_sprint4_ezdxf.py new file mode 100644 index 0000000..6f976fe --- /dev/null +++ b/tests/test_sprint4_ezdxf.py @@ -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 == [] diff --git a/tests/test_sprint4_server.py b/tests/test_sprint4_server.py new file mode 100644 index 0000000..c1eec22 --- /dev/null +++ b/tests/test_sprint4_server.py @@ -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 diff --git a/tests/test_sprint4_validator.py b/tests/test_sprint4_validator.py new file mode 100644 index 0000000..681d000 --- /dev/null +++ b/tests/test_sprint4_validator.py @@ -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