provenance/backend/tests/test_change_proposal.py

"""ChangeProposal: a proposal mutates nothing until an editor approves it, and
application goes through the editing services (privacy + audit). See
docs/design/change-proposal.md and CLAUDE.md non-negotiable #1.
"""

import uuid

from tests.conftest import auth, register


async def _tree(client, email):
    h = auth(await register(client, email))
    tid = (await client.post("/api/v1/trees", json={"name": "T"}, headers=h)).json()["id"]
    return h, tid


async def _propose(client, tid, headers, summary, operations, origin="assistant"):
    r = await client.post(
        f"/api/v1/trees/{tid}/proposals",
        json={"summary": summary, "origin": origin, "operations": operations},
        headers=headers,
    )
    assert r.status_code == 201, r.text
    return r.json()


async def test_proposal_not_applied_until_approved(client):
    h, tid = await _tree(client, "cp-owner@ex.com")
    cp = await _propose(
        client,
        tid,
        h,
        "Add Ada Lovelace",
        [{"op": "create", "entity_type": "person", "payload": {"given": "Ada", "surname": "Lovelace"}}],
    )
    assert cp["status"] == "pending"

    # The proposed person does NOT exist yet.
    people = (await client.get(f"/api/v1/trees/{tid}/persons", headers=h)).json()
    assert not any(p["primary_name"] == "Ada Lovelace" for p in people)

    # Approve → applied → the person now exists.
    a = await client.post(f"/api/v1/trees/{tid}/proposals/{cp['id']}/apply", headers=h)
    assert a.status_code == 200 and a.json()["status"] == "applied"
    people = (await client.get(f"/api/v1/trees/{tid}/persons", headers=h)).json()
    assert any(p["primary_name"] == "Ada Lovelace" for p in people)


async def test_reject_does_not_apply(client):
    h, tid = await _tree(client, "cp-reject@ex.com")
    cp = await _propose(
        client,
        tid,
        h,
        "Add Reject Me",
        [{"op": "create", "entity_type": "person", "payload": {"given": "Reject", "surname": "Me"}}],
    )
    rr = await client.post(
        f"/api/v1/trees/{tid}/proposals/{cp['id']}/reject", json={"note": "no"}, headers=h
    )
    assert rr.status_code == 200 and rr.json()["status"] == "rejected"
    people = (await client.get(f"/api/v1/trees/{tid}/persons", headers=h)).json()
    assert not any(p["primary_name"] == "Reject Me" for p in people)
    # A rejected proposal can't then be applied.
    assert (
        await client.post(f"/api/v1/trees/{tid}/proposals/{cp['id']}/apply", headers=h)
    ).status_code == 409


async def test_non_editor_member_can_see_but_not_apply(client):
    owner = auth(await register(client, "cp-o2@ex.com"))
    viewer = auth(await register(client, "cp-v2@ex.com"))
    tid = (
        await client.post("/api/v1/trees", json={"name": "Shared"}, headers=owner)
    ).json()["id"]
    await client.post(
        f"/api/v1/trees/{tid}/members", json={"email": "cp-v2@ex.com", "role": "viewer"}, headers=owner
    )
    cp = await _propose(
        client,
        tid,
        owner,
        "Add V P",
        [{"op": "create", "entity_type": "person", "payload": {"given": "V", "surname": "P"}}],
    )
    # A viewer (member) can see the proposal list...
    assert (await client.get(f"/api/v1/trees/{tid}/proposals", headers=viewer)).status_code == 200
    # ...but cannot apply it (not an editor).
    assert (
        await client.post(f"/api/v1/trees/{tid}/proposals/{cp['id']}/apply", headers=viewer)
    ).status_code == 403


async def test_multi_op_applies_all(client):
    h, tid = await _tree(client, "cp-multi@ex.com")
    pid = (
        await client.post(
            f"/api/v1/trees/{tid}/persons", json={"given": "Multi", "surname": "Op"}, headers=h
        )
    ).json()["id"]
    cp = await _propose(
        client,
        tid,
        h,
        "name + event on existing person",
        [
            {"op": "create", "entity_type": "name", "payload": {"person_id": pid, "name_type": "alias", "given": "Mo"}},
            {"op": "create", "entity_type": "event", "payload": {"event_type": "birth", "person_id": pid, "date_value": "1900"}},
        ],
    )
    assert (
        await client.post(f"/api/v1/trees/{tid}/proposals/{cp['id']}/apply", headers=h)
    ).status_code == 200
    names = (await client.get(f"/api/v1/trees/{tid}/persons/{pid}/names", headers=h)).json()
    assert any(n.get("given") == "Mo" for n in names)
    events = (await client.get(f"/api/v1/trees/{tid}/persons/{pid}/events", headers=h)).json()
    assert any(e["date_value"] == "1900" for e in events)


async def test_apply_with_edited_operations(client):
    h, tid = await _tree(client, "cp-edit@ex.com")
    cp = await _propose(
        client,
        tid,
        h,
        "Add Original",
        [{"op": "create", "entity_type": "person", "payload": {"given": "Original", "surname": "Name"}}],
    )
    edited = {
        "operations": [
            {"op": "create", "entity_type": "person", "payload": {"given": "Edited", "surname": "Name"}}
        ]
    }
    assert (
        await client.post(f"/api/v1/trees/{tid}/proposals/{cp['id']}/apply", json=edited, headers=h)
    ).status_code == 200
    names = {p["primary_name"] for p in (await client.get(f"/api/v1/trees/{tid}/persons", headers=h)).json()}
    assert "Edited Name" in names and "Original Name" not in names


async def test_apply_error_keeps_pending(client):
    h, tid = await _tree(client, "cp-err@ex.com")
    cp = await _propose(
        client,
        tid,
        h,
        "Bad update",
        [{"op": "update", "entity_type": "person", "entity_id": str(uuid.uuid4()), "payload": {"given": "X"}}],
    )
    a = await client.post(f"/api/v1/trees/{tid}/proposals/{cp['id']}/apply", headers=h)
    assert a.status_code == 409
    g = (await client.get(f"/api/v1/trees/{tid}/proposals/{cp['id']}", headers=h)).json()
    assert g["status"] == "pending"
    assert g["apply_error"]