leocrm/tests/test_api_tokens.py

"""Tests for Phase 5 — API Token Service and Delegation Token.

Covers:
- API Token: create, verify, revoke, list
- API Token: expired token rejected
- API Token: revoked token rejected
- API Token: inactive user rejected
- Delegation Token: create, verify, expiry, audience check
- Delegation Token: tampered token rejected
"""

from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.api_token import create_api_token, verify_api_token, revoke_api_token, list_api_tokens, _hash_token
from app.core.delegation_token import create_delegation_token, verify_delegation_token, DELEGATION_AUDIENCE
from app.models.auth import ApiToken
from app.models.tenant import Tenant
from app.models.user import User, UserTenant


async def _seed_tenant_and_user(db: AsyncSession) -> dict:
    from app.core.auth import hash_password

    tenant = Tenant(name="Test Tenant", slug="test-tenant-phase5")
    db.add(tenant)
    await db.flush()
    user = User(
        email="phase5@example.com",
        name="Phase5 User",
        password_hash=hash_password("TestPass123!"),
        is_active=True,
        preferences={},
    )
    db.add(user)
    await db.flush()
    ut = UserTenant(user_id=user.id, tenant_id=tenant.id, is_default=True, role="admin", status="active")
    db.add(ut)
    await db.flush()
    return {"tenant": tenant, "user": user}


# ─── API Token Tests ──────────────────────────────────────────


@pytest.mark.asyncio
async def test_create_api_token_returns_plaintext(db_session: AsyncSession):
    """create_api_token returns the plaintext token once."""
    seed = await _seed_tenant_and_user(db_session)
    result = await create_api_token(
        db_session, seed["tenant"].id, seed["user"].id, name="Test Token",
    )
    assert "token" in result
    assert len(result["token"]) > 20  # URL-safe token
    assert result["name"] == "Test Token"


@pytest.mark.asyncio
async def test_verify_api_token_valid(db_session: AsyncSession):
    """verify_api_token returns user context for a valid token."""
    seed = await _seed_tenant_and_user(db_session)
    result = await create_api_token(
        db_session, seed["tenant"].id, seed["user"].id, name="Test Token",
    )
    user_data = await verify_api_token(db_session, result["token"])
    assert user_data is not None
    assert user_data["user_id"] == str(seed["user"].id)
    assert user_data["tenant_id"] == str(seed["tenant"].id)
    assert user_data["_auth_method"] == "api_token"


@pytest.mark.asyncio
async def test_verify_api_token_invalid(db_session: AsyncSession):
    """verify_api_token returns None for an invalid token."""
    user_data = await verify_api_token(db_session, "invalid-token-string")
    assert user_data is None


@pytest.mark.asyncio
async def test_revoke_api_token(db_session: AsyncSession):
    """revoked tokens are rejected by verify_api_token."""
    seed = await _seed_tenant_and_user(db_session)
    result = await create_api_token(
        db_session, seed["tenant"].id, seed["user"].id, name="To Revoke",
    )
    token_id = uuid.UUID(result["id"])
    revoked = await revoke_api_token(db_session, seed["tenant"].id, token_id)
    assert revoked is True
    # Token should no longer verify
    user_data = await verify_api_token(db_session, result["token"])
    assert user_data is None


@pytest.mark.asyncio
async def test_verify_api_token_expired(db_session: AsyncSession):
    """expired tokens are rejected."""
    seed = await _seed_tenant_and_user(db_session)
    expires_at = datetime.now(UTC) - timedelta(seconds=1)  # Already expired
    result = await create_api_token(
        db_session, seed["tenant"].id, seed["user"].id, name="Expired",
        expires_at=expires_at,
    )
    user_data = await verify_api_token(db_session, result["token"])
    assert user_data is None


@pytest.mark.asyncio
async def test_list_api_tokens(db_session: AsyncSession):
    """list_api_tokens returns tokens without hashes."""
    seed = await _seed_tenant_and_user(db_session)
    await create_api_token(
        db_session, seed["tenant"].id, seed["user"].id, name="Token 1",
    )
    await create_api_token(
        db_session, seed["tenant"].id, seed["user"].id, name="Token 2",
    )
    tokens = await list_api_tokens(db_session, seed["tenant"].id, seed["user"].id)
    assert len(tokens) == 2
    assert "token" not in tokens[0]  # No plaintext in list
    assert "token_hash" not in tokens[0]  # No hash in list


@pytest.mark.asyncio
async def test_verify_api_token_inactive_user(db_session: AsyncSession):
    """inactive users are rejected."""
    seed = await _seed_tenant_and_user(db_session)
    # Deactivate user
    seed["user"].is_active = False
    await db_session.flush()
    result = await create_api_token(
        db_session, seed["tenant"].id, seed["user"].id, name="Inactive User",
    )
    user_data = await verify_api_token(db_session, result["token"])
    assert user_data is None


# ─── Delegation Token Tests ──────────────────────────────────


def test_create_delegation_token_returns_string():
    """create_delegation_token returns a signed string."""
    token = create_delegation_token(
        user_id="user-123", tenant_id="tenant-456",
    )
    assert isinstance(token, str)
    assert "." in token  # payload.signature format


def test_verify_delegation_token_valid():
    """verify_delegation_token returns payload for a valid token."""
    token = create_delegation_token(
        user_id="user-123", tenant_id="tenant-456",
    )
    payload = verify_delegation_token(token)
    assert payload is not None
    assert payload["user_id"] == "user-123"
    assert payload["tenant_id"] == "tenant-456"
    assert payload["audience"] == DELEGATION_AUDIENCE
    assert "expires_at" in payload
    assert "token_id" in payload


def test_verify_delegation_token_invalid():
    """verify_delegation_token returns None for invalid token."""
    payload = verify_delegation_token("invalid.token")
    assert payload is None


def test_verify_delegation_token_tampered():
    """tampered tokens are rejected."""
    token = create_delegation_token(
        user_id="user-123", tenant_id="tenant-456",
    )
    # Tamper with the payload part
    parts = token.split(".")
    tampered = parts[0] + "x." + parts[1]
    payload = verify_delegation_token(tampered)
    assert payload is None


def test_verify_delegation_token_wrong_audience():
    """tokens with wrong audience are rejected."""
    from app.core.delegation_token import _sign
    import json
    from datetime import UTC, datetime, timedelta
    import uuid as uuid_mod

    now = datetime.now(UTC)
    payload = {
        "user_id": "user-123",
        "tenant_id": "tenant-456",
        "agent_id": "test",
        "audience": "wrong-audience",
        "expires_at": (now + timedelta(seconds=30)).isoformat(),
        "token_id": str(uuid_mod.uuid4()),
    }
    token = _sign(payload)
    result = verify_delegation_token(token)
    assert result is None


def test_delegation_token_max_lifetime():
    """token lifetime is capped at MAX_TOKEN_LIFETIME."""
    from app.core.delegation_token import MAX_TOKEN_LIFETIME
    token = create_delegation_token(
        user_id="user-123", tenant_id="tenant-456",
        lifetime_seconds=3600,  # Request 1 hour
    )
    payload = verify_delegation_token(token)
    assert payload is not None
    # Should be capped at 60 seconds
    expires_at = datetime.fromisoformat(payload["expires_at"])
    now = datetime.now(UTC)
    lifetime = (expires_at - now).total_seconds()
    assert lifetime <= MAX_TOKEN_LIFETIME + 5  # Allow small timing variance