tests/test_rd_merge_patch.py
Rama master · réplica de lectura del repositorio original.
"""Regression tests for the PATCH /api/rd merge-patch contract.
`api_rd_patch` (api/routes_api.py) used to REPLACE the whole `cards` array
with whatever the client sent. Every client (rd.js, hq-groups.js,
hq-board.js, card-dialog.js) works off a snapshot of ALL cards and PATCHes
that snapshot back after touching one (or a few) — meanwhile the in-process
nudge loop writes SERVER-OWNED state (`card["nudge"]`) to cards on disk
roughly every 30s. A stale client snapshot silently reverted those writes.
The fix: `helpers._merge_cards` shallow-merges incoming cards onto disk by
id — an omitted field is preserved from disk, only fields the client
actually sends win. `api_rd_patch` is unimportable here (it pulls in
`fastapi`, not part of this lightweight dev environment — see
test_rd_lock.py's note on the same constraint), so these tests exercise the
merge function directly, plus a full disk round-trip through the real
`_load_rd`/`_save_rd` persistence path to prove the on-disk nudge subtree
survives a merge-patch the way rd.js's `save()` now sends one.
Pure unit tests — no live app, no container, no fastapi. Monkeypatch
helpers.DATA_DIR to a tmp dir like test_rd_lock.py.
"""
import copy
import json
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "api"))
import helpers # noqa: E402
def _seed(data_dir: Path, cards: list) -> None:
(data_dir / "rd.json").write_text(json.dumps(
{"columns": ["rd", "hq", "archives", "exile"], "cards": cards}))
@pytest.fixture
def data_dir(tmp_path, monkeypatch):
monkeypatch.setattr(helpers, "DATA_DIR", tmp_path)
monkeypatch.setattr(helpers, "_ACTIVITY_LOG", tmp_path / "activity_log.json")
helpers._json_cache.clear()
yield tmp_path
helpers._json_cache.clear()
_NUDGE = {
"stage": "nudging",
"graph": {
"nodes": [{"id": "n1", "label": "step one", "done": False, "est_min": 15}],
"edges": [],
},
"active_node": "n1",
"next_nudge_at": "2026-08-21T10:00:00",
}
# ── pure _merge_cards unit tests ────────────────────────────────────────────
def test_merge_preserves_omitted_nudge_field():
"""The core bug: a client that omits `nudge` on a card it didn't touch
must not erase the server-owned nudge subtree already on disk."""
disk = [{"id": "card-x", "title": "X", "column": "hq", "order": 0, "nudge": copy.deepcopy(_NUDGE)}]
# rd.js's new save() shape: only {id, column, order}, no nudge at all.
incoming = [{"id": "card-x", "column": "hq", "order": 0}]
merged = helpers._merge_cards(disk, incoming)
assert len(merged) == 1
assert merged[0]["nudge"] == _NUDGE
assert merged[0]["title"] == "X" # untouched field preserved too
def test_merge_applies_incoming_field_changes():
disk = [{"id": "card-y", "title": "Y", "column": "rd", "order": 0}]
incoming = [{"id": "card-y", "column": "hq", "order": 5}]
merged = helpers._merge_cards(disk, incoming)
assert merged[0]["column"] == "hq"
assert merged[0]["order"] == 5
assert merged[0]["title"] == "Y"
def test_merge_incoming_nudge_wins_wholesale_not_deep_merged():
"""Only a nudge-authoring client (card-dialog.js) sends `nudge`, and it
must replace the disk value wholesale — never deep-merged node-by-node."""
disk = [{"id": "card-z", "column": "hq", "nudge": copy.deepcopy(_NUDGE)}]
new_nudge = {"stage": "idle", "graph": {"nodes": [], "edges": []}, "active_node": None}
incoming = [{"id": "card-z", "column": "hq", "nudge": new_nudge}]
merged = helpers._merge_cards(disk, incoming)
assert merged[0]["nudge"] == new_nudge
def test_merge_new_card_kept_as_is():
disk = [{"id": "card-a", "column": "rd"}]
incoming = [{"id": "card-new", "title": "brand new", "column": "rd"}]
merged = helpers._merge_cards(disk, incoming)
ids = {c["id"] for c in merged}
assert ids == {"card-a", "card-new"}
new = next(c for c in merged if c["id"] == "card-new")
assert new["title"] == "brand new"
def test_merge_order_incoming_first_then_untouched_disk_order():
disk = [{"id": "a", "column": "rd"}, {"id": "b", "column": "rd"}, {"id": "c", "column": "rd"}]
incoming = [{"id": "b", "column": "hq"}]
merged = helpers._merge_cards(disk, incoming)
assert [c["id"] for c in merged] == ["b", "a", "c"]
def test_merge_empty_incoming_returns_disk_unchanged():
disk = [{"id": "a", "column": "rd", "nudge": copy.deepcopy(_NUDGE)}]
merged = helpers._merge_cards(disk, [])
assert merged == disk
# ── full disk round-trip through the real _load_rd/_save_rd path ───────────
def _simulate_patch(incoming_cards: list) -> None:
"""Mirrors api_rd_patch's write path minus the routes_api-only
scheduling/log side-effects (those operate on the merged list and don't
change what gets persisted for an untouched card's nudge subtree)."""
with helpers._RD_LOCK:
rd = helpers._load_rd()
rd["cards"] = helpers._merge_cards(rd.get("cards", []), incoming_cards)
helpers._save_rd(rd)
def test_patch_round_trip_leaves_untouched_cards_nudge_intact(data_dir):
"""The exact scenario in the finding: rd.js drags a card on the board and
PATCHes {id, column, order} for every card — including card X, which it
never decomposed and knows nothing about. X's nudge (written by the
server-side nudge loop) must survive; Y's column change must land."""
card_x = {"id": "card-x", "title": "X", "column": "hq", "order": 0,
"nudge": copy.deepcopy(_NUDGE)}
card_y = {"id": "card-y", "title": "Y", "column": "rd", "order": 0}
_seed(data_dir, [card_x, card_y])
# rd.js's save(): the whole board's {id, column, order} triples, no nudge.
_simulate_patch([
{"id": "card-x", "column": "hq", "order": 0},
{"id": "card-y", "column": "hq", "order": 1},
])
on_disk = {c["id"]: c for c in json.loads((data_dir / "rd.json").read_text())["cards"]}
assert on_disk["card-x"]["nudge"] == _NUDGE
assert on_disk["card-y"]["column"] == "hq"
assert on_disk["card-y"]["order"] == 1
def test_patch_round_trip_card_x_omitted_entirely(data_dir):
"""A client that only sends the one card it touched (hq-groups.js /
hq-board.js / card-dialog.js shape) must not disturb any other card."""
card_x = {"id": "card-x", "title": "X", "column": "hq",
"nudge": copy.deepcopy(_NUDGE)}
card_y = {"id": "card-y", "title": "Y", "column": "hq", "dir_start_min": 600}
_seed(data_dir, [card_x, card_y])
_simulate_patch([{"id": "card-y", "title": "Y", "column": "hq", "dir_start_min": 630}])
on_disk = {c["id"]: c for c in json.loads((data_dir / "rd.json").read_text())["cards"]}
assert on_disk["card-x"] == card_x # completely untouched
assert on_disk["card-y"]["dir_start_min"] == 630