jacobh2/kvitt
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity
Files
bfc3acbbe0477449c67a05892940971ac8b62012
kvitt/tools/contracts-check.py
T
jacobh2 6eaccec7ac M12b: category contract v2 on the server (schema 2, fourteen defaults, finer import mapping) 
- migrations/0002_categories_v2.sql, byte-identical to contracts/schema/migrations/0002_categories_v2.sql
  (SPEC §14.1.1), with an identity test; SCHEMA_VERSION = 2 (E-1 reports it, the importer's check follows).
- Seed test on a fresh database (the fourteen rows of AC-7.2, server_version 11 to 14) and an upgrade test of a
  v1 database in use (0001 applied through the migrator, then users, a custom, a renamed and an archived default,
  an account, a recurring template, expenses): every earlier row unchanged, the four rows above the old
  sequence, foreign_key_check empty, schema_version 2, a pull from the old cursor returns exactly the four rows.
- DefaultCategoryKey with the fourteen keys (wire enum and contracts/openapi.yaml); icons 33 and colours 14
  (§9.7) in validation.
- The two category examples M12a held back (E-26 response-default, sync/pull-page-categories-v2) and its
  contracts-check patch for them and the enum.
- Importer: splitwise_categories.csv per the §10.8 v0.8 table (D-053), DEFAULT_KEYS of fourteen, synthetic
  fixtures for the finer mapping (Insurance/Försäkringar to insurance among them), a schema-1 refusal test.
- Test harnesses seed users at server_version 15/16 (after the fourteen seeds); pull counts 10 -> 14.

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

1568 lines
85 KiB
Python
Executable File
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""Kvitt contract check (SPEC §11.2 M1 gate). Standard library only, Python 3.9+.
Checks, one output line each:
* every JSON file under contracts/ parses;
* every test-vector file matches the format of SPEC §14.5.5 and covers the minimum cases;
* a reference implementation of SPEC §9.1, §9.3, §9.4 and §14.5.2/§14.5.4 reproduces every expected value;
* every example under contracts/examples/ matches the wire shapes of SPEC §14.3, §14.4 and §14.6 by key set and type,
plus the money invariants of §9.1 / §9.4 / §14.3.3 (sums, sort orders);
* contracts/schema/server-v1.sql is byte-identical to the DDL in SPEC §14.1, loads in SQLite, has a clean
foreign_key_check and the seed rows;
* contracts/schema/migrations/0002_categories_v2.sql loads on top of server-v1.sql (foreign keys on, one
transaction, as sqlx runs it), on an empty database and on one with rows referring to categories, and leaves the
14 default categories with server_versions from the global sequence and schema_version 2;
* contracts/category-keywords.csv (the keyword table of the category suggestion) is well-formed: header, match
kinds, default keys, keywords already normalised, unique, sorted;
* contracts/openapi.yaml declares exactly the endpoints of SPEC §7 and the error codes of §14.3.2, and the
DefaultCategoryKey enum holds the 14 default keys (text check: the standard library has no YAML parser).
Exit status 0 when every check passes, 1 otherwise.
Usage: python3 tools/contracts-check.py [repository root]
"""
import csv
import datetime
import io
import json
import os
import re
import sqlite3
import sys
import unicodedata
ROOT = os.path.abspath(sys.argv[1]) if len(sys.argv) > 1 else os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
CONTRACTS = os.path.join(ROOT, "contracts")
FAILURES = []
PASSES = [0]
def report(ok, name, detail=""):
if ok:
PASSES[0] += 1
print("PASS %s%s" % (name, (": " + detail) if detail else ""))
else:
FAILURES.append(name)
print("FAIL %s: %s" % (name, detail))
def rel(path):
return os.path.relpath(path, ROOT)
# ----------------------------------------------------------------------------------------------------------------------
# Reference implementation of the pure domain functions (SPEC §9.1, §9.3, §9.4, §14.5.2, §14.5.4)
# ----------------------------------------------------------------------------------------------------------------------
class DomainError(Exception):
def __init__(self, token):
Exception.__init__(self, token)
self.token = token
AMOUNT_MIN, AMOUNT_MAX = 1, 1000000000
BP_TOTAL = 10000
MAX_ALLOCATIONS = 50
def compute_split(amount_ore, payer_user_id, user_ids, spec):
"""§9.1 and §14.5.2. Errors in SplitError enum order."""
mode = spec["mode"]
# 1 amount_out_of_range
if not (AMOUNT_MIN <= amount_ore <= AMOUNT_MAX):
raise DomainError("amount_out_of_range")
# 2 users_invalid
if len(user_ids) != 2 or len(set(user_ids)) != 2:
raise DomainError("users_invalid")
# 3 payer_unknown
if payer_user_id not in user_ids:
raise DomainError("payer_unknown")
if mode == "full":
bearer = spec["bearer_user_id"]
# 4-7 do not apply; 8 bearer_unknown; 9 share_users_mismatch cannot happen
if bearer not in user_ids:
raise DomainError("bearer_unknown")
bp = {u: (BP_TOTAL if u == bearer else 0) for u in user_ids}
return _ratio_rule(amount_ore, payer_user_id, user_ids, bp)
if mode == "ratio":
bp = spec["bp"]
# 4 bp_out_of_range, 5 bp_sum_not_10000
if any(not (0 <= v <= BP_TOTAL) for v in bp.values()):
raise DomainError("bp_out_of_range")
if sum(bp.values()) != BP_TOTAL:
raise DomainError("bp_sum_not_10000")
# 9 share_users_mismatch
if set(bp.keys()) != set(user_ids):
raise DomainError("share_users_mismatch")
return _ratio_rule(amount_ore, payer_user_id, user_ids, bp)
if mode == "exact":
ore = spec["ore"]
# 6 exact_negative, 7 exact_sum_mismatch
if any(v < 0 for v in ore.values()):
raise DomainError("exact_negative")
if sum(ore.values()) != amount_ore:
raise DomainError("exact_sum_mismatch")
if set(ore.keys()) != set(user_ids):
raise DomainError("share_users_mismatch")
return {u: ore[u] for u in user_ids}
raise ValueError("unknown split mode %r" % mode)
def _ratio_rule(amount_ore, payer, user_ids, bp):
non_payer = [u for u in user_ids if u != payer][0]
share_n = (amount_ore * bp[non_payer]) // BP_TOTAL # floor; all operands non-negative
return {non_payer: share_n, payer: amount_ore - share_n}
def compute_balances(user_ids, account_ids, expenses, settlements):
"""§9.3 and §14.5.2. Inputs hold only non-deleted rows."""
per = {}
def acc(a):
if a not in per:
per[a] = {u: 0 for u in user_ids}
return per[a]
for a in account_ids:
acc(a)
for e in expenses:
m = acc(e["account_id"])
m[e["payer_user_id"]] = m.get(e["payer_user_id"], 0) + e["amount_ore"]
for u, s in e["shares"].items():
m[u] = m.get(u, 0) - s
for s in settlements:
for a, amt in s["allocations"].items():
m = acc(a)
m[s["from_user_id"]] = m.get(s["from_user_id"], 0) + amt
m[s["to_user_id"]] = m.get(s["to_user_id"], 0) - amt
glob = {u: 0 for u in user_ids}
for m in per.values():
for u, v in m.items():
glob[u] = glob.get(u, 0) + v
return {"per_account": per, "global": glob}
def _account_order(a):
return (a["created_at"], a["account_id"])
def propose_allocation(amount_ore, accounts):
"""§9.4 algorithm, §14.5.2 semantics. Errors in AllocationError enum order."""
if not (AMOUNT_MIN <= amount_ore <= AMOUNT_MAX):
raise DomainError("amount_out_of_range")
if not accounts:
raise DomainError("no_accounts")
owing = sorted([a for a in accounts if a["payer_net_ore"] < 0], key=_account_order)
allocations = []
remaining = amount_ore
for a in owing:
if remaining <= 0:
break
x = min(remaining, -a["payer_net_ore"])
allocations.append({"account_id": a["account_id"], "amount_ore": x})
remaining -= x
excess = 0
over = []
if owing:
if remaining > 0:
allocations[-1]["amount_ore"] += remaining
excess = remaining
over = [allocations[-1]["account_id"]]
else:
active = sorted([a for a in accounts if not a["archived"]], key=_account_order)
target = active[0] if active else sorted(accounts, key=_account_order)[0]
allocations = [{"account_id": target["account_id"], "amount_ore": amount_ore}]
excess = amount_ore
over = [target["account_id"]]
return {"allocations": allocations, "excess_ore": excess, "over_settled_account_ids": over}
def validate_settlement(amount_ore, from_user_id, to_user_id, user_ids, allocations):
"""§9.4 settlement validity. Errors in SettlementError enum order. Returns None when valid."""
if not (AMOUNT_MIN <= amount_ore <= AMOUNT_MAX):
raise DomainError("amount_out_of_range")
if from_user_id == to_user_id:
raise DomainError("same_user")
if from_user_id not in user_ids or to_user_id not in user_ids:
raise DomainError("user_unknown")
if not allocations:
raise DomainError("no_allocations")
if len(allocations) > MAX_ALLOCATIONS:
raise DomainError("too_many_allocations")
if any(a["amount_ore"] < 1 for a in allocations):
raise DomainError("allocation_not_positive")
ids = [a["account_id"] for a in allocations]
if len(set(ids)) != len(ids):
raise DomainError("duplicate_account")
if sum(a["amount_ore"] for a in allocations) != amount_ore:
raise DomainError("allocation_sum_mismatch")
return None
def over_settled_accounts(allocations, accounts):
"""§9.4 over-settle warning, §14.5.2: unknown account counts as net 0."""
net = {a["account_id"]: a["payer_net_ore"] for a in accounts}
return [a["account_id"] for a in allocations if a["amount_ore"] > max(0, -net.get(a["account_id"], 0))]
def _d(s):
return datetime.date(int(s[0:4]), int(s[5:7]), int(s[8:10]))
def _month_key(d):
return "%04d-%02d" % (d.year, d.month)
def _week_key(d):
y, w, _ = d.isocalendar()
return "%04d-W%02d" % (y, w)
def _month_last(d):
nxt = datetime.date(d.year + (d.month // 12), d.month % 12 + 1, 1)
return nxt - datetime.timedelta(days=1)
def spend_series(expenses, range_, granularity, measure, group_by):
"""§14.5.4 spendSeries."""
start, end = _d(range_["start"]), _d(range_["end_inclusive"])
buckets = []
if granularity == "month":
cur = datetime.date(start.year, start.month, 1)
while cur <= end:
last = _month_last(cur)
buckets.append((_month_key(cur), max(cur, start), min(last, end)))
cur = last + datetime.timedelta(days=1)
keyf = _month_key
elif granularity == "week":
cur = start - datetime.timedelta(days=start.weekday())
while cur <= end:
last = cur + datetime.timedelta(days=6)
buckets.append((_week_key(cur), max(cur, start), min(last, end)))
cur = last + datetime.timedelta(days=1)
keyf = _week_key
else:
raise ValueError("granularity %r" % granularity)
totals = {k: {} for k, _, _ in buckets}
for e in expenses:
d = _d(e["date"])
if d < start or d > end:
continue
if measure == "household":
v = e["amount_ore"]
else:
v = e["shares"].get(measure["share"], 0)
g = e["category_id"] if group_by == "category" else e["account_id"]
t = totals[keyf(d)]
t[g] = t.get(g, 0) + v
out = []
group_totals = {}
for k, s, en in buckets:
t = {g: v for g, v in totals[k].items() if v != 0}
for g, v in t.items():
group_totals[g] = group_totals.get(g, 0) + v
out.append({"key": k, "start": s.isoformat(), "end_inclusive": en.isoformat(), "totals": t, "total": sum(t.values())})
group_totals = {g: v for g, v in group_totals.items() if v != 0}
return {"buckets": out, "group_totals": group_totals, "grand_total": sum(group_totals.values())}
def resolve_range(preset, today, earliest_expense, custom_start, custom_end):
"""§14.5.4 resolveRange."""
t = _d(today)
def months_back(n):
m = t.year * 12 + (t.month - 1) - (n - 1)
return datetime.date(m // 12, m % 12 + 1, 1)
if preset == "this_month":
s, e = datetime.date(t.year, t.month, 1), t
elif preset in ("last_3_months", "last_6_months", "last_12_months"):
s, e = months_back(int(preset.split("_")[1])), t
elif preset == "this_year":
s, e = datetime.date(t.year, 1, 1), t
elif preset == "all_time":
s, e = (_d(earliest_expense) if earliest_expense else t), t
elif preset == "custom":
s, e = _d(custom_start), _d(custom_end)
if s > e:
s, e = e, s
else:
raise ValueError("preset %r" % preset)
return {"start": s.isoformat(), "end_inclusive": e.isoformat()}
# Category suggestion (UX-3, category-suggestion.md CS-4 to CS-6; Kotlin only)
LETTER_CATEGORIES = ("Lu", "Ll", "Lt", "Lm", "Lo")
def normalize_description(s):
"""CS-4: NFC, lowercase without locale rules, delete ' ’ :, every other character that is not a letter, a decimal
digit or & becomes a space, collapse and trim. Works per code point."""
s = unicodedata.normalize("NFC", s).lower()
s = s.replace("'", "").replace("\u2019", "").replace(":", "")
kept = []
for c in s:
cat = unicodedata.category(c)
kept.append(c if (cat in LETTER_CATEGORIES or cat == "Nd" or c == "&") else " ")
return " ".join("".join(kept).split())
def _keyword_matches(entry, tokens):
kw, kind = entry["keyword"], entry["match"]
if kind == "token":
return kw in tokens
if kind == "prefix":
return any(t.startswith(kw) for t in tokens)
if kind == "phrase":
kt = kw.split(" ")
return any(tokens[i:i + len(kt)] == kt for i in range(len(tokens) - len(kt) + 1))
raise ValueError("match %r" % kind)
def _history_pick(rows, active, trace, stage):
"""CS-5 steps 1 and 2: most often among active categories, ties by the latest (expense_date, created_at) of each
tied category. A remaining tie (same count, same latest date and created_at) goes to the lower category id:
TODO(spec), category-suggestion.md does not say."""
stats = {}
for r in rows:
cid = r["category_id"]
if cid not in active:
trace["skipped_inactive"] = True
continue
recency = (r["expense_date"], r["created_at"])
n, latest = stats.get(cid, (0, None))
stats[cid] = (n + 1, recency if latest is None or recency > latest else latest)
if not stats:
return None
ranked = sorted(stats.items(), key=lambda kv: kv[0])
ranked.sort(key=lambda kv: kv[1][1], reverse=True)
ranked.sort(key=lambda kv: kv[1][0], reverse=True)
if len(ranked) > 1 and ranked[0][1][0] == ranked[1][1][0]:
trace["count_tie"] = stage
return ranked[0][0]
def suggest_category_traced(description, history, keywords, active_categories, fallback):
trace = {}
active = set(c["id"] for c in active_categories)
by_key = dict((c["default_key"], c["id"]) for c in active_categories if c["default_key"] is not None)
norm = normalize_description(description)
if not norm:
trace["stage"] = "empty"
return fallback, trace
tokens = norm.split(" ")
normalised = [(normalize_description(r["description"]), r) for r in history]
got = _history_pick([r for n, r in normalised if n == norm], active, trace, "history_exact")
if got is not None:
trace["stage"] = "history_exact"
return got, trace
got = _history_pick([r for n, r in normalised if n and n.split(" ")[0] == tokens[0]], active, trace, "history_first_token")
if got is not None:
trace["stage"] = "history_first_token"
return got, trace
matching = [(i, e) for i, e in enumerate(keywords) if _keyword_matches(e, tokens)]
matching.sort(key=lambda ie: (-len(ie[1]["keyword"]), ie[0]))
if len(matching) > 1 and len(matching[0][1]["keyword"]) == len(matching[1][1]["keyword"]):
trace["equal_length_tie"] = True
for n, (_, e) in enumerate(matching):
cid = by_key.get(e["default_key"])
if cid is not None:
trace["stage"] = "keyword"
trace["match"] = e["match"]
if n > 0:
trace["keyword_fell_through"] = True
return cid, trace
if matching:
trace["keyword_fell_through"] = True
trace["stage"] = "fallback"
return fallback, trace
def suggest_category(description, history, keywords, active_categories, fallback):
"""CS-4 to CS-6 as the pure Kotlin function suggestCategory (UX-3). `history` holds the non-deleted expenses,
`active_categories` the categories that are neither archived nor deleted, `fallback` the AC-8.6 default."""
return suggest_category_traced(description, history, keywords, active_categories, fallback)[0]
# ----------------------------------------------------------------------------------------------------------------------
# Test vectors (§14.5.5)
# ----------------------------------------------------------------------------------------------------------------------
SPLIT_ERRORS = ["amount_out_of_range", "users_invalid", "payer_unknown", "bp_out_of_range", "bp_sum_not_10000",
"exact_negative", "exact_sum_mismatch", "bearer_unknown", "share_users_mismatch"]
ALLOCATION_ERRORS = ["amount_out_of_range", "no_accounts"]
SETTLEMENT_ERRORS = ["amount_out_of_range", "same_user", "user_unknown", "no_allocations", "too_many_allocations",
"allocation_not_positive", "duplicate_account", "allocation_sum_mismatch"]
PRESETS = ["this_month", "last_3_months", "last_6_months", "last_12_months", "this_year", "all_time", "custom"]
VECTOR_FUNCTIONS = {
# function: (input keys = Rust parameter names, or snake_case Kotlin names for the Kotlin-only functions,
# allowed error tokens, reference)
"compute_split": (["amount_ore", "payer_user_id", "user_ids", "spec"], SPLIT_ERRORS,
lambda i: compute_split(i["amount_ore"], i["payer_user_id"], i["user_ids"], i["spec"])),
"compute_balances": (["user_ids", "account_ids", "expenses", "settlements"], [],
lambda i: compute_balances(i["user_ids"], i["account_ids"], i["expenses"], i["settlements"])),
"propose_allocation": (["amount_ore", "accounts"], ALLOCATION_ERRORS,
lambda i: propose_allocation(i["amount_ore"], i["accounts"])),
"validate_settlement": (["amount_ore", "from_user_id", "to_user_id", "user_ids", "allocations"], SETTLEMENT_ERRORS,
lambda i: validate_settlement(i["amount_ore"], i["from_user_id"], i["to_user_id"],
i["user_ids"], i["allocations"])),
"over_settled_accounts": (["allocations", "accounts"], [],
lambda i: over_settled_accounts(i["allocations"], i["accounts"])),
"spend_series": (["expenses", "range", "granularity", "measure", "group_by"], [],
lambda i: spend_series(i["expenses"], i["range"], i["granularity"], i["measure"], i["group_by"])),
"resolve_range": (["preset", "today", "earliest_expense", "custom_start", "custom_end"], [],
lambda i: resolve_range(i["preset"], i["today"], i["earliest_expense"], i["custom_start"],
i["custom_end"])),
"suggest_category": (["description", "history", "keywords", "active_categories", "fallback"], [],
lambda i: suggest_category(i["description"], i["history"], i["keywords"],
i["active_categories"], i["fallback"])),
}
SUGGEST_MIN_CASES = 30
DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
TS_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$")
PERIOD_RE = re.compile(r"^\d{4}-(0[1-9]|1[0-2])$")
WEEK_RE = re.compile(r"^\d{4}-W(0[1-9]|[1-4]\d|5[0-3])$")
UUID_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$")
def is_int(v):
return isinstance(v, int) and not isinstance(v, bool)
def valid_date(s):
if not isinstance(s, str) or not DATE_RE.match(s):
return False
try:
_d(s)
return True
except ValueError:
return False
def valid_ts(s):
if not isinstance(s, str) or not TS_RE.match(s):
return False
try:
datetime.datetime.strptime(s[:19], "%Y-%m-%dT%H:%M:%S")
return True
except ValueError:
return False
def check_vector_input(fn, inp):
"""Type check of a case input against the §14.5.1 types. Returns an error string or None."""
def int_map(m):
return isinstance(m, dict) and all(isinstance(k, str) and is_int(v) for k, v in m.items())
def str_list(x):
return isinstance(x, list) and all(isinstance(v, str) for v in x)
def position(a):
return (isinstance(a, dict) and set(a) == {"account_id", "created_at", "archived", "payer_net_ore"}
and isinstance(a["account_id"], str) and valid_ts(a["created_at"])
and isinstance(a["archived"], bool) and is_int(a["payer_net_ore"]))
def allocation(a):
return isinstance(a, dict) and set(a) == {"account_id", "amount_ore"} and isinstance(a["account_id"], str) and is_int(a["amount_ore"])
if fn == "compute_split":
sp = inp["spec"]
if not (is_int(inp["amount_ore"]) and isinstance(inp["payer_user_id"], str) and str_list(inp["user_ids"])):
return "scalar types"
if not isinstance(sp, dict) or sp.get("mode") not in ("ratio", "exact", "full"):
return "spec.mode must be ratio, exact or full"
want = {"ratio": {"mode", "bp"}, "exact": {"mode", "ore"}, "full": {"mode", "bearer_user_id"}}[sp["mode"]]
if set(sp) != want:
return "spec keys %s, expected %s" % (sorted(sp), sorted(want))
if sp["mode"] == "ratio" and not int_map(sp["bp"]):
return "spec.bp must map user id to int"
if sp["mode"] == "exact" and not int_map(sp["ore"]):
return "spec.ore must map user id to int"
if sp["mode"] == "full" and not isinstance(sp["bearer_user_id"], str):
return "spec.bearer_user_id must be a string"
elif fn == "compute_balances":
if not (str_list(inp["user_ids"]) and str_list(inp["account_ids"])):
return "id lists"
for e in inp["expenses"]:
if not (isinstance(e, dict) and set(e) == {"account_id", "amount_ore", "payer_user_id", "shares"}
and is_int(e["amount_ore"]) and int_map(e["shares"])):
return "expense %r" % (e,)
for s in inp["settlements"]:
if not (isinstance(s, dict) and set(s) == {"from_user_id", "to_user_id", "allocations"} and int_map(s["allocations"])):
return "settlement %r" % (s,)
elif fn == "propose_allocation":
if not is_int(inp["amount_ore"]) or not all(position(a) for a in inp["accounts"]):
return "amount or account positions"
elif fn == "validate_settlement":
if not (is_int(inp["amount_ore"]) and str_list(inp["user_ids"]) and all(allocation(a) for a in inp["allocations"])):
return "amount, user ids or allocations"
elif fn == "over_settled_accounts":
if not (all(allocation(a) for a in inp["allocations"]) and all(position(a) for a in inp["accounts"])):
return "allocations or account positions"
elif fn == "spend_series":
for e in inp["expenses"]:
if not (isinstance(e, dict) and set(e) == {"date", "account_id", "category_id", "amount_ore", "shares"}
and valid_date(e["date"]) and is_int(e["amount_ore"]) and int_map(e["shares"])):
return "expense %r" % (e,)
r = inp["range"]
if not (isinstance(r, dict) and set(r) == {"start", "end_inclusive"} and valid_date(r["start"]) and valid_date(r["end_inclusive"])):
return "range"
if inp["granularity"] not in ("month", "week") or inp["group_by"] not in ("category", "account"):
return "granularity or group_by"
m = inp["measure"]
if not (m == "household" or (isinstance(m, dict) and set(m) == {"share"} and isinstance(m["share"], str))):
return "measure must be \"household\" or {\"share\": id}"
elif fn == "resolve_range":
if inp["preset"] not in PRESETS or not valid_date(inp["today"]):
return "preset or today"
for k in ("earliest_expense", "custom_start", "custom_end"):
if inp[k] is not None and not valid_date(inp[k]):
return k
elif fn == "suggest_category":
if not isinstance(inp["description"], str):
return "description must be a string"
for h in inp["history"]:
if not (isinstance(h, dict) and set(h) == {"description", "category_id", "expense_date", "created_at"}
and isinstance(h["description"], str) and h["description"] and UUID_RE.match(str(h["category_id"]))
and valid_date(h["expense_date"]) and valid_ts(h["created_at"])):
return "history row %r" % (h,)
for k in inp["keywords"]:
if not (isinstance(k, dict) and set(k) == {"keyword", "match", "default_key"}
and isinstance(k["keyword"], str) and k["match"] in KEYWORD_MATCHES and k["default_key"] in DEFAULT_KEYS):
return "keyword %r" % (k,)
for c in inp["active_categories"]:
if not (isinstance(c, dict) and set(c) == {"id", "default_key"} and UUID_RE.match(str(c["id"]))
and (c["default_key"] is None or c["default_key"] in DEFAULT_KEYS)):
return "active category %r" % (c,)
ids = [c["id"] for c in inp["active_categories"]]
keys = [c["default_key"] for c in inp["active_categories"] if c["default_key"] is not None]
if len(set(ids)) != len(ids) or len(set(keys)) != len(keys):
return "active_categories: duplicate id or default_key"
if inp["fallback"] is not None and not UUID_RE.match(str(inp["fallback"])):
return "fallback must be a category id or null"
return None
def check_vector_ok_shape(fn, ok):
if fn == "compute_split":
return isinstance(ok, dict) and all(is_int(v) for v in ok.values())
if fn == "compute_balances":
return isinstance(ok, dict) and set(ok) == {"per_account", "global"}
if fn == "propose_allocation":
return (isinstance(ok, dict) and set(ok) == {"allocations", "excess_ore", "over_settled_account_ids"}
and all(isinstance(a, dict) and set(a) == {"account_id", "amount_ore"} for a in ok["allocations"]))
if fn == "validate_settlement":
return ok is None
if fn == "over_settled_accounts":
return isinstance(ok, list)
if fn == "spend_series":
return (isinstance(ok, dict) and set(ok) == {"buckets", "group_totals", "grand_total"}
and all(isinstance(b, dict) and set(b) == {"key", "start", "end_inclusive", "totals", "total"} for b in ok["buckets"]))
if fn == "resolve_range":
return isinstance(ok, dict) and set(ok) == {"start", "end_inclusive"}
if fn == "suggest_category":
return ok is None or (isinstance(ok, str) and bool(UUID_RE.match(ok)))
return False
def coverage(fn, cases):
"""Minimum cases of §14.5.5 that can be checked mechanically. Returns a list of missing items."""
missing = []
errors = set(c["expected"].get("error") for c in cases if "error" in c["expected"])
if fn == "compute_split":
missing += ["error " + e for e in SPLIT_ERRORS if e not in errors]
rows = [(1001, "a", 501, 500), (1, "a", 1, 0), (100, "b", 40, 60), (999, "a", 333, 666), (999, "b", 667, 332),
(50000, "a", 0, 50000), (50000, "a", 50000, 0)]
for amt, payer, sp, sn in rows:
hit = any(c["input"]["amount_ore"] == amt and c["input"]["payer_user_id"] == payer
and c["expected"].get("ok") is not None and c["expected"]["ok"].get(payer) == sp
and sorted(c["expected"]["ok"].values()) == sorted([sp, sn]) for c in cases)
if not hit:
missing.append("§9.1 row %d/%s" % (amt, payer))
modes = set((c["input"]["spec"]["mode"], "ok" if "ok" in c["expected"] else "err") for c in cases)
for m in [("full", "ok"), ("exact", "ok")]:
if m not in modes:
missing.append("%s %s" % m)
if not any(c["input"]["spec"]["mode"] == "full" and c["input"]["spec"]["bearer_user_id"] == c["input"]["payer_user_id"]
and "ok" in c["expected"] for c in cases):
missing.append("full on payer")
if not any(c["input"]["spec"]["mode"] == "full" and c["input"]["spec"]["bearer_user_id"] != c["input"]["payer_user_id"]
and "ok" in c["expected"] for c in cases):
missing.append("full on non-payer")
amounts = set(c["input"]["amount_ore"] for c in cases)
for a in (0, 1000000001, 1000000000):
if a not in amounts:
missing.append("amount %d" % a)
elif fn == "propose_allocation":
missing += ["error " + e for e in ALLOCATION_ERRORS if e not in errors]
oks = [c["expected"]["ok"] for c in cases if "ok" in c["expected"]]
if not any(o["excess_ore"] > 0 and len(o["allocations"]) > 1 for o in oks):
missing.append("excess added to the last owing account")
if not any(o["excess_ore"] == 0 and o["allocations"] for o in oks):
missing.append("no excess")
elif fn == "validate_settlement":
missing += ["error " + e for e in SETTLEMENT_ERRORS if e not in errors]
valid_sizes = set(len(c["input"]["allocations"]) for c in cases if "ok" in c["expected"])
for n in (3, 50):
if n not in valid_sizes:
missing.append("valid case with %d allocations" % n)
elif fn == "over_settled_accounts":
oks = [c["expected"]["ok"] for c in cases]
if [] not in oks:
missing.append("none")
if not any(len(o) == 1 for o in oks):
missing.append("one")
elif fn == "spend_series":
seen = set()
for c in cases:
i = c["input"]
seen.add(i["granularity"])
seen.add("household" if i["measure"] == "household" else "share")
seen.add(i["group_by"])
for b in c["expected"]["ok"]["buckets"]:
seen.add(b["key"])
if b["total"] == 0:
seen.add("empty bucket")
for want in ("month", "week", "household", "share", "category", "account", "2026-W53", "empty bucket"):
if want not in seen:
missing.append(want)
elif fn == "resolve_range":
presets = set(c["input"]["preset"] for c in cases)
missing += ["preset " + p for p in PRESETS if p not in presets]
if not any(c["input"]["preset"] == "all_time" and c["input"]["earliest_expense"] is None for c in cases):
missing.append("all_time without earliest expense")
if not any(c["input"]["preset"] == "all_time" and c["input"]["earliest_expense"] is not None for c in cases):
missing.append("all_time with earliest expense")
if not any(c["input"]["preset"] == "custom" and c["input"]["custom_start"] > c["input"]["custom_end"] for c in cases):
missing.append("custom swapped")
if not all(any(c["input"]["preset"] == p and c["input"]["today"] == "2026-10-17" for c in cases) for p in PRESETS if p != "custom"):
missing.append("every preset with today 2026-10-17")
elif fn == "suggest_category":
if len(cases) < SUGGEST_MIN_CASES:
missing.append("at least %d cases (has %d)" % (SUGGEST_MIN_CASES, len(cases)))
seen = set()
for c in cases:
i = c["input"]
_, t = suggest_category_traced(i["description"], i["history"], i["keywords"], i["active_categories"], i["fallback"])
seen.add(t["stage"])
if i["description"] == "":
seen.add("empty description")
if t["stage"] == "keyword":
seen.add("keyword " + t["match"])
for k in ("skipped_inactive", "equal_length_tie", "keyword_fell_through"):
if t.get(k):
seen.add(k)
if t.get("count_tie"):
seen.add("tie " + t["count_tie"])
if c["expected"]["ok"] is None:
seen.add("null result")
if normalize_description(i["description"]) != " ".join(i["description"].split()).lower():
seen.add("normalisation changes the description")
for want in ("empty description", "empty", "history_exact", "history_first_token", "keyword", "fallback",
"keyword token", "keyword prefix", "keyword phrase", "skipped_inactive", "equal_length_tie",
"keyword_fell_through", "tie history_exact", "null result", "normalisation changes the description"):
if want not in seen:
missing.append(want)
return missing
def check_vectors():
d = os.path.join(CONTRACTS, "test-vectors")
expected_files = set(f + ".json" for f in VECTOR_FUNCTIONS)
present = set(f for f in os.listdir(d) if f.endswith(".json")) if os.path.isdir(d) else set()
report(present == expected_files, "vectors: file set is exactly §14.5.5",
"missing %s, extra %s" % (sorted(expected_files - present), sorted(present - expected_files)) if present != expected_files else "%d files" % len(present))
counts = {}
for fn in sorted(VECTOR_FUNCTIONS):
path = os.path.join(d, fn + ".json")
if not os.path.exists(path):
continue
try:
with open(path, encoding="utf-8") as f:
doc = json.load(f)
except ValueError as e:
report(False, "vectors %s format" % fn, "does not parse: %s" % e)
continue
keys, errs, ref = VECTOR_FUNCTIONS[fn]
problems = []
if not isinstance(doc, dict) or set(doc) != {"format", "function", "cases"}:
problems.append("top-level keys must be format, function, cases")
else:
if doc["format"] != 1:
problems.append("format must be 1")
if doc["function"] != fn:
problems.append("function %r does not match file name" % doc["function"])
if not isinstance(doc["cases"], list) or not doc["cases"]:
problems.append("cases must be a non-empty list")
cases = doc.get("cases") if isinstance(doc, dict) and isinstance(doc.get("cases"), list) else []
names = set()
for n, c in enumerate(cases):
where = "case %d" % n
if not isinstance(c, dict) or set(c) != {"name", "input", "expected"}:
problems.append("%s: keys must be name, input, expected" % where)
continue
if not isinstance(c["name"], str) or not c["name"]:
problems.append("%s: name must be a non-empty string" % where)
elif c["name"] in names:
problems.append("%s: duplicate name %r" % (where, c["name"]))
names.add(c["name"])
if not isinstance(c["input"], dict) or set(c["input"]) != set(keys):
problems.append("%s (%s): input keys %s, expected %s" % (where, c["name"], sorted(c["input"]) if isinstance(c["input"], dict) else "?", sorted(keys)))
continue
bad = check_vector_input(fn, c["input"])
if bad:
problems.append("%s (%s): input %s" % (where, c["name"], bad))
exp = c["expected"]
if not isinstance(exp, dict) or len(exp) != 1 or list(exp)[0] not in ("ok", "error"):
problems.append("%s (%s): expected must be {\"ok\": …} or {\"error\": …}" % (where, c["name"]))
continue
if "error" in exp and exp["error"] not in errs:
problems.append("%s (%s): error token %r not allowed for %s" % (where, c["name"], exp["error"], fn))
if "ok" in exp and not check_vector_ok_shape(fn, exp["ok"]):
problems.append("%s (%s): ok value has the wrong shape" % (where, c["name"]))
report(not problems, "vectors %s.json format (§14.5.5)" % fn, "; ".join(problems) if problems else "%d cases" % len(cases))
if problems:
continue
missing = coverage(fn, cases)
report(not missing, "vectors %s.json minimum cases (§14.5.5)" % fn, "missing: " + ", ".join(missing) if missing else "covered")
mismatches = []
for c in cases:
try:
got = {"ok": ref(c["input"])}
except DomainError as e:
got = {"error": e.token}
if got != c["expected"]:
mismatches.append("%r: expected %s, reference gives %s" % (c["name"], json.dumps(c["expected"], ensure_ascii=False), json.dumps(got, ensure_ascii=False)))
report(not mismatches, "vectors %s.json reproduced by the reference implementation" % fn,
" | ".join(mismatches) if mismatches else "%d/%d cases" % (len(cases), len(cases)))
counts[fn] = len(cases)
return counts
# ----------------------------------------------------------------------------------------------------------------------
# Wire shapes (§14.3.3, §14.3.4, §14.4.2, §14.4.3, §14.6)
# ----------------------------------------------------------------------------------------------------------------------
# A type is: "uuid" | "string" | "int" | "bool" | "ts" | "date" | "period" | "any" | "hex32"
# | ("enum", [values]) | ("null", type) | ("array", type, min, max) | "<ShapeName>"
# A shape is {"fields": {key: type}, "optional": set of keys that may be absent, "patch": bool (all optional, ≥ 1)}
def N(t):
return ("null", t)
def A(t, lo=0, hi=None):
return ("array", t, lo, hi)
def E(*vals):
return ("enum", list(vals))
ROLE = E("admin", "member")
INVITE_KIND = E("admin_setup", "partner", "device")
DEFAULT_KEYS = ["groceries", "restaurants", "home", "utilities", "transport", "travel", "entertainment", "health",
"shopping", "other", "housing", "insurance", "children", "fees"]
DEFAULT_KEY = E(*DEFAULT_KEYS)
KEYWORD_MATCHES = ["token", "prefix", "phrase"]
SPLIT_MODE = E("default", "ratio", "exact", "full")
ENTITY_TYPE = E("expense", "settlement")
REASON = E("stale", "deleted", "validation_failed", "reference_not_found", "clock_skew")
ERROR_CODES = ["invalid_request", "invite_invalid", "unauthorized", "device_revoked", "forbidden", "not_found",
"household_full", "household_incomplete", "name_taken", "account_not_empty", "category_in_use",
"category_is_default", "cannot_revoke_current_device", "no_accounts", "cursor_ahead", "payload_too_large",
"unsupported_media_type", "validation_failed", "image_invalid", "batch_too_large", "protocol_unsupported",
"rate_limited", "internal", "not_implemented", "unavailable"]
PUSH_TYPES = ["expense_created", "expense_updated", "expense_deleted", "settlement_created", "settlement_updated",
"settlement_deleted", "recurring_expense_created", "sync_batch"]
ICONS = ["shopping_cart", "restaurant", "local_cafe", "home", "bolt", "water_drop", "wifi", "directions_car", "train",
"local_gas_station", "flight", "hotel", "luggage", "movie", "music_note", "sports_esports", "fitness_center",
"medical_services", "medication", "shopping_bag", "checkroom", "pets", "child_care", "school", "card_giftcard",
"celebration", "build", "chair", "subscriptions", "category", "key", "shield", "account_balance"]
COLORS = ["green", "orange", "brown", "yellow", "blue", "teal", "purple", "red", "pink", "grey", "indigo", "lime",
"slate", "amber"]
def S(fields, optional=(), patch=False):
return {"fields": fields, "optional": set(optional), "patch": patch}
TEMPLATE_WRITABLE = {
"account_id": "uuid", "category_id": "uuid", "amount_ore": "int", "description": "string", "notes": "string",
"payer_user_id": "uuid", "split_mode": SPLIT_MODE, "splits": A("TemplateSplit", 0, 2), "day_of_month": "int",
"start_period": "period", "end_period": N("period"), "paused": "bool",
}
SHAPES = {
# §14.3.3 wire objects
"User": S({"id": "uuid", "role": ROLE, "display_name": "string", "picture_version": N("int"), "created_at": "ts",
"updated_at": "ts", "server_version": "int"}),
"Device": S({"id": "uuid", "user_id": "uuid", "name": "string", "app_version_code": N("int"), "created_at": "ts",
"last_seen_at": "ts", "revoked_at": N("ts"), "is_current": "bool"}),
"Invite": S({"id": "uuid", "kind": INVITE_KIND, "target_user_id": N("uuid"), "code": N("string"), "link": N("string"),
"created_at": "ts", "expires_at": "ts"}),
"Share": S({"user_id": "uuid", "share_bp": "int"}),
"Account": S({"id": "uuid", "name": "string", "default_shares": A("Share", 2, 2), "created_by_user_id": "uuid",
"created_at": "ts", "updated_at": "ts", "archived_at": N("ts"), "deleted_at": N("ts"), "server_version": "int"}),
"Category": S({"id": "uuid", "default_key": N(DEFAULT_KEY), "name": N("string"), "icon": "string", "color": "string",
"sort_order": "int", "created_at": "ts", "updated_at": "ts", "archived_at": N("ts"), "deleted_at": N("ts"),
"server_version": "int"}),
"Split": S({"user_id": "uuid", "share_ore": "int", "share_bp": N("int")}),
"TemplateSplit": S({"user_id": "uuid", "share_bp": N("int"), "share_ore": N("int")}),
"Expense": S({"id": "uuid", "account_id": "uuid", "category_id": "uuid", "amount_ore": "int", "description": "string",
"notes": "string", "expense_date": "date", "payer_user_id": "uuid", "split_mode": SPLIT_MODE,
"splits": A("Split", 2, 2), "recurring_template_id": N("uuid"), "recurring_period": N("period"),
"created_by_user_id": "uuid", "updated_by_user_id": "uuid", "created_at": "ts", "updated_at": "ts",
"deleted_at": N("ts"), "server_version": "int"}),
"Allocation": S({"account_id": "uuid", "amount_ore": "int"}),
"Settlement": S({"id": "uuid", "from_user_id": "uuid", "to_user_id": "uuid", "amount_ore": "int",
"settlement_date": "date", "note": "string", "allocations": A("Allocation", 1, 50),
"created_by_user_id": "uuid", "updated_by_user_id": "uuid", "created_at": "ts", "updated_at": "ts",
"deleted_at": N("ts"), "server_version": "int"}),
"RecurringTemplate": S(dict(TEMPLATE_WRITABLE, **{
"id": "uuid", "last_generated_period": N("period"), "next_occurrence_date": N("date"),
"created_by_user_id": "uuid", "updated_by_user_id": "uuid", "created_at": "ts", "updated_at": "ts",
"deleted_at": N("ts"), "server_version": "int"})),
"AuditChange": S({"field": "string", "old": "any", "new": "any"}),
"AuditEntry": S({"id": "uuid", "entity_type": ENTITY_TYPE, "entity_id": "uuid", "action": E("create", "update", "delete"),
"source": E("app", "recurring", "import"), "actor_user_id": "uuid", "client_at": "ts", "server_at": "ts",
"changes": A("AuditChange"), "server_version": "int"}),
"Error": S({"code": E(*ERROR_CODES), "message": "string", "field": N("string")}),
# §14.3.4 request and response bodies
"Health": S({"status": E("ok", "degraded"), "server_version": "string", "schema_version": "int", "db_ok": "bool"}),
"SetupStatus": S({"initialized": "bool"}),
"AcceptInviteRequest": S({"code": "string", "device_name": "string", "display_name": "string"}, optional=["display_name"]),
"AcceptInviteResponse": S({"token": "string", "user": "User", "device": "Device", "household": A("User", 1, 2)}),
"ApkLatest": S({"version_code": "int", "version_name": "string", "file_name": "string", "size_bytes": "int",
"sha256": "string", "released_at": "ts"}),
"MeResponse": S({"user": "User", "device": "Device", "household": A("User", 1, 2), "public_url": "string"}),
"PatchMeRequest": S({"display_name": "string"}, patch=True),
"CreateInviteRequest": S({"kind": E("partner", "device"), "target_user_id": N("uuid")}, optional=["target_user_id"]),
"InviteList": S({"invites": A("Invite")}),
"DeviceList": S({"devices": A("Device")}),
"PushTokenRequest": S({"fcm_token": N("string")}),
"CreateAccountRequest": S({"name": "string", "default_shares": A("Share", 2, 2)}),
"PatchAccountRequest": S({"name": "string", "default_shares": A("Share", 2, 2)}, patch=True),
"CreateCategoryRequest": S({"name": "string", "icon": "string", "color": "string"}),
"PatchCategoryRequest": S({"name": N("string"), "icon": "string", "color": "string", "sort_order": "int"}, patch=True),
"CreateRecurringTemplateRequest": S(dict(TEMPLATE_WRITABLE)),
"PatchRecurringTemplateRequest": S(dict(TEMPLATE_WRITABLE), patch=True),
"ProposalRequest": S({"payer_user_id": "uuid", "amount_ore": "int", "exclude_settlement_id": N("uuid")},
optional=["exclude_settlement_id"]),
"ProposalResponse": S({"allocations": A("Allocation"), "excess_ore": "int", "over_settled_account_ids": A("uuid")}),
# §14.4.2 and §14.4.3 sync
"SyncPullResponse": S({"epoch": "hex32", "since": "int", "next_since": "int", "has_more": "bool", "server_time": "ts",
"users": A("User"), "categories": A("Category"), "accounts": A("Account"),
"recurring_templates": A("RecurringTemplate"), "expenses": A("Expense"),
"settlements": A("Settlement"), "audit_entries": A("AuditEntry")}),
"PushedExpense": S({"id": "uuid", "account_id": "uuid", "category_id": "uuid", "amount_ore": "int", "description": "string",
"notes": "string", "expense_date": "date", "payer_user_id": "uuid", "split_mode": SPLIT_MODE,
"splits": A("Split", 2, 2), "created_at": "ts", "updated_at": "ts", "deleted_at": N("ts")}),
"PushedSettlement": S({"id": "uuid", "from_user_id": "uuid", "to_user_id": "uuid", "amount_ore": "int",
"settlement_date": "date", "note": "string", "allocations": A("Allocation", 1, 50),
"created_at": "ts", "updated_at": "ts", "deleted_at": N("ts")}),
"ExpenseMutation": S({"mutation_id": "uuid", "entity_type": E("expense"), "entity": "PushedExpense"}),
"SettlementMutation": S({"mutation_id": "uuid", "entity_type": E("settlement"), "entity": "PushedSettlement"}),
"SyncPushRequest": S({"mutations": A("Mutation", 1, 100)}),
"AcceptedExpense": S({"mutation_id": "uuid", "entity_type": E("expense"), "entity_id": "uuid", "status": E("accepted"),
"server_version": "int", "entity": "Expense"}),
"AcceptedSettlement": S({"mutation_id": "uuid", "entity_type": E("settlement"), "entity_id": "uuid", "status": E("accepted"),
"server_version": "int", "entity": "Settlement"}),
"RejectedExpense": S({"mutation_id": "uuid", "entity_type": E("expense"), "entity_id": "uuid", "status": E("rejected"),
"reason": REASON, "message": "string", "current": N("Expense")}),
"RejectedSettlement": S({"mutation_id": "uuid", "entity_type": E("settlement"), "entity_id": "uuid", "status": E("rejected"),
"reason": REASON, "message": "string", "current": N("Settlement")}),
"SyncPushResponse": S({"server_time": "ts", "results": A("MutationResult", 1, 100)}),
# §14.6 push payload
"FcmMessage": S({"message": "FcmBody"}),
"FcmBody": S({"token": "string", "android": "FcmAndroid", "data": "FcmData"}),
"FcmAndroid": S({"priority": E("high"), "ttl": E("86400s")}),
"FcmData": S({"v": E("1"), "type": E(*PUSH_TYPES), "entity_type": E("expense", "settlement", ""), "entity_id": "string",
"actor_user_id": "uuid", "count": "string"}),
}
UNIONS = {
"Mutation": lambda o: {"expense": "ExpenseMutation", "settlement": "SettlementMutation"}.get(o.get("entity_type")),
"MutationResult": lambda o: {("expense", "accepted"): "AcceptedExpense", ("settlement", "accepted"): "AcceptedSettlement",
("expense", "rejected"): "RejectedExpense", ("settlement", "rejected"): "RejectedSettlement"
}.get((o.get("entity_type"), o.get("status"))),
}
def validate(value, typ, path, errs, seen):
"""Validate value against typ; append messages to errs; record (shape, object) pairs in seen."""
if isinstance(typ, tuple):
kind = typ[0]
if kind == "null":
if value is not None:
validate(value, typ[1], path, errs, seen)
return
if kind == "enum":
if value not in typ[1]:
errs.append("%s: %r is not one of %s" % (path, value, typ[1]))
return
if kind == "array":
if not isinstance(value, list):
errs.append("%s: expected an array" % path)
return
lo, hi = typ[2], typ[3]
if len(value) < lo or (hi is not None and len(value) > hi):
errs.append("%s: array length %d outside %s..%s" % (path, len(value), lo, hi))
for i, v in enumerate(value):
validate(v, typ[1], "%s[%d]" % (path, i), errs, seen)
return
if typ in UNIONS:
if not isinstance(value, dict):
errs.append("%s: expected an object" % path)
return
member = UNIONS[typ](value)
if member is None:
errs.append("%s: no %s variant for this object" % (path, typ))
return
validate(value, member, path, errs, seen)
return
if typ in SHAPES:
shape = SHAPES[typ]
if not isinstance(value, dict):
errs.append("%s: expected a %s object" % (path, typ))
return
keys, fields = set(value), set(shape["fields"])
if shape["patch"]:
if not keys or not keys <= fields:
errs.append("%s: %s keys %s must be a non-empty subset of %s" % (path, typ, sorted(keys), sorted(fields)))
else:
required = fields - shape["optional"]
if not (required <= keys <= fields):
errs.append("%s: %s keys differ: missing %s, unknown %s" % (path, typ, sorted(required - keys), sorted(keys - fields)))
for k in keys & fields:
validate(value[k], shape["fields"][k], "%s.%s" % (path, k), errs, seen)
seen.append((typ, value, path))
return
ok = {
"uuid": lambda v: isinstance(v, str) and bool(UUID_RE.match(v)),
"string": lambda v: isinstance(v, str),
"int": is_int,
"bool": lambda v: isinstance(v, bool),
"ts": valid_ts,
"date": valid_date,
"period": lambda v: isinstance(v, str) and bool(PERIOD_RE.match(v)),
"hex32": lambda v: isinstance(v, str) and bool(re.match(r"^[0-9a-f]{32}$", v)),
"any": lambda v: True,
}.get(typ)
if ok is None:
errs.append("%s: unknown type %r in the checker" % (path, typ))
elif not ok(value):
errs.append("%s: %r is not a valid %s" % (path, value, typ))
def semantic(shape, o, path, errs):
"""Invariants of §9.1, §9.4, §9.6, §9.7, §14.3.3, §14.6 on objects that passed the shape check."""
def sorted_by(lst, key):
ids = [x[key] for x in lst]
return ids == sorted(ids)
if shape in ("Expense", "PushedExpense"):
sp = o["splits"]
if not sorted_by(sp, "user_id"):
errs.append("%s.splits: not sorted by user_id" % path)
if sum(s["share_ore"] for s in sp) != o["amount_ore"] and shape == "Expense":
errs.append("%s.splits: shares do not sum to amount_ore" % path)
if shape == "Expense" and len(sp) == 2:
users = [s["user_id"] for s in sp]
try:
if o["split_mode"] == "exact":
if any(s["share_bp"] is not None for s in sp):
errs.append("%s.splits: share_bp must be null for mode exact" % path)
spec = {"mode": "exact", "ore": {s["user_id"]: s["share_ore"] for s in sp}}
else:
spec = {"mode": "ratio", "bp": {s["user_id"]: s["share_bp"] for s in sp}}
if any(s["share_bp"] is None for s in sp):
raise DomainError("share_bp null in mode %s" % o["split_mode"])
if o["split_mode"] == "full" and sorted(s["share_bp"] for s in sp) != [0, 10000]:
errs.append("%s.splits: mode full needs bp 10000 and 0" % path)
want = compute_split(o["amount_ore"], o["payer_user_id"], users, spec)
got = {s["user_id"]: s["share_ore"] for s in sp}
if want != got:
errs.append("%s.splits: share_ore %s, §9.1 gives %s" % (path, got, want))
except DomainError as e:
errs.append("%s.splits: %s" % (path, e.token))
if shape == "Expense" and (o["recurring_template_id"] is None) != (o["recurring_period"] is None):
errs.append("%s: recurring_template_id and recurring_period must be both null or both set" % path)
if shape in ("Settlement", "PushedSettlement"):
al = o["allocations"]
if not sorted_by(al, "account_id"):
errs.append("%s.allocations: not sorted by account_id" % path)
try:
validate_settlement(o["amount_ore"], o["from_user_id"], o["to_user_id"], [o["from_user_id"], o["to_user_id"]], al)
except DomainError as e:
errs.append("%s: %s" % (path, e.token))
if shape in ("Account", "CreateAccountRequest", "PatchAccountRequest") and "default_shares" in o:
ds = o["default_shares"]
if not sorted_by(ds, "user_id"):
errs.append("%s.default_shares: not sorted by user_id" % path)
if sum(s["share_bp"] for s in ds) != 10000 or any(not 0 <= s["share_bp"] <= 10000 for s in ds):
errs.append("%s.default_shares: bp must be 0..10000 and sum to 10000" % path)
if shape in ("Category", "CreateCategoryRequest", "PatchCategoryRequest"):
if "icon" in o and o["icon"] not in ICONS:
errs.append("%s.icon: %r is not a §9.7 icon token" % (path, o["icon"]))
if "color" in o and o["color"] not in COLORS:
errs.append("%s.color: %r is not a §9.7 colour token" % (path, o["color"]))
if shape == "Category" and o["name"] is None and o["default_key"] is None:
errs.append("%s: name and default_key cannot both be null" % path)
if shape in ("RecurringTemplate", "CreateRecurringTemplateRequest", "PatchRecurringTemplateRequest") and "splits" in o:
mode, sp = o.get("split_mode"), o["splits"]
if not sorted_by(sp, "user_id"):
errs.append("%s.splits: not sorted by user_id" % path)
for s in sp:
if (s["share_bp"] is None) == (s["share_ore"] is None):
errs.append("%s.splits: exactly one of share_bp / share_ore must be set" % path)
if mode == "default" and sp:
errs.append("%s.splits: must be empty for mode default" % path)
if mode in ("ratio", "full") and (len(sp) != 2 or any(s["share_bp"] is None for s in sp) or sum(s["share_bp"] or 0 for s in sp) != 10000):
errs.append("%s.splits: two bp entries summing to 10000 required for mode %s" % (path, mode))
if mode == "exact" and (len(sp) != 2 or any(s["share_ore"] is None for s in sp) or sum(s["share_ore"] or 0 for s in sp) != o.get("amount_ore")):
errs.append("%s.splits: two öre entries summing to the amount required for mode exact" % path)
if shape in ("AcceptedExpense", "AcceptedSettlement"):
if o["entity"]["id"] != o["entity_id"] or o["entity"]["server_version"] != o["server_version"]:
errs.append("%s: entity id / server_version do not match the result" % path)
if shape in ("RejectedExpense", "RejectedSettlement") and o["current"] is not None and o["current"]["id"] != o["entity_id"]:
errs.append("%s: current.id does not match entity_id" % path)
if shape == "SyncPullResponse":
for k in ("users", "categories", "accounts", "recurring_templates", "expenses", "settlements", "audit_entries"):
if not sorted_by(o[k], "server_version"):
errs.append("%s.%s: not sorted by server_version" % (path, k))
versions = [r["server_version"] for k in ("users", "categories", "accounts", "recurring_templates", "expenses",
"settlements", "audit_entries") for r in o[k]]
if versions and max(versions) != o["next_since"]:
errs.append("%s: next_since must be the highest server_version in the page" % path)
if not versions and o["next_since"] != o["since"]:
errs.append("%s: next_since must equal since for an empty page" % path)
if any(v <= o["since"] for v in versions):
errs.append("%s: rows must have server_version > since" % path)
if shape == "AcceptInviteResponse" and not re.match(r"^kvt_[A-Za-z0-9_-]{43}$", o["token"]):
errs.append("%s.token: must be kvt_ + 43 base64url characters (§8.2)" % path)
if shape == "Invite" and o["code"] is not None and not re.match(r"^[0-9ABCDEFGHJKMNPQRSTVWXYZ]{12}$", o["code"]):
errs.append("%s.code: must be 12 Crockford base32 characters (§8.3)" % path)
if shape == "FcmData":
if o["type"] == "sync_batch":
if o["entity_type"] != "" or o["entity_id"] != "" or not re.match(r"^[1-9][0-9]*$", o["count"]):
errs.append("%s: sync_batch needs empty entity_type and entity_id and a positive count" % path)
else:
want = "settlement" if o["type"].startswith("settlement") else "expense"
if o["entity_type"] != want or not UUID_RE.match(o["entity_id"]) or o["count"] != "1":
errs.append("%s: %s needs entity_type %s, a uuid entity_id and count \"1\"" % (path, o["type"], want))
EXAMPLES = {
"E-1-health/response.json": "Health",
"E-2-setup-status/response.json": "SetupStatus",
"E-3-invites-accept/request.json": "AcceptInviteRequest",
"E-3-invites-accept/response.json": "AcceptInviteResponse",
"E-5-apk-latest/response.json": "ApkLatest",
"E-7-me/response.json": "MeResponse",
"E-8-patch-me/request.json": "PatchMeRequest",
"E-8-patch-me/response.json": "User",
"E-9-put-me-picture/response.json": "User",
"E-10-delete-me-picture/response.json": "User",
"E-12-create-invite/request.json": "CreateInviteRequest",
"E-12-create-invite/response.json": "Invite",
"E-13-list-invites/response.json": "InviteList",
"E-15-list-devices/response.json": "DeviceList",
"E-17-push-token/request.json": "PushTokenRequest",
"E-18-create-account/request.json": "CreateAccountRequest",
"E-18-create-account/response.json": "Account",
"E-19-patch-account/request.json": "PatchAccountRequest",
"E-19-patch-account/response.json": "Account",
"E-20-archive-account/response.json": "Account",
"E-21-unarchive-account/response.json": "Account",
"E-22-delete-account/response.json": "Account",
"E-23-create-category/request.json": "CreateCategoryRequest",
"E-23-create-category/response.json": "Category",
"E-24-patch-category/request.json": "PatchCategoryRequest",
"E-24-patch-category/response.json": "Category",
"E-25-archive-category/response.json": "Category",
"E-26-unarchive-category/response.json": "Category",
"E-26-unarchive-category/response-default.json": "Category",
"E-27-delete-category/response.json": "Category",
"E-28-create-recurring-template/request.json": "CreateRecurringTemplateRequest",
"E-28-create-recurring-template/response.json": "RecurringTemplate",
"E-29-patch-recurring-template/request.json": "PatchRecurringTemplateRequest",
"E-29-patch-recurring-template/response.json": "RecurringTemplate",
"E-30-delete-recurring-template/response.json": "RecurringTemplate",
"E-31-sync-pull/response.json": "SyncPullResponse",
"E-32-sync-push/request.json": "SyncPushRequest",
"E-32-sync-push/response.json": "SyncPushResponse",
"E-33-settlement-proposal/request.json": "ProposalRequest",
"E-33-settlement-proposal/response.json": "ProposalResponse",
"sync/pull-page.json": "SyncPullResponse",
"sync/pull-empty-page.json": "SyncPullResponse",
"sync/pull-page-categories-v2.json": "SyncPullResponse",
"sync/push-request.json": "SyncPushRequest",
"sync/push-response.json": "SyncPushResponse",
"sync/push-request-delete.json": "SyncPushRequest",
"sync/push-response-delete.json": "SyncPushResponse",
"sync/push-request-rejections.json": "SyncPushRequest",
"sync/push-response-rejections.json": "SyncPushResponse",
}
EXAMPLES.update({"errors/%s.json" % c: "Error" for c in ERROR_CODES})
EXAMPLES.update({"push/%s.json" % t: "FcmMessage" for t in PUSH_TYPES})
def check_examples():
d = os.path.join(CONTRACTS, "examples")
present = set()
for dirpath, _, files in os.walk(d):
for f in files:
present.add(os.path.relpath(os.path.join(dirpath, f), d).replace(os.sep, "/"))
missing, extra = sorted(set(EXAMPLES) - present), sorted(present - set(EXAMPLES))
report(not missing and not extra, "examples: file set (one per JSON body of §14.3.4, error code, sync and push payload)",
"missing %s, unmapped %s" % (missing, extra) if (missing or extra) else "%d files" % len(present))
seen_reasons, seen_status = set(), set()
def order(s):
m = re.match(r"^E-(\d+)-", s)
return (0, int(m.group(1)), s) if m else (1, 0, s)
for name in sorted(set(EXAMPLES) & present, key=order):
shape = EXAMPLES[name]
try:
with open(os.path.join(d, name), encoding="utf-8") as f:
doc = json.load(f)
except ValueError as e:
report(False, "example %s" % name, "does not parse: %s" % e)
continue
errs, seen = [], []
validate(doc, shape, "$", errs, seen)
if not errs:
for sh, o, p in seen:
semantic(sh, o, p, errs)
if sh.startswith("Rejected"):
seen_reasons.add(o["reason"])
if sh.startswith("Accepted") or sh.startswith("Rejected"):
seen_status.add(o["status"])
if name.startswith("errors/") and not errs and doc["code"] != name[len("errors/"):-len(".json")]:
errs.append("code does not match the file name")
if name.startswith("push/") and not errs and doc["message"]["data"]["type"] != name[len("push/"):-len(".json")]:
errs.append("type does not match the file name")
report(not errs, "example %s matches %s" % (name, shape), "; ".join(errs[:5]))
reasons = set(REASON[1])
report(seen_reasons == reasons and seen_status == {"accepted", "rejected"},
"examples: every push result status and rejection reason of §14.4.6 appears",
"missing %s" % sorted((reasons - seen_reasons) | ({"accepted", "rejected"} - seen_status)) if seen_reasons != reasons else "")
# ----------------------------------------------------------------------------------------------------------------------
# SQLite schema (§14.1)
# ----------------------------------------------------------------------------------------------------------------------
def check_schema():
path = os.path.join(CONTRACTS, "schema", "server-v1.sql")
if not os.path.exists(path):
report(False, "schema server-v1.sql exists", "missing")
return
with open(path, encoding="utf-8") as f:
sql = f.read()
with open(os.path.join(ROOT, "SPEC.md"), encoding="utf-8") as f:
spec = f.read()
try:
i = spec.index("### 14.1 SQLite schema v1")
j = spec.index("```sql\n", i) + len("```sql\n")
block = spec[j:spec.index("```\n", j)]
report(block == sql, "schema server-v1.sql is byte-identical to the SPEC §14.1 DDL",
"differs from SPEC §14.1" if block != sql else "%d bytes" % len(sql.encode("utf-8")))
except ValueError:
report(False, "schema server-v1.sql is byte-identical to the SPEC §14.1 DDL", "§14.1 sql block not found in SPEC.md")
if sqlite3.sqlite_version_info < (3, 37, 0):
report(False, "schema server-v1.sql loads in SQLite", "SQLite %s is older than 3.37 (STRICT tables)" % sqlite3.sqlite_version)
return
db = sqlite3.connect(":memory:")
try:
db.execute("PRAGMA foreign_keys = ON")
db.executescript(sql)
report(True, "schema server-v1.sql loads in SQLite", "SQLite %s" % sqlite3.sqlite_version)
except sqlite3.Error as e:
report(False, "schema server-v1.sql loads in SQLite", str(e))
return
fk = db.execute("PRAGMA foreign_key_check").fetchall()
report(not fk, "schema PRAGMA foreign_key_check is clean", "%r" % fk if fk else "")
tables = set(r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type = 'table'"))
want = {"meta", "sync_sequence", "users", "devices", "invites", "accounts", "account_default_shares", "categories",
"recurring_templates", "recurring_template_splits", "expenses", "expense_splits", "settlements",
"settlement_allocations", "audit_log", "sync_mutations"}
report(tables == want, "schema tables are the 16 of §14.1", "got %s" % sorted(tables) if tables != want else "")
meta = dict(db.execute("SELECT key, value FROM meta"))
seq = db.execute("SELECT id, value FROM sync_sequence").fetchall()
cats = db.execute("SELECT default_key, icon, color, sort_order, server_version FROM categories ORDER BY sort_order").fetchall()
want_cats = [("groceries", "shopping_cart", "green", 10, 1), ("restaurants", "restaurant", "orange", 20, 2),
("home", "home", "brown", 30, 3), ("utilities", "bolt", "yellow", 40, 4),
("transport", "directions_car", "blue", 50, 5), ("travel", "flight", "teal", 60, 6),
("entertainment", "movie", "purple", 70, 7), ("health", "medical_services", "red", 80, 8),
("shopping", "shopping_bag", "pink", 90, 9), ("other", "category", "grey", 100, 10)]
ok = (meta.get("schema_version") == "1" and re.match(r"^[0-9a-f]{32}$", meta.get("sync_epoch", "")) is not None
and seq == [(1, 10)] and cats == want_cats)
report(ok, "schema seed rows (schema_version 1, sync_epoch, sync_sequence 10, ten default categories of AC-7.2)",
"meta %r, sequence %r, categories %r" % (meta, seq, cats) if not ok else "")
db.close()
V1_DEFAULT_CATEGORIES = [
("01920000-0000-7000-8000-000000000001", "groceries", "shopping_cart", "green", 10),
("01920000-0000-7000-8000-000000000002", "restaurants", "restaurant", "orange", 20),
("01920000-0000-7000-8000-000000000003", "home", "home", "brown", 30),
("01920000-0000-7000-8000-000000000004", "utilities", "bolt", "yellow", 40),
("01920000-0000-7000-8000-000000000005", "transport", "directions_car", "blue", 50),
("01920000-0000-7000-8000-000000000006", "travel", "flight", "teal", 60),
("01920000-0000-7000-8000-000000000007", "entertainment", "movie", "purple", 70),
("01920000-0000-7000-8000-000000000008", "health", "medical_services", "red", 80),
("01920000-0000-7000-8000-000000000009", "shopping", "shopping_bag", "pink", 90),
("01920000-0000-7000-8000-00000000000a", "other", "category", "grey", 100),
]
V2_NEW_CATEGORIES = [
("01920000-0000-7000-8000-000000000011", "housing", "key", "indigo", 25),
("01920000-0000-7000-8000-000000000012", "insurance", "shield", "slate", 45),
("01920000-0000-7000-8000-000000000013", "children", "child_care", "lime", 85),
("01920000-0000-7000-8000-000000000014", "fees", "account_balance", "amber", 95),
]
MIGRATION_0002 = os.path.join("schema", "migrations", "0002_categories_v2.sql")
def _migrate_0002(db, sql):
"""Applies 0002 the way sqlx does: one transaction, foreign keys on (set before it starts)."""
db.isolation_level = None
db.execute("BEGIN")
try:
for stmt in _split_sql(sql):
db.execute(stmt)
db.execute("COMMIT")
except sqlite3.Error:
db.execute("ROLLBACK")
raise
def _split_sql(sql):
out, buf = [], ""
for line in sql.splitlines(True):
buf += line
if sqlite3.complete_statement(buf):
out.append(buf)
buf = ""
if buf.strip() and not all(l.strip().startswith("--") or not l.strip() for l in buf.splitlines()):
out.append(buf)
return out
def check_migration_0002():
path = os.path.join(CONTRACTS, MIGRATION_0002)
v1 = os.path.join(CONTRACTS, "schema", "server-v1.sql")
if not os.path.exists(path):
report(False, "schema 0002_categories_v2.sql exists", "missing")
return
with open(path, encoding="utf-8") as f:
sql = f.read()
with open(v1, encoding="utf-8") as f:
v1sql = f.read()
if re.search(r"^\s*(BEGIN|COMMIT|END|ROLLBACK)\b", sql, re.M | re.I) or re.search(r"PRAGMA\s+foreign_keys\b", sql, re.I):
report(False, "schema 0002 has no transaction control or foreign_keys pragma (sqlx wraps it)", "found one")
else:
report(True, "schema 0002 has no transaction control or foreign_keys pragma (sqlx wraps it)")
def fresh():
db = sqlite3.connect(":memory:")
db.execute("PRAGMA foreign_keys = ON")
db.executescript(v1sql)
return db
# 1. empty v1 database
db = fresh()
try:
_migrate_0002(db, sql)
except sqlite3.Error as e:
report(False, "schema 0002 loads on top of server-v1.sql (foreign keys on, one transaction)", str(e))
return
fk = db.execute("PRAGMA foreign_key_check").fetchall()
integ = db.execute("PRAGMA integrity_check").fetchone()[0]
report(not fk and integ == "ok", "schema 0002 loads on top of server-v1.sql (foreign keys on, one transaction)",
"foreign_key_check %r, integrity %s" % (fk, integ) if (fk or integ != "ok") else "SQLite %s" % sqlite3.sqlite_version)
meta = dict(db.execute("SELECT key, value FROM meta"))
seq = db.execute("SELECT id, value FROM sync_sequence").fetchall()
rows = db.execute("SELECT id, default_key, name, icon, color, sort_order, created_at, updated_at, archived_at, "
"deleted_at, server_version FROM categories ORDER BY sort_order").fetchall()
want = sorted([(i, k, None, ic, co, so, "2026-10-17T00:00:00.000Z", "2026-10-17T00:00:00.000Z", None, None, n + 1)
for n, (i, k, ic, co, so) in enumerate(V1_DEFAULT_CATEGORIES)]
+ [(i, k, None, ic, co, so, "2026-10-18T00:00:00.000Z", "2026-10-18T00:00:00.000Z", None, None, 11 + n)
for n, (i, k, ic, co, so) in enumerate(V2_NEW_CATEGORIES)], key=lambda r: r[5])
ok = meta.get("schema_version") == "2" and seq == [(1, 14)] and rows == want
report(ok, "schema 0002 on a fresh database: 14 default categories, new server_versions 11..14, sync_sequence 14, schema_version 2",
"meta %r, sequence %r, categories %r" % (meta, seq, rows) if not ok else "")
keys = [r[1] for r in rows]
report(sorted(keys) == sorted(DEFAULT_KEYS) and all(r[3] in ICONS and r[4] in COLORS for r in rows),
"schema 0002 default keys, icons and colours are the checker's §9.7 / AC-7.2 lists", "keys %r" % keys)
try:
db.execute("INSERT INTO categories (id, default_key, name, icon, color, sort_order, created_at, updated_at, server_version) "
"VALUES ('x', 'pets', NULL, 'pets', 'lime', 1, 't', 't', 999)")
report(False, "schema 0002 keeps a default_key CHECK", "an unknown default_key was accepted")
except sqlite3.IntegrityError:
report(True, "schema 0002 keeps a default_key CHECK")
seed_sql = (
"INSERT INTO users VALUES ('u1', 'admin', 'A', NULL, 't', 't', 11);"
"INSERT INTO accounts VALUES ('a1', 'Vardag', 'vardag', 'u1', 't', 't', NULL, NULL, 12);"
"INSERT INTO categories VALUES ('0192a007-0000-7000-8000-000000000001', NULL, 'Husdjur', 'pets', 'lime', 110, 't', 't', 't', NULL, 13);"
"INSERT INTO expenses (id, account_id, category_id, amount_ore, description, expense_date, payer_user_id, split_mode,"
" created_by_user_id, updated_by_user_id, created_at, updated_at, server_received_at, server_version)"
" VALUES ('e1', 'a1', '01920000-0000-7000-8000-000000000001', 100, 'Ica', '2026-10-01', 'u1', 'default', 'u1', 'u1', 't', 't', 't', 14),"
" ('e2', 'a1', '0192a007-0000-7000-8000-000000000001', 200, 'Hundmat', '2026-10-02', 'u1', 'default', 'u1', 'u1', 't', 't', 't', 15);"
"INSERT INTO recurring_templates (id, account_id, category_id, amount_ore, description, 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 ('r1', 'a1', '01920000-0000-7000-8000-000000000004', 300, 'El', 'u1', 'default', 1, '2026-10', 'u1', 'u1', 't', 't', 16);"
"UPDATE categories SET name = 'Matvaror', server_version = 17 WHERE default_key = 'groceries';"
"UPDATE sync_sequence SET value = 917 WHERE id = 1;")
db.close()
# 2. a v1 database in use: renamed default, custom archived category, expenses and a template referring to categories
db = fresh()
db.executescript(seed_sql)
before = db.execute("SELECT * FROM categories ORDER BY id").fetchall()
try:
_migrate_0002(db, sql)
fk = db.execute("PRAGMA foreign_key_check").fetchall()
after = db.execute("SELECT * FROM categories WHERE id NOT IN (%s) ORDER BY id" % ",".join("'%s'" % c[0] for c in V2_NEW_CATEGORIES)).fetchall()
new = db.execute("SELECT default_key, server_version FROM categories WHERE created_at = '2026-10-18T00:00:00.000Z' ORDER BY server_version").fetchall()
seq = db.execute("SELECT value FROM sync_sequence").fetchone()[0]
refs = db.execute("SELECT count(*) FROM expenses e JOIN categories c ON c.id = e.category_id").fetchone()[0]
defer = db.execute("PRAGMA defer_foreign_keys").fetchone()[0]
ok = (not fk and after == before and new == [("housing", 918), ("insurance", 919), ("children", 920), ("fees", 921)]
and seq == 921 and refs == 2 and defer == 0)
report(ok, "schema 0002 on a database in use: rows kept, references intact, new rows take 918..921 from the sequence",
"fk %r, kept %s, new %r, sequence %r, refs %r, defer %r" % (fk, after == before, new, seq, refs, defer) if not ok else "")
except sqlite3.Error as e:
report(False, "schema 0002 on a database in use: rows kept, references intact, new rows take 918..921 from the sequence", str(e))
db.close()
# 3. the deferred check still catches a dangling reference: drop the referenced row inside the same transaction
db = fresh()
db.executescript(seed_sql)
try:
_migrate_0002(db, sql + "\nDELETE FROM categories WHERE id = '0192a007-0000-7000-8000-000000000001';\n")
report(False, "schema 0002 keeps foreign keys enforced (a dangling reference fails the migration)", "a dangling expense reference was committed")
except sqlite3.IntegrityError:
report(True, "schema 0002 keeps foreign keys enforced (a dangling reference fails the migration)")
db.close()
server_copy = os.path.join(ROOT, "server", "crates", "kvitt-server", "migrations", "0002_categories_v2.sql")
if os.path.exists(server_copy):
with open(server_copy, encoding="utf-8") as f:
same = f.read() == sql
report(same, "schema server migrations/0002_categories_v2.sql is byte-identical to the contract", "" if same else "differs")
# ----------------------------------------------------------------------------------------------------------------------
# Keyword table of the category suggestion (UX-3, CS-6)
# ----------------------------------------------------------------------------------------------------------------------
KEYWORDS_CSV = "category-keywords.csv"
KEYWORDS_HEADER = ["keyword", "match", "default_key", "source"]
def load_keywords():
path = os.path.join(CONTRACTS, KEYWORDS_CSV)
if not os.path.exists(path):
return None, None
with open(path, "rb") as f:
raw = f.read()
return raw, list(csv.reader(io.StringIO(raw.decode("utf-8"), newline="")))
def check_keywords():
raw, rows = load_keywords()
if raw is None:
report(False, "keywords %s exists" % KEYWORDS_CSV, "missing")
return []
problems = []
if raw.startswith(b"\xef\xbb\xbf") or b"\r" in raw or not raw.endswith(b"\n") or raw.endswith(b"\n\n"):
problems.append("must be UTF-8 without BOM, LF line ends, one final newline")
if not rows or rows[0] != KEYWORDS_HEADER:
problems.append("header must be %s" % ",".join(KEYWORDS_HEADER))
entries = []
for n, r in enumerate(rows[1:], start=2):
if len(r) != 4:
problems.append("line %d: %d columns" % (n, len(r)))
continue
kw, kind, key, source = r
if kind not in KEYWORD_MATCHES:
problems.append("line %d: match %r" % (n, kind))
if key not in DEFAULT_KEYS or key == "other":
problems.append("line %d: default_key %r (one of the 14 keys, never other)" % (n, key))
if source not in ("data", "curated"):
problems.append("line %d: source %r" % (n, source))
if not kw or normalize_description(kw) != kw:
problems.append("line %d: keyword %r is not normalised (CS-4 gives %r)" % (n, kw, normalize_description(kw)))
if (kind == "phrase") != (" " in kw):
problems.append("line %d: %r: a phrase has two or more tokens, a token or prefix exactly one" % (n, kw))
entries.append({"keyword": kw, "match": kind, "default_key": key, "source": source})
kws = [e["keyword"] for e in entries]
dup = sorted(set(k for k in kws if kws.count(k) > 1))
if dup:
problems.append("duplicate keywords %s" % dup)
order = [(e["default_key"], e["keyword"]) for e in entries]
if order != sorted(order):
bad = [b for a, b in zip(order, order[1:]) if a > b]
problems.append("not sorted by default_key, then keyword (code points); first out of order %r" % (bad[:1],))
unused = sorted(set(k for k in DEFAULT_KEYS if k != "other") - set(e["default_key"] for e in entries))
detail = "%d entries (%s), every default key but other covered%s" % (
len(entries), ", ".join("%d %s" % (sum(1 for e in entries if e["match"] == m), m) for m in KEYWORD_MATCHES),
"" if not unused else "; keys without entries: %s" % unused)
report(not problems, "keywords %s: header, match kinds, default keys, normalised, unique, sorted" % KEYWORDS_CSV,
"; ".join(problems[:6]) if problems else detail)
return entries
def check_vector_keywords(table):
"""Every keyword list in suggest_category.json is a subsequence of the table (same entries, table order)."""
path = os.path.join(CONTRACTS, "test-vectors", "suggest_category.json")
if not table or not os.path.exists(path):
return
try:
with open(path, encoding="utf-8") as f:
cases = json.load(f)["cases"]
except (ValueError, KeyError, TypeError):
return
index = dict((e["keyword"], (n, e["match"], e["default_key"])) for n, e in enumerate(table))
bad = []
for c in cases:
try:
pos = [index.get(k["keyword"]) for k in c["input"]["keywords"]]
ok = (all(p is not None and (p[1], p[2]) == (k["match"], k["default_key"]) for p, k in zip(pos, c["input"]["keywords"]))
and [p[0] for p in pos] == sorted(set(p[0] for p in pos)))
except (KeyError, TypeError):
ok = False
if not ok:
bad.append(c.get("name"))
report(not bad, "vectors suggest_category.json keyword lists are taken from %s in table order" % KEYWORDS_CSV,
"cases %s" % bad[:5] if bad else "%d cases" % len(cases))
# ----------------------------------------------------------------------------------------------------------------------
# OpenAPI (§7, §14.3) — text-level check
# ----------------------------------------------------------------------------------------------------------------------
ENDPOINTS = [
("E-1", "get", "/health"), ("E-2", "get", "/setup/status"), ("E-3", "post", "/invites/accept"),
("E-4", "get", "/i/{code}"), ("E-5", "get", "/apk/latest"), ("E-6", "get", "/apk/files/{file_name}"),
("E-7", "get", "/me"), ("E-8", "patch", "/me"), ("E-9", "put", "/me/picture"), ("E-10", "delete", "/me/picture"),
("E-11", "get", "/users/{user_id}/picture"), ("E-12", "post", "/invites"), ("E-13", "get", "/invites"),
("E-14", "delete", "/invites/{invite_id}"), ("E-15", "get", "/devices"), ("E-16", "delete", "/devices/{device_id}"),
("E-17", "put", "/devices/current/push-token"), ("E-18", "post", "/accounts"),
("E-19", "patch", "/accounts/{account_id}"), ("E-20", "post", "/accounts/{account_id}/archive"),
("E-21", "post", "/accounts/{account_id}/unarchive"), ("E-22", "delete", "/accounts/{account_id}"),
("E-23", "post", "/categories"), ("E-24", "patch", "/categories/{category_id}"),
("E-25", "post", "/categories/{category_id}/archive"), ("E-26", "post", "/categories/{category_id}/unarchive"),
("E-27", "delete", "/categories/{category_id}"), ("E-28", "post", "/recurring-templates"),
("E-29", "patch", "/recurring-templates/{template_id}"), ("E-30", "delete", "/recurring-templates/{template_id}"),
("E-31", "get", "/sync"), ("E-32", "post", "/sync"), ("E-33", "post", "/settlements/proposal"),
]
UNAUTHENTICATED = {"E-1", "E-2", "E-3", "E-4", "E-5", "E-6"}
def check_openapi():
path = os.path.join(CONTRACTS, "openapi.yaml")
if not os.path.exists(path):
report(False, "openapi.yaml exists", "missing")
return
with open(path, encoding="utf-8") as f:
lines = f.read().splitlines()
text = "\n".join(lines)
head_ok = (re.search(r"^openapi: 3\.1\.0$", text, re.M) and re.search(r"^ title: Kvitt$", text, re.M)
and re.search(r"^ version: '1'$", text, re.M) and "url: https://icehack.asuscomm.com:8443" in text)
report(bool(head_ok), "openapi header (3.1.0, title Kvitt, version 1, one server)", "" if head_ok else "header fields differ from §14.3")
sec_ok = re.search(r"^ device:\n type: http\n scheme: bearer$", text, re.M) is not None
report(sec_ok, "openapi security scheme device = HTTP bearer (§8.2)", "" if sec_ok else "components.securitySchemes.device not found")
# paths: path keys at indent 2, method keys at indent 4, each operation carries `summary: E-n …`
ops, in_paths, cur_path, cur_method = {}, False, None, None
for ln in lines:
if re.match(r"^\S", ln):
in_paths = ln.startswith("paths:")
continue
if not in_paths:
continue
m = re.match(r"^ (/\S*):\s*$", ln)
if m:
cur_path = m.group(1)
continue
m = re.match(r"^ (get|put|post|patch|delete|head|options|trace):\s*$", ln)
if m and cur_path:
cur_method = m.group(1)
ops[(cur_method, cur_path)] = {"eid": None, "security": None}
continue
m = re.match(r"^ summary: '?(E-\d+)\b", ln)
if m and cur_method:
ops[(cur_method, cur_path)]["eid"] = m.group(1)
m = re.match(r"^ security: (.*)$", ln)
if m and cur_method:
ops[(cur_method, cur_path)]["security"] = m.group(1).strip()
want = {(meth, p): eid for eid, meth, p in ENDPOINTS}
got = {k: v["eid"] for k, v in ops.items()}
report(got == want, "openapi operations are exactly E-1..E-33 of §7",
"missing %s, extra %s, wrong ids %s" % (sorted(set(want) - set(got)), sorted(set(got) - set(want)),
sorted(k for k in want if k in got and got[k] != want[k])) if got != want else "%d operations" % len(got))
bad_sec = sorted(eid for (meth, p), eid in want.items() if (meth, p) in ops and
((ops[(meth, p)]["security"] == "[]") != (eid in UNAUTHENTICATED)))
report(not bad_sec, "openapi security: E-1..E-6 unauthenticated, all others device", "wrong: %s" % bad_sec if bad_sec else "")
m = re.search(r"^ ErrorCode:\n((?: .*\n|\s*\n)+)", text + "\n", re.M)
codes = re.findall(r"^ - (\w+)$", m.group(1), re.M) if m else []
report(codes == ERROR_CODES, "openapi ErrorCode enum is the exhaustive list of §14.3.2",
"got %s" % codes if codes != ERROR_CODES else "%d codes" % len(codes))
m = re.search(r"^ RejectionReason:\n((?: .*\n|\s*\n)+)", text + "\n", re.M)
reasons = re.findall(r"^ - (\w+)$", m.group(1), re.M) if m else []
report(reasons == REASON[1], "openapi RejectionReason enum is the exhaustive list of §14.4.6", "got %s" % reasons if reasons != REASON[1] else "")
m = re.search(r"^ DefaultCategoryKey:\n type: string\n enum: \[([^\]]*)\]$", text, re.M)
keys = [k.strip() for k in m.group(1).split(",")] if m else []
report(keys == DEFAULT_KEYS, "openapi DefaultCategoryKey enum is the 14 default keys (AC-7.2, UX-8)",
"got %s" % keys if keys != DEFAULT_KEYS else "%d keys" % len(keys))
leaked = [t for t in ("ColorToken", "IconToken") if re.search(r"^ %s:" % t, text, re.M)]
icon_plain = re.search(r"^ icon:\n type: string\n", text, re.M) is not None
color_plain = re.search(r"^ color:\n type: string\n", text, re.M) is not None
report(not leaked and icon_plain and color_plain,
"openapi icon and color stay plain strings (§9.7 tokens, incl. key, shield, account_balance, slate, amber, are checked in code)",
"token enums %s, icon plain %s, color plain %s" % (leaked, icon_plain, color_plain) if (leaked or not icon_plain or not color_plain) else "")
schemas = set(re.findall(r"^ ([A-Z]\w*):$", text, re.M))
need = {"User", "Device", "Invite", "Share", "Account", "Category", "Split", "TemplateSplit", "Expense", "Allocation",
"Settlement", "RecurringTemplate", "AuditEntry", "Error"}
report(need <= schemas, "openapi declares every wire object of §14.3.3", "missing %s" % sorted(need - schemas) if not need <= schemas else "")
# ----------------------------------------------------------------------------------------------------------------------
def check_json_parses():
bad, n = [], 0
for dirpath, _, files in os.walk(CONTRACTS):
for f in files:
if f.endswith(".json"):
n += 1
p = os.path.join(dirpath, f)
try:
with open(p, encoding="utf-8") as fh:
json.load(fh)
except ValueError as e:
bad.append("%s (%s)" % (rel(p), e))
report(not bad and n > 0, "json: every file under contracts/ parses", "; ".join(bad) if bad else "%d files" % n)
def main():
if not os.path.isdir(CONTRACTS):
report(False, "contracts/ exists", "no such directory: %s" % CONTRACTS)
else:
check_json_parses()
counts = check_vectors()
table = check_keywords()
check_vector_keywords(table)
check_examples()
check_schema()
check_migration_0002()
check_openapi()
if counts:
print("vector cases: " + ", ".join("%s %d" % (k, v) for k, v in sorted(counts.items())))
print("%d checks passed, %d failed" % (PASSES[0], len(FAILURES)))
return 1 if FAILURES else 0
if __name__ == "__main__":
sys.exit(main())
Reference in New Issue View Git Blame Copy Permalink