fix(task-2): find_replace rewrites every surface of a multi-line attribute; COM explode propagates a failing read and undoes a failed burst

Headless text_find_replace matched and wrote only dxf.text of an ATTRIB/ATTDEF, so on a multi-line attribute it reported replaced: 1 while block_get_attributes and block_explode still carried the old embedded MTEXT (and a needle on the second line was never found). The ATTRIB/ATTDEF branch now reads through _attrib_value and writes through _write_attrib_value, the same pair block_get_attributes / block_set_attributes use.

COM block_explode swallowed a failing GetAttributes() (attrs = ()), GetConstantAttributes() and Invisible, then exploded, deleted every ATTDEF placeholder and the reference and returned ok: True with no attribute text -- every value destroyed silently on a transient RPC_E_CALL_REJECTED. Those reads now happen before Explode() and propagate with nothing written; a failure after the explode deletes the members it added and the TEXTs written so far and leaves the reference in place (_undo_explode), so the drawing is never left half-burst under an exception. The fakes gain the members a live AcadBlockReference / AcadAttributeDefinition always expose (GetConstantAttributes, Constant, Invisible) instead of relying on the swallow.

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-17 23:32:56 +03:00
co-authored by Claude Opus 5
parent 5f33613460
commit b753a12bb6
4 changed files with 304 additions and 36 deletions
+55 -21
View File
@@ -3431,13 +3431,16 @@ class ComBackend(AutoCADBackend):
A multi-line attribute (``MTextAttribute`` true) keeps its content in A multi-line attribute (``MTextAttribute`` true) keeps its content in
``MTextAttributeContent`` -- ``TextString`` is one flattened string -- ``MTextAttributeContent`` -- ``TextString`` is one flattened string --
and its box width in ``MTextBoundaryWidth``; ``mtext`` marks it so the and its box width in ``MTextBoundaryWidth``; ``mtext`` marks it so the
writer builds an MTEXT. An object that does not expose the member is writer builds an MTEXT. An object that does not expose *that* member
single-line. is single-line (the member is younger than the ActiveX surface, and a
``TextString`` fallback loses nothing).
``Invisible`` is *not* guessed: a read failure used to default to
visible, so a hidden attribute's value (a link, a cost code) became
visible text with ``ok: True``. The failure propagates, and the
caller of this helper reads it before anything is written.
""" """
try:
invisible = bool(attr.Invisible) invisible = bool(attr.Invisible)
except Exception:
invisible = False
try: try:
is_mtext = bool(attr.MTextAttribute) is_mtext = bool(attr.MTextAttribute)
except Exception: except Exception:
@@ -3547,6 +3550,21 @@ class ComBackend(AutoCADBackend):
text.Visible = False text.Visible = False
return text return text
@staticmethod
def _undo_explode(exploded, texts) -> None:
"""Delete what a failed burst has already written (best effort).
An object the burst deleted itself (an ATTDEF placeholder) raises on a
second ``Delete``; that is skipped, every other failure is logged and
the next object is tried, so the original exception -- the one the
caller is about to re-raise -- stays the one the caller sees.
"""
for obj in tuple(texts) + tuple(exploded):
try:
obj.Delete()
except Exception as exc:
log.debug("explode rollback: %r not deleted: %s", obj, exc)
@staticmethod @staticmethod
def _is_constant_attdef(obj, constant_tags: set) -> bool: def _is_constant_attdef(obj, constant_tags: set) -> bool:
"""Whether an exploded ``AcDbAttributeDefinition`` is a constant attribute. """Whether an exploded ``AcDbAttributeDefinition`` is a constant attribute.
@@ -3560,10 +3578,7 @@ class ComBackend(AutoCADBackend):
return bool(obj.Constant) return bool(obj.Constant)
except Exception: except Exception:
pass pass
try:
return str(obj.TagString) in constant_tags return str(obj.TagString) in constant_tags
except Exception:
return False
async def block_explode(self, handle) -> dict: async def block_explode(self, handle) -> dict:
"""Explode an INSERT through ActiveX; ATTRIB values survive as TEXT. """Explode an INSERT through ActiveX; ATTRIB values survive as TEXT.
@@ -3618,6 +3633,16 @@ class ComBackend(AutoCADBackend):
and an xref (``Blocks.Item(Name).IsXRef``; ActiveX ``Explode()`` raises and an xref (``Blocks.Item(Name).IsXRef``; ActiveX ``Explode()`` raises
on one, the headless engine used to delete it) -- the same three on one, the headless engine used to delete it) -- the same three
refusals the headless engine names. refusals the headless engine names.
No read that decides what gets written is guessed: ``GetAttributes()``,
``GetConstantAttributes()`` and each attribute's ``Invisible`` are read
before ``Explode()`` and a failure propagates with nothing written
(a swallowed ``GetAttributes()`` used to burst *no* value and report
``ok: True``). A failure after the explode -- a constant ATTDEF whose
members cannot be read, an ``AddText`` that is refused -- undoes what
the explode added and the TEXTs written so far and leaves the
reference in place (``_undo_explode``), so the drawing is never left
half-burst under an exception.
""" """
def _sync(): def _sync():
@@ -3645,21 +3670,23 @@ class ComBackend(AutoCADBackend):
"be exploded (bind it into the drawing first)" "be exploded (bind it into the drawing first)"
) )
owner = _owner_layout_block(doc, ent, handle) owner = _owner_layout_block(doc, ent, handle)
try: # Every read that decides what the burst writes happens *before*
# ``Explode()`` and propagates: swallowing a failing
# ``GetAttributes()`` (a transient ``RPC_E_CALL_REJECTED`` while
# AutoCAD is busy is enough) went on to explode, delete every
# ATTDEF placeholder and the reference, and report ``ok: True``
# with ``attribute_texts: []`` -- every value destroyed silently,
# the defect this method exists to remove. Nothing has been
# written yet, so raising here costs nothing.
attrs = ent.GetAttributes() attrs = ent.GetAttributes()
except Exception as exc:
log.debug("GetAttributes failed before explode: %s", exc)
attrs = ()
captured = [self._capture_attrib(attr) for attr in attrs] captured = [self._capture_attrib(attr) for attr in attrs]
constant_tags: set = set() constant_tags: set = {str(attr.TagString) for attr in ent.GetConstantAttributes() or ()}
try: exploded = tuple(ent.Explode() or ())
for attr in ent.GetConstantAttributes() or ():
constant_tags.add(str(attr.TagString))
except Exception as exc:
log.debug("GetConstantAttributes failed before explode: %s", exc)
exploded = ent.Explode()
inserted = [] inserted = []
for obj in exploded or (): attribute_texts = []
texts = []
try:
for obj in exploded:
if obj.ObjectName == "AcDbAttributeDefinition": if obj.ObjectName == "AcDbAttributeDefinition":
if self._is_constant_attdef(obj, constant_tags): if self._is_constant_attdef(obj, constant_tags):
# A constant attribute's only text; already WCS here. # A constant attribute's only text; already WCS here.
@@ -3667,10 +3694,17 @@ class ComBackend(AutoCADBackend):
obj.Delete() # the tag placeholder EXPLODE leaves; the value is below obj.Delete() # the tag placeholder EXPLODE leaves; the value is below
continue continue
inserted.append(str(obj.Handle)) inserted.append(str(obj.Handle))
attribute_texts = []
for item in captured: for item in captured:
text = self._add_text_from_capture(owner, item) text = self._add_text_from_capture(owner, item)
texts.append(text)
attribute_texts.append(str(text.Handle)) attribute_texts.append(str(text.Handle))
except Exception:
# ActiveX ``Explode()`` leaves the reference in place, so the
# drawing is put back to exactly that: the members it added and
# the TEXTs written so far are removed, the reference stays,
# and the failure propagates instead of a half-burst symbol.
self._undo_explode(exploded, texts)
raise
ent.Delete() ent.Delete()
_regen() _regen()
return { return {
+15 -1
View File
@@ -3958,13 +3958,27 @@ class EzdxfBackend(AutoCADBackend):
seen.add(key) seen.add(key)
if layer and entity.dxf.get("layer", "0") != layer: if layer and entity.dxf.get("layer", "0") != layer:
continue continue
current = entity.text if dxftype == "MTEXT" else entity.dxf.get("text", "") # An ATTRIB/ATTDEF is read and written through the attribute
# helpers: a multi-line one keeps its content in an embedded
# MTEXT and ``dxf.text`` is only the first line, so matching
# and writing ``dxf.text`` alone reported ``replaced: 1`` while
# ``block_get_attributes`` and ``block_explode`` still carried
# the old value. One value, every surface -- the same rule
# ``block_set_attributes`` follows.
if dxftype == "MTEXT":
current = entity.text
elif dxftype in ("ATTRIB", "ATTDEF"):
current = self._attrib_value(entity)
else:
current = entity.dxf.get("text", "")
if not current or not pattern.search(current): if not current or not pattern.search(current):
continue continue
updated = pattern.sub(replace, current) updated = pattern.sub(replace, current)
if not dry_run: if not dry_run:
if dxftype == "MTEXT": if dxftype == "MTEXT":
entity.text = updated entity.text = updated
elif dxftype in ("ATTRIB", "ATTDEF"):
self._write_attrib_value(entity, updated)
else: else:
entity.dxf.text = updated entity.dxf.text = updated
changed.append( changed.append(
+3 -1
View File
@@ -2417,7 +2417,9 @@ async def text_find_replace(
never searched" are different answers. Block *definitions* are included, so never searched" are different answers. Block *definitions* are included, so
the next insert does not reintroduce the old text. DIMENSION text is out of the next insert does not reintroduce the old text. DIMENSION text is out of
scope: its text field holds the `<>` override placeholder rather than the scope: its text field holds the `<>` override placeholder rather than the
measurement, so editing it would break the association. measurement, so editing it would break the association. A multi-line
attribute is matched and rewritten as the whole value (`Line1\\PLine2`),
the same value `block_get_attributes` reports and `block_explode` bursts.
""" """
await ctx.info(f"Replacing {find!r} with {replace!r}{' (dry run)' if dry_run else ''}") await ctx.info(f"Replacing {find!r} with {replace!r}{' (dry run)' if dry_run else ''}")
return await _backend(ctx).text_find_replace(find, replace, layer or None, match_case, dry_run) return await _backend(ctx).text_find_replace(find, replace, layer or None, match_case, dry_run)
+219 -1
View File
@@ -328,7 +328,7 @@ async def test_explode_refuses_a_non_insert_without_writing(backend):
async def test_com_explode_deletes_attdefs_adds_text_per_attrib_and_deletes_the_reference(com): async def test_com_explode_deletes_attdefs_adds_text_per_attrib_and_deletes_the_reference(com):
backend, document, space = com backend, document, space = com
attdef = _FakeObject("AcDbAttributeDefinition", "D1") attdef = _FakeObject("AcDbAttributeDefinition", "D1", Constant=False)
line = _FakeObject("AcDbLine", "L1") line = _FakeObject("AcDbLine", "L1")
visible = _FakeObject( visible = _FakeObject(
"AcDbAttribute", "AcDbAttribute",
@@ -358,6 +358,7 @@ async def test_com_explode_deletes_attdefs_adds_text_per_attrib_and_deletes_the_
Name="TB", Name="TB",
OwnerID=1, OwnerID=1,
GetAttributes=lambda: (visible, hidden), GetAttributes=lambda: (visible, hidden),
GetConstantAttributes=lambda: (),
Explode=lambda: (line, attdef), Explode=lambda: (line, attdef),
) )
document.objects["2F"] = ref document.objects["2F"] = ref
@@ -412,6 +413,7 @@ async def test_com_explode_adds_the_text_to_the_owner_block_not_the_active_layou
Name="TB", Name="TB",
OwnerID=2, OwnerID=2,
GetAttributes=lambda: (title,), GetAttributes=lambda: (title,),
GetConstantAttributes=lambda: (),
Explode=lambda: (line,), Explode=lambda: (line,),
) )
document.objects["4B"] = ref document.objects["4B"] = ref
@@ -439,6 +441,7 @@ async def test_com_explode_refuses_a_nested_reference_before_explode(com):
Name="TB", Name="TB",
OwnerID=3, OwnerID=3,
GetAttributes=lambda: (), GetAttributes=lambda: (),
GetConstantAttributes=lambda: (),
Explode=_explode, Explode=_explode,
) )
document.objects["5C"] = ref document.objects["5C"] = ref
@@ -482,6 +485,7 @@ async def test_com_explode_carries_the_attrib_frame_and_style_onto_the_text(com)
Name="TB", Name="TB",
OwnerID=1, OwnerID=1,
GetAttributes=lambda: (attrib,), GetAttributes=lambda: (attrib,),
GetConstantAttributes=lambda: (),
Explode=lambda: (), Explode=lambda: (),
) )
document.objects["6D"] = ref document.objects["6D"] = ref
@@ -555,6 +559,7 @@ async def test_com_explode_skips_frame_members_the_attrib_does_not_expose(com):
Name="TB", Name="TB",
OwnerID=1, OwnerID=1,
GetAttributes=lambda: (attrib,), GetAttributes=lambda: (attrib,),
GetConstantAttributes=lambda: (),
Explode=lambda: (), Explode=lambda: (),
) )
document.objects["7E"] = ref document.objects["7E"] = ref
@@ -647,6 +652,7 @@ async def test_com_explode_falls_back_to_the_constant_tag_list_when_constant_is_
Height=2.0, Height=2.0,
Rotation=0.0, Rotation=0.0,
Layer="0", Layer="0",
Invisible=False,
) )
placeholder = _FakeObject("AcDbAttributeDefinition", "D1", TagString="TAG") placeholder = _FakeObject("AcDbAttributeDefinition", "D1", TagString="TAG")
ref = _FakeObject( ref = _FakeObject(
@@ -973,6 +979,7 @@ async def test_com_explode_turns_a_multi_line_attrib_into_mtext(com):
Name="TB", Name="TB",
OwnerID=1, OwnerID=1,
GetAttributes=lambda: (note, plain), GetAttributes=lambda: (note, plain),
GetConstantAttributes=lambda: (),
Explode=lambda: (line,), Explode=lambda: (line,),
) )
document.objects["6D"] = ref document.objects["6D"] = ref
@@ -1028,6 +1035,7 @@ async def test_com_explode_multi_line_attrib_takes_the_frame_before_the_angle(co
Name="TB", Name="TB",
OwnerID=1, OwnerID=1,
GetAttributes=lambda: (note,), GetAttributes=lambda: (note,),
GetConstantAttributes=lambda: (),
Explode=lambda: (), Explode=lambda: (),
) )
document.objects["7E"] = ref document.objects["7E"] = ref
@@ -1071,6 +1079,7 @@ async def test_com_explode_refuses_an_xref_before_explode(com):
Name="XREF_DETAIL", Name="XREF_DETAIL",
OwnerID=1, OwnerID=1,
GetAttributes=lambda: (), GetAttributes=lambda: (),
GetConstantAttributes=lambda: (),
Explode=_explode, Explode=_explode,
) )
document.objects["8A"] = ref document.objects["8A"] = ref
@@ -1097,6 +1106,7 @@ async def test_com_explode_refuses_a_minsert_by_name_before_explode(com):
Rows=2, Rows=2,
Columns=3, Columns=3,
GetAttributes=lambda: (), GetAttributes=lambda: (),
GetConstantAttributes=lambda: (),
Explode=_explode, Explode=_explode,
) )
document.objects["9B"] = grid document.objects["9B"] = grid
@@ -1286,3 +1296,211 @@ async def test_explode_opaque_members_is_declared_on_both_engines():
headless = EzdxfBackend().capabilities().features["explode_opaque_members"] headless = EzdxfBackend().capabilities().features["explode_opaque_members"]
assert headless.supported is False and "OLE2FRAME" in (headless.reason or "") assert headless.supported is False and "OLE2FRAME" in (headless.reason or "")
assert ComBackend().capabilities().features["explode_opaque_members"].supported is True assert ComBackend().capabilities().features["explode_opaque_members"].supported is True
# ── Task 2 (review 3): find_replace on a multi-line attribute; COM reads propagate ─
async def test_find_replace_rewrites_every_surface_of_a_multi_line_attrib(backend):
"""``text_find_replace`` matched and wrote only ``dxf.text`` of an ATTRIB,
so on a multi-line attribute it reported ``replaced: 1, 'OLD ONE' ->
'NEW ONE'`` while ``block_get_attributes`` still read ``OLD ONE\\POLD TWO``
and ``block_explode`` burst it. The ATTRIB branch goes through the same
reader and writer the attribute tools use: one value, every surface."""
ref, attrib = await _insert_multi_line(backend)
result = await backend.text_find_replace("OLD", "NEW")
assert result["replaced"] == 1
(entry,) = result["entities"]
assert entry["type"] == "ATTRIB"
assert entry["before"] == "OLD ONE\\POLD TWO", "the whole value is what was matched"
assert entry["after"] == "NEW ONE\\PNEW TWO", "every line is replaced, not the first"
assert attrib.has_embedded_mtext_entity, "still a multi-line attribute"
assert attrib.virtual_mtext_entity().text == "NEW ONE\\PNEW TWO"
assert attrib.dxf.text == "NEW ONE\\PNEW TWO"
assert await backend.block_get_attributes(ref.handle) == {"NOTE": "NEW ONE\\PNEW TWO"}
exploded = await backend.block_explode(ref.handle)
(handle,) = exploded["attribute_texts"]
raw = backend._doc.entitydb[handle]
assert raw.dxftype() == "MTEXT" and raw.text == "NEW ONE\\PNEW TWO"
assert backend._doc.audit().errors == []
async def test_find_replace_matches_a_line_below_the_first_of_a_multi_line_attrib(backend):
"""A needle that only occurs on the second line was invisible to a search
of ``dxf.text`` (the first line alone): ``replaced: 0`` on a value the
reader plainly reports."""
ref, attrib = await _insert_multi_line(backend)
result = await backend.text_find_replace("TWO", "2")
assert result["replaced"] == 1
assert await backend.block_get_attributes(ref.handle) == {"NOTE": "OLD ONE\\POLD 2"}
assert attrib.virtual_mtext_entity().text == "OLD ONE\\POLD 2"
async def test_find_replace_dry_run_reports_the_whole_multi_line_value_and_writes_nothing(
backend,
):
ref, attrib = await _insert_multi_line(backend)
result = await backend.text_find_replace("OLD", "NEW", dry_run=True)
assert result["dry_run"] is True and result["replaced"] == 1
assert result["entities"][0]["before"] == "OLD ONE\\POLD TWO"
assert result["entities"][0]["after"] == "NEW ONE\\PNEW TWO"
assert attrib.virtual_mtext_entity().text == "OLD ONE\\POLD TWO"
assert attrib.dxf.text == "OLD ONE"
assert await backend.block_get_attributes(ref.handle) == {"NOTE": "OLD ONE\\POLD TWO"}
async def test_find_replace_rewrites_a_multi_line_attdef_inside_a_block_definition(backend):
"""The ATTDEF twin: the default a new INSERT is prompted with lives in the
block definition, and a multi-line one keeps it in an embedded MTEXT."""
blk = backend._doc.blocks.new(name="MLD")
blk.add_line((0, 0), (10, 0))
attdef = blk.add_attdef("NOTE", insert=(0, 3), text="x", dxfattribs={"height": 2.5})
_embed_multi_line(attdef, ["OLD ONE", "OLD TWO"], (0, 3))
result = await backend.text_find_replace("OLD", "NEW")
assert result["replaced"] == 1 and result["entities"][0]["type"] == "ATTDEF"
assert result["entities"][0]["after"] == "NEW ONE\\PNEW TWO"
assert attdef.has_embedded_mtext_entity
assert attdef.virtual_mtext_entity().text == "NEW ONE\\PNEW TWO"
assert attdef.dxf.text == "NEW ONE\\PNEW TWO"
assert backend._doc.audit().errors == []
async def test_find_replace_leaves_a_single_line_attrib_single_line(backend):
_define_tagged_block(backend, "TB3")
ref = await backend.block_insert("TB3", 0, 0, attributes={"TAG": "OLD-1"})
attrib = backend._doc.entitydb[ref.handle].attribs[0]
result = await backend.text_find_replace("OLD", "NEW")
assert result["replaced"] == 1 and result["entities"][0]["after"] == "NEW-1"
assert attrib.dxf.text == "NEW-1" and not attrib.has_embedded_mtext_entity
assert await backend.block_get_attributes(ref.handle) == {"TAG": "NEW-1"}
def _boom(*_args):
raise RuntimeError("RPC_E_CALL_REJECTED: the application is busy")
@pytest.mark.parametrize(
"broken",
["GetAttributes", "GetConstantAttributes", "Invisible"],
ids=["get_attributes", "get_constant_attributes", "invisible"],
)
async def test_com_explode_propagates_a_failing_read_before_explode(com, broken):
"""A failing ``GetAttributes()`` used to be swallowed (``attrs = ()``), and
the burst went on to explode, delete every ATTDEF placeholder and the
reference, and return ``ok: True, attribute_texts: []`` -- every value
destroyed silently, the defect the BURST rule exists to remove. A
transient ``RPC_E_CALL_REJECTED`` while AutoCAD is busy is enough to
trigger it. The same for ``GetConstantAttributes()`` (a constant ATTDEF
would then be deleted as a placeholder) and for ``Invisible`` (a hidden
value would appear as visible text). Nothing has been written at that
point, so the failure propagates and the drawing is untouched."""
backend, document, space = com
dispatched = []
attdef = _FakeObject("AcDbAttributeDefinition", "D1", Constant=False)
line = _FakeObject("AcDbLine", "L1")
def _explode():
dispatched.append("Explode")
return (line, attdef)
attrib_members = dict(
TagString="TAG",
TextString="P-101",
InsertionPoint=(7.0, 15.2, 0.0),
Height=5.0,
Rotation=0.0,
Layer="PID-TAG",
Invisible=True,
)
if broken == "Invisible":
attrib_members.pop("Invisible") # the read raises AttributeError on the fake
attrib = _FakeObject("AcDbAttribute", "A1", **attrib_members)
ref_members = dict(
Name="TB",
OwnerID=1,
GetAttributes=lambda: (attrib,),
GetConstantAttributes=lambda: (),
Explode=_explode,
)
if broken in ref_members:
ref_members[broken] = _boom
ref = _FakeObject("AcDbBlockReference", "2F", **ref_members)
document.objects["2F"] = ref
with pytest.raises((RuntimeError, AttributeError)):
await backend.block_explode("2F")
assert dispatched == [], "the read failed before Explode() was dispatched"
assert space.calls == [] and ref.deleted is False and attdef.deleted is False
class _SpaceFailingSecondAddText(_FakeSpace):
def AddText(self, text, point, height):
if self.texts:
raise RuntimeError("E_FAIL: AddText")
return super().AddText(text, point, height)
async def test_com_explode_undoes_a_burst_that_fails_after_explode(com):
"""ActiveX ``Explode()`` leaves the reference in place, so a failure after
it (here ``AddText`` on the second attribute; a constant ATTDEF whose
members cannot be read is the other way in) is put back to that: the
members the explode added and the TEXTs written so far are deleted, the
reference stays, and the failure propagates rather than a half-burst
symbol with ``ok: True``."""
backend, document, space = com
sheet = _SpaceFailingSecondAddText(name="Sheet")
document.owners[2] = sheet
first = _FakeObject(
"AcDbAttribute",
"A1",
TagString="TAG",
TextString="P-101",
InsertionPoint=(7.0, 15.2, 0.0),
Height=5.0,
Rotation=0.0,
Layer="0",
Invisible=False,
)
second = _FakeObject(
"AcDbAttribute",
"A2",
TagString="LINK",
TextString="L-1",
InsertionPoint=(1.0, 1.0, 0.0),
Height=2.0,
Rotation=0.0,
Layer="0",
Invisible=False,
)
placeholder = _FakeObject("AcDbAttributeDefinition", "D1", Constant=False)
line = _FakeObject("AcDbLine", "L1")
ref = _FakeObject(
"AcDbBlockReference",
"4C",
Name="TB",
OwnerID=2,
GetAttributes=lambda: (first, second),
GetConstantAttributes=lambda: (),
Explode=lambda: (line, placeholder),
)
document.objects["4C"] = ref
with pytest.raises(RuntimeError, match="E_FAIL: AddText"):
await backend.block_explode("4C")
assert ref.deleted is False, "the reference Explode() left in place is kept"
assert line.deleted is True, "the member Explode() added is removed"
assert placeholder.deleted is True
assert [t.deleted for t in sheet.texts] == [True], "the TEXT written before the failure"
assert space.calls == []