tests/test_api_keys.py

"""Tests for API key creation, listing, revocation, and authentication."""
import os
import sys
import uuid
import math
import wave
import struct
import io

import pytest
from fastapi.testclient import TestClient

sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

from backend.main import app
from tests.conftest import TestingSessionLocal

# conftest.py already sets app.dependency_overrides[get_db]
client = TestClient(app)


def _register():
    email = f"u_{uuid.uuid4().hex[:8]}@test.com"
    resp = client.post("/api/auth/register", json={"email": email, "password": "Pass12345X!"})
    assert resp.status_code == 201
    return resp.json()["access_token"], email


def _headers(token):
    return {"Authorization": f"Bearer {token}"}


def _wav():
    n = int(0.5 * 22050)
    buf = io.BytesIO()
    with wave.open(buf, "wb") as wf:
        wf.setnchannels(1); wf.setsampwidth(2); wf.setframerate(22050)
        wf.writeframes(struct.pack(f"<{n}h", *[int(32767 * 0.3 * math.sin(2*math.pi*440*i/22050)) for i in range(n)]))
    return buf.getvalue()


# ── Create ─────────────────────────────────────────────────────────────────

def test_create_api_key():
    token, _ = _register()
    resp = client.post("/api/keys", json={"label": "Test Key"}, headers=_headers(token))
    assert resp.status_code == 201
    data = resp.json()
    assert data["key"].startswith("msk_")
    assert len(data["key"]) > 20
    assert "warning" in data
    assert data["label"] == "Test Key"


def test_create_api_key_default_label():
    token, _ = _register()
    resp = client.post("/api/keys", json={}, headers=_headers(token))
    assert resp.status_code == 201
    assert resp.json()["label"] == "My API Key"


def test_create_api_key_unauthenticated():
    resp = client.post("/api/keys", json={"label": "x"})
    assert resp.status_code == 401


# ── List ───────────────────────────────────────────────────────────────────

def test_list_api_keys_empty():
    token, _ = _register()
    resp = client.get("/api/keys", headers=_headers(token))
    assert resp.status_code == 200
    assert resp.json()["keys"] == []


def test_list_api_keys_shows_prefix_only():
    token, _ = _register()
    create_resp = client.post("/api/keys", json={"label": "K1"}, headers=_headers(token))
    raw_key = create_resp.json()["key"]

    list_resp = client.get("/api/keys", headers=_headers(token))
    keys = list_resp.json()["keys"]
    assert len(keys) == 1
    # Should show prefix with ellipsis, NOT the full key
    assert keys[0]["prefix"].endswith("…")
    assert raw_key not in str(keys[0])


def test_list_api_keys_multiple():
    token, _ = _register()
    for i in range(3):
        client.post("/api/keys", json={"label": f"Key {i}"}, headers=_headers(token))
    resp = client.get("/api/keys", headers=_headers(token))
    assert len(resp.json()["keys"]) == 3


# ── Revoke ─────────────────────────────────────────────────────────────────

def test_revoke_api_key():
    token, _ = _register()
    create_resp = client.post("/api/keys", json={"label": "Temp"}, headers=_headers(token))
    key_id = create_resp.json()["id"]

    del_resp = client.delete(f"/api/keys/{key_id}", headers=_headers(token))
    assert del_resp.status_code == 204

    # Should show as inactive
    list_resp = client.get("/api/keys", headers=_headers(token))
    key = next(k for k in list_resp.json()["keys"] if k["id"] == key_id)
    assert key["is_active"] is False


def test_revoke_other_users_key():
    token_a, _ = _register()
    token_b, _ = _register()
    create_resp = client.post("/api/keys", json={"label": "A Key"}, headers=_headers(token_a))
    key_id = create_resp.json()["id"]

    resp = client.delete(f"/api/keys/{key_id}", headers=_headers(token_b))
    assert resp.status_code == 404


# ── Auth via X-API-Key header ──────────────────────────────────────────────

def test_auth_via_api_key():
    token, _ = _register()
    create_resp = client.post("/api/keys", json={"label": "Auth Test"}, headers=_headers(token))
    raw_key = create_resp.json()["key"]

    # Use X-API-Key instead of JWT
    resp = client.get("/api/user/me", headers={"X-API-Key": raw_key})
    assert resp.status_code == 200


def test_auth_via_revoked_api_key():
    token, _ = _register()
    create_resp = client.post("/api/keys", json={"label": "Revoked"}, headers=_headers(token))
    raw_key = create_resp.json()["key"]
    key_id = create_resp.json()["id"]

    # Revoke it
    client.delete(f"/api/keys/{key_id}", headers=_headers(token))

    # Should now fail
    resp = client.get("/api/user/me", headers={"X-API-Key": raw_key})
    assert resp.status_code == 401


def test_api_key_can_analyze():
    token, _ = _register()
    create_resp = client.post("/api/keys", json={"label": "Analyze Key"}, headers=_headers(token))
    raw_key = create_resp.json()["key"]

    resp = client.post(
        "/api/analyze?mode=offline&rights_attested=true",
        files={"file": ("test.wav", _wav(), "audio/wav")},
        headers={"X-API-Key": raw_key},
    )
    assert resp.status_code == 200


def test_invalid_api_key_rejected():
    resp = client.get("/api/user/me", headers={"X-API-Key": "msk_invalidkey"})
    assert resp.status_code == 401


# ── Admin ──────────────────────────────────────────────────────────────────

def test_admin_stats_requires_admin():
    token, _ = _register()
    resp = client.get("/api/admin/stats", headers=_headers(token))
    assert resp.status_code == 403


def test_admin_stats_accessible_to_admin():
    token, email = _register()

    # Promote to admin via DB
    db = TestingSessionLocal()
    from backend.models import User as UserModel
    user = db.query(UserModel).filter(UserModel.email == email).first()
    user.is_admin = True
    db.commit()
    db.close()

    resp = client.get("/api/admin/stats", headers=_headers(token))
    assert resp.status_code == 200
    data = resp.json()
    assert "total_users" in data
    assert "total_analyses" in data
    assert "total_revenue_usd" in data
    assert "training_dataset" in data


def test_admin_list_users():
    token, email = _register()

    db = TestingSessionLocal()
    from backend.models import User as UserModel
    user = db.query(UserModel).filter(UserModel.email == email).first()
    user.is_admin = True
    db.commit()
    db.close()

    resp = client.get("/api/admin/users", headers=_headers(token))
    assert resp.status_code == 200
    data = resp.json()
    assert "users" in data
    assert "total" in data
    assert data["total"] >= 1


def test_admin_add_credits():
    token, email = _register()

    db = TestingSessionLocal()
    from backend.models import User as UserModel
    user = db.query(UserModel).filter(UserModel.email == email).first()
    user.is_admin = True
    target_id = user.id
    db.commit()
    db.close()

    resp = client.post(f"/api/admin/users/{target_id}/add-credits?amount=50",
                       headers=_headers(token))
    assert resp.status_code == 200
    assert resp.json()["credits"] >= 50


# ── /api/user/me returns is_admin ──────────────────────────────────────────

def test_me_returns_is_admin_false():
    token, _ = _register()
    resp = client.get("/api/user/me", headers=_headers(token))
    assert resp.status_code == 200
    assert resp.json()["is_admin"] is False


def test_me_returns_is_admin_true():
    token, email = _register()

    db = TestingSessionLocal()
    from backend.models import User as UserModel
    user = db.query(UserModel).filter(UserModel.email == email).first()
    user.is_admin = True
    db.commit()
    db.close()

    resp = client.get("/api/user/me", headers=_headers(token))
    assert resp.status_code == 200
    assert resp.json()["is_admin"] is True


# ── Change password ────────────────────────────────────────────────────────

def test_change_password_success():
    token, _ = _register()
    resp = client.patch(
        "/api/user/me/password",
        json={"current_password": "Pass12345X!", "new_password": "NewPass999!"},
        headers=_headers(token),
    )
    assert resp.status_code == 204


def test_change_password_wrong_current():
    token, _ = _register()
    resp = client.patch(
        "/api/user/me/password",
        json={"current_password": "wrongpassword", "new_password": "newpass999"},
        headers=_headers(token),
    )
    assert resp.status_code == 400
    assert "incorrect" in resp.json()["detail"].lower()


def test_change_password_too_short():
    token, _ = _register()
    resp = client.patch(
        "/api/user/me/password",
        json={"current_password": "Pass12345X!", "new_password": "short"},
        headers=_headers(token),
    )
    assert resp.status_code == 422


def test_change_password_unauthenticated():
    resp = client.patch(
        "/api/user/me/password",
        json={"current_password": "pass12345", "new_password": "newpass999"},
    )
    assert resp.status_code == 401


def test_change_password_then_login_with_new():
    token, email = _register()
    # Change password
    client.patch(
        "/api/user/me/password",
        json={"current_password": "Pass12345X!", "new_password": "NewPass999!"},
        headers=_headers(token),
    )
    # Login with new password
    resp = client.post("/api/auth/login", json={"email": email, "password": "NewPass999!"})
    assert resp.status_code == 200
    assert "access_token" in resp.json()


# ── Password reset ─────────────────────────────────────────────────────────

def test_forgot_password_always_202():
    """Should return 202 even for unknown emails (no enumeration)."""
    resp = client.post("/api/auth/forgot-password",
                       json={"email": "nobody@nowhere.com"})
    assert resp.status_code == 202


def test_forgot_password_known_email():
    _, email = _register()
    resp = client.post("/api/auth/forgot-password", json={"email": email})
    assert resp.status_code == 202
    assert "reset link" in resp.json()["message"].lower()


def test_reset_password_invalid_token():
    resp = client.post("/api/auth/reset-password",
                       json={"token": "badtoken", "new_password": "newpass999"})
    assert resp.status_code == 400
    assert "invalid" in resp.json()["detail"].lower()


def test_reset_password_full_flow():
    """Forgot → get token from DB → reset → login with new password."""
    _, email = _register()

    # Trigger forgot-password
    client.post("/api/auth/forgot-password", json={"email": email})

    # Get the raw token from DB
    from backend.models import PasswordResetToken, User as UserModel
    db = TestingSessionLocal()
    user = db.query(UserModel).filter(UserModel.email == email).first()
    record = db.query(PasswordResetToken).filter(
        PasswordResetToken.user_id == user.id,
        PasswordResetToken.used == False,  # noqa: E712
    ).first()
    token = record.token
    db.close()

    # Reset password
    resp = client.post("/api/auth/reset-password",
                       json={"token": token, "new_password": "brandnew999"})
    assert resp.status_code == 204

    # Login with new password
    resp = client.post("/api/auth/login",
                       json={"email": email, "password": "brandnew999"})
    assert resp.status_code == 200


def test_reset_token_cannot_be_reused():
    _, email = _register()
    client.post("/api/auth/forgot-password", json={"email": email})

    from backend.models import PasswordResetToken, User as UserModel
    db = TestingSessionLocal()
    user = db.query(UserModel).filter(UserModel.email == email).first()
    record = db.query(PasswordResetToken).filter(
        PasswordResetToken.user_id == user.id,
        PasswordResetToken.used == False,  # noqa: E712
    ).first()
    token = record.token
    db.close()

    client.post("/api/auth/reset-password",
                json={"token": token, "new_password": "first999pw"})
    # Second use should fail
    resp = client.post("/api/auth/reset-password",
                       json={"token": token, "new_password": "second999pw"})
    assert resp.status_code == 400


def test_reset_password_too_short():
    _, email = _register()
    client.post("/api/auth/forgot-password", json={"email": email})

    from backend.models import PasswordResetToken, User as UserModel
    db = TestingSessionLocal()
    user = db.query(UserModel).filter(UserModel.email == email).first()
    record = db.query(PasswordResetToken).filter(
        PasswordResetToken.user_id == user.id,
        PasswordResetToken.used == False,  # noqa: E712
    ).first()
    token = record.token
    db.close()

    resp = client.post("/api/auth/reset-password",
                       json={"token": token, "new_password": "short"})
    assert resp.status_code == 422


# ── Delete account (GDPR) ──────────────────────────────────────────────────

def test_delete_account():
    token, email = _register()
    resp = client.delete("/api/user/me", headers=_headers(token))
    assert resp.status_code == 204


def test_deleted_account_cannot_login():
    token, email = _register()
    client.delete("/api/user/me", headers=_headers(token))
    resp = client.post("/api/auth/login", json={"email": email, "password": "Pass12345X!"})
    assert resp.status_code == 401


def test_delete_account_unauthenticated():
    resp = client.delete("/api/user/me")
    assert resp.status_code == 401