inkwell/tests/test_sync.py

from datetime import datetime, timezone

from inkwell.sync import (
    DEFAULT_LIMIT,
    MAX_LIMIT,
    MIN_CLIENT_PROTOCOL_VERSION,
    SYNC_FEATURES,
    SYNC_PROTOCOL_VERSION,
    _clamp_limit,
    _page_cursor,
    _parse_since,
    client_wins,
    protocol_advertisement,
)


async def test_changes_requires_auth(app):
    client = app.test_client()
    resp = await client.get("/api/sync/changes")
    assert resp.status_code == 401


async def test_push_requires_auth(app):
    client = app.test_client()
    resp = await client.post("/api/sync/push", json={"changes": []})
    assert resp.status_code == 401


async def test_attachment_upload_requires_auth(app):
    client = app.test_client()
    resp = await client.put("/api/sync/attachments/00000000-0000-0000-0000-000000000000", data=b"x")
    assert resp.status_code == 401


def test_client_wins():
    older = datetime(2026, 7, 20, tzinfo=timezone.utc)
    newer = datetime(2026, 7, 22, tzinfo=timezone.utc)
    assert client_wins(newer, older) is True  # newer client edit wins
    assert client_wins(older, newer) is False  # older client edit loses (server kept)
    assert client_wins(older, older) is True  # tie → client applies (idempotent)
    assert client_wins(None, older) is False  # unknown client time can't overwrite a real edit
    assert client_wins(older, None) is True  # new/unknown server side yields
    assert client_wins(None, None) is True


def test_parse_since():
    assert _parse_since(None) == 0
    assert _parse_since("42") == 42
    assert _parse_since("-5") == 0  # negative clamps to 0
    assert _parse_since("garbage") == 0


def test_clamp_limit():
    assert _clamp_limit(None) == DEFAULT_LIMIT
    assert _clamp_limit("10") == 10
    assert _clamp_limit("0") == 1  # floor of 1
    assert _clamp_limit("999999") == MAX_LIMIT
    assert _clamp_limit("nope") == DEFAULT_LIMIT


def test_page_cursor_all_drained():
    # Neither stream is full → cursor is the max revision seen; nothing more to page.
    cursor, more = _page_cursor([[1, 3, 5], [2, 4], []], since=0, limit=500)
    assert cursor == 5
    assert more is False


def test_page_cursor_empty():
    # No changes since the cursor → cursor stays put, no more pages.
    cursor, more = _page_cursor([[], [], []], since=7, limit=500)
    assert cursor == 7
    assert more is False


def test_page_cursor_one_stream_full_advances_to_its_boundary():
    # Notes came back full (limit=3) → truncate at its boundary; later labels defer.
    cursor, more = _page_cursor([[1, 2, 3], [4, 5], []], since=0, limit=3)
    assert cursor == 3
    assert more is True


def test_page_cursor_both_full_uses_min_boundary():
    # Both full → advance only to the SMALLER boundary so neither stream skips a gap.
    cursor, more = _page_cursor([[1, 2, 10], [3, 4, 5], []], since=0, limit=3)
    assert cursor == 5
    assert more is True


def test_page_cursor_a_full_revocation_stream_holds_the_cursor_back():
    # Revocations page on the same cursor: a full one truncates like any other stream.
    cursor, more = _page_cursor([[1, 9], [], [2, 3, 4]], since=0, limit=3)
    assert cursor == 4
    assert more is True


# --- protocol handshake -----------------------------------------------------


def test_protocol_advertisement_shape():
    ad = protocol_advertisement()
    assert ad["sync_protocol_version"] == SYNC_PROTOCOL_VERSION
    assert ad["min_client_protocol_version"] == MIN_CLIENT_PROTOCOL_VERSION
    # A list, not a tuple — it has to survive jsonify as a JSON array.
    assert isinstance(ad["sync_features"], list)
    assert ad["sync_features"] == list(SYNC_FEATURES)


def test_protocol_floor_never_exceeds_current():
    # A server can't demand a client protocol newer than the one it speaks itself —
    # that would lock out every client, including a perfectly current one.
    assert MIN_CLIENT_PROTOCOL_VERSION <= SYNC_PROTOCOL_VERSION


def test_protocol_features_are_unique_nonempty_names():
    # Clients match capabilities by exact name, so duplicates or blanks would make
    # a feature check silently meaningless.
    assert all(f and f.strip() == f for f in SYNC_FEATURES)
    assert len(set(SYNC_FEATURES)) == len(SYNC_FEATURES)