anatoly/autocad-mcp-pro
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

fix(task-13): read revision/balloon state from the space it is drawn on, and never at 200

Browse Source 
Three findings, one family: a guard that reads the WRONG SET silently stops
guarding and the drawing gets a duplicate or a wrong number with no refusal.

1. Layout-scoped state was read from the wrong space. add_revision called
   read_revisions and add_balloon called read_balloons BEFORE enter_layout;
   both read entity_list, which follows the current space, so the guard
   inspected Model while the rows and balloons lived on the layout -- and the
   layout is the sheet workflow (sheet_frame and titleblock_apply both take
   one). Measured headlessly: two revision_add(rev="B", layout="SHEET1") calls
   both returned created=True, handles 46 and 5A, both bbox
   {min [230,273], max [410,287]} -- a second heading band and a second B row
   drawn exactly on top of the first, twelve TEXTs on the tab instead of six.
   Both reads now happen inside the entered layout, and the whole body sits in
   the try/ so the caller's tab comes back on the refusal path too.

2. bom_extract truncated the parts list at 200 INSERTs and printed a wrong QTY
   into a real TABLE. extract_records passed its own limit=200 default straight
   to entity_list, the tool exposed no limit, and nothing in the payload said
   the list had been cut: 250 bolts reported [("ISO 4014 - M12x60", 200)].
   all_entities now pages entity_list (offset loop, PAGE_SIZE 1000, plus a
   no-new-handles guard so a backend that ignored offset cannot spin) and
   extract_records(limit=None) reads the whole drawing. bom_extract exposes
   limit (0 = no cap) and reports truncated measured, by asking for one record
   past the cap.

3. The balloon clash guard did not fire in two routine cases. (a) targets is
   optional in the tool, so the clash test compared () != () -- False -- and
   two untargeted balloons carried item 1 with no refusal. The renumber arm
   above has already returned for the balloon that stands for these targets, so
   anything still carrying the number is a clash, whatever its targets; the
   message names the handle and says "no targets" when it has none. (b)
   read_balloons and read_revisions took entity_list's 200-entity default, so a
   drawing whose first 200 CIRCLEs are a hole pattern read back no balloons at
   all and every guard passed blind -- and index = len(existing) restarted at
   0, stacking revision bands. Both readers now go through all_entities.

Evidence, headless (ezdxf), each finding's own repro before -> after:
  1  created True/True, handles 46+5A, 12 TEXTs -> created True/False, one
     handle 46, 6 TEXTs; renumbered False/False -> False/True, same handle 33
  2  records 200, rows [(..., 200)] -> records 250, rows [(..., 250)]
  3a two balloons handles 2F+35, no refusal -> refused, naming balloon 2F
  3b read_balloons() == () with 206 circles -> the balloon past the 200th
     circle is found, the rerun renumbers, the item-1/BBB clash is refused
  3c 200 notes + rev A twice + rev B -> created True/False/True, row 1 then
     row 2 (was: every band at row 1, on top of the last)

Six regression tests added: one per finding plus the layout-scoped twin of
each, which is what would have caught all three.

The live engine was UNAVAILABLE for this round:
win32com.client.GetActiveObject("AutoCAD.Application") did not return within
90 s against the running acad.exe (PID 40068), and the ComBackend probe timed
out on its first call at both COM_CALL_TIMEOUT=60 and 180. No second AutoCAD
was launched and no document was opened. No new ActiveX call is introduced by
this fix -- it only reorders existing calls and pages entity_list(offset=),
which both engines already implement.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd
This commit is contained in:
Umut CAN
2026-09-22 22:39:10 +03:00
co-authored by Claude Opus 5
parent 499b642fbd
commit de8dc8f813
5 changed files with 363 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files
engineering/sheet/bom.py
+122 -56
View File
@@ -49,8 +49,10 @@ __all__ = [
"DIRECTIONS",
"LIST_STANDARDS",
"LIST_WIDTH",
"PAGE_SIZE",
"ROW_HEIGHT",
"add_balloon",
"all_entities",
"balloon_prims",
"draw_bom_table",
"extract_records",
@@ -104,6 +106,43 @@ COLUMN_SHARES: dict[str, float] = {
}
#: How many entities one `entity_list` call asks for while paging. It is a page
#: size, never a cap: `all_entities` keeps asking until a short page comes back.
PAGE_SIZE = 1000
async def all_entities(backend: AutoCADBackend, *, type_filter=None, layer=None):
"""Every entity of a type in the CURRENT space -- not the first 200 of them.
`entity_list`'s `limit` defaults to 200 on both engines, so a single call is
a silent truncation on any real drawing: a parts list that stops at the
200th INSERT prints a wrong QTY, and a duplicate-balloon guard that only
sees the first 200 CIRCLEs is blind on a drawing whose hole pattern fills
them. Everything in this module that *counts* or *guards* reads through
here instead.
The loop stops on a short page, and also when a page brings nothing new --
a backend that ignored `offset` would otherwise hand back the same page for
ever. Handles are unique per drawing, so that guard cannot drop a real
entity.
"""
found: list = []
seen: set[str] = set()
offset = 0
while True:
page = await backend.entity_list(
type_filter=type_filter, layer_filter=layer, limit=PAGE_SIZE, offset=offset
)
fresh = [entity for entity in page if entity.handle not in seen]
if not fresh:
return tuple(found)
seen.update(entity.handle for entity in fresh)
found.extend(fresh)
if len(page) < PAGE_SIZE:
return tuple(found)
offset += len(page)
def _check_columns(columns) -> tuple[str, ...]:
chosen = tuple(DEFAULT_COLUMNS if columns is None else columns)
unknown = [c for c in chosen if c not in BOM_COLUMNS]
@@ -304,7 +343,7 @@ def _from_attributes(attributes: dict) -> dict:
return record
async def extract_records(backend: AutoCADBackend, *, layer=None, limit: int = 200):
async def extract_records(backend: AutoCADBackend, *, layer=None, limit: int | None = None):
"""Walk the INSERTs and return one parts-list record per block reference.
Reads only: the entity listing, each INSERT's attributes and its
@@ -312,8 +351,17 @@ async def extract_records(backend: AutoCADBackend, *, layer=None, limit: int = 2
is not ours. `source` is "xdata" when the payload supplied the row and
"attributes" when the block's ATTRIBs did; a block with neither is skipped
rather than guessed at.
`limit` is None by default and the whole drawing is read, by paging
`entity_list` rather than taking its 200-entity default: a parts list that
stops at the 200th INSERT does not print a shorter list, it prints a wrong
QTY, and nothing on the sheet says so. A caller that wants a cap passes one
and reads `len(records)` back against it -- `bom_extract` asks for one more
record than the cap so it can report `truncated` exactly.
"""
inserts = await backend.entity_list(type_filter="INSERT", layer_filter=layer, limit=limit)
if limit is not None and int(limit) <= 0:
raise ValueError(f"bom_extract: limit must be a positive number of records; got {limit!r}")
inserts = await all_entities(backend, type_filter="INSERT", layer=layer)
records: list[dict] = []
for entity in inserts:
payload: dict = {}
@@ -353,6 +401,8 @@ async def extract_records(backend: AutoCADBackend, *, layer=None, limit: int = 2
"source": "xdata" if payload else "attributes",
}
)
if limit is not None and len(records) >= int(limit):
break
return tuple(records)
@@ -415,9 +465,15 @@ async def draw_bom_table(
async def read_balloons(backend: AutoCADBackend):
"""Every ISO 6433 balloon this server drew, from its ACADMCP_MECH payload."""
"""Every ISO 6433 balloon this server drew in the CURRENT space, from its
ACADMCP_MECH payload.
Paged, not capped: a drawing whose first 200 CIRCLEs are a hole pattern --
routine -- used to read back no balloons at all, which made every guard in
`add_balloon` silently pass.
"""
found: list[dict] = []
for entity in await backend.entity_list(type_filter="CIRCLE"):
for entity in await all_entities(backend, type_filter="CIRCLE"):
try:
raw = await backend.entity_get_xdata(entity.handle, APP_ID)
payload = decode(raw.get("xdata", {}).get(APP_ID, [])) or {}
@@ -453,69 +509,79 @@ async def add_balloon(
A balloon already pointing at the same targets is *renumbered* -- its
numeral is edited and its payload rewritten -- so running the balloon pass
again after the parts list is regrouped does not litter the sheet with
duplicates. An item number already used for a different set of targets is
refused before anything is drawn: ISO 129-1 and ISO 6433 both want one
duplicates. An item number already on the sheet is refused before anything
is drawn unless it is that renumber: ISO 129-1 and ISO 6433 both want one
reference per item.
Both guards read the space the balloon is going ON. `read_balloons` lists
the current space, so reading it before `enter_layout` inspected Model
while the balloons lived on the layout -- and the layout is the sheet
workflow, so the guards were off exactly where they matter.
"""
number = int(item)
wanted = tuple(sorted(str(t) for t in targets))
prims = balloon_prims(number, at, leader_to, radius=radius) # validates first
existing = await read_balloons(backend)
for balloon in existing:
if tuple(sorted(balloon["targets"])) == wanted and wanted:
if balloon["text"]:
await backend.entity_edit_text(balloon["text"], text=str(number))
payload = {
"v": PAYLOAD_VERSION,
"kind": "balloon",
"item": number,
"targets": list(wanted),
"text": balloon["text"],
}
await backend.entity_set_xdata(balloon["handle"], APP_ID, to_values(payload))
return {
"ok": True,
"renumbered": True,
"item": number,
"handle": balloon["handle"],
"text": balloon["text"],
"targets": list(wanted),
"layout": layout or "Model",
}
clash = next(
(b for b in existing if b["item"] == number and tuple(sorted(b["targets"])) != wanted),
None,
)
if clash is not None:
raise ValueError(
f"item reference {number} is already used by balloon {clash['handle']} for "
f"targets {clash['targets']}; ISO 6433 gives each item one reference"
)
previous = await enter_layout(backend, layout, caller="balloon_add")
try:
existing = await read_balloons(backend)
for balloon in existing:
if wanted and tuple(sorted(balloon["targets"])) == wanted:
if balloon["text"]:
await backend.entity_edit_text(balloon["text"], text=str(number))
payload = {
"v": PAYLOAD_VERSION,
"kind": "balloon",
"item": number,
"targets": list(wanted),
"text": balloon["text"],
}
await backend.entity_set_xdata(balloon["handle"], APP_ID, to_values(payload))
return {
"ok": True,
"renumbered": True,
"item": number,
"handle": balloon["handle"],
"text": balloon["text"],
"targets": list(wanted),
"layout": layout or "Model",
}
# Anything still carrying this number is a clash: the renumber arm above
# has already returned for the one balloon that stands for these
# targets. Comparing the target tuples here instead let two untargeted
# balloons -- `targets` is optional in the tool, so `() != ()` -- carry
# the same item number with no refusal at all.
clash = next((b for b in existing if b["item"] == number), None)
if clash is not None:
has_targets = f"targets {clash['targets']}" if clash["targets"] else "no targets"
raise ValueError(
f"item reference {number} is already used by balloon {clash['handle']} "
f"({has_targets}); ISO 6433 gives each item one reference. Pass that "
"balloon's targets to renumber it in place, or use a free number."
)
handles = await draw_sheet_prims(
backend, prims, layer_map={"visible": layer, "text": layer, "center": layer}
)
payload = {
"v": PAYLOAD_VERSION,
"kind": "balloon",
"item": number,
"targets": list(wanted),
"text": handles[3],
}
await backend.entity_set_xdata(handles[0], APP_ID, to_values(payload))
return {
"ok": True,
"renumbered": False,
"item": number,
"handle": handles[0],
"leader": handles[1],
"dot": handles[2],
"text": handles[3],
"targets": list(wanted),
"layout": layout or "Model",
}
finally:
if layout is not None:
await backend.layout_set_current(previous)
payload = {
"v": PAYLOAD_VERSION,
"kind": "balloon",
"item": number,
"targets": list(wanted),
"text": handles[3],
}
await backend.entity_set_xdata(handles[0], APP_ID, to_values(payload))
return {
"ok": True,
"renumbered": False,
"item": number,
"handle": handles[0],
"leader": handles[1],
"dot": handles[2],
"text": handles[3],
"targets": list(wanted),
"layout": layout or "Model",
}
engineering/sheet/revision.py
+83 -70
View File
@@ -29,7 +29,7 @@ from typing import TYPE_CHECKING
from backends.xdata_specs import MAX_BYTES, app_size
from engineering.mech.primitives import Line, Poly, Prim, Pt, Text
from engineering.sheet.bom import LIST_WIDTH, ROW_HEIGHT, TEXT_HEIGHT
from engineering.sheet.bom import LIST_WIDTH, ROW_HEIGHT, TEXT_HEIGHT, all_entities
from engineering.sheet.frames import (
SHEET_LAYER,
draw_sheet_prims,
@@ -226,9 +226,16 @@ def revision_block_prims(rows, *, at: Pt, height: float = ROW_HEIGHT) -> tuple[P
async def read_revisions(backend: AutoCADBackend):
"""Every revision row this server wrote, from its ACADMCP_SHEET_REV payload."""
"""Every revision row this server wrote in the CURRENT space, from its
ACADMCP_SHEET_REV payload.
Paged, not capped: `entity_list`'s 200-entity default made a lettered
drawing read back no revisions at all, which both broke the idempotency
guard and restarted `index = len(existing)` at 0 -- a second heading band
and a second row drawn exactly on top of the first.
"""
found: list[dict] = []
for entity in await backend.entity_list(type_filter="TEXT"):
for entity in await all_entities(backend, type_filter="TEXT"):
try:
raw = await backend.entity_get_xdata(entity.handle, REV_APP_ID)
payload = rev_payload(raw.get("xdata", {}).get(REV_APP_ID, [])) or {}
@@ -272,6 +279,12 @@ async def add_revision(
cannot stack duplicate rows. Refusals fire before the first entity: a
malformed code, a cloud region that is not four finite numbers, and -- on
an engine whose `revcloud` capability is false -- any request for clouds.
The idempotency guard reads the space the row is going ON. `read_revisions`
lists the current space, so reading it before `enter_layout` inspected
Model while the block lived on the layout, and every re-run of a
`layout=`-scoped revision pass stacked a second heading band and a second
row exactly on top of the first.
"""
code = _check_rev(rev)
regions = [[float(v) for v in region] for region in clouds]
@@ -291,37 +304,37 @@ async def add_revision(
"`clouds` to record the row, or draw the sheet on the ezdxf engine.",
)
existing = await read_revisions(backend)
match = next((row for row in existing if row["rev"] == code), None)
if match is not None:
return {
"ok": True,
"created": False,
"rev": code,
"handle": match["handle"],
"row": len(existing),
"clouds": [],
"tags": [],
"layout": layout or "Model",
}
index = len(existing)
new_row = {"rev": code, "description": description, "date": date, "by": by}
metrics = frame_metrics(size, orientation=orientation)
_x0, _y0, x1, y1 = metrics["frame"]
anchor = (
(float(origin[0]) + x1 - LIST_WIDTH, float(origin[1]) + y1)
if at is None
else (float(at[0]), float(at[1]))
)
# Append-only: the heading band is drawn once, on the first revision, and
# every later revision adds exactly one band below the last. Nothing is
# redrawn, so a revision can never stack a second copy of the block.
head_prims = list(revision_heading_prims(at=anchor)) if index == 0 else []
row_prims = list(revision_row_prims(new_row, at=anchor, index=index))
previous = await enter_layout(backend, layout, caller="revision_add")
try:
existing = await read_revisions(backend)
match = next((row for row in existing if row["rev"] == code), None)
if match is not None:
return {
"ok": True,
"created": False,
"rev": code,
"handle": match["handle"],
"row": len(existing),
"clouds": [],
"tags": [],
"layout": layout or "Model",
}
index = len(existing)
new_row = {"rev": code, "description": description, "date": date, "by": by}
metrics = frame_metrics(size, orientation=orientation)
_x0, _y0, x1, y1 = metrics["frame"]
anchor = (
(float(origin[0]) + x1 - LIST_WIDTH, float(origin[1]) + y1)
if at is None
else (float(at[0]), float(at[1]))
)
# Append-only: the heading band is drawn once, on the first revision, and
# every later revision adds exactly one band below the last. Nothing is
# redrawn, so a revision can never stack a second copy of the block.
head_prims = list(revision_heading_prims(at=anchor)) if index == 0 else []
row_prims = list(revision_row_prims(new_row, at=anchor, index=index))
block_handles = await draw_sheet_prims(backend, head_prims + row_prims)
cloud_handles = []
for region in regions:
@@ -340,44 +353,44 @@ async def add_revision(
for point in points:
drawn = await draw_sheet_prims(backend, revision_tag_prims(code, (point[0], point[1])))
tag_handles.append({"at": [point[0], point[1]], "handles": drawn})
# The payload rides on the row band's first Text primitive, which is its
# REV cell -- `rev` is validated non-empty, so that cell is always drawn.
row_handles = block_handles[len(head_prims) :]
row_handle = next(
handle
for handle, prim in zip(row_handles, row_prims, strict=True)
if isinstance(prim, Text)
)
await backend.entity_set_xdata(
row_handle,
REV_APP_ID,
rev_values(
{
"v": REV_PAYLOAD_VERSION,
"rev": code,
"description": description,
"date": date,
"by": by,
}
),
)
return {
"ok": True,
"created": True,
"rev": code,
"handle": row_handle,
"row": index + 1,
"block_handles": block_handles,
"clouds": cloud_handles,
"tags": tag_handles,
"layout": layout or "Model",
"bbox": {
"min": [anchor[0], anchor[1] - ROW_HEIGHT * (index + 2)],
"max": [anchor[0] + LIST_WIDTH, anchor[1]],
},
}
finally:
if layout is not None:
await backend.layout_set_current(previous)
# The payload rides on the row band's first Text primitive, which is its
# REV cell -- `rev` is validated non-empty, so that cell is always drawn.
row_handles = block_handles[len(head_prims) :]
row_handle = next(
handle
for handle, prim in zip(row_handles, row_prims, strict=True)
if isinstance(prim, Text)
)
await backend.entity_set_xdata(
row_handle,
REV_APP_ID,
rev_values(
{
"v": REV_PAYLOAD_VERSION,
"rev": code,
"description": description,
"date": date,
"by": by,
}
),
)
return {
"ok": True,
"created": True,
"rev": code,
"handle": row_handle,
"row": index + 1,
"block_handles": block_handles,
"clouds": cloud_handles,
"tags": tag_handles,
"layout": layout or "Model",
"bbox": {
"min": [anchor[0], anchor[1] - ROW_HEIGHT * (index + 2)],
"max": [anchor[0] + LIST_WIDTH, anchor[1]],
},
}
server.py
+28 -3
View File
@@ -8392,6 +8392,15 @@ async def bom_extract(
description="Columns to return; default item, qty, designation, standard, material.",
),
] = None,
limit: Annotated[
int,
Field(
default=0,
ge=0,
description="Stop after this many item records; 0 (the default) reads the whole "
"drawing. `truncated` says whether the cap cut the list.",
),
] = 0,
ctx: Context = None,
) -> dict:
"""Walk the INSERTs and return the ISO 7573 item rows. Never modifies the
@@ -8400,19 +8409,35 @@ async def bom_extract(
Each row's source is reported: `xdata` when the ACADMCP_MECH payload
supplied it (what `std_part_insert` writes), `attributes` when the block's
own ATTRIBs did. A block with neither a payload nor a DESIGNATION attribute
is skipped rather than guessed at, and `skipped` counts them. Refusals: an
unknown column or group_by field, named with the list of the ones there are.
is skipped rather than guessed at, and `skipped` counts them.
The whole drawing is read by default — `entity_list`'s 200-entity page is
paged through, not taken as a cap, because an item list that stops at the
200th INSERT prints a wrong QTY rather than a shorter one. Pass `limit` to
cap it deliberately; `truncated` is then true if and only if the cap cut
the list. Refusals: an unknown column or group_by field, named with the
list of the ones there are.
"""
from engineering.sheet.bom import extract_records, rows_from_records
backend = _backend(ctx)
records = await extract_records(backend, layer=layer or None)
cap = int(limit) or None
# One record past the cap, so `truncated` is measured rather than guessed:
# exactly `cap` records on the drawing is not a truncation.
records = await extract_records(
backend, layer=layer or None, limit=None if cap is None else cap + 1
)
truncated = cap is not None and len(records) > cap
if truncated:
records = records[:cap]
rows = rows_from_records(records, columns=columns, group_by=group_by or None)
return {
"ok": True,
"standard": "ISO 7573",
"rows": [dict(row) | {"handles": list(row["handles"])} for row in rows],
"records": len(records),
"limit": cap,
"truncated": truncated,
"group_by": group_by or None,
}
tests/test_sheet_bom.py
+93
View File
@@ -294,3 +294,96 @@ async def test_reusing_an_item_number_for_different_targets_is_refused(backend):
await add_balloon(backend, item=1, at=(100.0, 40.0), leader_to=(60.0, 0.0), targets=(two,))
assert "1" in str(excinfo.value)
assert len(await backend.entity_list()) == before
@pytest.mark.asyncio
async def test_a_second_balloon_with_the_same_number_and_no_targets_is_refused(backend):
"""`targets` is optional in the tool, so the clash test used to compare
`() != ()` -- False -- and never fire. Two balloons carrying item 1."""
first = await add_balloon(backend, item=1, at=(40.0, 40.0), leader_to=(0.0, 0.0))
before = len(await backend.entity_list())
with pytest.raises(ValueError) as excinfo:
await add_balloon(backend, item=1, at=(90.0, 40.0), leader_to=(60.0, 0.0))
assert first["handle"] in str(excinfo.value)
assert len(await backend.entity_list()) == before
assert [b["item"] for b in await read_balloons(backend)] == [1]
@pytest.mark.asyncio
async def test_the_balloon_guards_read_past_the_entity_list_page(backend):
"""A routine hole pattern fills `entity_list`'s 200-entity default, which
used to make `read_balloons` return nothing and every guard pass."""
for index in range(200):
await backend.entity_create_circle(float(index) * 10.0, 0.0, 2.0)
await add_balloon(backend, item=1, at=(40.0, 400.0), leader_to=(0.0, 380.0), targets=("AAA",))
assert [b["item"] for b in await read_balloons(backend)] == [1]
again = await add_balloon(
backend, item=9, at=(40.0, 400.0), leader_to=(0.0, 380.0), targets=("AAA",)
)
assert again["renumbered"] is True, "the guard must see a balloon past the 200th circle"
with pytest.raises(ValueError):
await add_balloon(
backend, item=9, at=(200.0, 400.0), leader_to=(160.0, 380.0), targets=("BBB",)
)
@pytest.mark.asyncio
async def test_extract_reads_every_insert_not_the_first_page(backend):
"""250 bolts must report QTY 250. The 200-entity `entity_list` default used
to print 200 into a real TABLE with nothing saying it had been cut."""
await backend.block_define("BOLT", [{"type": "circle", "cx": 0.0, "cy": 0.0, "r": 3.0}], [])
for index in range(250):
insert = await backend.block_insert("BOLT", float(index) * 10.0, 0.0)
await backend.entity_set_xdata(
insert.handle,
APP_ID,
to_values(
{
"v": 1,
"kind": "std_part",
"designation": "ISO 4014 - M12x60",
"standard": "ISO 4014",
"size": "M12x60",
"qty": 1,
"material": "8.8",
}
),
)
records = await extract_records(backend)
assert len(records) == 250
rows = rows_from_records(records)
assert [(r["designation"], r["qty"]) for r in rows] == [("ISO 4014 - M12x60", 250)]
# A cap is still honoured when one is asked for, explicitly.
assert len(await extract_records(backend, limit=17)) == 17
with pytest.raises(ValueError):
await extract_records(backend, limit=0)
@pytest.mark.asyncio
async def test_the_balloon_guards_read_the_layout_they_draw_on(backend):
"""`read_balloons` follows the current space, so reading it before
`enter_layout` inspected Model while the balloons went on the layout."""
await backend.layout_create("SHEET1")
first = await add_balloon(
backend, item=1, at=(40.0, 40.0), leader_to=(0.0, 0.0), targets=("AAA",), layout="SHEET1"
)
again = await add_balloon(
backend, item=5, at=(40.0, 40.0), leader_to=(0.0, 0.0), targets=("AAA",), layout="SHEET1"
)
assert again["renumbered"] is True
assert again["handle"] == first["handle"]
with pytest.raises(ValueError):
await add_balloon(
backend,
item=5,
at=(90.0, 40.0),
leader_to=(60.0, 0.0),
targets=("BBB",),
layout="SHEET1",
)
assert (await backend.layout_list())["current"] == "Model", "the caller's tab must come back"
await backend.layout_set_current("SHEET1")
try:
assert len(await backend.entity_list(type_filter="CIRCLE", limit=10_000)) == 2
finally:
await backend.layout_set_current("Model")
tests/test_sheet_revision.py
+37
View File
@@ -146,3 +146,40 @@ async def test_clouds_are_refused_before_any_row_when_the_engine_has_no_revcloud
assert excinfo.value.capability == "revcloud"
assert len(await backend.entity_list()) == before
assert (await read_revisions(backend)) == ()
@pytest.mark.asyncio
async def test_a_revision_on_a_layout_is_still_written_once_per_letter(backend):
"""`read_revisions` follows the current space, so reading it before
`enter_layout` inspected Model while the block went on the layout -- and
every rerun stacked a second heading band on top of the first."""
await backend.layout_create("SHEET1")
first = await add_revision(backend, rev="B", description="bore 40H7", layout="SHEET1")
assert first["created"] is True
again = await add_revision(backend, rev="B", description="bore 40H7", layout="SHEET1")
assert again["created"] is False
assert again["handle"] == first["handle"]
assert (await backend.layout_list())["current"] == "Model", "the caller's tab must come back"
await backend.layout_set_current("SHEET1")
try:
texts = [e.properties["text"] for e in await backend.entity_list(type_filter="TEXT")]
assert texts == ["REV", "DESCRIPTION", "DATE", "BY", "B", "bore 40H7"]
rows = await read_revisions(backend)
assert [r["rev"] for r in rows] == ["B"]
finally:
await backend.layout_set_current("Model")
@pytest.mark.asyncio
async def test_revisions_are_read_past_the_entity_list_page(backend):
"""200 notes fill `entity_list`'s default page. The idempotency guard used
to see no revisions at all, and `index = len(existing)` restarted at 0."""
for index in range(200):
await backend.entity_create_text(text=f"NOTE {index}", x=float(index), y=500.0, height=2.5)
first = await add_revision(backend, rev="A", description="first issue")
assert first["row"] == 1
assert (await add_revision(backend, rev="A", description="first issue"))["created"] is False
second = await add_revision(backend, rev="B", description="bore 40H7")
assert second["row"] == 2, "the second band must sit below the first, not on top of it"
assert second["bbox"]["min"][1] < first["bbox"]["min"][1]
assert [r["rev"] for r in await read_revisions(backend)] == ["A", "B"]