jacobh2/kvitt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
c9abd24c772854bf5814d4e7f4a4ab042e73b5cf
kvitt/server/crates/kvitt-server/tests/sync_api/push_rules.rs
T
jacobh2 8cb42f85a9 Server overhaul: one test harness in tests/common for every test binary 
core_api/common.rs becomes tests/common/mod.rs and sync_api/m4.rs tests/common/sync_fixtures.rs; each binary
includes them through #[path] (the binaries of SPEC §14.7 stay as they are). TempDir's Default moves into the
harness instead of three copies; the skeleton helpers take ids, tokens and TempDir from it. Milestone wording in
test docs and the server README replaced by what the code does; README seams and test layout updated.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TQd3pEP9UeN2uifbHknQ7B
2026-10-18 22:43:54 +02:00

1043 lines
36 KiB
Rust
Raw Blame History

//! E-32 push: SPEC §14.4.3 and every step of §14.4.4 in order, each rejection reason of §14.4.6, batch limits and
//! push batching (AC-13.4).
use axum::http::Method;
use serde_json::{Value, json};
use crate::common::*;
use crate::sync_fixtures::*;
/// A change applied to a pushed entity.
type Mutator = Box<dyn Fn(&mut Value)>;
const T1: &str = "2026-10-17T11:00:00.000Z";
const T2: &str = "2026-10-17T11:30:00.000Z";
// ---- whole-request rules ----------------------------------------------------------------------------------------
#[tokio::test]
async fn batch_of_101_is_batch_too_large_and_0_too_and_100_is_fine() {
let app = app().await;
let many = |n: usize| {
(0..n)
.map(|_| exp(expense(&new_id(), T1)))
.collect::<Vec<_>>()
};
let reply = push(&app, &admin_token(), many(101)).await;
assert_error(&reply, 422, "batch_too_large");
assert_eq!(count(&app, "expenses").await, 0);
let reply = push(&app, &admin_token(), Vec::new()).await;
assert_error(&reply, 422, "batch_too_large");
let reply = push(&app, &admin_token(), many(100)).await;
assert_eq!(reply.status, 200, "{}", reply.text());
assert_eq!(reply.json["results"].as_array().unwrap().len(), 100);
assert_eq!(count(&app, "expenses").await, 100);
}
#[tokio::test]
async fn body_over_1_mib_is_payload_too_large() {
let app = app().await;
let mut e = expense(&new_id(), T1);
e["notes"] = json!("x".repeat(1024 * 1024 + 1));
let reply = push(&app, &admin_token(), vec![exp(e)]).await;
assert_error(&reply, 413, "payload_too_large");
}
#[tokio::test]
async fn malformed_body_or_envelope_is_invalid_request() {
let app = app().await;
let reply = app
.call(
Method::POST,
"/sync",
&admin_token(),
Some(json!({"nope": []})),
)
.await;
assert_error(&reply, 400, "invalid_request");
// Without a usable mutation_id, entity_type or entity.id there is no per-mutation result to give.
for bad in [
json!({"mutation_id": "x", "entity_type": "expense", "entity": expense(&new_id(), T1)}),
json!({"mutation_id": new_id(), "entity_type": "account", "entity": expense(&new_id(), T1)}),
json!({"mutation_id": new_id(), "entity_type": "expense", "entity": {"id": 5}}),
json!({"mutation_id": new_id().to_uppercase(), "entity_type": "expense", "entity": expense(&new_id(), T1)}),
] {
let reply = push(&app, &admin_token(), vec![exp(expense(&new_id(), T1)), bad]).await;
assert_error(&reply, 400, "invalid_request");
}
assert_eq!(count(&app, "expenses").await, 0);
}
#[tokio::test]
async fn step2_household_incomplete_fails_the_whole_request() {
let app = TestApp::admin_only().await;
let reply = push(&app, &admin_token(), vec![exp(expense(&new_id(), T1))]).await;
assert_error(&reply, 409, "household_incomplete");
assert_eq!(count(&app, "sync_mutations").await, 0);
}
#[tokio::test]
async fn results_come_in_request_order_with_exact_key_sets() {
let app = app().await;
let ids: Vec<String> = (0..3).map(|_| new_id()).collect();
let mut bad = expense(&ids[1], T1);
bad["amount_ore"] = json!(0);
let mutations = vec![
exp(expense(&ids[0], T1)),
exp(bad),
set(settlement(&ids[2], T1)),
];
let reply = push(&app, &admin_token(), mutations.clone()).await;
assert_eq!(reply.status, 200);
assert_eq!(reply.json["server_time"], NOW);
let results = reply.json["results"].as_array().unwrap();
for (r, m) in results.iter().zip(&mutations) {
assert_eq!(r["mutation_id"], m["mutation_id"]);
assert_eq!(r["entity_id"], m["entity"]["id"]);
assert_eq!(r["entity_type"], m["entity_type"]);
}
let keys: Vec<_> = results[0].as_object().unwrap().keys().cloned().collect();
assert_eq!(
keys,
[
"entity",
"entity_id",
"entity_type",
"mutation_id",
"server_version",
"status"
]
);
rejected(&results[1], "validation_failed");
accepted(&results[2]);
}
// ---- step 1: replay ---------------------------------------------------------------------------------------------
#[tokio::test]
async fn step1_replay_returns_the_stored_result_unchanged_with_no_other_effect() {
let app = app().await;
let id = new_id();
let m = exp(expense(&id, T1));
let first = push_one(&app, &admin_token(), m.clone()).await;
let version = sequence(&app).await;
let audits = count(&app, "audit_log").await;
let pushed = pushes(&app).len();
// The same mutation id with different content (and later in time) replays the first result.
let mut changed = m.clone();
changed["entity"]["amount_ore"] = json!(1);
changed["entity"]["updated_at"] = json!(T2);
app.set_now("2026-10-18T12:00:00.000Z");
let again = push_one(&app, &admin_token(), changed).await;
assert_eq!(again, first);
assert_eq!(sequence(&app).await, version);
assert_eq!(count(&app, "audit_log").await, audits);
assert_eq!(pushes(&app).len(), pushed);
assert_eq!(count(&app, "sync_mutations").await, 1);
let amount: i64 = app.scalar_i64("SELECT amount_ore FROM expenses").await;
assert_eq!(amount, 41200);
// A rejected result replays too.
let mut bad = exp(expense(&new_id(), T1));
bad["entity"]["description"] = json!("");
let r1 = push_one(&app, &admin_token(), bad.clone()).await;
rejected(&r1, "validation_failed");
bad["entity"]["description"] = json!("fixed");
let r2 = push_one(&app, &admin_token(), bad).await;
assert_eq!(r1, r2);
}
#[tokio::test]
async fn duplicate_mutation_ids_in_one_request_replay_the_first() {
let app = app().await;
let m = exp(expense(&new_id(), T1));
let reply = push(&app, &admin_token(), vec![m.clone(), m]).await;
let results = reply.json["results"].as_array().unwrap();
assert_eq!(results[0], results[1]);
assert_eq!(count(&app, "expenses").await, 1);
}
// ---- step 3: shape ----------------------------------------------------------------------------------------------
#[tokio::test]
async fn step3_shape_errors_are_rejected_validation_failed_per_mutation() {
let app = app().await;
let cases: Vec<(&str, Mutator)> = vec![
(
"missing field",
Box::new(|e| {
e.as_object_mut().unwrap().remove("description");
}),
),
(
"missing deleted_at",
Box::new(|e| {
e.as_object_mut().unwrap().remove("deleted_at");
}),
),
("wrong type", Box::new(|e| e["amount_ore"] = json!("41200"))),
(
"unknown enum",
Box::new(|e| e["split_mode"] = json!("half")),
),
(
"not a uuid",
Box::new(|e| e["account_id"] = json!("vardag")),
),
(
"uppercase uuid",
Box::new(|e| e["account_id"] = json!(ACCOUNT_1.to_uppercase())),
),
(
"split user not a uuid",
Box::new(|e| e["splits"][0]["user_id"] = json!("jacob")),
),
(
"timestamp form",
Box::new(|e| e["updated_at"] = json!("2026-10-17T11:00:00Z")),
),
(
"deleted_at form",
Box::new(|e| e["deleted_at"] = json!("yesterday")),
),
];
for (name, mutate) in cases {
let mut e = expense(&new_id(), T1);
mutate(&mut e);
let r = push_one(&app, &admin_token(), exp(e)).await;
assert_eq!(r["reason"], "validation_failed", "{name}");
rejected(&r, "validation_failed");
assert_eq!(r["current"], Value::Null, "{name}");
}
let mut s = settlement(&new_id(), T1);
s["allocations"][0]["account_id"] = json!(1);
rejected(
&push_one(&app, &admin_token(), set(s)).await,
"validation_failed",
);
assert_eq!(count(&app, "expenses").await, 0);
assert_eq!(count(&app, "sync_mutations").await, 10);
}
#[tokio::test]
async fn server_owned_fields_are_ignored_if_sent() {
let app = app().await;
let mut e = expense(&new_id(), T1);
e["server_version"] = json!(999_999);
e["created_by_user_id"] = json!(MEMBER_ID);
e["recurring_template_id"] = json!(UNKNOWN_ID);
e["recurring_period"] = json!("2026-10");
let entity = accepted(&push_one(&app, &admin_token(), exp(e)).await);
assert_eq!(entity["created_by_user_id"], ADMIN_ID);
assert_eq!(entity["recurring_template_id"], Value::Null);
assert_eq!(entity["recurring_period"], Value::Null);
assert_ne!(entity["server_version"], 999_999);
}
// ---- step 4: clock ----------------------------------------------------------------------------------------------
#[tokio::test]
async fn step4_updated_at_more_than_10_minutes_ahead_is_clock_skew() {
let app = app().await;
let r = push_one(
&app,
&admin_token(),
exp(expense(&new_id(), "2026-10-17T12:10:00.001Z")),
)
.await;
rejected(&r, "clock_skew");
assert_eq!(r["current"], Value::Null);
let r = push_one(
&app,
&admin_token(),
exp(expense(&new_id(), "2026-10-17T12:10:00.000Z")),
)
.await;
accepted(&r);
// Clock skew comes before the tombstone and validation checks: the stored row is the current.
let id = new_id();
push_one(&app, &admin_token(), exp(expense(&id, T1))).await;
let mut e = expense(&id, "2026-10-18T12:00:00.000Z");
e["amount_ore"] = json!(0);
let r = push_one(&app, &admin_token(), exp(e)).await;
rejected(&r, "clock_skew");
assert_eq!(r["current"]["id"], id);
}
// ---- steps 5, 8, 9: load, validate, insert -----------------------------------------------------------------------
#[tokio::test]
async fn step9_create_inserts_with_server_fields_audit_create_and_push() {
let app = app().await;
let id = new_id();
let before = sequence(&app).await;
let mut pushed = expense(&id, T1);
pushed["description"] = json!(" ICA Maxi ");
pushed["created_at"] = json!("2026-10-17T10:59:00.000Z");
let r = push_one(&app, &member_token(), exp(pushed)).await;
let entity = accepted(&r);
assert_eq!(entity["description"], "ICA Maxi", "stored trimmed");
assert_eq!(entity["created_by_user_id"], MEMBER_ID);
assert_eq!(entity["updated_by_user_id"], MEMBER_ID);
assert_eq!(entity["created_at"], "2026-10-17T10:59:00.000Z");
assert_eq!(entity["updated_at"], T1);
assert!(entity["server_version"].as_i64().unwrap() > before);
let (device, received): (Option<String>, String) = sqlx::query_as(
"SELECT updated_by_device_id, server_received_at FROM expenses WHERE id = ?",
)
.bind(&id)
.fetch_one(&app.state.db)
.await
.unwrap();
assert_eq!(device.as_deref(), Some(MEMBER_DEVICE_ID));
assert_eq!(received, NOW);
let audit = audit_of(&app, &id).await;
assert_eq!(audit.len(), 1);
let a = &audit[0];
assert_eq!((a.action.as_str(), a.source.as_str()), ("create", "app"));
assert_eq!(a.actor_user_id, MEMBER_ID);
assert_eq!(a.actor_device_id.as_deref(), Some(MEMBER_DEVICE_ID));
assert_eq!((a.client_at.as_str(), a.server_at.as_str()), (T1, NOW));
let fields: Vec<_> = a
.changes
.as_array()
.unwrap()
.iter()
.map(|c| c["field"].as_str().unwrap())
.collect();
assert_eq!(
fields,
[
"account_id",
"category_id",
"payer_user_id",
"amount_ore",
"description",
"notes",
"expense_date",
"split_mode",
"splits"
]
);
assert!(
a.changes
.as_array()
.unwrap()
.iter()
.all(|c| c["old"].is_null())
);
assert_eq!(a.changes[4]["new"], "ICA Maxi");
assert_eq!(
pushes(&app),
[(
"expense_created".to_owned(),
Some(id.clone()),
MEMBER_ID.to_owned(),
1
)]
);
let sid = new_id();
let entity = accepted(&push_one(&app, &member_token(), set(settlement(&sid, T1))).await);
assert_eq!(entity["allocations"].as_array().unwrap().len(), 2);
let audit = audit_of(&app, &sid).await;
let fields: Vec<_> = audit[0]
.changes
.as_array()
.unwrap()
.iter()
.map(|c| c["field"].as_str().unwrap())
.collect();
assert_eq!(
fields,
[
"from_user_id",
"to_user_id",
"amount_ore",
"settlement_date",
"note",
"allocations"
]
);
assert_eq!(pushes(&app)[1].0, "settlement_created");
}
/// One field rule of §9.6 / §9.1 / §9.4 broken: validation_failed.
async fn assert_invalid(app: &TestApp, entity_type: &str, entity: Value, why: &str) {
let r = push_one(app, &admin_token(), mutation(entity_type, entity)).await;
assert_eq!(r["reason"], "validation_failed", "{why}: {r:#}");
}
#[tokio::test]
async fn step8_expense_rules_of_9_6_and_9_1() {
let app = app().await;
let e = || expense(&new_id(), T1);
let with = |f: &dyn Fn(&mut Value)| {
let mut v = e();
f(&mut v);
v
};
let cases: Vec<(&str, Value)> = vec![
(
"amount 0",
with(&|v| {
v["amount_ore"] = json!(0);
}),
),
(
"amount over max",
with(&|v| {
v["amount_ore"] = json!(1_000_000_001);
v["splits"][0]["share_ore"] = json!(500_000_001);
v["splits"][1]["share_ore"] = json!(500_000_000);
}),
),
(
"empty description",
with(&|v| {
v["description"] = json!(" ");
}),
),
(
"description 101",
with(&|v| {
v["description"] = json!("é".repeat(101));
}),
),
(
"notes 1001",
with(&|v| {
v["notes"] = json!("n".repeat(1001));
}),
),
(
"notes control char",
with(&|v| {
v["notes"] = json!("a\tb");
}),
),
(
"date before 2000",
with(&|v| {
v["expense_date"] = json!("1999-12-31");
}),
),
(
"date over a year ahead",
with(&|v| {
v["expense_date"] = json!("2027-10-18");
}),
),
(
"date invalid",
with(&|v| {
v["expense_date"] = json!("2026-02-30");
}),
),
(
"share mismatch",
with(&|v| {
v["splits"][0]["share_ore"] = json!(20000);
v["splits"][1]["share_ore"] = json!(21200);
}),
),
(
"bp sum",
with(&|v| {
v["splits"][0]["share_bp"] = json!(6000);
}),
),
(
"bp out of range",
with(&|v| {
v["splits"][0]["share_bp"] = json!(-1);
v["splits"][1]["share_bp"] = json!(10001);
}),
),
(
"bp missing in ratio mode",
with(&|v| {
v["split_mode"] = json!("ratio");
v["splits"][0]["share_bp"] = json!(null);
}),
),
(
"exact with bp",
with(&|v| {
v["split_mode"] = json!("exact");
}),
),
(
"exact sum",
with(&|v| {
v["split_mode"] = json!("exact");
v["splits"][0]["share_bp"] = json!(null);
v["splits"][1]["share_bp"] = json!(null);
v["splits"][1]["share_ore"] = json!(1);
}),
),
(
"full with 50/50",
with(&|v| {
v["split_mode"] = json!("full");
}),
),
(
"unknown split user",
with(&|v| {
v["splits"][1]["user_id"] = json!(UNKNOWN_ID);
}),
),
(
"one split",
with(&|v| {
v["splits"].as_array_mut().unwrap().pop();
}),
),
(
"three splits",
with(&|v| {
let first = v["splits"][0].clone();
v["splits"].as_array_mut().unwrap().push(first);
}),
),
(
"payer not in household",
with(&|v| {
v["payer_user_id"] = json!(UNKNOWN_ID);
}),
),
];
for (why, entity) in cases {
assert_invalid(&app, "expense", entity, why).await;
}
assert_eq!(count(&app, "expenses").await, 0);
assert_eq!(count(&app, "audit_log").await, 0);
assert!(pushes(&app).is_empty());
}
#[tokio::test]
async fn step8_valid_edges_are_accepted() {
let app = app().await;
let ok = |v: Value| async { accepted(&push_one(&app, &admin_token(), exp(v)).await) };
let mut v = expense(&new_id(), T1);
v["expense_date"] = json!("2027-10-17"); // today (Stockholm) + 365
v["description"] = json!("é".repeat(100));
v["notes"] = json!("rad ett\nrad två");
ok(v).await;
let mut v = expense(&new_id(), T1);
v["expense_date"] = json!("2000-01-01");
ok(v).await;
// Mode default: the bp are not compared with the account's current default (here 60/40 vs the account's 50/50).
let mut v = expense(&new_id(), T1);
v["splits"] = json!([
{"user_id": ADMIN_ID, "share_ore": 24720, "share_bp": 6000},
{"user_id": MEMBER_ID, "share_ore": 16480, "share_bp": 4000}
]);
ok(v).await;
// Ratio rule: the payer absorbs the remainder; exact; full; splits in any order.
let mut v = expense(&new_id(), T1);
v["amount_ore"] = json!(1001);
v["split_mode"] = json!("ratio");
v["splits"] = json!([
{"user_id": MEMBER_ID, "share_ore": 500, "share_bp": 5000},
{"user_id": ADMIN_ID, "share_ore": 501, "share_bp": 5000}
]);
let entity = ok(v).await;
assert_eq!(
entity["splits"][0]["user_id"], ADMIN_ID,
"sorted by user_id"
);
let mut v = expense(&new_id(), T1);
v["split_mode"] = json!("exact");
v["splits"] = json!([
{"user_id": ADMIN_ID, "share_ore": 41000, "share_bp": null},
{"user_id": MEMBER_ID, "share_ore": 200, "share_bp": null}
]);
ok(v).await;
let mut v = expense(&new_id(), T1);
v["split_mode"] = json!("full");
v["splits"] = json!([
{"user_id": ADMIN_ID, "share_ore": 0, "share_bp": 0},
{"user_id": MEMBER_ID, "share_ore": 41200, "share_bp": 10000}
]);
ok(v).await;
}
#[tokio::test]
async fn step8_settlement_rules_of_9_4() {
let app = app().await;
let s = || settlement(&new_id(), T1);
let with = |f: &dyn Fn(&mut Value)| {
let mut v = s();
f(&mut v);
v
};
let cases: Vec<(&str, Value)> = vec![
(
"same user",
with(&|v| {
v["to_user_id"] = json!(MEMBER_ID);
}),
),
(
"unknown user",
with(&|v| {
v["to_user_id"] = json!(UNKNOWN_ID);
}),
),
(
"amount 0",
with(&|v| {
v["amount_ore"] = json!(0);
}),
),
(
"no allocations",
with(&|v| {
v["allocations"] = json!([]);
}),
),
(
"allocation 0",
with(&|v| {
v["allocations"][0]["amount_ore"] = json!(0);
v["allocations"][1]["amount_ore"] = json!(50000);
}),
),
(
"sum mismatch",
with(&|v| {
v["allocations"][0]["amount_ore"] = json!(1);
}),
),
(
"duplicate account",
with(&|v| {
v["allocations"][1]["account_id"] = json!(ACCOUNT_1);
}),
),
(
"51 allocations",
with(&|v| {
v["amount_ore"] = json!(51);
v["allocations"] = Value::Array(
(0..51)
.map(|_| json!({"account_id": new_id(), "amount_ore": 1}))
.collect(),
);
}),
),
(
"note 201",
with(&|v| {
v["note"] = json!("n".repeat(201));
}),
),
(
"date",
with(&|v| {
v["settlement_date"] = json!("16/10/2026");
}),
),
];
for (why, entity) in cases {
assert_invalid(&app, "settlement", entity, why).await;
}
assert_eq!(count(&app, "settlements").await, 0);
}
#[tokio::test]
async fn step8_deleted_references_are_reference_not_found_archived_are_fine() {
let app = app().await;
tombstone(&app, "accounts", ACCOUNT_2).await;
tombstone(&app, "categories", HOME).await;
let mut e = expense(&new_id(), T1);
e["account_id"] = json!(ACCOUNT_2);
let r = push_one(&app, &admin_token(), exp(e)).await;
rejected(&r, "reference_not_found");
assert_eq!(r["message"], "account_id");
let mut e = expense(&new_id(), T1);
e["account_id"] = json!(UNKNOWN_ID);
rejected(
&push_one(&app, &admin_token(), exp(e)).await,
"reference_not_found",
);
let mut e = expense(&new_id(), T1);
e["category_id"] = json!(HOME);
let r = push_one(&app, &admin_token(), exp(e)).await;
rejected(&r, "reference_not_found");
assert_eq!(r["message"], "category_id");
let r = push_one(&app, &admin_token(), set(settlement(&new_id(), T1))).await;
rejected(&r, "reference_not_found");
assert_eq!(r["message"], "allocations");
// §9.10: archived accounts and categories accept writes.
sqlx::query("UPDATE accounts SET archived_at = ? WHERE id = ?")
.bind(NOW)
.bind(ACCOUNT_1)
.execute(&app.state.db)
.await
.unwrap();
sqlx::query("UPDATE categories SET archived_at = ? WHERE id = ?")
.bind(NOW)
.bind(GROCERIES)
.execute(&app.state.db)
.await
.unwrap();
accepted(&push_one(&app, &admin_token(), exp(expense(&new_id(), T1))).await);
}
#[tokio::test]
async fn step8_rejection_of_an_existing_row_carries_it_as_current() {
let app = app().await;
let id = new_id();
let stored = accepted(&push_one(&app, &admin_token(), exp(expense(&id, T1))).await);
let mut e = expense(&id, T2);
e["amount_ore"] = json!(0);
let r = push_one(&app, &admin_token(), exp(e)).await;
rejected(&r, "validation_failed");
assert_eq!(r["current"], stored);
}
// ---- step 6: stored tombstone -----------------------------------------------------------------------------------
#[tokio::test]
async fn step6_tombstone_accepts_a_repeated_delete_and_rejects_anything_else_as_deleted() {
let app = app().await;
let id = new_id();
push_one(&app, &admin_token(), exp(expense(&id, T1))).await;
let mut del = expense(&id, T2);
del["deleted_at"] = json!(T2);
let tomb = accepted(&push_one(&app, &admin_token(), exp(del.clone())).await);
let version = sequence(&app).await;
let audits = count(&app, "audit_log").await;
let pushed = pushes(&app).len();
// Idempotent delete: accepted, nothing changes.
let mut del2 = del.clone();
del2["deleted_at"] = json!("2026-10-17T11:45:00.000Z");
del2["updated_at"] = json!("2026-10-17T11:45:00.000Z");
let r = push_one(&app, &member_token(), exp(del2)).await;
assert_eq!(accepted(&r), tomb);
assert_eq!(sequence(&app).await, version);
assert_eq!(count(&app, "audit_log").await, audits);
assert_eq!(pushes(&app).len(), pushed);
// An edit of the tombstone, even a newer one: deleted, with the tombstone as current.
let mut edit = expense(&id, "2026-10-17T11:59:00.000Z");
edit["amount_ore"] = json!(100);
edit["splits"][0]["share_ore"] = json!(50);
edit["splits"][1]["share_ore"] = json!(50);
let r = push_one(&app, &member_token(), exp(edit)).await;
rejected(&r, "deleted");
assert_eq!(r["current"], tomb);
assert_eq!(sequence(&app).await, version);
}
// ---- step 7: pushed delete -------------------------------------------------------------------------------------
#[tokio::test]
async fn step7_delete_of_a_stored_row_always_wins_and_keeps_other_fields() {
let app = app().await;
let id = new_id();
let stored = accepted(&push_one(&app, &admin_token(), exp(expense(&id, T2))).await);
// An older delete with different content: still applied; other fields keep the stored values.
let mut del = expense(&id, T1);
del["deleted_at"] = json!("2026-10-17T11:10:00.000Z");
del["amount_ore"] = json!(999);
del["description"] = json!("changed");
let entity = accepted(&push_one(&app, &member_token(), exp(del)).await);
assert_eq!(entity["deleted_at"], "2026-10-17T11:10:00.000Z");
assert_eq!(entity["updated_at"], T2, "the later of stored and pushed");
assert_eq!(entity["amount_ore"], 41200);
assert_eq!(entity["description"], "ICA Maxi");
assert_eq!(entity["updated_by_user_id"], MEMBER_ID);
assert_eq!(entity["created_by_user_id"], ADMIN_ID);
assert!(
entity["server_version"].as_i64().unwrap() > stored["server_version"].as_i64().unwrap()
);
let audit = audit_of(&app, &id).await;
assert_eq!(audit.len(), 2);
assert_eq!(audit[1].action, "delete");
assert_eq!(audit[1].client_at, "2026-10-17T11:10:00.000Z");
assert_eq!(
audit[1].changes,
json!([{"field": "deleted_at", "old": null, "new": "2026-10-17T11:10:00.000Z"}])
);
assert_eq!(
pushes(&app)[1],
(
"expense_deleted".to_owned(),
Some(id),
MEMBER_ID.to_owned(),
1
)
);
// A newer pushed updated_at is taken.
let sid = new_id();
push_one(&app, &admin_token(), set(settlement(&sid, T1))).await;
let mut del = settlement(&sid, T2);
del["deleted_at"] = json!(T2);
let entity = accepted(&push_one(&app, &admin_token(), set(del)).await);
assert_eq!(entity["updated_at"], T2);
assert_eq!(pushes(&app).last().unwrap().0, "settlement_deleted");
}
#[tokio::test]
async fn step7_delete_of_a_never_synced_row_stores_a_tombstone_without_audit_or_push() {
let app = app().await;
let id = new_id();
let mut del = expense(&id, T1);
del["deleted_at"] = json!(T1);
let entity = accepted(&push_one(&app, &admin_token(), exp(del)).await);
assert_eq!(entity["deleted_at"], T1);
assert_eq!(entity["created_by_user_id"], ADMIN_ID);
assert!(audit_of(&app, &id).await.is_empty());
assert!(pushes(&app).is_empty());
// References are still checked.
tombstone(&app, "accounts", ACCOUNT_2).await;
let mut del = expense(&new_id(), T1);
del["deleted_at"] = json!(T1);
del["account_id"] = json!(ACCOUNT_2);
let r = push_one(&app, &admin_token(), exp(del)).await;
rejected(&r, "reference_not_found");
assert_eq!(r["current"], Value::Null);
}
// ---- step 10: stale ---------------------------------------------------------------------------------------------
#[tokio::test]
async fn step10_equal_or_older_updated_at_is_stale_ties_go_to_the_server() {
let app = app().await;
let id = new_id();
let stored = accepted(&push_one(&app, &admin_token(), exp(expense(&id, T2))).await);
for at in [T2, T1] {
let mut e = expense(&id, at);
e["description"] = json!("Other");
let r = push_one(&app, &member_token(), exp(e)).await;
rejected(&r, "stale");
assert_eq!(r["message"], "a newer version exists");
assert_eq!(r["current"], stored);
}
assert_eq!(audit_of(&app, &id).await.len(), 1);
}
// ---- step 11: update --------------------------------------------------------------------------------------------
#[tokio::test]
async fn step11_newer_update_writes_the_diff_and_keeps_server_owned_fields() {
let app = app().await;
// A generated expense: recurring fields must survive an edit.
let id = new_id();
accepted(&push_one(&app, &admin_token(), exp(expense(&id, T1))).await);
sqlx::query("INSERT INTO recurring_templates (id, account_id, category_id, amount_ore, description, notes, payer_user_id, split_mode, day_of_month, start_period, created_by_user_id, updated_by_user_id, created_at, updated_at, server_version) VALUES (?, ?, ?, 41200, 'ICA', '', ?, 'default', 1, '2026-10', ?, ?, ?, ?, 900000)")
.bind(UNKNOWN_ID).bind(ACCOUNT_1).bind(GROCERIES).bind(ADMIN_ID).bind(ADMIN_ID).bind(ADMIN_ID).bind(NOW).bind(NOW)
.execute(&app.state.db).await.unwrap();
sqlx::query(
"UPDATE expenses SET recurring_template_id = ?, recurring_period = '2026-10' WHERE id = ?",
)
.bind(UNKNOWN_ID)
.bind(&id)
.execute(&app.state.db)
.await
.unwrap();
let mut e = expense(&id, T2);
e["amount_ore"] = json!(41300);
e["splits"][0]["share_ore"] = json!(20650);
e["splits"][1]["share_ore"] = json!(20650);
e["notes"] = json!("pant");
e["created_at"] = json!("2026-01-01T00:00:00.000Z");
let entity = accepted(&push_one(&app, &member_token(), exp(e)).await);
assert_eq!(entity["amount_ore"], 41300);
assert_eq!(entity["updated_at"], T2);
assert_eq!(entity["updated_by_user_id"], MEMBER_ID);
assert_eq!(entity["created_by_user_id"], ADMIN_ID);
assert_eq!(entity["recurring_template_id"], UNKNOWN_ID);
assert_eq!(entity["recurring_period"], "2026-10");
assert_eq!(
entity["created_at"], T1,
"created_at keeps the stored value"
);
let audit = audit_of(&app, &id).await;
assert_eq!(audit[1].action, "update");
assert_eq!(audit[1].client_at, T2);
assert_eq!(
audit[1].changes,
json!([
{"field": "amount_ore", "old": 41200, "new": 41300},
{"field": "notes", "old": "", "new": "pant"},
{"field": "splits",
"old": [{"user_id": ADMIN_ID, "share_ore": 20600, "share_bp": 5000}, {"user_id": MEMBER_ID, "share_ore": 20600, "share_bp": 5000}],
"new": [{"user_id": ADMIN_ID, "share_ore": 20650, "share_bp": 5000}, {"user_id": MEMBER_ID, "share_ore": 20650, "share_bp": 5000}]}
])
);
assert_eq!(
pushes(&app)[1],
(
"expense_updated".to_owned(),
Some(id),
MEMBER_ID.to_owned(),
1
)
);
// Settlement update with new allocations replaces the children.
let sid = new_id();
push_one(&app, &admin_token(), set(settlement(&sid, T1))).await;
let mut s = settlement(&sid, T2);
s["allocations"] = json!([{"account_id": ACCOUNT_2, "amount_ore": 50000}]);
let entity = accepted(&push_one(&app, &admin_token(), set(s)).await);
assert_eq!(
entity["allocations"],
json!([{"account_id": ACCOUNT_2, "amount_ore": 50000}])
);
let n: i64 = app
.scalar_i64("SELECT count(*) FROM settlement_allocations")
.await;
assert_eq!(n, 1);
assert_eq!(
audit_of(&app, &sid).await[1].changes[0]["field"],
"allocations"
);
}
#[tokio::test]
async fn step11_empty_diff_changes_nothing_not_even_updated_at() {
let app = app().await;
let id = new_id();
let stored = accepted(&push_one(&app, &admin_token(), exp(expense(&id, T1))).await);
let version = sequence(&app).await;
// Newer updated_at, a different created_at and splits in another order: no diff field differs.
let mut e = expense(&id, T2);
e["splits"].as_array_mut().unwrap().reverse();
e["description"] = json!(" ICA Maxi ");
let r = push_one(&app, &member_token(), exp(e)).await;
assert_eq!(accepted(&r), stored);
assert_eq!(sequence(&app).await, version);
assert_eq!(audit_of(&app, &id).await.len(), 1);
assert_eq!(pushes(&app).len(), 1);
}
// ---- step 12: sync_mutations ------------------------------------------------------------------------------------
#[tokio::test]
async fn step12_every_result_is_recorded_with_device_and_time() {
let app = app().await;
let ok = exp(expense(&new_id(), T1));
let mut bad = exp(expense(&new_id(), T1));
bad["entity"]["amount_ore"] = json!(-5);
let reply = push(&app, &member_token(), vec![ok.clone(), bad.clone()]).await;
let rows: Vec<(String, String, String, String, String, String)> = sqlx::query_as(
"SELECT mutation_id, device_id, entity_type, entity_id, result, received_at FROM sync_mutations ORDER BY rowid",
)
.fetch_all(&app.state.db)
.await
.unwrap();
assert_eq!(rows.len(), 2);
for (row, (m, result)) in rows.iter().zip(
[ok, bad]
.iter()
.zip(reply.json["results"].as_array().unwrap()),
) {
assert_eq!(row.0, m["mutation_id"].as_str().unwrap());
assert_eq!(row.1, MEMBER_DEVICE_ID);
assert_eq!(row.2, "expense");
assert_eq!(row.3, m["entity"]["id"].as_str().unwrap());
assert_eq!(serde_json::from_str::<Value>(&row.4).unwrap(), *result);
assert_eq!(row.5, NOW);
}
}
// ---- pushes after commit, sync_batch -----------------------------------------------------------------------------
#[tokio::test]
async fn more_than_3_notifiable_changes_send_one_sync_batch_3_send_each() {
let app = app().await;
let three: Vec<Value> = (0..3).map(|_| exp(expense(&new_id(), T1))).collect();
push(&app, &admin_token(), three).await;
let p = pushes(&app);
assert_eq!(p.len(), 3);
assert!(p.iter().all(|m| m.0 == "expense_created" && m.3 == 1));
// Four notifiable changes plus a rejection and a no-op (not notifiable): one sync_batch with count 4.
let id = new_id();
push_one(&app, &admin_token(), exp(expense(&id, T1))).await;
let mut bad = expense(&new_id(), T1);
bad["amount_ore"] = json!(0);
let mut batch: Vec<Value> = (0..3).map(|_| exp(expense(&new_id(), T1))).collect();
batch.push(set(settlement(&new_id(), T1)));
batch.push(exp(bad));
batch.push(exp(expense(&id, T2)));
push(&app, &member_token(), batch).await;
let p = pushes(&app);
assert_eq!(p.len(), 5);
assert_eq!(
p[4],
("sync_batch".to_owned(), None, MEMBER_ID.to_owned(), 4)
);
}
// ---- the contract examples ---------------------------------------------------------------------------------------
/// The `contracts/examples/sync/push-*` scenario: the stored rows of the examples, then each request, compared with
/// its response by value (ids, versions and times of rows created here by type only).
#[tokio::test]
async fn push_examples_reproduce_their_responses() {
let app = app().await;
app.set_now("2026-10-17T08:00:01.000Z");
// Fika (exact) and the settlement stored before, as in the pull-page example.
let fika = example("sync/push-request-delete.json")["mutations"][0]["entity"].clone();
let mut stored = fika.clone();
stored["updated_at"] = stored["created_at"].clone();
stored["deleted_at"] = Value::Null;
accepted(&push_one(&app, &member_token(), exp(stored)).await);
let reply = app
.call(
Method::POST,
"/sync",
&admin_token(),
Some(example("sync/push-request-delete.json")),
)
.await;
assert_like_example(
&reply.json,
"sync/push-response-delete.json",
&["server_version"],
);
app.set_now(NOW);
let mut s = example("sync/push-response.json")["results"][1]["current"].clone();
for key in ["created_by_user_id", "updated_by_user_id", "server_version"] {
s.as_object_mut().unwrap().remove(key);
}
accepted(&push_one(&app, &member_token(), set(s)).await);
let reply = app
.call(
Method::POST,
"/sync",
&admin_token(),
Some(example("sync/push-request.json")),
)
.await;
assert_like_example(&reply.json, "sync/push-response.json", &["server_version"]);
tombstone(&app, "accounts", ACCOUNT_2).await;
let reply = app
.call(
Method::POST,
"/sync",
&admin_token(),
Some(example("sync/push-request-rejections.json")),
)
.await;
assert_like_example(
&reply.json,
"sync/push-response-rejections.json",
&["server_version"],
);
}
Reference in New Issue View Git Blame Copy Permalink