fix(task-21): registry identity is normcase(abspath) everywhere — a case variant reloads instead of registering twice, resolves instead of being refused, and drawing_close(save) checks the file it writes
aeab950 claimed one registry entry per file but only the two save paths
compared with normcase. _register_document keyed by the plain
os.path.abspath string, so on Windows drawing_open of an already-open
file spelled with another case registered a second entry (no `reloaded`,
open_documents: 2); from there every document_activate/close of either
entry was refused as ambiguous, and drawing_close(save=True) — which
bypasses the resolver — wrote the second entry's bytes to the shared
file while the surviving row reported saved: True. The resolver's
by_path route had the same plain-string comparison, so over the wire a
lowercase drive letter was refused 'no open document named ...' while
ComBackend._com_find_document lowercases and accepts it.
One helper, _file_identity(path) = normcase(abspath(path)), now decides
every registry comparison: _register_document consults _path_holder so
another spelling of an open file reloads it; the _doc_path setter re-keys
only when the file changes, so a case-variant save keeps the key a caller
holds; _resolve_document_key's by_path compares identities;
_close_document_sync refuses at the write when another entry holds the
file; _clear_quarantine's same-path warning uses the same rule. Keys stay
spelled as the caller spelled them — the identity is for comparing, not
for showing.
Five tests pin it (four skip on a case-sensitive filesystem, where two
spellings of one file cannot exist), including the wire route through
fastmcp.Client that the review measured.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTdThis commit is contained in:
Umut CAN
co-authored by
Claude Opus 5
Claude Opus 5
parent
aeab950d2d
commit
2805697d16
+53
-24
@@ -1103,26 +1103,48 @@ class EzdxfBackend(AutoCADBackend):
|
||||
)
|
||||
state = self._state
|
||||
state.path = value
|
||||
if new_key != self._active_key:
|
||||
# Re-key only when the *file* changes: another spelling of the file
|
||||
# the entry already holds (a case variant on Windows) keeps the key a
|
||||
# caller may be holding.
|
||||
if not self._same_file(new_key, self._active_key):
|
||||
del self._docs[self._active_key]
|
||||
state.key = new_key
|
||||
self._docs[new_key] = state
|
||||
self._active_key = new_key
|
||||
|
||||
def _path_holder(self, abs_path: str, *, exclude: str | None = None) -> str | None:
|
||||
"""Key of the entry (other than ``exclude``) that holds ``abs_path``, or None.
|
||||
@staticmethod
|
||||
def _file_identity(path: str) -> str:
|
||||
"""The string two spellings of one file share: ``normcase(abspath(path))``.
|
||||
|
||||
Compared with ``os.path.normcase`` so that on Windows two spellings of
|
||||
one file count as the same file; on POSIX that is plain equality.
|
||||
Every registry comparison goes through this. On Windows it folds case
|
||||
and separators, so ``C:\\Work\\Gear.dxf`` and ``c:/work/gear.dxf`` are one
|
||||
file; on POSIX ``normcase`` is the identity and only ``..`` / ``.`` /
|
||||
duplicate separators are folded. Keys themselves stay spelled as the
|
||||
caller spelled them - the identity is for comparing, not for showing.
|
||||
"""
|
||||
wanted = os.path.normcase(abs_path)
|
||||
try:
|
||||
return os.path.normcase(os.path.abspath(path))
|
||||
except (OSError, ValueError): # a path this OS cannot normalise
|
||||
return os.path.normcase(path)
|
||||
|
||||
@classmethod
|
||||
def _same_file(cls, a: str, b: str) -> bool:
|
||||
return cls._file_identity(a) == cls._file_identity(b)
|
||||
|
||||
def _path_holder(self, path: str, *, exclude: str | None = None) -> str | None:
|
||||
"""Key of the entry (other than ``exclude``) that holds ``path``, or None.
|
||||
|
||||
Compared by ``_file_identity`` so that on Windows two spellings of one
|
||||
file count as the same file; on POSIX that is plain equality.
|
||||
"""
|
||||
wanted = self._file_identity(path)
|
||||
for key, state in self._docs.items():
|
||||
if key == exclude:
|
||||
continue
|
||||
candidates = [key]
|
||||
if state.path:
|
||||
candidates.append(os.path.abspath(state.path))
|
||||
if any(os.path.normcase(c) == wanted for c in candidates):
|
||||
candidates.append(state.path)
|
||||
if any(self._file_identity(c) == wanted for c in candidates):
|
||||
return key
|
||||
return None
|
||||
|
||||
@@ -1132,7 +1154,7 @@ class EzdxfBackend(AutoCADBackend):
|
||||
Runs before ``doc.saveas`` so a refusal leaves the disk, the registry
|
||||
and the active document exactly as they were.
|
||||
"""
|
||||
holder = self._path_holder(os.path.abspath(path), exclude=self._active_key)
|
||||
holder = self._path_holder(path, exclude=self._active_key)
|
||||
if holder is None:
|
||||
return
|
||||
label = self._docs[holder].label
|
||||
@@ -1273,7 +1295,10 @@ class EzdxfBackend(AutoCADBackend):
|
||||
Returns ``(key, reloaded)``. A path that is already open is *reloaded*:
|
||||
its old entry (and its snapshots) is dropped and the fresh read takes
|
||||
its place — the pre-registry behaviour of ``drawing_open``, kept so a
|
||||
save-then-reopen verification still reads the disk.
|
||||
save-then-reopen verification still reads the disk. "Already open" is
|
||||
decided by ``_path_holder`` (file identity), not by the key string:
|
||||
the same file spelled with another case used to register a second
|
||||
entry, after which every lookup of either was refused as ambiguous.
|
||||
"""
|
||||
if path is None:
|
||||
self._untitled_counter += 1
|
||||
@@ -1281,10 +1306,12 @@ class EzdxfBackend(AutoCADBackend):
|
||||
reloaded = False
|
||||
else:
|
||||
key = os.path.abspath(path)
|
||||
previous = self._docs.pop(key, None)
|
||||
reloaded = previous is not None
|
||||
if previous is not None:
|
||||
previous.discard_snapshots()
|
||||
previous_key = self._path_holder(key)
|
||||
reloaded = previous_key is not None
|
||||
if previous_key is not None:
|
||||
self._docs.pop(previous_key).discard_snapshots()
|
||||
if self._active_key == previous_key:
|
||||
self._active_key = None
|
||||
self._docs[key] = _DocState(key=key, doc=doc, path=path)
|
||||
self._activate_key(key)
|
||||
return key, reloaded
|
||||
@@ -1317,14 +1344,14 @@ class EzdxfBackend(AutoCADBackend):
|
||||
# say so rather than hand back whichever entry the key names.
|
||||
self._refuse_shared_path(wanted, self._docs[wanted])
|
||||
return wanted
|
||||
try:
|
||||
as_path = os.path.abspath(wanted)
|
||||
except (OSError, ValueError):
|
||||
as_path = wanted
|
||||
# Paths compare by file identity (``normcase`` on Windows), as the
|
||||
# live engine's ``_com_find_document`` lowercases before comparing: a
|
||||
# lowercase drive letter or a case variant names the same document.
|
||||
as_path = self._file_identity(wanted)
|
||||
by_path = [
|
||||
key
|
||||
for key, state in self._docs.items()
|
||||
if state.path and os.path.abspath(state.path) == as_path
|
||||
if state.path and self._file_identity(state.path) == as_path
|
||||
]
|
||||
if len(by_path) == 1:
|
||||
self._refuse_shared_path(wanted, self._docs[by_path[0]])
|
||||
@@ -1355,7 +1382,7 @@ class EzdxfBackend(AutoCADBackend):
|
||||
if not state.path:
|
||||
return
|
||||
abs_path = os.path.abspath(state.path)
|
||||
other = self._path_holder(abs_path, exclude=state.key)
|
||||
other = self._path_holder(state.path, exclude=state.key)
|
||||
if other is not None:
|
||||
raise ValueError(
|
||||
f"document {wanted!r} is ambiguous: entries {sorted([state.key, other])} "
|
||||
@@ -1407,6 +1434,11 @@ class EzdxfBackend(AutoCADBackend):
|
||||
f"document_close: {label!r} has never been saved, so save=True has "
|
||||
"nowhere to write; drawing_save_as(path) first, or pass discard=True"
|
||||
)
|
||||
# ``drawing_close`` reaches here without the resolver, so the
|
||||
# one-entry-per-file check is made at the write itself: a
|
||||
# save onto a file another entry holds would leave that
|
||||
# entry claiming ``saved: True`` over someone else's bytes.
|
||||
self._refuse_shared_path(label, state)
|
||||
state.doc.saveas(state.path)
|
||||
state.dirty = False
|
||||
saved = True
|
||||
@@ -1554,10 +1586,7 @@ class EzdxfBackend(AutoCADBackend):
|
||||
record.call,
|
||||
)
|
||||
if new_path and record.document_path:
|
||||
try:
|
||||
same = os.path.abspath(new_path) == os.path.abspath(record.document_path)
|
||||
except (OSError, ValueError): # a path this OS cannot normalise
|
||||
same = new_path == record.document_path
|
||||
same = self._same_file(new_path, record.document_path)
|
||||
if same:
|
||||
log.warning(
|
||||
"%s is reopening %s, the same path the abandoned %r call was working "
|
||||
|
||||
@@ -179,6 +179,91 @@ async def test_resolver_refuses_a_shared_path_even_on_an_exact_key_hit(backend,
|
||||
assert len(await backend.document_list()) == 2, "a refusal closes nothing"
|
||||
|
||||
|
||||
# Two spellings of one file only exist where the filesystem folds case;
|
||||
# ``os.path.normcase`` is the identity on POSIX, so these tests skip there.
|
||||
_case_folding_fs = pytest.mark.skipif(
|
||||
os.path.normcase("A") == "A", reason="filesystem is case-sensitive"
|
||||
)
|
||||
|
||||
|
||||
@_case_folding_fs
|
||||
async def test_opening_a_case_variant_of_an_open_file_reloads_it(backend, tmp_path):
|
||||
"""Registration keys by file identity, not by spelling.
|
||||
|
||||
``drawing_open`` of an already-open file spelled with another case used to
|
||||
register a *second* entry (key compared as plain strings), and from there
|
||||
every lookup of either entry was refused as ambiguous.
|
||||
"""
|
||||
path = tmp_path / "Gear.dxf"
|
||||
await backend.drawing_save(str(path))
|
||||
await backend.entity_create_circle(0, 0, 5) # unsaved edit
|
||||
reopened = await backend.drawing_open(str(path).lower())
|
||||
assert reopened["reloaded"] is True
|
||||
assert reopened["open_documents"] == 1
|
||||
rows = await backend.document_list()
|
||||
assert len(rows) == 1 and rows[0]["entity_count"] == 0, "reload reads the disk"
|
||||
assert (await backend.document_activate(str(path)))["ok"] is True
|
||||
|
||||
|
||||
@_case_folding_fs
|
||||
async def test_resolver_accepts_any_spelling_of_a_registered_path(backend, tmp_path):
|
||||
"""``document_activate`` / ``document_close`` compare paths as the OS does.
|
||||
|
||||
The live engine lowercases before comparing (``_com_find_document``); a
|
||||
lowercase drive letter or a case variant used to be refused headless as
|
||||
'no open document named ...' - a dual-engine parity break.
|
||||
"""
|
||||
path = tmp_path / "Gear.dxf"
|
||||
await backend.drawing_save(str(path))
|
||||
key = os.path.abspath(str(path))
|
||||
await backend.drawing_new()
|
||||
spelled = str(path)
|
||||
for variant in (
|
||||
spelled.lower(),
|
||||
spelled[0].lower() + spelled[1:],
|
||||
spelled[0].upper() + spelled[1:],
|
||||
spelled.upper(),
|
||||
):
|
||||
await backend.document_activate("untitled-2")
|
||||
assert (await backend.document_activate(variant))["active"] == key, variant
|
||||
assert len(await backend.document_list()) == 2
|
||||
closed = await backend.document_close(spelled.lower())
|
||||
assert closed["closed"] == key and closed["open_documents"] == 1
|
||||
|
||||
|
||||
@_case_folding_fs
|
||||
async def test_saving_to_a_case_variant_of_the_own_path_keeps_one_entry(backend, tmp_path):
|
||||
path = tmp_path / "Self.dxf"
|
||||
await backend.drawing_save(str(path))
|
||||
key = os.path.abspath(str(path))
|
||||
await backend.entity_create_circle(0, 0, 1)
|
||||
assert (await backend.drawing_save_as(str(path).lower()))["ok"] is True
|
||||
rows = await backend.document_list()
|
||||
assert len(rows) == 1 and rows[0]["saved"] is True
|
||||
assert backend._active_key == key, "the same file keeps its key"
|
||||
assert (await backend.document_activate(key))["ok"] is True
|
||||
|
||||
|
||||
async def test_drawing_close_with_save_refuses_a_shared_path(backend, tmp_path):
|
||||
"""``drawing_close`` bypasses the resolver, so the close itself must check.
|
||||
|
||||
With two entries on one file, ``drawing_close(save=True)`` wrote the active
|
||||
entry's bytes over the file the other entry holds and reported
|
||||
``saved: True`` - the surviving row then claimed the disk held its content.
|
||||
"""
|
||||
path = tmp_path / "dup.dxf"
|
||||
await backend.drawing_save(str(path))
|
||||
await backend.entity_create_line(0, 0, 1, 1) # the owner is dirty
|
||||
await backend.drawing_new()
|
||||
await backend.entity_create_circle(0, 0, 5)
|
||||
backend._docs[backend._active_key].path = str(path) # corrupt the registry behind its back
|
||||
before = path.read_bytes()
|
||||
with pytest.raises(ValueError, match="share the file"):
|
||||
await backend.drawing_close(save=True)
|
||||
assert path.read_bytes() == before, "a refusal writes nothing"
|
||||
assert len(await backend.document_list()) == 2, "a refusal closes nothing"
|
||||
|
||||
|
||||
async def test_activate_by_basename_full_path_or_key(backend, tmp_path):
|
||||
path = tmp_path / "gear.dxf"
|
||||
await backend.drawing_save(str(path))
|
||||
@@ -303,6 +388,30 @@ async def test_document_list_tool_over_the_wire(monkeypatch):
|
||||
assert closed["closed"] == "untitled-1" and closed["active"] == "untitled-2"
|
||||
|
||||
|
||||
@_case_folding_fs
|
||||
async def test_document_activate_accepts_a_lowercase_drive_over_the_wire(monkeypatch, tmp_path):
|
||||
"""The wire canonicalises ``drawing_open`` paths (``validate_path`` resolves
|
||||
them to disk casing) but not ``document_activate`` / ``document_close``
|
||||
names, so a client spelling the drive letter in lowercase used to be
|
||||
refused headless while the live engine accepted it."""
|
||||
monkeypatch.setenv("AUTOCAD_MCP_BACKEND", "ezdxf")
|
||||
path = tmp_path / "Gear.dxf"
|
||||
spelled = str(path)
|
||||
lowered_drive = spelled[0].lower() + spelled[1:]
|
||||
async with Client(server.mcp) as client:
|
||||
await client.call_tool("drawing_new", {"bootstrap": False})
|
||||
await client.call_tool("drawing_save", {"path": spelled})
|
||||
await client.call_tool("drawing_new", {"bootstrap": False})
|
||||
activated = (
|
||||
await client.call_tool("document_activate", {"name_or_path": lowered_drive})
|
||||
).structured_content
|
||||
assert activated["name"] == "Gear.dxf" and activated["previous"] == "untitled-2"
|
||||
closed = (
|
||||
await client.call_tool("document_close", {"name_or_path": spelled.lower()})
|
||||
).structured_content
|
||||
assert closed["closed"] == os.path.abspath(spelled) and closed["open_documents"] == 1
|
||||
|
||||
|
||||
# ── live engine, against a fake ActiveX surface ──────────────────────────────
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user