Skip to content
Merged
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
13 changes: 11 additions & 2 deletions apps/api/services/workbook/column_deps.py
Original file line number Diff line number Diff line change
Expand Up @@ -94,9 +94,14 @@ def referencing_columns(cols: List[dict], column: dict) -> List[dict]:
"""
normalize = lambda value: str(value).strip().lower().replace("_", " ")
aliases = {normalize(column[key]) for key in ("id", "name") if column.get(key)}
stable_ids = {candidate.get("id") for candidate in cols if candidate.get("id")}
return [candidate for candidate in cols
if candidate.get("id") != column.get("id")
and aliases.intersection(normalize(ref) for ref in _refs_in(candidate))]
and any(
(ref == column.get("id") if ref in stable_ids
else normalize(ref) in aliases)
for ref in _refs_in(candidate)
)]


def broken_rename_references(before: List[dict], after: List[dict]) -> List[dict]:
Expand All @@ -116,8 +121,12 @@ def broken_rename_references(before: List[dict], after: List[dict]) -> List[dict
aliases = {normalize(column[key]) for key in ("id", "name") if column.get(key)}
if name not in aliases:
lost.add(name)
# Stable IDs take precedence over display aliases during execution. A
# retained ID cannot be broken by renaming another column's display name.
retained_ids = {column.get("id") for column in after if column.get("id")}
return [column for column in after
if lost.intersection(normalize(ref) for ref in _refs_in(column))]
if any(ref not in retained_ids and normalize(ref) in lost
for ref in _refs_in(column))]


def cycle_blocked_columns(columns: List[dict]) -> set:
Expand Down
29 changes: 29 additions & 0 deletions tests/test_column_deps.py
Original file line number Diff line number Diff line change
Expand Up @@ -166,3 +166,32 @@ def test_formula_dependencies_share_escaped_and_triple_quote_rules():
formula = 'concat("escaped \\"{literal}\\"", """{multiline}\ntext""", { Owner\'s Name })'
assert _refs_in({"formula": formula}) == {"Owner's Name"}
assert _refs_in({"formula": "{self}"}) == {"self"}


def test_structure_guards_preserve_stable_id_precedence_over_display_names():
from apps.api.services.workbook.column_deps import referencing_columns, broken_rename_references
columns = [
{"id": "display", "name": "source", "type": "input"},
{"id": "source", "name": "Real", "type": "input"},
{"id": "derived", "type": "formula", "formula": "{source}"},
]
assert referencing_columns(columns, columns[0]) == []
assert referencing_columns(columns, columns[1]) == [columns[2]]
# Whole-config replacement checks retain the same ID authority.
assert referencing_columns(columns[1:], columns[0]) == []
renamed = [dict(column, name="Renamed") if column["id"] == "display"
else dict(column) for column in columns]
assert broken_rename_references(columns, renamed) == []


def test_structure_guards_still_block_ambiguous_display_alias_retargeting():
from apps.api.services.workbook.column_deps import referencing_columns, broken_rename_references
columns = [
{"id": "first", "name": "Shared", "type": "input"},
{"id": "second", "name": "Shared", "type": "input"},
{"id": "derived", "type": "formula", "formula": "{Shared}"},
]
assert referencing_columns(columns, columns[0]) == [columns[2]]
renamed = [dict(column, name="Renamed") if column["id"] == "first"
else dict(column) for column in columns]
assert broken_rename_references(columns, renamed) == [renamed[2]]
24 changes: 24 additions & 0 deletions tests/test_workbook_views.py
Original file line number Diff line number Diff line change
Expand Up @@ -2845,3 +2845,27 @@ def test_duplicate_display_alias_rejects_ambiguous_public_reference(client, comp
assert row.enrichments["result"]["status"] == "error"
assert row.enrichments["result"]["error"] == "ambiguous_column_reference"
assert row.enrichments["result"]["value"] is None


@pytest.mark.parametrize("operation", ["delete", "rename", "replace"])
def test_structure_edit_does_not_redirect_or_block_another_stable_id(client, operation):
tc, Session, _ = client
columns = [
{"id": "display", "name": "source", "type": "input", "width": 200},
{"id": "source", "name": "Actual source", "type": "input", "width": 200},
{"id": "derived", "name": "Derived", "type": "formula", "formula": "{source}", "width": 200},
]
wid = _mk_workbook(Session, columns)
if operation == "delete":
response = tc.delete(f"/api/workbooks/{wid}/columns/display")
elif operation == "rename":
response = tc.patch(f"/api/workbooks/{wid}/columns/display/settings", json={
"changes": {"name": "Renamed"}, "expected": {"name": "source"}})
else:
response = tc.put(f"/api/workbooks/{wid}", json={"columns_config": columns[1:]})
assert response.status_code == 200, response.text
with Session() as db:
stored = db.get(Workbook, wid).columns_config
assert next(c for c in stored if c["id"] == "source") == columns[1]
assert next(c for c in stored if c["id"] == "derived")["formula"] == "{source}"
assert db.query(Job).count() == 0
Loading