Umut CANandClaude Opus 5
f4c2f27bee
encode_values() let strings and nan/inf through float() into 1010 point
coordinates and never finiteness-checked bare 1040 reals, so
entity_set_xdata(handle, app, [["nan", 1]]) or [float("inf")] wrote a
literal "nan"/"inf" into the DXF (and every undo snapshot) and read back
as null. Numbers now go through the same rule block_specs._scalar applies:
int/float only, bool refused, math.isfinite, error naming values[i][j].
The advertised 16 KB limit was per call per app; AutoCAD's is per object
across every application (xdroom/xdsize), so three 15 KB apps on one
entity passed headless and would fail SetXData live. check_entity_budget()
now sums the entity's existing XDATA under every other app (ACAD's DSTYLE
overrides included) plus each 1001 marker with AutoCAD's resbuf sizes,
and both engines run it before registering the APPID or writing; removal
is never refused. The result's "bytes" is that same accounting.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd
357 lines
15 KiB
Python
357 lines
15 KiB
Python
"""XDATA round trips, typing by group code, limits, removal — both engines."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import types
|
||
|
||
import pytest
|
||
|
||
from backends.xdata_specs import (
|
||
MAX_BYTES,
|
||
MAX_STRING,
|
||
app_size,
|
||
check_entity_budget,
|
||
decode_values,
|
||
encode_values,
|
||
split_by_app,
|
||
validate_app_name,
|
||
)
|
||
|
||
pytestmark = pytest.mark.asyncio
|
||
|
||
|
||
# ── pure encoding ────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_values_are_typed_by_group_code():
|
||
assert encode_values(["json", 7, 2.5, [1, 2], [1, 2, 3]]) == [
|
||
(1000, "json"),
|
||
(1071, 7),
|
||
(1040, 2.5),
|
||
(1010, (1.0, 2.0, 0.0)),
|
||
(1010, (1.0, 2.0, 3.0)),
|
||
]
|
||
|
||
|
||
def test_decode_is_the_inverse():
|
||
tags = encode_values(["a", 1, 1.5, [3, 4]])
|
||
assert decode_values(tags) == ["a", 1, 1.5, [3.0, 4.0, 0.0]]
|
||
|
||
|
||
@pytest.mark.parametrize("bad", [True, None, {"a": 1}, [1], [1, 2, 3, 4], ["x", "y"], 2**31])
|
||
def test_untypeable_values_are_refused(bad):
|
||
with pytest.raises((TypeError, ValueError)):
|
||
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"):
|
||
encode_values(["x" * (MAX_STRING + 1)])
|
||
|
||
|
||
@pytest.mark.parametrize("sep", ["\n", "\r", "\r\n"])
|
||
def test_line_separators_in_strings_are_refused(sep):
|
||
# DXF is "code\nvalue\n" lines and ezdxf writes 1000 tags verbatim, so a
|
||
# separator inside the value splits the tag pair and corrupts the file.
|
||
with pytest.raises(ValueError, match=r"values\[1\].*line separators are not allowed"):
|
||
encode_values(["fine", f"Inspect weld{sep}before paint"])
|
||
|
||
|
||
def test_other_control_characters_are_still_allowed():
|
||
# Only LF/CR split a DXF tag; tabs and the exotic separators round-trip.
|
||
assert encode_values(["a\tb", "a\x0bb", "a
b"]) == [
|
||
(1000, "a\tb"),
|
||
(1000, "a\x0bb"),
|
||
(1000, "a
b"),
|
||
]
|
||
|
||
|
||
def test_total_size_over_16k_is_refused():
|
||
with pytest.raises(ValueError, match="16"):
|
||
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):
|
||
validate_app_name(bad)
|
||
|
||
|
||
def test_split_by_app_groups_a_flat_activex_stream():
|
||
codes = [1001, 1000, 1071, 1001, 1040]
|
||
values = ["APP_A", "hello", 3, "APP_B", 1.5]
|
||
assert split_by_app(codes, values) == {"APP_A": ["hello", 3], "APP_B": [1.5]}
|
||
|
||
|
||
# ── headless engine ─────────────────────────────────────────────────────────
|
||
|
||
|
||
async def test_roundtrip_registers_the_appid(backend):
|
||
line = await backend.entity_create_line(0, 0, 1, 1)
|
||
result = await backend.entity_set_xdata(line.handle, "ACADMCP_PID", ["json", 42, 1.5, [1, 2]])
|
||
assert result["ok"] is True and result["value_count"] == 4 and result["removed"] is False
|
||
assert "ACADMCP_PID" in backend._doc.appids
|
||
got = await backend.entity_get_xdata(line.handle, "ACADMCP_PID")
|
||
assert got["xdata"] == {"ACADMCP_PID": ["json", 42, 1.5, [1.0, 2.0, 0.0]]}
|
||
|
||
|
||
async def test_get_without_app_returns_every_app(backend):
|
||
line = await backend.entity_create_line(0, 0, 1, 1)
|
||
await backend.entity_set_xdata(line.handle, "APP_A", ["a"])
|
||
await backend.entity_set_xdata(line.handle, "APP_B", ["b"])
|
||
got = await backend.entity_get_xdata(line.handle)
|
||
assert got["xdata"] == {"APP_A": ["a"], "APP_B": ["b"]}
|
||
|
||
|
||
async def test_unknown_app_is_empty_not_an_error(backend):
|
||
line = await backend.entity_create_line(0, 0, 1, 1)
|
||
got = await backend.entity_get_xdata(line.handle, "NOPE")
|
||
assert got["xdata"] == {}
|
||
|
||
|
||
async def test_set_replaces_and_empty_removes(backend):
|
||
line = await backend.entity_create_line(0, 0, 1, 1)
|
||
await backend.entity_set_xdata(line.handle, "APP_A", ["one", "two"])
|
||
await backend.entity_set_xdata(line.handle, "APP_A", ["three"])
|
||
assert (await backend.entity_get_xdata(line.handle, "APP_A"))["xdata"] == {"APP_A": ["three"]}
|
||
result = await backend.entity_set_xdata(line.handle, "APP_A", [])
|
||
assert result["removed"] is True
|
||
assert (await backend.entity_get_xdata(line.handle))["xdata"] == {}
|
||
|
||
|
||
async def test_limits_are_enforced_before_writing(backend):
|
||
line = await backend.entity_create_line(0, 0, 1, 1)
|
||
with pytest.raises(ValueError, match="255"):
|
||
await backend.entity_set_xdata(line.handle, "APP_A", ["x" * 256])
|
||
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.
|
||
line = await backend.entity_create_line(0, 0, 1, 1)
|
||
with pytest.raises(ValueError, match="line separators are not allowed"):
|
||
await backend.entity_set_xdata(line.handle, "NOTES", ["Inspect weld\nbefore paint"])
|
||
assert (await backend.entity_get_xdata(line.handle))["xdata"] == {}
|
||
|
||
await backend.entity_set_xdata(line.handle, "NOTES", ["Inspect weld before paint"])
|
||
await backend.transaction_begin()
|
||
circle = await backend.entity_create_circle(5, 5, 1)
|
||
await backend.transaction_rollback()
|
||
assert (await backend.entity_get_xdata(line.handle, "NOTES"))["xdata"] == {
|
||
"NOTES": ["Inspect weld before paint"]
|
||
}
|
||
with pytest.raises(RuntimeError, match="not found"):
|
||
await backend.entity_get(circle.handle)
|
||
|
||
path = str(tmp_path / "notes.dxf")
|
||
await backend.drawing_save(path)
|
||
await backend.drawing_open(path)
|
||
assert (await backend.entity_get_xdata(line.handle, "NOTES"))["xdata"] == {
|
||
"NOTES": ["Inspect weld before paint"]
|
||
}
|
||
|
||
|
||
# ── live engine, fake ActiveX ────────────────────────────────────────────────
|
||
|
||
|
||
class _FakeEntity:
|
||
def __init__(self):
|
||
self.stored: tuple | None = None
|
||
|
||
def SetXData(self, types_, values):
|
||
self.stored = (list(types_.value), list(values.value))
|
||
|
||
def GetXData(self, app):
|
||
if self.stored is None:
|
||
return ([], [])
|
||
codes, values = self.stored
|
||
if app:
|
||
return (codes, values)
|
||
return (codes, values)
|
||
|
||
|
||
@pytest.fixture
|
||
def com_backend(monkeypatch):
|
||
pytest.importorskip("win32com.client", reason="pywin32 not installed")
|
||
from backends import com_backend as module
|
||
|
||
entity = _FakeEntity()
|
||
registered: list[str] = []
|
||
document = types.SimpleNamespace(
|
||
HandleToObject=lambda handle: entity,
|
||
RegisteredApplications=types.SimpleNamespace(Add=registered.append),
|
||
)
|
||
monkeypatch.setattr(module, "_acad_doc", lambda: document)
|
||
backend = module.ComBackend()
|
||
|
||
async def _run_inline(func, *args, **kwargs):
|
||
return func(*args, **kwargs)
|
||
|
||
monkeypatch.setattr(backend, "_run", _run_inline)
|
||
return backend, entity, registered
|
||
|
||
|
||
async def test_com_set_writes_the_1001_marker_first(com_backend):
|
||
backend, entity, registered = com_backend
|
||
await backend.entity_set_xdata("2F", "ACADMCP_PID", ["json", 3, [1, 2]])
|
||
assert registered == ["ACADMCP_PID"]
|
||
codes, values = entity.stored
|
||
assert codes == [1001, 1000, 1071, 1010]
|
||
assert values[0] == "ACADMCP_PID" and values[1] == "json" and values[2] == 3
|
||
assert tuple(values[3].value) == (1.0, 2.0, 0.0)
|
||
|
||
|
||
async def test_com_get_splits_by_app(com_backend):
|
||
backend, entity, _ = com_backend
|
||
await backend.entity_set_xdata("2F", "ACADMCP_PID", ["json"])
|
||
got = await backend.entity_get_xdata("2F", "ACADMCP_PID")
|
||
assert got["xdata"] == {"ACADMCP_PID": ["json"]}
|
||
|
||
|
||
async def test_com_empty_values_write_only_the_marker(com_backend):
|
||
backend, entity, _ = com_backend
|
||
result = await backend.entity_set_xdata("2F", "ACADMCP_PID", [])
|
||
assert result["removed"] is True
|
||
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"):
|
||
await backend.entity_set_xdata("2F", "NOTES", ["a\r\nb"])
|
||
assert entity.stored is None and registered == []
|