tests/test_admin_tools.py

"""Tests for admin impersonation, bulk credits, and API key scopes."""
import pytest
from fastapi.testclient import TestClient

from tests.conftest import TestingSessionLocal, app
from backend.auth import create_access_token, hash_password
from backend.models import AuditLog, User, APIKey


client = TestClient(app, raise_server_exceptions=True)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def make_user(email, password="testpass", is_admin=False, credits=10):
    db = TestingSessionLocal()
    try:
        existing = db.query(User).filter(User.email == email).first()
        if existing:
            return existing.id
        user = User(
            email=email,
            password_hash=hash_password(password),
            is_admin=is_admin,
            credits=credits,
        )
        db.add(user)
        db.commit()
        db.refresh(user)
        return user.id
    finally:
        db.close()


def get_token(user_id):
    return create_access_token(user_id)


def auth_headers(user_id):
    return {"Authorization": f"Bearer {get_token(user_id)}"}


def get_user_credits(user_id):
    db = TestingSessionLocal()
    try:
        return db.query(User).filter(User.id == user_id).first().credits
    finally:
        db.close()


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------

@pytest.fixture(scope="module")
def admin_id():
    return make_user("admin_tools@example.com", is_admin=True)


@pytest.fixture(scope="module")
def regular_id():
    return make_user("regular_tools@example.com", is_admin=False, credits=5)


@pytest.fixture(scope="module")
def other_admin_id():
    return make_user("other_admin_tools@example.com", is_admin=True)


@pytest.fixture(scope="module")
def target_id():
    return make_user("target_tools@example.com", is_admin=False, credits=0)


@pytest.fixture(scope="module")
def bulk_user_ids():
    ids = []
    for i in range(3):
        uid = make_user(f"bulk{i}_tools@example.com", credits=0)
        ids.append(uid)
    return ids


# ---------------------------------------------------------------------------
# Impersonation tests
# ---------------------------------------------------------------------------

class TestImpersonation:

    def test_admin_can_impersonate(self, admin_id, target_id):
        resp = client.post(
            f"/api/admin/tools/impersonate/{target_id}",
            headers=auth_headers(admin_id),
        )
        assert resp.status_code == 200
        data = resp.json()
        assert "access_token" in data
        assert data["expires_in"] == 3600
        assert data["impersonating"] == target_id

    def test_non_admin_cannot_impersonate(self, regular_id, target_id):
        resp = client.post(
            f"/api/admin/tools/impersonate/{target_id}",
            headers=auth_headers(regular_id),
        )
        assert resp.status_code == 403

    def test_impersonate_returns_valid_jwt(self, admin_id, target_id):
        resp = client.post(
            f"/api/admin/tools/impersonate/{target_id}",
            headers=auth_headers(admin_id),
        )
        assert resp.status_code == 200
        token = resp.json()["access_token"]
        # The returned token should let us access /api/user/me as the target user
        me_resp = client.get(
            "/api/user/me",
            headers={"Authorization": f"Bearer {token}"},
        )
        assert me_resp.status_code == 200
        assert me_resp.json()["id"] == target_id

    def test_impersonated_token_works_as_target_user(self, admin_id, target_id):
        resp = client.post(
            f"/api/admin/tools/impersonate/{target_id}",
            headers=auth_headers(admin_id),
        )
        token = resp.json()["access_token"]
        me_resp = client.get(
            "/api/user/me",
            headers={"Authorization": f"Bearer {token}"},
        )
        assert me_resp.json()["email"] == "target_tools@example.com"

    def test_cannot_impersonate_admin(self, admin_id, other_admin_id):
        resp = client.post(
            f"/api/admin/tools/impersonate/{other_admin_id}",
            headers=auth_headers(admin_id),
        )
        assert resp.status_code == 403
        assert "admin" in resp.json()["detail"].lower()

    def test_impersonate_nonexistent_user(self, admin_id):
        resp = client.post(
            "/api/admin/tools/impersonate/999999",
            headers=auth_headers(admin_id),
        )
        assert resp.status_code == 404

    def test_impersonation_logged_to_audit_log(self, admin_id, target_id):
        resp = client.post(
            f"/api/admin/tools/impersonate/{target_id}",
            headers=auth_headers(admin_id),
        )
        assert resp.status_code == 200
        db = TestingSessionLocal()
        try:
            log = db.query(AuditLog).filter(
                AuditLog.action == "admin.impersonate",
                AuditLog.resource_type == "user",
                AuditLog.resource_id == str(target_id),
                AuditLog.user_id == admin_id,
            ).first()
            assert log is not None
            assert log.user_email == "admin_tools@example.com"
        finally:
            db.close()


# ---------------------------------------------------------------------------
# Bulk grant credits tests
# ---------------------------------------------------------------------------

class TestBulkGrantCredits:

    def test_admin_can_bulk_grant_credits(self, admin_id, bulk_user_ids):
        resp = client.post(
            "/api/admin/tools/bulk-grant-credits",
            json={"user_ids": bulk_user_ids, "credits": 10, "reason": "test grant"},
            headers=auth_headers(admin_id),
        )
        assert resp.status_code == 200
        data = resp.json()
        assert data["credits_granted"] == 10
        assert len(data["results"]) == len(bulk_user_ids)
        for r in data["results"]:
            assert r["status"] == "ok"

    def test_bulk_grant_updates_all_user_credits(self, admin_id, bulk_user_ids):
        # Reset credits to known value
        db = TestingSessionLocal()
        try:
            for uid in bulk_user_ids:
                user = db.query(User).filter(User.id == uid).first()
                user.credits = 0
            db.commit()
        finally:
            db.close()

        resp = client.post(
            "/api/admin/tools/bulk-grant-credits",
            json={"user_ids": bulk_user_ids, "credits": 5, "reason": "verification"},
            headers=auth_headers(admin_id),
        )
        assert resp.status_code == 200
        for uid in bulk_user_ids:
            assert get_user_credits(uid) == 5

    def test_non_admin_cannot_bulk_grant(self, regular_id, bulk_user_ids):
        resp = client.post(
            "/api/admin/tools/bulk-grant-credits",
            json={"user_ids": bulk_user_ids, "credits": 10, "reason": "hack"},
            headers=auth_headers(regular_id),
        )
        assert resp.status_code == 403

    def test_bulk_grant_more_than_100_users_returns_400(self, admin_id):
        user_ids = list(range(1, 102))  # 101 ids
        resp = client.post(
            "/api/admin/tools/bulk-grant-credits",
            json={"user_ids": user_ids, "credits": 1, "reason": "overflow"},
            headers=auth_headers(admin_id),
        )
        assert resp.status_code == 400
        assert "100" in resp.json()["detail"]


# ---------------------------------------------------------------------------
# API key scopes tests
# ---------------------------------------------------------------------------

class TestAPIKeyScopes:

    def test_api_key_created_with_custom_scopes(self, regular_id):
        resp = client.post(
            "/api/keys",
            json={"label": "Scoped Key", "scopes": "analyze:read keys:read"},
            headers=auth_headers(regular_id),
        )
        assert resp.status_code == 201
        data = resp.json()
        # Verify in DB that scopes were stored
        db = TestingSessionLocal()
        try:
            key_obj = db.query(APIKey).filter(APIKey.id == data["id"]).first()
            assert key_obj is not None
            assert key_obj.scopes == "analyze:read keys:read"
        finally:
            db.close()

    def test_get_api_keys_response_includes_scopes(self, regular_id):
        # Ensure at least one key exists
        client.post(
            "/api/keys",
            json={"label": "Scope Check Key", "scopes": "analyze:read"},
            headers=auth_headers(regular_id),
        )
        resp = client.get("/api/keys", headers=auth_headers(regular_id))
        assert resp.status_code == 200
        keys = resp.json()["keys"]
        assert len(keys) > 0
        for k in keys:
            assert "scopes" in k

    def test_ip_allowlist_stored_when_creating_key(self, regular_id):
        resp = client.post(
            "/api/keys",
            json={
                "label": "IP-Restricted Key",
                "scopes": "analyze:read",
                "ip_allowlist": "10.0.0.0/8,192.168.1.0/24",
            },
            headers=auth_headers(regular_id),
        )
        assert resp.status_code == 201
        key_id = resp.json()["id"]
        db = TestingSessionLocal()
        try:
            key_obj = db.query(APIKey).filter(APIKey.id == key_id).first()
            assert key_obj.ip_allowlist == "10.0.0.0/8,192.168.1.0/24"
        finally:
            db.close()

    def test_api_key_default_scopes(self, regular_id):
        resp = client.post(
            "/api/keys",
            json={"label": "Default Scopes Key"},
            headers=auth_headers(regular_id),
        )
        assert resp.status_code == 201
        key_id = resp.json()["id"]
        db = TestingSessionLocal()
        try:
            key_obj = db.query(APIKey).filter(APIKey.id == key_id).first()
            assert "analyze:read" in key_obj.scopes
            assert "analyze:write" in key_obj.scopes
        finally:
            db.close()

    def test_get_keys_includes_ip_allowlist_field(self, regular_id):
        resp = client.get("/api/keys", headers=auth_headers(regular_id))
        assert resp.status_code == 200
        keys = resp.json()["keys"]
        assert len(keys) > 0
        # ip_allowlist field should be present (may be None)
        for k in keys:
            assert "ip_allowlist" in k