Contracts for 14 default categories and the category suggestion (UX-3, UX-8)
- contracts/schema/migrations/0002_categories_v2.sql: server migration 0002 rebuilds categories with the 14-key default_key CHECK, inserts housing, insurance, children and fees with server_versions taken from the global sequence, sets schema_version 2. server-v1.sql unchanged. - contracts/category-keywords.csv: the keyword table of the suggestion (148 entries, from docs/ux). - contracts/test-vectors/suggest_category.json: 65 Kotlin-only cases of CS-4 to CS-6. - contracts/examples/E-1-health: schema_version 2. - tools/contracts-check.py: reference suggestCategory, keyword-table checks, 0002 load checks on a fresh and an in-use database, plain icon/color check, new icon and colour tokens. - android/domain: suggestCategory (CS-4 to CS-6, code points, platform NFC and categories through UnicodeText), vector runner for suggest_category, eight vector files. - server/crates/kvitt-domain/tests/vectors.rs: suggest_category is Kotlin-only. The OpenAPI DefaultCategoryKey extension and the two new category examples are held back until the server's wire types know the new keys (M12b). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TQd3pEP9UeN2uifbHknQ7B
This commit is contained in:
+407
-5
@@ -9,19 +9,27 @@ Checks, one output line each:
|
||||
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/openapi.yaml declares exactly the endpoints of SPEC §7 and the error codes of §14.3.2 (text check:
|
||||
the standard library has no YAML parser).
|
||||
* 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 icon and
|
||||
color stay plain strings (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")
|
||||
@@ -283,6 +291,100 @@ def resolve_range(preset, today, earliest_expense, custom_start, custom_end):
|
||||
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)
|
||||
# ----------------------------------------------------------------------------------------------------------------------
|
||||
@@ -313,7 +415,11 @@ VECTOR_FUNCTIONS = {
|
||||
"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$")
|
||||
@@ -415,6 +521,28 @@ def check_vector_input(fn, inp):
|
||||
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
|
||||
|
||||
|
||||
@@ -435,6 +563,8 @@ def check_vector_ok_shape(fn, ok):
|
||||
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
|
||||
|
||||
|
||||
@@ -510,6 +640,32 @@ def coverage(fn, 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
|
||||
|
||||
|
||||
@@ -607,7 +763,10 @@ def E(*vals):
|
||||
|
||||
ROLE = E("admin", "member")
|
||||
INVITE_KIND = E("admin_setup", "partner", "device")
|
||||
DEFAULT_KEY = E("groceries", "restaurants", "home", "utilities", "transport", "travel", "entertainment", "health", "shopping", "other")
|
||||
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")
|
||||
@@ -621,8 +780,9 @@ PUSH_TYPES = ["expense_created", "expense_updated", "expense_deleted", "settleme
|
||||
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"]
|
||||
COLORS = ["green", "orange", "brown", "yellow", "blue", "teal", "purple", "red", "pink", "grey", "indigo", "lime"]
|
||||
"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):
|
||||
@@ -1044,6 +1204,239 @@ def check_schema():
|
||||
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
|
||||
# ----------------------------------------------------------------------------------------------------------------------
|
||||
@@ -1117,6 +1510,12 @@ def check_openapi():
|
||||
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 "")
|
||||
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"}
|
||||
@@ -1146,8 +1545,11 @@ def main():
|
||||
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())))
|
||||
|
||||
Reference in New Issue
Block a user