brain-of-reese/tests/unit/test_tokens.py

"""Unit: the API-token service (phase 79, task 02).

Covers ``app.core.tokens`` — the create/lookup/revoke service behind the
admin API (task 02's endpoints) and the future token-auth login (task
03):

* ``generate_token`` — the ``bor_`` + 32-hex shape, two calls differ;
* ``hash_token`` — deterministic 64-hex digest of the FULL string (a
  stripped prefix can never collide);
* ``create_token`` / ``find_active_by_token`` — the round-trip (the
  plaintext resolves to its active row) and the GENERIC-MISS contract:
  revoked, unknown, empty, short and wrong-prefix candidates all return
  the same ``None`` (hash matches no row — there is no "almost" path,
  which is what makes the task-03 one-generic-401 safe);
* ``revoke`` — stamps ``revoked_at`` once (idempotent re-call keeps the
  original stamp; False only for a missing id);
* ``mark_used`` — bumps ``last_used_at``;
* ``regenerate_token`` (phase 101) — the atomic rotation matrix: an
  active row is revoked (first stamp) and its successor (same label,
  fresh well-formed plaintext, hash-only row) takes its place — the new
  plaintext round-trips, the old one no longer authenticates; a missing
  id returns ``None``; an already-revoked id raises
  ``TokenAlreadyRevoked``; and the service NEVER commits — the caller's
  rollback undoes both writes.

House DB-test pattern (the ``test_sources_meta`` precedent): the service
is a thin session wrapper whose contract (server-default ``created_at``,
the unique hash index) only holds against a real database — runs against
the local compose Postgres, skips with clear instructions when the stack
is not up. The service flushes, never commits: the tests commit, as the
endpoints do.
"""
from __future__ import annotations

import re
import uuid
from collections.abc import Iterator
from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core import tokens as tok
from app.models import ApiToken

TOKEN_SHAPE = re.compile(r"^bor_[0-9a-f]{32}$")


@pytest.fixture(autouse=True)
def clean_tokens(db: Session) -> Iterator[None]:
    """api_tokens is global state: reset around every test."""
    db.execute(text("TRUNCATE api_tokens"))
    db.commit()
    yield
    db.execute(text("TRUNCATE api_tokens"))
    db.commit()


def _create_and_commit(db: Session, label: str = "alice") -> tuple[ApiToken, str]:
    """Service create + the endpoint's commit/refresh (the house split)."""
    row, plaintext = tok.create_token(db, label)
    db.commit()
    db.refresh(row)
    return row, plaintext


def test_generate_token_shape_and_uniqueness() -> None:
    """``bor_`` + exactly 32 lowercase hex chars (128-bit), no reuse."""
    a = tok.generate_token()
    b = tok.generate_token()
    assert TOKEN_SHAPE.fullmatch(a), a
    assert TOKEN_SHAPE.fullmatch(b), b
    assert a != b


def test_hash_token_deterministic_64_hex_full_string() -> None:
    """Same token → same digest; 64 hex chars; the FULL string is hashed
    (hashing ``bor_X`` ≠ hashing ``X`` — a stripped prefix can never
    collide with another token's hash)."""
    plain = tok.generate_token()
    h1 = tok.hash_token(plain)
    h2 = tok.hash_token(plain)
    assert h1 == h2
    assert re.fullmatch(r"[0-9a-f]{64}", h1), h1
    assert tok.hash_token(plain.removeprefix("bor_")) != h1


def test_create_and_find_round_trip(db: Session) -> None:
    """create → the row carries ONLY the hash; the plaintext resolves
    back to the same active row (the task-03 login path)."""
    row, plaintext = _create_and_commit(db, "  alice  ")

    # The service strips the label; the row only ever carries the hash.
    assert row.label == "alice"
    assert row.token_hash == tok.hash_token(plaintext)
    assert row.token_hash != plaintext
    assert row.last_used_at is None
    assert row.revoked_at is None

    hit = tok.find_active_by_token(db, plaintext)
    assert hit is not None
    assert hit.id == row.id
    assert hit.label == "alice"


@pytest.mark.parametrize(
    ("candidate", "why"),
    [
        ("", "empty string"),
        ("bor_ab", "short string"),
        ("bor_" + "f" * 31, "31 hex chars (one short)"),
        (
            "zzz_" + "0" * 32,
            "wrong prefix",
        ),
    ],
    ids=["empty", "short", "31-hex", "wrong-prefix"],
)
def test_find_active_by_token_generic_miss(db: Session, candidate: str, why: str) -> None:
    """Every malformed/unknown candidate is the SAME miss — the hash
    simply matches no row (no "almost" path, no per-shape error)."""
    _, plaintext = _create_and_commit(db)  # the table is NOT empty
    assert tok.find_active_by_token(db, candidate) is None, why
    # …while the stored token still resolves (the miss is specific).
    assert tok.find_active_by_token(db, plaintext) is not None


def test_find_active_by_token_unknown_well_formed(db: Session) -> None:
    """A well-formed token that was never stored (or belongs to another
    admin) misses — no enumeration surface beyond the unique-index hit."""
    unknown = tok.generate_token()  # never persisted
    assert tok.find_active_by_token(db, unknown) is None


def test_find_active_by_token_revoked_misses(db: Session) -> None:
    """A revoked row never resolves: the hash still matches the row, but
    ``revoked_at IS NULL`` is part of the contract (A4 — dead is dead,
    enforced immediately)."""
    row, plaintext = _create_and_commit(db)
    assert tok.revoke(db, row.id) is True
    db.commit()
    assert tok.find_active_by_token(db, plaintext) is None


def test_revoke_stamps_once_and_false_only_for_missing(db: Session) -> None:
    """First revoke stamps ``revoked_at`` (True); the second call is
    idempotent (True, the ORIGINAL stamp kept — no re-stamp); False only
    when the row does not exist."""
    row, _ = _create_and_commit(db)
    assert tok.revoke(db, row.id) is True
    db.commit()
    db.refresh(row)
    first_stamp = row.revoked_at
    assert first_stamp is not None

    # Idempotent: True again, and the first stamp survives.
    assert tok.revoke(db, row.id) is True
    db.commit()
    db.refresh(row)
    assert row.revoked_at == first_stamp

    # A missing id is the only False.
    assert tok.revoke(db, uuid.uuid4()) is False


def test_mark_used_stamps_last_used_at(db: Session) -> None:
    """NULL until first use, then "now" (UTC, tz-aware)."""
    row, _ = _create_and_commit(db)
    assert row.last_used_at is None

    before = datetime.now(UTC)
    tok.mark_used(row)
    after = datetime.now(UTC)

    assert row.last_used_at is not None
    assert row.last_used_at.tzinfo is not None
    assert before - timedelta(seconds=1) <= row.last_used_at <= after + timedelta(
        seconds=1
    )
    db.commit()


# ---------- phase 101, task 01: the atomic rotation ----------


def test_regenerate_token_rotates_an_active_row(db: Session) -> None:
    """Active row → a NEW row (different id) with the SAME label + a
    well-formed new plaintext (``bor_`` + 32 hex) whose sha256 IS the
    stored hash; the OLD row is stamped with its first ``revoked_at``;
    the new plaintext round-trips through ``find_active_by_token`` and
    the old plaintext no longer authenticates (dead is dead)."""
    old_row, old_plain = _create_and_commit(db, "alice")

    before = datetime.now(UTC)
    rotated = tok.regenerate_token(db, old_row.id)
    assert rotated is not None
    new_row, new_plain = rotated
    db.commit()  # the caller owns the commit (the endpoint's job)
    db.refresh(old_row)
    db.refresh(new_row)

    # The successor: a different id, the SAME hand-out label, a fresh
    # well-formed plaintext (never a reuse of the old one).
    assert new_row.id != old_row.id
    assert new_row.label == old_row.label == "alice"
    assert TOKEN_SHAPE.fullmatch(new_plain), new_plain
    assert new_plain != old_plain

    # The successor row stores ONLY the sha256 of the NEW plaintext.
    assert new_row.token_hash == tok.hash_token(new_plain)
    assert new_row.token_hash != tok.hash_token(old_plain)
    assert new_row.last_used_at is None
    assert new_row.revoked_at is None

    # The old row carries its first (and only) revocation stamp.
    assert old_row.revoked_at is not None
    assert old_row.revoked_at.tzinfo is not None
    assert before - timedelta(seconds=1) <= old_row.revoked_at <= (
        datetime.now(UTC) + timedelta(seconds=1)
    )

    # The rotation takes effect immediately, both directions.
    hit = tok.find_active_by_token(db, new_plain)
    assert hit is not None
    assert hit.id == new_row.id
    assert tok.find_active_by_token(db, old_plain) is None


def test_regenerate_token_missing_id_returns_none(db: Session) -> None:
    """Unknown id → ``None`` (the endpoint maps it to 404) — nothing is
    created, no stamp anywhere."""
    _create_and_commit(db)  # the table is NOT empty — the miss is by id
    assert tok.regenerate_token(db, uuid.uuid4()) is None
    db.rollback()
    assert db.execute(text("SELECT count(*) FROM api_tokens")).scalar_one() == 1


def test_regenerate_token_already_revoked_raises(db: Session) -> None:
    """A dead token cannot be rotated: an already-revoked id raises
    ``TokenAlreadyRevoked`` (the endpoint's 409) and creates NOTHING —
    no successor row, the original stamp untouched."""
    row, _ = _create_and_commit(db)
    assert tok.revoke(db, row.id) is True
    db.commit()
    db.refresh(row)
    original_stamp = row.revoked_at
    assert original_stamp is not None

    with pytest.raises(tok.TokenAlreadyRevoked):
        tok.regenerate_token(db, row.id)
    db.rollback()

    assert db.execute(text("SELECT count(*) FROM api_tokens")).scalar_one() == 1
    db.refresh(row)
    assert row.revoked_at == original_stamp


def test_regenerate_token_service_never_commits(db: Session) -> None:
    """Atomicity + the house commit split: the service flushes, never
    commits — the caller's rollback AFTER the rotation undoes BOTH
    writes (the stamp AND the successor), so the old token stays live."""
    old_row, old_plain = _create_and_commit(db, "carol")

    rotated = tok.regenerate_token(db, old_row.id)
    assert rotated is not None
    new_row, _ = rotated
    db.rollback()  # the caller refuses the rotation
    db.expire_all()

    survivor = db.get(ApiToken, old_row.id)
    assert survivor is not None
    assert survivor.revoked_at is None  # the stamp rolled back
    assert db.get(ApiToken, new_row.id) is None  # the successor rolled back
    # The old credential is alive again — the rotation never happened.
    hit = tok.find_active_by_token(db, old_plain)
    assert hit is not None
    assert hit.id == old_row.id
    db.commit()