Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 31 additions & 0 deletions src/specify_cli/workflows/overlays/_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -207,6 +207,37 @@ def workflow_overlay_add(
target_path = _ensure_contained_path(
target_dir / f"{overlay.id}.yml", _overlay_root(project_root)
)
# Overlay identity is the manifest ``id``, not the filename (see
# ``_find_overlay_file``), so ``<id>.yml`` can legitimately already hold
# a DIFFERENT overlay. Committing onto it would destroy that overlay
# permanently -- the commit renames the victim to a ``.bak`` and the
# success path then discards that backup -- while reporting success.
if target_path.is_file():
# Fail closed: refuse unless the occupant is provably this same
# overlay. Reaching here means ``_find_overlay_file`` did not match
# this path, and it skips exactly the files whose identity cannot be
# established -- unreadable, malformed, non-mapping, or missing a
# usable ``id``. Letting those through would destroy the user's file
# just as permanently as overwriting a valid one, only without even
# being able to name what was lost.
occupant, read_errors = _read_overlay(target_path)
occupant_id = occupant.get("id") if isinstance(occupant, dict) else None
if not (isinstance(occupant_id, str) and occupant_id == overlay.id):
if isinstance(occupant_id, str) and occupant_id:
detail = f"already holds overlay {_escape_markup(repr(occupant_id))}"
elif read_errors:
detail = (
"could not be parsed as an overlay "
f"({_escape_markup('; '.join(read_errors))})"
)
else:
detail = "is not a readable overlay manifest (no usable 'id')"
err_console.print(
f"[red]Error:[/red] {_escape_markup(str(target_path))} {detail}. "
f"Rename or remove it before adding overlay "
f"{_escape_markup(repr(overlay.id))}."
)
return None

backup: Path | None = None
try:
Expand Down
124 changes: 124 additions & 0 deletions tests/workflows/test_overlay_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -893,3 +893,127 @@ def test_duplicate_manifest_id_is_rejected(self, project_dir, monkeypatch):

with pytest.raises(typer.Exit):
_find_overlay_file(project_dir, "wf", "lint")


class TestOverlayAddDoesNotClobber:
"""`overlay add` must not destroy a different overlay sitting at <id>.yml.

Overlay identity is the manifest `id`, not the filename (see
`_find_overlay_file`), so `lint.yml` can legitimately contain
`id: format`. When `_find_overlay_file` found no file carrying the new
overlay's id, the fallback target was derived purely from the filename and
committed onto unconditionally — permanently destroying the occupant, since
the commit renames it to a `.bak` and the success path then discards that
backup. Exit code 0, no warning.
"""

def _setup(self, project_dir: Path, occupant_id: str | None) -> tuple[Path, Path]:
_write_workflow(
project_dir,
"wf",
{
"schema_version": "1.0",
"workflow": {"id": "wf", "name": "WF", "version": "1.0.0"},
"steps": [{"id": "a", "type": "command", "command": "echo"}],
},
)
ov_dir = project_dir / ".specify" / "workflows" / "overlays" / "wf"
ov_dir.mkdir(parents=True, exist_ok=True)
if occupant_id is not None:
(ov_dir / "lint.yml").write_text(
yaml.safe_dump(
{
"id": occupant_id,
"extends": "wf",
"priority": 3,
"edits": [{"remove": "a"}],
}
),
encoding="utf-8",
)
incoming = project_dir / "incoming.yml"
incoming.write_text(
yaml.safe_dump(
{
"id": "lint",
"extends": "wf",
"priority": 10,
"edits": [{"remove": "a"}],
}
),
encoding="utf-8",
)
return ov_dir, incoming

def test_add_does_not_clobber_a_different_overlay(self, project_dir, monkeypatch):
monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir)
ov_dir, incoming = self._setup(project_dir, occupant_id="format")

result = runner.invoke(app, ["workflow", "overlay", "add", str(incoming)])

assert result.exit_code == 1, result.output
# The victim must be untouched, and no backup left lying around.
survivor = yaml.safe_load((ov_dir / "lint.yml").read_text(encoding="utf-8"))
assert survivor["id"] == "format", survivor
assert survivor["priority"] == 3, survivor
assert [p.name for p in ov_dir.iterdir() if "bak" in p.name] == []

def test_add_still_updates_the_same_overlay_in_place(
self, project_dir, monkeypatch
):
"""The guard must only fire for a *different* overlay id."""
monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir)
ov_dir, incoming = self._setup(project_dir, occupant_id="lint")

result = runner.invoke(app, ["workflow", "overlay", "add", str(incoming)])

assert result.exit_code == 0, result.output
updated = yaml.safe_load((ov_dir / "lint.yml").read_text(encoding="utf-8"))
assert updated["id"] == "lint"
assert updated["priority"] == 10

@pytest.mark.parametrize(
"raw",
[
"id: [1, 2\n bad: yaml:\n",
"- just\n- a\n- sequence\n",
"just a scalar\n",
"extends: wf\npriority: 3\n",
"id: 5\nextends: wf\npriority: 3\n",
],
ids=["malformed", "sequence", "scalar", "missing_id", "non_string_id"],
)
def test_add_fails_closed_when_the_occupant_cannot_be_identified(
self, project_dir, monkeypatch, raw
):
"""An unidentifiable occupant must be refused, not silently destroyed.

`_find_overlay_file` matches on the manifest `id` and skips exactly the
files whose identity cannot be established — unreadable, malformed,
non-mapping, or missing a usable `id`. Those therefore fall through to
the filename-derived target, so a guard that only refuses a *different
valid* id would still let `_commit_workflow_file` discard the user's
file. It is destroyed just as permanently as a valid overlay, only
without even being able to name what was lost.
"""
monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir)
ov_dir, incoming = self._setup(project_dir, occupant_id=None)
occupant = ov_dir / "lint.yml"
occupant.write_text(raw, encoding="utf-8")

result = runner.invoke(app, ["workflow", "overlay", "add", str(incoming)])

assert result.exit_code == 1, result.output
# Byte-for-byte survival, and no backup left behind.
assert occupant.read_text(encoding="utf-8") == raw
assert [p.name for p in ov_dir.iterdir() if "bak" in p.name] == []

def test_add_creates_the_file_when_absent(self, project_dir, monkeypatch):
monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir)
ov_dir, incoming = self._setup(project_dir, occupant_id=None)

result = runner.invoke(app, ["workflow", "overlay", "add", str(incoming)])

assert result.exit_code == 0, result.output
created = yaml.safe_load((ov_dir / "lint.yml").read_text(encoding="utf-8"))
assert created["id"] == "lint"