microfish/backend/tests/test_admin_users_api.py

import json

from flask import Flask
from sqlalchemy import create_engine

from app.api.admin import admin_bp
from app.api.auth import auth_bp
from app.db import Base, create_session_factory
from app.services.identity import IdentityRepository, PasswordService


def make_admin_app():
    engine = create_engine("sqlite+pysqlite:///:memory:")
    Base.metadata.create_all(engine)
    session_factory = create_session_factory(engine)

    app = Flask(__name__)
    app.config.update(TESTING=True, SECRET_KEY="test-secret", SESSION_COOKIE_SECURE=False)
    app.extensions["crowdsight_session_factory"] = session_factory
    app.register_blueprint(auth_bp, url_prefix="/api/auth")
    app.register_blueprint(admin_bp, url_prefix="/api/admin")

    with session_factory() as session:
        repo = IdentityRepository(session)
        org = repo.create_organization(name="Org A", slug="org-a")
        admin = repo.create_user(
            email="admin@example.com",
            password_hash=PasswordService.hash_password("correct horse battery staple"),
        )
        repo.create_membership(admin.id, org.id, "admin")
        user = repo.create_user(
            email="user@example.com",
            password_hash=PasswordService.hash_password("correct horse battery staple"),
        )
        repo.create_membership(user.id, org.id, "user")
        session.commit()

    return app, engine


def login(client, email):
    response = client.post(
        "/api/auth/login",
        json={"email": email, "password": "correct horse battery staple"},
    )
    assert response.status_code == 200


def _csrf_headers(client):
    return {"X-CSRF-Token": client.get_cookie("crowdsight_csrf").value}


def test_admin_lists_and_creates_users_only_in_current_organization():
    app, engine = make_admin_app()
    try:
        client = app.test_client()
        login(client, "admin@example.com")

        listed = client.get("/api/admin/users")
        assert listed.status_code == 200
        body = listed.get_json()
        assert body["data"]["count"] == 2
        assert all("password_hash" not in user for user in body["data"]["users"])

        created = client.post(
            "/api/admin/users",
            json={"email": "new-user@example.com"},
            headers=_csrf_headers(client),
        )
        assert created.status_code == 201
        assert created.get_json()["data"]["role"] == "user"
    finally:
        engine.dispose()


def test_admin_cannot_grant_admin_and_user_cannot_manage_users():
    app, engine = make_admin_app()
    try:
        admin_client = app.test_client()
        login(admin_client, "admin@example.com")
        forbidden = admin_client.post(
            "/api/admin/users",
            json={"email": "new-admin@example.com", "role": "admin"},
            headers=_csrf_headers(admin_client),
        )
        assert forbidden.status_code == 403
        assert forbidden.get_json()["error_code"] == "admin_role_grant_forbidden"

        user_client = app.test_client()
        login(user_client, "user@example.com")
        user_forbidden = user_client.get("/api/admin/users")
        assert user_forbidden.status_code == 403
    finally:
        engine.dispose()

def test_admin_duplicate_email_returns_conflict_without_db_error():
    app, engine = make_admin_app()
    try:
        client = app.test_client()
        login(client, "admin@example.com")
        response = client.post(
            "/api/admin/users",
            json={"email": "user@example.com"},
            headers=_csrf_headers(client),
        )
        assert response.status_code == 409
        assert response.get_json()["error_code"] == "user_exists"
    finally:
        engine.dispose()

def test_state_changing_admin_request_requires_csrf_token():
    app, engine = make_admin_app()
    try:
        client = app.test_client()
        login(client, "admin@example.com")
        response = client.post(
            "/api/admin/users",
            json={"email": "csrf-user@example.com"},
        )
        assert response.status_code == 403
        assert response.get_json()["error_code"] == "csrf_failed"
    finally:
        engine.dispose()


def test_admin_can_update_user_role_within_allowed_domain():
    app, engine = make_admin_app()
    try:
        client = app.test_client()
        login(client, "admin@example.com")

        # Find the target user's id.
        listed = client.get("/api/admin/users").get_json()
        target = next(u for u in listed["data"]["users"] if u["email"] == "user@example.com")

        # Admin may manage a USER role only (policy: user→user is allowed).
        response = client.patch(
            f"/api/admin/users/{target['id']}",
            json={"role": "user"},
            headers=_csrf_headers(client),
        )
        assert response.status_code == 200
        assert response.get_json()["data"]["role"] == "user"
    finally:
        engine.dispose()


def test_admin_cannot_promote_to_super_admin():
    app, engine = make_admin_app()
    try:
        client = app.test_client()
        login(client, "admin@example.com")
        listed = client.get("/api/admin/users").get_json()
        target = next(u for u in listed["data"]["users"] if u["email"] == "user@example.com")

        response = client.patch(
            f"/api/admin/users/{target['id']}",
            json={"role": "super_admin"},
            headers=_csrf_headers(client),
        )
        assert response.status_code == 403
        assert response.get_json()["error_code"] == "admin_role_grant_forbidden"
    finally:
        engine.dispose()


def test_settings_endpoint_requires_super_admin():
    app, engine = make_admin_app()
    try:
        client = app.test_client()
        login(client, "admin@example.com")
        # An org admin (ADMIN) must NOT be able to read or mutate platform settings.
        forbidden_get = client.get("/api/admin/settings")
        assert forbidden_get.status_code == 403
        forbidden_put = client.put(
            "/api/admin/settings",
            json={"provider": "openai", "model": "gpt-4o"},
            headers=_csrf_headers(client),
        )
        assert forbidden_put.status_code == 403
    finally:
        engine.dispose()


def test_super_admin_can_save_and_read_settings():
    app, engine = make_admin_app()
    try:
        # SettingsService derives its Fernet key from Config.SECRET_KEY; set it
        # directly (it was evaluated from env at import time).
        from app.config import Config

        Config.SECRET_KEY = "test-encryption-secret-" * 3

        # Promote the existing admin membership to super_admin in its org.
        with app.extensions["crowdsight_session_factory"]() as session:
            from sqlalchemy import select

            from app.models.saas import Membership, Organization
            from app.services.identity import IdentityRepository

            repo = IdentityRepository(session)
            admin_user = repo.get_user_by_email("admin@example.com")
            org = session.execute(
                select(Organization).where(Organization.slug == "org-a")
            ).scalar_one()
            membership = session.execute(
                select(Membership).where(
                    Membership.user_id == admin_user.id,
                    Membership.organization_id == org.id,
                )
            ).scalar_one()
            membership.role = "super_admin"
            session.commit()

        client = app.test_client()
        login(client, "admin@example.com")

        saved = client.put(
            "/api/admin/settings",
            json={"provider": "openai", "model": "gpt-4o", "api_key": "sk-secret-value"},
            headers=_csrf_headers(client),
        )
        assert saved.status_code == 200
        data = saved.get_json()["data"]
        assert data["settings"]["model"] == "gpt-4o"
        # API key is masked (never plaintext). It may be a bare mask or a
        # masked-with-prefix variant; the important invariant is no plaintext.
        assert "sk-secret-value" not in json.dumps(saved.get_json())

        # Reading back returns masked value only; no plaintext secret leaked.
        got = client.get("/api/admin/settings")
        assert got.status_code == 200
        assert "sk-secret-value" not in json.dumps(got.get_json())
    finally:
        engine.dispose()