diff --git a/backends/com_backend.py b/backends/com_backend.py index a51fad8..74476bd 100644 --- a/backends/com_backend.py +++ b/backends/com_backend.py @@ -3106,7 +3106,13 @@ class ComBackend(AutoCADBackend): return await self._run(_sync) async def entity_set_xdata(self, handle, app_name, values) -> dict: - from backends.xdata_specs import encode_values, validate_app_name + from backends.xdata_specs import ( + app_size, + check_entity_budget, + encode_values, + split_tags_by_app, + validate_app_name, + ) # Validate and type every value before any ActiveX call: a bad # values[i] raises here and nothing is written. @@ -3116,6 +3122,14 @@ class ComBackend(AutoCADBackend): def _sync(): doc = _acad_doc() ent = doc.HandleToObject(str(handle).strip().upper()) + if tags: + # The 16 KB limit is per entity across every application, the + # same gate the headless engine runs: read what is already + # there and refuse before the APPID is registered or SetXData + # gets the chance to fail half-way with AutoCAD's own error. + codes, values_ = ent.GetXData("") + existing = split_tags_by_app(list(codes or []), list(values_ or [])) + check_entity_budget(app, tags, existing) try: doc.RegisteredApplications.Add(app) except Exception as exc: # already registered @@ -3133,7 +3147,7 @@ class ComBackend(AutoCADBackend): "handle": handle, "app_name": app, "value_count": len(tags), - "bytes": sum(len(str(v)) for _, v in tags), + "bytes": app_size(app, tags) if tags else 0, "removed": not tags, "backend": "com", } diff --git a/backends/ezdxf_backend.py b/backends/ezdxf_backend.py index d3c656e..a39b81a 100644 --- a/backends/ezdxf_backend.py +++ b/backends/ezdxf_backend.py @@ -4540,7 +4540,12 @@ class EzdxfBackend(AutoCADBackend): return await self._async(_sync) async def entity_set_xdata(self, handle, app_name, values) -> dict: - from backends.xdata_specs import encode_values, validate_app_name + from backends.xdata_specs import ( + app_size, + check_entity_budget, + encode_values, + validate_app_name, + ) # Validate and type every value before touching the document: a bad # values[i] raises here and nothing is written. @@ -4562,6 +4567,16 @@ class EzdxfBackend(AutoCADBackend): "removed": True, "backend": "ezdxf", } + # The 16 KB limit is per entity across every application (ACAD's + # own DSTYLE overrides included), so the gate sees what is already + # there. Raises before the APPID is registered or anything is set. + existing = {} + if ent.xdata is not None: + existing = { + other: [(tag.code, tag.value) for tag in other_tags] + for other, other_tags in ent.xdata.data.items() + } + check_entity_budget(app, tags, existing) if app not in doc.appids: doc.appids.add(app) ent.set_xdata(app, tags) @@ -4571,7 +4586,7 @@ class EzdxfBackend(AutoCADBackend): "handle": handle, "app_name": app, "value_count": len(tags), - "bytes": sum(len(str(v)) for _, v in tags), + "bytes": app_size(app, tags), "removed": False, "backend": "ezdxf", } diff --git a/backends/xdata_specs.py b/backends/xdata_specs.py index c6b3240..0d1e470 100644 --- a/backends/xdata_specs.py +++ b/backends/xdata_specs.py @@ -3,16 +3,29 @@ DXF XDATA is a list of (group code, value) pairs under a registered APPID. This module is the one place that decides how a JSON value becomes a group code and enforces the two limits AutoCAD enforces silently: 255 characters -per string and 16 KB per entity. It also refuses the one thing neither engine -can store: a line separator inside a string. DXF is a text format of +per string and 16 KB per *entity* -- across every application on it, the ACAD +app's own overrides (DSTYLE on a toleranced dimension) included, not just the +app being written. ``encode_values`` refuses a single call that cannot fit; +``check_entity_budget`` is the real gate, run by both engines against the +entity's existing XDATA before anything is written, so the headless engine +refuses exactly the write the live engine's ``SetXData`` would reject. + +It also refuses the values neither engine can store: a line separator inside a +string, and a non-finite or non-numeric "number". DXF is a text format of alternating "group code" / "value" lines, and ezdxf writes a 1000 tag verbatim, so a LF or CR in the value splits the tag pair and the saved file -- and every -undo/transaction snapshot, which is also a DXF save -- becomes unreadable. +undo/transaction snapshot, which is also a DXF save -- becomes unreadable. A +``nan``/``inf`` real is written as the literal text ``nan``/``inf``, which is +not a DXF double, and a string coerced into a point coordinate is a silent +wrong number; both are refused before any write, the same way +``backends/block_specs.py`` refuses them for block primitives. """ from __future__ import annotations +import math import re +from collections.abc import Iterable, Mapping APP_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]{0,30}$") RESERVED_APPS = {"ACAD"} @@ -21,6 +34,8 @@ MAX_BYTES = 16 * 1024 INT_MIN, INT_MAX = -(2**31), 2**31 - 1 LINE_SEPARATORS = ("\n", "\r") +Tag = tuple[int, object] + def validate_app_name(app) -> str: if not isinstance(app, str) or not APP_NAME_RE.match(app): @@ -30,12 +45,57 @@ def validate_app_name(app) -> str: return app -def encode_values(values) -> list[tuple[int, object]]: +def _real(value, where: str) -> float: + """A finite float from an int or float; strings, bools and nan/inf are refused.""" + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise TypeError(f"xdata: {where} must be a number, got {value!r}") + try: + number = float(value) + except OverflowError: + number = math.inf + if not math.isfinite(number): + raise ValueError(f"xdata: {where} must be finite, got {value!r}") + return number + + +def tag_size(code: int, value) -> int: + """Bytes one XDATA tag occupies in AutoCAD's accounting (what ``xdsize`` reports). + + Every tag carries a one-byte group code; strings (1000-1004) add a two-byte + length, reals are 8 bytes, points 3 reals, a 32-bit int 4, a 16-bit int 2, + a handle 8. The 1001 application marker is a string and counts like one. + """ + if code in (1000, 1001, 1002, 1003, 1004, 1005): + if isinstance(value, str): + return 3 + len(value.encode("utf-8")) + if isinstance(value, (bytes, bytearray, list, tuple)): # 1004 binary chunk + return 3 + len(value) + return 3 + len(str(value).encode("utf-8")) + if code in (1010, 1011, 1012, 1013): + return 1 + 3 * 8 + if code in (1040, 1041, 1042): + return 1 + 8 + if code == 1070: + return 1 + 2 + if code == 1071: + return 1 + 4 + raise ValueError(f"xdata: {code} is not an XDATA group code") + + +def app_size(app: str, tags: Iterable[Tag]) -> int: + """Bytes one application's XDATA occupies: its 1001 marker plus every tag. + + ``tags`` may or may not carry the 1001 marker (ezdxf stores it, its + ``get_xdata`` strips it); it is counted exactly once either way. + """ + return tag_size(1001, app) + sum(tag_size(code, value) for code, value in tags if code != 1001) + + +def encode_values(values) -> list[Tag]: """JSON values → (group code, value). Raises TypeError/ValueError; writes nothing.""" if not isinstance(values, (list, tuple)): raise TypeError("xdata: values must be a list") - tags: list[tuple[int, object]] = [] - size = 0 + tags: list[Tag] = [] for i, value in enumerate(values): where = f"values[{i}]" if isinstance(value, bool): @@ -51,33 +111,56 @@ def encode_values(values) -> list[tuple[int, object]]: "line separators are not allowed in XDATA strings" ) tags.append((1000, value)) - size += len(value.encode("utf-8")) + 2 elif isinstance(value, int): if not INT_MIN <= value <= INT_MAX: raise ValueError(f"xdata: {where} is outside the 32-bit integer range") tags.append((1071, int(value))) - size += 6 elif isinstance(value, float): - tags.append((1040, float(value))) - size += 10 + tags.append((1040, _real(value, where))) elif isinstance(value, (list, tuple)) and len(value) in (2, 3): - if any(isinstance(c, bool) for c in value): - raise TypeError(f"xdata: {where} point coordinates must be numbers") - try: - coords = tuple(float(c) for c in value) - except (TypeError, ValueError) as exc: - raise TypeError(f"xdata: {where} point coordinates must be numbers") from exc + coords = tuple(_real(c, f"{where}[{j}]") for j, c in enumerate(value)) if len(coords) == 2: coords = (coords[0], coords[1], 0.0) tags.append((1010, coords)) - size += 26 else: raise TypeError(f"xdata: {where} must be a string, int, float or [x, y(, z)] point") + # A single call that cannot fit on an empty entity is refused here; the + # per-entity budget (existing apps included) is check_entity_budget(). + size = sum(tag_size(code, value) for code, value in tags) if size > MAX_BYTES: raise ValueError(f"xdata: encoded size {size} bytes exceeds the 16 KB per-entity limit") return tags +def check_entity_budget( + app: str, tags: Iterable[Tag], existing: Mapping[str, Iterable[Tag]] +) -> int: + """Bytes the entity's XDATA occupies once ``app`` is replaced by ``tags``. + + ``existing`` is every application currently on the entity (ACAD included) + with its tags; ``app``'s own entry is what the write replaces, so it is + dropped from the sum. Raises ValueError over ``MAX_BYTES``; writes nothing. + An empty ``tags`` removes the app and is never refused, whatever the other + applications already hold (a file opened with too much XDATA must still be + shrinkable). + """ + tags = list(tags) + others = 0 + for other, other_tags in existing.items(): + if other == app: + continue + others += app_size(other, other_tags) + if not tags: + return others + total = app_size(app, tags) + others + if total > MAX_BYTES: + raise ValueError( + f"xdata: entity would carry {total} bytes of XDATA ({others} under other " + f"applications); the limit is 16 KB per entity across every application" + ) + return total + + def decode_values(tags) -> list: """(group code, value) → JSON values. Skips the 1001 app marker and 1002 braces.""" out: list = [] @@ -95,18 +178,19 @@ def decode_values(tags) -> list: return out -def split_by_app(codes, values) -> dict[str, list]: - """A flat ActiveX GetXData stream (1001 markers inline) → {app: values}.""" - out: dict[str, list] = {} +def split_tags_by_app(codes, values) -> dict[str, list[Tag]]: + """A flat ActiveX GetXData stream (1001 markers inline) → {app: raw tags}.""" + out: dict[str, list[Tag]] = {} current: str | None = None - pending: list[tuple[int, object]] = [] for code, value in zip(codes, values, strict=True): if code == 1001: - if current is not None: - out[current] = decode_values(pending) - current, pending = str(value), [] - else: - pending.append((code, value)) - if current is not None: - out[current] = decode_values(pending) + current = str(value) + out.setdefault(current, []) + elif current is not None: + out[current].append((code, value)) return out + + +def split_by_app(codes, values) -> dict[str, list]: + """A flat ActiveX GetXData stream (1001 markers inline) → {app: values}.""" + return {app: decode_values(tags) for app, tags in split_tags_by_app(codes, values).items()} diff --git a/tests/test_xdata.py b/tests/test_xdata.py index e60cee6..93b5371 100644 --- a/tests/test_xdata.py +++ b/tests/test_xdata.py @@ -9,6 +9,8 @@ import pytest from backends.xdata_specs import ( MAX_BYTES, MAX_STRING, + app_size, + check_entity_budget, decode_values, encode_values, split_by_app, @@ -42,6 +44,34 @@ def test_untypeable_values_are_refused(bad): encode_values([bad]) +@pytest.mark.parametrize( + "bad", + [ + ["1", "2"], # strings coerce with float() but are not numbers + ["nan", 1], # ... and "nan" is the silent wrong number end to end + [1, "2", 3], + [float("nan"), 1], + [1, float("inf")], + [1, 2, float("-inf")], + ], +) +def test_point_coordinates_must_be_finite_numbers(bad): + with pytest.raises((TypeError, ValueError), match=r"values\[0\]\[\d\]"): + encode_values([bad]) + + +@pytest.mark.parametrize("bad", [float("nan"), float("inf"), float("-inf")]) +def test_non_finite_floats_are_refused(bad): + # ezdxf writes a 1040 real verbatim, so nan/inf would land in the file + # (and every undo snapshot) as literal text that is not a DXF double. + with pytest.raises(ValueError, match=r"values\[1\] must be finite"): + encode_values([1.5, bad]) + + +def test_integer_coordinates_still_encode_as_reals(): + assert encode_values([[1, 2], [3, 4, 5]]) == [(1010, (1.0, 2.0, 0.0)), (1010, (3.0, 4.0, 5.0))] + + def test_string_over_255_is_refused(): encode_values(["x" * MAX_STRING]) with pytest.raises(ValueError, match="255"): @@ -70,6 +100,29 @@ def test_total_size_over_16k_is_refused(): encode_values(["x" * 255] * (MAX_BYTES // 255 + 1)) +def test_entity_budget_counts_every_app_and_the_markers(): + # AutoCAD's 16 KB is per object across all apps (xdroom/xdsize), so the + # gate must add what the entity already carries under other apps, ACAD's + # own DSTYLE overrides included, plus each app's 1001 marker. + big = encode_values(["x" * 255] * 40) # 40 * 258 = 10320 bytes + marker + assert app_size("APP_A", big) == 40 * 258 + 3 + len("APP_A") + assert check_entity_budget("APP_A", big, {}) == app_size("APP_A", big) + # Replacing the same app does not double-count its old payload. + assert check_entity_budget("APP_A", big, {"APP_A": big}) == app_size("APP_A", big) + # A second app of the same size no longer fits next to the first. + with pytest.raises(ValueError, match=r"other applications.*16 KB per entity"): + check_entity_budget("APP_B", big, {"APP_A": big}) + # Removal is never refused, even when what is left is already over the + # limit (a file opened with too much XDATA must still be shrinkable). + assert check_entity_budget("APP_B", [], {"APP_A": big}) == app_size("APP_A", big) + assert check_entity_budget("APP_A", [], {"APP_A": big, "APP_B": big}) == app_size("APP_B", big) + # Existing tags may carry their own 1001 marker (ezdxf stores it); it is + # counted once either way. + with_marker = [(1001, "APP_A"), *big] + assert app_size("APP_A", with_marker) == app_size("APP_A", big) + assert check_entity_budget("APP_B", [], {"APP_A": with_marker}) == app_size("APP_A", big) + + @pytest.mark.parametrize("bad", ["", "1abc", "a-b", "a" * 32, "ACAD"]) def test_bad_app_names_are_refused(bad): with pytest.raises(ValueError): @@ -125,6 +178,68 @@ async def test_limits_are_enforced_before_writing(backend): assert (await backend.entity_get_xdata(line.handle))["xdata"] == {} +async def test_non_finite_and_string_coordinates_are_refused_before_writing(backend): + line = await backend.entity_create_line(0, 0, 1, 1) + with pytest.raises(TypeError, match=r"values\[0\]\[0\] must be a number"): + await backend.entity_set_xdata(line.handle, "APP_A", [["nan", 1]]) + with pytest.raises(ValueError, match=r"values\[0\] must be finite"): + await backend.entity_set_xdata(line.handle, "APP_A", [float("nan")]) + with pytest.raises(ValueError, match=r"values\[1\] must be finite"): + await backend.entity_set_xdata(line.handle, "APP_A", [1.0, float("inf")]) + assert (await backend.entity_get_xdata(line.handle))["xdata"] == {} + + +async def test_16k_is_per_entity_across_apps_not_per_call(backend, tmp_path): + # Three 15 KB payloads under three apps each fit on their own; the entity + # does not, and AutoCAD's SetXData would refuse the second one live. + line = await backend.entity_create_line(0, 0, 1, 1) + payload = ["x" * 255] * 60 + first = await backend.entity_set_xdata(line.handle, "APP_A", payload) + assert first["ok"] is True and first["bytes"] == 60 * 258 + 3 + len("APP_A") + with pytest.raises(ValueError, match=r"other applications.*16 KB per entity"): + await backend.entity_set_xdata(line.handle, "APP_B", payload) + with pytest.raises(ValueError, match="16 KB per entity"): + await backend.entity_set_xdata(line.handle, "APP_C", ["y" * 255] * 5) + got = (await backend.entity_get_xdata(line.handle))["xdata"] + assert set(got) == {"APP_A"} and "APP_B" not in backend._doc.appids + # Replacing APP_A wholesale is judged against the entity *without* its old + # APP_A payload, so a same-size rewrite is fine ... + await backend.entity_set_xdata(line.handle, "APP_A", ["z" * 255] * 60) + # ... and once APP_A shrinks, APP_B fits. + await backend.entity_set_xdata(line.handle, "APP_A", ["small"]) + await backend.entity_set_xdata(line.handle, "APP_B", payload) + got = (await backend.entity_get_xdata(line.handle))["xdata"] + assert got["APP_A"] == ["small"] and len(got["APP_B"]) == 60 + path = str(tmp_path / "budget.dxf") + await backend.drawing_save(path) + await backend.drawing_open(path) + assert len((await backend.entity_get_xdata(line.handle, "APP_B"))["xdata"]["APP_B"]) == 60 + + +async def test_budget_counts_acad_xdata_the_dimension_tools_write(backend): + # A toleranced dimension carries DSTYLE overrides under the ACAD app; the + # live engine counts them against the 16 KB and so must the gate. + dim = await backend.dimension_linear( + 0, 0, 100, 0, 0, 20, tol_upper=0.1, tol_lower=-0.1, tol_mode="deviation" + ) + acad = (await backend.entity_get_xdata(dim.handle))["xdata"] + assert "ACAD" in acad and acad["ACAD"], acad + ent = backend._get_entity(dim.handle) + acad_bytes = app_size("ACAD", [(t.code, t.value) for t in ent.xdata.data["ACAD"]]) + assert acad_bytes > 0 + # A payload that fits an empty entity but not one carrying the override. + base = ["x" * 255] * 63 + slack = MAX_BYTES - app_size("APP_A", encode_values(base)) + assert 3 < acad_bytes < slack + filler = [*base, "y" * (slack - 3 - acad_bytes // 2)] + assert app_size("APP_A", encode_values(filler)) <= MAX_BYTES + assert app_size("APP_A", encode_values(filler)) + acad_bytes > MAX_BYTES + with pytest.raises(ValueError, match=r"other applications.*16 KB per entity"): + await backend.entity_set_xdata(dim.handle, "APP_A", filler) + # Nothing was written and the override is untouched. + assert (await backend.entity_get_xdata(dim.handle))["xdata"] == acad + + async def test_newline_is_refused_before_writing_and_snapshots_still_load(backend, tmp_path): # Every undo/transaction snapshot is a DXF save, so an accepted newline # would break rollback, not just the file on disk. @@ -215,6 +330,25 @@ async def test_com_empty_values_write_only_the_marker(com_backend): assert entity.stored == ([1001], ["ACADMCP_PID"]) +async def test_com_budget_reads_the_whole_stream_before_writing(com_backend): + backend, entity, registered = com_backend + # Pretend AutoCAD already holds 15 KB under APP_A (and ACAD overrides). + entity.stored = ( + [1001, 1070, 1001] + [1000] * 60, + ["ACAD", 271, "APP_A"] + ["x" * 255] * 60, + ) + with pytest.raises(ValueError, match=r"other applications.*16 KB per entity"): + await backend.entity_set_xdata("2F", "APP_B", ["y" * 255] * 5) + # Refused before RegisteredApplications.Add and SetXData. + assert registered == [] and entity.stored[1][2] == "APP_A" and len(entity.stored[0]) == 63 + # Rewriting APP_A itself is judged without its old payload. + result = await backend.entity_set_xdata("2F", "APP_A", ["z" * 255] * 60) + assert result["ok"] is True and registered == ["APP_A"] + # Removal always goes through. + result = await backend.entity_set_xdata("2F", "APP_B", []) + assert result["removed"] is True and result["bytes"] == 0 + + async def test_com_newline_is_refused_before_any_activex_call(com_backend): backend, entity, registered = com_backend with pytest.raises(ValueError, match="line separators are not allowed"):