Umut CANandClaude Opus 5
dcc8f07bf4
Layer states are ACADMCP_LAYERSTATES XRECORD chunks (<= 255 ASCII chars each), in the file and not in AutoCAD's Layer States Manager. Headless view restore writes the *Active VPORT (ezdxf's header has no $VIEWCTR; measured). UCS is stored and made current; tool coordinates stay WCS. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd
445 lines
16 KiB
Python
445 lines
16 KiB
Python
"""Portable layer states: save → change → restore equals the snapshot.
|
|
|
|
These are the server's own layer states — JSON chunks in an XRECORD under
|
|
``ACADMCP_LAYERSTATES`` in the drawing's named object dictionary. They live
|
|
in the file and travel with it; they do NOT appear in AutoCAD's Layer States
|
|
Manager. The codec is engine-neutral, so the ezdxf tests prove the format and
|
|
the COM tests prove the ActiveX call shapes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import types
|
|
|
|
import pytest
|
|
|
|
from backends.base import LayerInfo
|
|
from engineering.environment.layer_states import (
|
|
DICT_NAME,
|
|
PROPERTIES,
|
|
decode_state,
|
|
diff_snapshot,
|
|
encode_state,
|
|
snapshot_from_layers,
|
|
validate_properties,
|
|
)
|
|
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
|
|
def _layer(name, **overrides) -> LayerInfo:
|
|
base = dict(
|
|
name=name,
|
|
color=7,
|
|
linetype="Continuous",
|
|
lineweight=-3,
|
|
is_on=True,
|
|
is_frozen=False,
|
|
is_locked=False,
|
|
is_current=False,
|
|
)
|
|
base.update(overrides)
|
|
return LayerInfo(**base)
|
|
|
|
|
|
# ── codec ───────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_snapshot_records_every_property_and_the_current_layer():
|
|
state = snapshot_from_layers(
|
|
[_layer("0"), _layer("HIDDEN", color=8, linetype="HIDDEN", lineweight=35, is_frozen=True)],
|
|
"HIDDEN",
|
|
plot={"HIDDEN": False},
|
|
description="before plotting",
|
|
)
|
|
assert state == {
|
|
"version": 1,
|
|
"description": "before plotting",
|
|
"current_layer": "HIDDEN",
|
|
"layers": {
|
|
"0": {
|
|
"on": True,
|
|
"frozen": False,
|
|
"locked": False,
|
|
"color": 7,
|
|
"linetype": "Continuous",
|
|
"lineweight": -3,
|
|
"plot": True,
|
|
},
|
|
"HIDDEN": {
|
|
"on": True,
|
|
"frozen": True,
|
|
"locked": False,
|
|
"color": 8,
|
|
"linetype": "HIDDEN",
|
|
"lineweight": 35,
|
|
"plot": False,
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def test_encode_chunks_are_at_most_255_ascii_characters_and_round_trip():
|
|
layers = [_layer(f"LAYER-{i:03d}-{'x' * 40}") for i in range(60)]
|
|
state = snapshot_from_layers(layers, "0")
|
|
chunks = encode_state(state)
|
|
assert len(chunks) > 5
|
|
assert all(len(c) <= 255 and c.isascii() for c in chunks)
|
|
assert decode_state(chunks) == state
|
|
|
|
|
|
def test_unicode_layer_names_survive_the_ascii_chunking():
|
|
state = snapshot_from_layers([_layer("KÖRPER-Ø20")], "KÖRPER-Ø20")
|
|
chunks = encode_state(state)
|
|
assert all(c.isascii() for c in chunks), "escaped so a chunk boundary never splits a code point"
|
|
assert decode_state(chunks)["current_layer"] == "KÖRPER-Ø20"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"chunks, fragment",
|
|
[
|
|
(["not json"], "not valid JSON"),
|
|
([json.dumps({"version": 2, "layers": {}})], "version"),
|
|
([json.dumps({"version": 1, "layers": {"A": {"on": True}}})], "lacks"),
|
|
([json.dumps([1, 2])], "version"),
|
|
],
|
|
)
|
|
def test_decode_refuses_anything_that_is_not_a_state(chunks, fragment):
|
|
with pytest.raises(ValueError, match=fragment):
|
|
decode_state(chunks)
|
|
|
|
|
|
def test_diff_reports_missing_and_new_layers_sorted():
|
|
state = snapshot_from_layers([_layer("A"), _layer("B"), _layer("C")], "0")
|
|
missing, new = diff_snapshot(state, [_layer("C"), _layer("Z"), _layer("A"), _layer("Y")])
|
|
assert missing == ["B"] and new == ["Y", "Z"]
|
|
|
|
|
|
def test_validate_properties_defaults_to_all_and_names_the_bad_one():
|
|
assert validate_properties(None) == PROPERTIES
|
|
assert validate_properties(["color", "on", "color"]) == ("color", "on")
|
|
with pytest.raises(ValueError, match=r"properties\[1\]: 'colour'"):
|
|
validate_properties(["on", "colour"])
|
|
with pytest.raises(TypeError, match="list"):
|
|
validate_properties("on")
|
|
with pytest.raises(ValueError, match="at least one"):
|
|
validate_properties([])
|
|
|
|
|
|
# ── headless engine ─────────────────────────────────────────────────────────
|
|
|
|
|
|
async def _seed(backend):
|
|
await backend.layer_create("GEOMETRY", color=7, linetype="Continuous", lineweight=50)
|
|
await backend.layer_create("HIDDEN", color=8, linetype="HIDDEN", lineweight=35)
|
|
await backend.layer_create("CENTER", color=1, linetype="CENTER", lineweight=25)
|
|
await backend.layer_set_current("GEOMETRY")
|
|
await backend.layer_freeze("CENTER")
|
|
await backend.layer_hide("HIDDEN")
|
|
|
|
|
|
async def _table(backend) -> dict:
|
|
return {
|
|
lyr.name: (
|
|
lyr.color,
|
|
lyr.linetype,
|
|
lyr.lineweight,
|
|
lyr.is_on,
|
|
lyr.is_frozen,
|
|
lyr.is_locked,
|
|
lyr.is_current,
|
|
)
|
|
for lyr in await backend.layer_list()
|
|
}
|
|
|
|
|
|
async def test_save_change_restore_equals_the_snapshot(backend):
|
|
await _seed(backend)
|
|
before = await _table(backend)
|
|
saved = await backend.layer_state_save("PLOT", description="plot set")
|
|
# measured: ezdxf.new() already has "0" and "Defpoints", so the seeded
|
|
# table is five layers and serialises to three <= 255-character chunks
|
|
assert saved == {
|
|
"ok": True,
|
|
"name": "PLOT",
|
|
"layer_count": 5,
|
|
"replaced": False,
|
|
"chunks": 3,
|
|
"backend": "ezdxf",
|
|
}
|
|
|
|
await backend.layer_thaw("CENTER")
|
|
await backend.layer_show("HIDDEN")
|
|
await backend.layer_lock("GEOMETRY")
|
|
await backend.layer_modify("HIDDEN", color=3, linetype="Continuous", lineweight=13)
|
|
await backend.layer_set_current("0")
|
|
assert await _table(backend) != before
|
|
|
|
restored = await backend.layer_state_restore("PLOT")
|
|
assert restored["applied"] == {
|
|
"layers": 5,
|
|
"properties": list(PROPERTIES),
|
|
"current_layer": "GEOMETRY",
|
|
}
|
|
assert restored["missing_layers"] == [] and restored["new_layers"] == []
|
|
assert await _table(backend) == before
|
|
|
|
|
|
async def test_restore_applies_only_the_requested_properties(backend):
|
|
await _seed(backend)
|
|
await backend.layer_state_save("S")
|
|
await backend.layer_thaw("CENTER")
|
|
await backend.layer_modify("CENTER", color=5)
|
|
result = await backend.layer_state_restore("S", properties=["color"])
|
|
assert result["applied"]["properties"] == ["color"]
|
|
center = next(lyr for lyr in await backend.layer_list() if lyr.name == "CENTER")
|
|
assert center.color == 1 and center.is_frozen is False, "frozen was not requested"
|
|
|
|
|
|
async def test_restore_reports_missing_and_new_layers_and_never_creates(backend):
|
|
await _seed(backend)
|
|
await backend.layer_state_save("S")
|
|
await backend.layer_delete("CENTER")
|
|
await backend.layer_create("NEW-ONE")
|
|
result = await backend.layer_state_restore("S")
|
|
assert result["missing_layers"] == ["CENTER"] and result["new_layers"] == ["NEW-ONE"]
|
|
assert "CENTER" not in [lyr.name for lyr in await backend.layer_list()]
|
|
|
|
|
|
async def test_plot_flag_and_current_layer_round_trip(backend):
|
|
await _seed(backend)
|
|
backend._doc.layers.get("HIDDEN").dxf.plot = 0
|
|
await backend.layer_state_save("S")
|
|
backend._doc.layers.get("HIDDEN").dxf.plot = 1
|
|
await backend.layer_set_current("0")
|
|
await backend.layer_state_restore("S", properties=["plot", "current"])
|
|
assert backend._doc.layers.get("HIDDEN").dxf.plot == 0
|
|
assert backend._current_layer == "GEOMETRY"
|
|
assert backend._doc.header["$CLAYER"] == "GEOMETRY"
|
|
|
|
|
|
async def test_state_survives_save_and_reopen(backend, tmp_path):
|
|
await _seed(backend)
|
|
await backend.layer_state_save("PLOT", description="d")
|
|
path = tmp_path / "states.dxf"
|
|
await backend.drawing_save(str(path))
|
|
await backend.document_close(None)
|
|
await backend.drawing_open(str(path))
|
|
assert await backend.layer_state_list() == [
|
|
{"name": "PLOT", "description": "d", "layer_count": 5}
|
|
]
|
|
await backend.layer_thaw("CENTER")
|
|
await backend.layer_state_restore("PLOT")
|
|
center = next(lyr for lyr in await backend.layer_list() if lyr.name == "CENTER")
|
|
assert center.is_frozen is True
|
|
|
|
|
|
async def test_xrecord_chunks_stay_under_256_characters(backend):
|
|
for i in range(80):
|
|
await backend.layer_create(f"LONG-LAYER-NAME-{i:03d}-{'x' * 30}")
|
|
saved = await backend.layer_state_save("BIG")
|
|
assert saved["chunks"] > 10
|
|
xrecord = backend._doc.rootdict.get(DICT_NAME).get("BIG")
|
|
tags = [t for t in xrecord.tags if t.code == 1000]
|
|
assert len(tags) == saved["chunks"]
|
|
assert all(len(t.value) <= 255 for t in tags)
|
|
|
|
|
|
async def test_replace_delete_and_missing_names(backend):
|
|
await backend.layer_state_save("S")
|
|
again = await backend.layer_state_save("s")
|
|
assert again["replaced"] is True, "AutoCAD dictionaries are case-insensitive"
|
|
assert [row["name"] for row in await backend.layer_state_list()] == ["s"]
|
|
assert await backend.layer_state_delete("S") == {"ok": True, "deleted": "s", "backend": "ezdxf"}
|
|
assert await backend.layer_state_list() == []
|
|
with pytest.raises(ValueError, match="no layer state named 'S'"):
|
|
await backend.layer_state_restore("S")
|
|
with pytest.raises(ValueError, match="no layer state named 'S'"):
|
|
await backend.layer_state_delete("S")
|
|
|
|
|
|
async def test_refusals_happen_before_any_write(backend):
|
|
with pytest.raises(ValueError, match="layer state name"):
|
|
await backend.layer_state_save(" ")
|
|
with pytest.raises(TypeError, match="description"):
|
|
await backend.layer_state_save("S", description=3)
|
|
assert DICT_NAME not in backend._doc.rootdict, "a refused save creates no dictionary"
|
|
await backend.layer_state_save("S")
|
|
with pytest.raises(ValueError, match=r"properties\[0\]: 'plottable'"):
|
|
await backend.layer_state_restore("S", properties=["plottable"])
|
|
assert len(list(backend._doc.rootdict.get(DICT_NAME).keys())) == 1
|
|
|
|
|
|
# ── live engine, against a fake ActiveX surface ──────────────────────────────
|
|
|
|
|
|
class _FakeXRecord:
|
|
def __init__(self, name):
|
|
self.name = name
|
|
self.types = None
|
|
self.values = None
|
|
self.deleted = False
|
|
|
|
def SetXRecordData(self, types_, values):
|
|
self.types, self.values = list(types_), list(values)
|
|
|
|
def GetXRecordData(self):
|
|
return self.types, self.values
|
|
|
|
def Delete(self):
|
|
self.deleted = True
|
|
|
|
|
|
class _FakeDict:
|
|
def __init__(self, name):
|
|
self.Name = name
|
|
self.entries: list[_FakeXRecord] = []
|
|
self.calls: list[tuple[str, tuple]] = []
|
|
|
|
@property
|
|
def Count(self):
|
|
return len(self.entries)
|
|
|
|
def Item(self, index):
|
|
return self.entries[index]
|
|
|
|
def GetName(self, obj):
|
|
return obj.name
|
|
|
|
def AddXRecord(self, keyword):
|
|
self.calls.append(("AddXRecord", (keyword,)))
|
|
rec = _FakeXRecord(keyword)
|
|
self.entries.append(rec)
|
|
return rec
|
|
|
|
def Remove(self, name):
|
|
self.calls.append(("Remove", (name,)))
|
|
rec = next(e for e in self.entries if e.name == name)
|
|
self.entries.remove(rec)
|
|
return rec
|
|
|
|
|
|
class _FakeDictionaries:
|
|
def __init__(self):
|
|
self.dicts: dict[str, _FakeDict] = {}
|
|
self.calls: list[tuple[str, tuple]] = []
|
|
|
|
def Item(self, name):
|
|
self.calls.append(("Item", (name,)))
|
|
if name not in self.dicts:
|
|
raise RuntimeError(f"no dictionary {name}")
|
|
return self.dicts[name]
|
|
|
|
def Add(self, name):
|
|
self.calls.append(("Add", (name,)))
|
|
self.dicts[name] = _FakeDict(name)
|
|
return self.dicts[name]
|
|
|
|
|
|
class _FakeLayer:
|
|
def __init__(self, name, **props):
|
|
self.Name = name
|
|
self.Color = props.get("color", 7)
|
|
self.Linetype = props.get("linetype", "Continuous")
|
|
self.LineWeight = props.get("lineweight", -3)
|
|
self.LayerOn = props.get("on", True)
|
|
self.Freeze = props.get("frozen", False)
|
|
self.Lock = props.get("locked", False)
|
|
self.Plottable = props.get("plot", True)
|
|
|
|
|
|
class _FakeLayers:
|
|
def __init__(self, layers):
|
|
self._layers = layers
|
|
|
|
@property
|
|
def Count(self):
|
|
return len(self._layers)
|
|
|
|
def Item(self, key):
|
|
if isinstance(key, int):
|
|
return self._layers[key]
|
|
for lyr in self._layers:
|
|
if lyr.Name == key:
|
|
return lyr
|
|
raise RuntimeError(f"no layer {key}")
|
|
|
|
|
|
@pytest.fixture
|
|
def com_backend(monkeypatch):
|
|
pytest.importorskip("win32com.client", reason="pywin32 not installed")
|
|
from backends import com_backend as module
|
|
|
|
layers = [
|
|
_FakeLayer("0"),
|
|
_FakeLayer("GEOMETRY", lineweight=50),
|
|
_FakeLayer("HIDDEN", color=8, linetype="HIDDEN", frozen=True, plot=False),
|
|
]
|
|
document = types.SimpleNamespace(
|
|
Dictionaries=_FakeDictionaries(),
|
|
Layers=_FakeLayers(layers),
|
|
ActiveLayer=layers[1],
|
|
)
|
|
monkeypatch.setattr(module, "_acad_doc", lambda: document)
|
|
monkeypatch.setattr(module, "_ensure_linetype_loaded", lambda name: None)
|
|
monkeypatch.setattr(module, "_regen", lambda: None)
|
|
monkeypatch.setattr(module, "_ai", lambda values: list(values))
|
|
monkeypatch.setattr(module, "_avar", lambda values: list(values))
|
|
backend = module.ComBackend()
|
|
|
|
async def _run_inline(func, *args, **kwargs):
|
|
return func(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(backend, "_run", _run_inline)
|
|
return backend, document
|
|
|
|
|
|
async def test_com_save_creates_the_dictionary_and_one_xrecord_of_1000_chunks(com_backend):
|
|
backend, document = com_backend
|
|
result = await backend.layer_state_save("PLOT", description="d")
|
|
assert result["ok"] and result["backend"] == "com" and result["replaced"] is False
|
|
assert document.Dictionaries.calls == [("Item", (DICT_NAME,)), ("Add", (DICT_NAME,))]
|
|
states = document.Dictionaries.dicts[DICT_NAME]
|
|
assert states.calls == [("AddXRecord", ("PLOT",))]
|
|
(rec,) = states.entries
|
|
assert rec.types == [1000] * len(rec.values)
|
|
assert all(len(v) <= 255 for v in rec.values)
|
|
state = decode_state(rec.values)
|
|
assert state["current_layer"] == "GEOMETRY"
|
|
assert state["layers"]["HIDDEN"]["plot"] is False
|
|
again = await backend.layer_state_save("plot")
|
|
assert again["replaced"] is True and again["name"] == "PLOT"
|
|
assert len(states.entries) == 1, "SetXRecordData on the existing record, no second AddXRecord"
|
|
|
|
|
|
async def test_com_restore_reads_the_xrecord_and_writes_layer_properties(com_backend):
|
|
backend, document = com_backend
|
|
await backend.layer_state_save("S")
|
|
hidden = document.Layers.Item("HIDDEN")
|
|
hidden.Freeze = False
|
|
hidden.Color = 3
|
|
hidden.Plottable = True
|
|
document.ActiveLayer = document.Layers.Item("0")
|
|
result = await backend.layer_state_restore("S")
|
|
assert result["applied"]["layers"] == 3
|
|
assert result["applied"]["current_layer"] == "GEOMETRY"
|
|
assert hidden.Freeze is True and hidden.Color == 8 and hidden.Plottable is False
|
|
assert document.ActiveLayer.Name == "GEOMETRY"
|
|
assert result["missing_layers"] == [] and result["new_layers"] == []
|
|
|
|
|
|
async def test_com_list_and_delete(com_backend):
|
|
backend, document = com_backend
|
|
await backend.layer_state_save("A", description="one")
|
|
await backend.layer_state_save("B")
|
|
assert await backend.layer_state_list() == [
|
|
{"name": "A", "description": "one", "layer_count": 3},
|
|
{"name": "B", "description": None, "layer_count": 3},
|
|
]
|
|
assert (await backend.layer_state_delete("a"))["deleted"] == "A"
|
|
states = document.Dictionaries.dicts[DICT_NAME]
|
|
assert states.calls[-1] == ("Remove", ("A",))
|
|
with pytest.raises(ValueError, match="no layer state named 'A'"):
|
|
await backend.layer_state_delete("A")
|