"""HTTP contracts for the M5 write endpoints (T-ENT-42: structured errors; T-AUD-001: mutations audit): PATCH cell edit round-trip, manual create + delete + undo, the bulk body contract incl. the safety stop over the wire, and the entities manage routes. These run over the migrated reporting scratch AFTER the read-API tests (file order); every test restores the state it touched.""" import uuid from sqlalchemy import select from sqlalchemy.orm import Session from app.models import AuditLog, TimeEntry TE2 = "0a000000-0000-0000-0000-000000000002" # March work B (pending) TE1 = "0a000000-0000-0000-0000-000000000001" # invoice-locked def test_patch_cell_edit_round_trip(as_rian): resp = as_rian.patch(f"/api/entries/{TE2}", json={"description": "renamed via API"}) assert resp.status_code == 200 body = resp.json() assert body["row"]["description"] == "renamed via API" assert body["row"]["id"] == TE2 # restore resp = as_rian.patch(f"/api/entries/{TE2}", json={"description": "March work B"}) assert resp.json()["row"]["description"] == "March work B" def test_patch_absent_fields_untouched(as_rian): """PATCH semantics: only SET fields apply — sending billable alone must not clear the description (model_fields_set plumbing).""" before = as_rian.get("/api/entries", params={"status": "all", "q": "March work B"}) row = before.json()["rows"][0] resp = as_rian.patch(f"/api/entries/{row['id']}", json={"billable": False}) updated = resp.json()["row"] assert updated["billable"] is False assert updated["description"] == row["description"] as_rian.patch(f"/api/entries/{row['id']}", json={"billable": True}) def test_locked_row_manager_409_structured(as_adi): resp = as_adi.patch(f"/api/entries/{TE1}", json={"description": "nope"}) assert resp.status_code == 409 body = resp.json() assert body["error_code"] == "ENTRY_LOCKED" assert "Only an owner can edit it" in body["summary"] def test_create_delete_undo_cycle(as_rian, rpt_engine): create = as_rian.post( "/api/entries", json={ "date": "2026-07-04", "duration": "0:30:00", "description": "api manual entry", "worker_party_id": _party_id(rpt_engine, "info@adipramono.com"), }, ) assert create.status_code == 200, create.text row = create.json()["row"] assert row["worker"] == "Adi Pramono" assert row["billable"] is True delete = as_rian.delete(f"/api/entries/{row['id']}") assert delete.status_code == 200 undo_id = delete.json()["undo_id"] undo = as_rian.post(f"/api/entries/undo/{undo_id}") assert undo.status_code == 200 assert undo.json()["reinserted"] == 1 # double-undo refused again = as_rian.post(f"/api/entries/undo/{undo_id}") assert again.status_code == 409 assert again.json()["error_code"] == "ALREADY_UNDONE" # clean up the re-inserted row for later suites with Session(rpt_engine) as s: entry = s.get(TimeEntry, uuid.UUID(row["id"])) s.delete(entry) s.commit() def test_create_validation_message_over_the_wire(as_rian): resp = as_rian.post("/api/entries", json={"date": "bogus"}) assert resp.status_code == 422 assert resp.json()["summary"] == "Date must be YYYY-MM-DD." def test_bulk_safety_stop_over_the_wire(as_rian): """The T-ENT-17 contract end-to-end: all_matching + the display's own filter params + a stale expected_count -> 409 SAFETY_STOP with the exact message, zero rows changed.""" options = as_rian.get("/api/entries/edit-options").json() worker = next(w for w in options["workers"] if w["email"] == "gary@rian.ca") resp = as_rian.post( "/api/entries/bulk/edit", json={ "all_matching": True, "filters": {"status": "pending", "operator": "PlusROI"}, "expected_count": 343, "worker_party_id": worker["party_id"], }, ) assert resp.status_code == 409 body = resp.json() assert body["error_code"] == "SAFETY_STOP" assert body["summary"].startswith("Safety stop: the filter currently matches") assert "you confirmed 343. Nothing was changed." in body["summary"] def test_bulk_recalculate_verified_counts_and_undo(as_rian): totals = as_rian.get("/api/entries/totals", params={"status": "all"}).json() resp = as_rian.post( "/api/entries/bulk/recalculate", json={"all_matching": True, "filters": {"status": "all"}, "expected_count": totals["row_count"]}, ) assert resp.status_code == 200 body = resp.json() assert body["matched"] == totals["row_count"] assert body["skipped_locked"] == 2 undo = as_rian.post(f"/api/entries/undo/{body['undo_id']}") assert undo.status_code == 200 assert undo.json()["restored"] == body["changed"] def test_bulk_requires_selection(as_rian): resp = as_rian.post( "/api/entries/bulk/recalculate", json={"expected_count": 0, "ids": []} ) assert resp.status_code == 422 assert resp.json()["error_code"] == "NO_SELECTION" def test_edit_options_shape(as_rian): body = as_rian.get("/api/entries/edit-options").json() assert {w["email"] for w in body["workers"]} >= {"info@adipramono.com", "gary@rian.ca"} assert any(p["label"].startswith("Brentwood : ") for p in body["projects"]) assert body["default_worker_party_id"] is not None # rian matches by email def test_entities_manage_and_rename_round_trip(as_rian, as_adi): payload = as_rian.get("/api/entities/manage").json() assert payload["can_manage"] is True site = next(p for p in payload["projects"] if p["name"] == "Site Rebuild") assert site["entries"] > 0 renamed = as_rian.post( f"/api/entities/projects/{site['id']}/rename", json={"name": "Site Rebuild v2"} ) assert renamed.status_code == 200 and renamed.json()["changed"] is True back = as_rian.post( f"/api/entities/projects/{site['id']}/rename", json={"name": "Site Rebuild"} ) assert back.json()["changed"] is True def test_mutations_write_audit_rows(as_rian, rpt_engine): as_rian.patch(f"/api/entries/{TE2}", json={"description": "March work B"}) with Session(rpt_engine) as s: audit = s.scalars( select(AuditLog).where(AuditLog.action == "entries.update") ).all() assert audit, "T-AUD-001: mutation must write an audit row" assert audit[-1].before is not None def _party_id(engine, email: str) -> str: from app.models import Party with Session(engine) as s: return str( s.scalars( select(Party.id).where(Party.kind == "person", Party.email == email) ).one() )