"""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``. 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()