import json
from contextlib import contextmanager
from urllib.parse import quote

import pytest
import requests

import mlflow
from mlflow import MlflowClient, MlflowException
from mlflow.environment_variables import (
    MLFLOW_AUTH_CONFIG_PATH,
    MLFLOW_FLASK_SERVER_SECRET_KEY,
    MLFLOW_TRACKING_PASSWORD,
    MLFLOW_TRACKING_USERNAME,
)
from mlflow.protos.databricks_pb2 import (
    BAD_REQUEST,
    INVALID_PARAMETER_VALUE,
    PERMISSION_DENIED,
    RESOURCE_DOES_NOT_EXIST,
    UNAUTHENTICATED,
    ErrorCode,
)
from mlflow.server.auth.client import AuthServiceClient
from mlflow.utils.os import is_windows

from tests.helper_functions import random_str
from tests.server.auth.auth_test_utils import (
    ADMIN_PASSWORD,
    ADMIN_USERNAME,
    User,
    write_isolated_auth_config,
)
from tests.tracking.integration_test_utils import _init_server


@pytest.fixture(autouse=True)
def clear_credentials(monkeypatch):
    monkeypatch.delenv(MLFLOW_TRACKING_USERNAME.name, raising=False)
    monkeypatch.delenv(MLFLOW_TRACKING_PASSWORD.name, raising=False)


@pytest.fixture
def client(tmp_path):
    auth_config_path = write_isolated_auth_config(tmp_path)
    path = tmp_path.joinpath("sqlalchemy.db").as_uri()
    backend_uri = ("sqlite://" if is_windows() else "sqlite:////") + path[len("file://") :]

    with _init_server(
        backend_uri=backend_uri,
        root_artifact_uri=tmp_path.joinpath("artifacts").as_uri(),
        app="mlflow.server.auth:create_app",
        extra_env={
            MLFLOW_FLASK_SERVER_SECRET_KEY.name: "my-secret-key",
            MLFLOW_AUTH_CONFIG_PATH.name: str(auth_config_path),
        },
        server_type="flask",
    ) as url:
        yield AuthServiceClient(url)


@contextmanager
def assert_unauthenticated():
    with pytest.raises(MlflowException, match=r"You are not authenticated.") as exception_context:
        yield
    assert exception_context.value.error_code == ErrorCode.Name(UNAUTHENTICATED)


@contextmanager
def assert_unauthorized():
    with pytest.raises(MlflowException, match=r"Permission denied.") as exception_context:
        yield
    assert exception_context.value.error_code == ErrorCode.Name(PERMISSION_DENIED)


def test_get_client():
    client = mlflow.server.get_app_client("basic-auth", "uri:/fake")
    assert isinstance(client, AuthServiceClient)


def test_create_user(client, monkeypatch):
    username = random_str()
    password = random_str()

    with assert_unauthenticated():
        client.create_user(username, password)

    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        user = client.create_user(username, password)
    assert user.username == username
    assert user.is_admin is False

    username2 = random_str()
    password2 = random_str()
    with User(username, password, monkeypatch), assert_unauthorized():
        client.create_user(username2, password2)


def test_get_user(client, monkeypatch):
    username = random_str()
    password = random_str()
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.create_user(username, password)

    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        user = client.get_user(username)
    assert user.username == username

    with assert_unauthenticated():
        client.get_user(username)

    username2 = random_str()
    password2 = random_str()
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.create_user(username2, password2)
    with User(username2, password2, monkeypatch), assert_unauthorized():
        client.get_user(username)


def test_get_current_user(client, monkeypatch):
    # /users/current returns minimal identity for whoever the request is
    # authenticated as. The admin UI relies on the response shape (id,
    # username, is_admin) to gate admin-only links.
    username = random_str()
    password = random_str()
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        created = client.create_user(username, password)

    url = f"{client.tracking_uri}/api/2.0/mlflow/users/current"

    resp = requests.get(url, auth=(username, password))
    assert resp.status_code == 200
    body = resp.json()
    assert body["user"] == {
        "id": created.id,
        "username": username,
        "is_admin": False,
    }
    # ``is_basic_auth`` lets the frontend gate Basic-Auth-only flows
    # (logout XHR, change-password) when a custom authorization_function
    # is configured.
    assert body["is_basic_auth"] is True

    resp = requests.get(url, auth=(ADMIN_USERNAME, ADMIN_PASSWORD))
    assert resp.status_code == 200
    admin_payload = resp.json()["user"]
    assert admin_payload["username"] == ADMIN_USERNAME
    assert admin_payload["is_admin"] is True

    resp = requests.get(url)
    assert resp.status_code == 401


@pytest.mark.parametrize(
    ("path", "method"),
    [
        ("/api/2.0/mlflow/experiments/permissions/get", "GET"),
        ("/api/2.0/mlflow/experiments/permissions/create", "POST"),
        ("/api/2.0/mlflow/experiments/permissions/update", "PATCH"),
        ("/api/2.0/mlflow/experiments/permissions/delete", "DELETE"),
        ("/api/2.0/mlflow/registered-models/permissions/get", "GET"),
        ("/api/2.0/mlflow/registered-models/permissions/create", "POST"),
        ("/api/2.0/mlflow/registered-models/permissions/update", "PATCH"),
        ("/api/2.0/mlflow/registered-models/permissions/delete", "DELETE"),
        ("/api/3.0/mlflow/scorers/permissions/get", "GET"),
        ("/api/3.0/mlflow/scorers/permissions/create", "POST"),
        ("/api/3.0/mlflow/scorers/permissions/update", "PATCH"),
        ("/api/3.0/mlflow/scorers/permissions/delete", "DELETE"),
        ("/api/3.0/mlflow/gateway/secrets/permissions/get", "GET"),
        ("/api/3.0/mlflow/gateway/secrets/permissions/create", "POST"),
        ("/api/3.0/mlflow/gateway/secrets/permissions/update", "PATCH"),
        ("/api/3.0/mlflow/gateway/secrets/permissions/delete", "DELETE"),
        ("/api/3.0/mlflow/gateway/endpoints/permissions/get", "GET"),
        ("/api/3.0/mlflow/gateway/endpoints/permissions/create", "POST"),
        ("/api/3.0/mlflow/gateway/endpoints/permissions/update", "PATCH"),
        ("/api/3.0/mlflow/gateway/endpoints/permissions/delete", "DELETE"),
        ("/api/3.0/mlflow/gateway/model-definitions/permissions/get", "GET"),
        ("/api/3.0/mlflow/gateway/model-definitions/permissions/create", "POST"),
        ("/api/3.0/mlflow/gateway/model-definitions/permissions/update", "PATCH"),
        ("/api/3.0/mlflow/gateway/model-definitions/permissions/delete", "DELETE"),
    ],
)
def test_legacy_permission_endpoints_return_404(client, path, method):
    # Regression guard: the deprecated per-resource permission endpoints were
    # removed (RFC mprahl review M6+M9 — "rip the band-aid"). Replacement is
    # the ``grant_user_permission`` / ``revoke_user_permission`` /
    # ``check_user_permission`` convenience APIs.
    resp = requests.request(
        method, client.tracking_uri + path, auth=(ADMIN_USERNAME, ADMIN_PASSWORD)
    )
    assert resp.status_code == 404, (
        f"{method} {path} unexpectedly returned {resp.status_code} — legacy "
        "permission endpoints must be removed"
    )


def _register_scorer(tracking_uri: str, experiment_id: str, name: str, auth) -> None:
    resp = requests.post(
        tracking_uri + "/api/3.0/mlflow/scorers/register",
        json={
            "experiment_id": experiment_id,
            "name": name,
            "serialized_scorer": json.dumps({"v": 1}),
        },
        auth=auth,
    )
    resp.raise_for_status()


def test_list_scorers_cross_experiment(client, monkeypatch):
    # ``ListScorers`` with no ``experiment_id`` returns scorers across every
    # experiment in the active workspace, populates ``experiment_name`` per
    # row, and the SqlAlchemyStore output is globally sorted on
    # ``(experiment_id, scorer_name)`` (pinned in store-level tests; smoke
    # check here that the order survives through to the response).
    url = client.tracking_uri + "/ajax-api/3.0/mlflow/scorers/list"

    # Unauthenticated callers get 401 — ``_before_request`` runs before any
    # per-route validator.
    assert requests.get(url).status_code == 401

    exp_a = _create_experiment(client.tracking_uri, monkeypatch, f"xex-a-{random_str()}")
    exp_b = _create_experiment(client.tracking_uri, monkeypatch, f"xex-b-{random_str()}")
    admin_auth = (ADMIN_USERNAME, ADMIN_PASSWORD)
    _register_scorer(client.tracking_uri, exp_a, "alpha", admin_auth)
    _register_scorer(client.tracking_uri, exp_b, "beta", admin_auth)
    # Scorer name with a ``/`` exercises the URL-encoding contract; the
    # client-side ``scorerResourcePattern`` must match the persisted grant
    # byte-for-byte.
    _register_scorer(client.tracking_uri, exp_b, "with/slash", admin_auth)

    resp = requests.get(url, auth=admin_auth)
    assert resp.status_code == 200, resp.text
    scorers = resp.json()["scorers"]

    by_name = {s["scorer_name"]: s for s in scorers}
    assert set(by_name) >= {"alpha", "beta", "with/slash"}
    assert str(by_name["alpha"]["experiment_id"]) == exp_a
    # Reconstruct the resource pattern client-side and verify it matches the
    # encoding the auth-side ``_scorer_pattern`` uses.
    assert f"{by_name['with/slash']['experiment_id']}/{quote('with/slash', safe='')}" == (
        f"{exp_b}/{quote('with/slash', safe='')}"
    )

    tuples = [(s["experiment_id"], s["scorer_name"]) for s in scorers]
    assert tuples == sorted(tuples)


def test_list_scorers_cross_experiment_pattern_round_trip(client, monkeypatch):
    # A scorer literally named ``*`` is the encoding edge case: Python's
    # ``quote(safe='')`` encodes it as ``%2A`` while JS ``encodeURIComponent``
    # leaves it bare. The picker reconstructs the composite pattern via the
    # frontend ``scorerResourcePattern`` helper (which matches Python's
    # encoding); this test pins that the backend's ``_scorer_pattern`` uses
    # the same encoding, so a grant placed on the picker-submitted pattern
    # resolves on subsequent permission checks.
    admin_auth = (ADMIN_USERNAME, ADMIN_PASSWORD)
    exp_id = _create_experiment(client.tracking_uri, monkeypatch, f"star-{random_str()}")
    _register_scorer(client.tracking_uri, exp_id, "*", admin_auth)

    # Grant on the composite pattern the picker would submit.
    expected_pattern = f"{exp_id}/{quote('*', safe='')}"
    target, _ = _new_user(client, monkeypatch)
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.grant_user_permission(target, "scorer", expected_pattern, "READ")
        roles = client.list_user_roles(target)
    synthetic = next(r for r in roles if r.name.startswith("__user_"))
    grants = [(p.resource_type, p.resource_pattern, p.permission) for p in synthetic.permissions]
    assert ("scorer", expected_pattern, "READ") in grants

    # Cross-experiment ListScorers should carry the ``*``-named row. Reconstruct
    # the composite pattern client-side and verify it matches the granted one
    # byte-for-byte.
    url = client.tracking_uri + "/ajax-api/3.0/mlflow/scorers/list"
    resp = requests.get(url, auth=admin_auth)
    assert resp.status_code == 200, resp.text
    by_name = {s["scorer_name"]: s for s in resp.json()["scorers"]}
    assert "*" in by_name
    reconstructed = f"{by_name['*']['experiment_id']}/{quote('*', safe='')}"
    assert reconstructed == expected_pattern


def test_update_user_password(client, monkeypatch):
    username = random_str()
    password = random_str()
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.create_user(username, password)

    new_password = random_str()
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.update_user_password(username, new_password)

    with User(username, password, monkeypatch), assert_unauthenticated():
        client.get_user(username)

    with User(username, new_password, monkeypatch):
        client.get_user(username)

    with assert_unauthenticated():
        client.update_user_password(username, new_password)

    username2 = random_str()
    password2 = random_str()
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.create_user(username2, password2)
    with User(username2, password2, monkeypatch), assert_unauthorized():
        client.update_user_password(username, new_password)


def test_update_user_password_self_service_rejects_same_password(client, monkeypatch):
    # Self-service: rotating to the existing value is a no-op and almost
    # always a UI mistake. Cheap equality check against the supplied
    # ``current_password`` rather than re-authenticating.
    username = random_str()
    password = random_str()
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.create_user(username, password)
    with (
        User(username, password, monkeypatch),
        pytest.raises(MlflowException, match=r"differ from the current password") as exc,
    ):
        client.update_user_password(username, password, current_password=password)
    assert exc.value.error_code == ErrorCode.Name(INVALID_PARAMETER_VALUE)


def test_update_user_password_admin_allows_same_password(client, monkeypatch):
    # Admin path has no ``current_password``, so we can't equality-check
    # cheaply; rejecting via bcrypt would leak a same-vs-different oracle
    # to any admin probing a candidate. Allow the silent no-op instead.
    username = random_str()
    password = random_str()
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.create_user(username, password)
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.update_user_password(username, password)
    with User(username, password, monkeypatch):
        client.get_user(username)


def test_self_service_password_change_requires_current_password(client, monkeypatch):
    # Defense-in-depth: a user changing their own password must re-assert the
    # current password. Admins changing someone else's password don't (and
    # can't) supply it — that path is exercised in `test_update_user_password`.
    username = random_str()
    password = random_str()
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.create_user(username, password)

    new_password = random_str()

    with User(username, password, monkeypatch):
        # Missing current_password: rejected.
        with pytest.raises(MlflowException, match="Current password is required"):
            client.update_user_password(username, new_password)

        # Wrong current_password: rejected.
        with pytest.raises(MlflowException, match="Current password does not match"):
            client.update_user_password(
                username, new_password, current_password="not-the-current-password"
            )

        # Correct current_password: accepted.
        client.update_user_password(username, new_password, current_password=password)

    # Old password no longer authenticates; new one does.
    with User(username, password, monkeypatch), assert_unauthenticated():
        client.get_user(username)
    with User(username, new_password, monkeypatch):
        client.get_user(username)


def test_create_user_with_null_or_missing_json_body_returns_400(client, monkeypatch):
    # Defensive: a JSON-typed POST whose body is literal ``null`` (or empty)
    # used to crash ``_get_request_param`` with ``None | dict`` and surface as
    # a 500. Make sure it's a clean 400 instead.
    url = f"{client.tracking_uri}/api/2.0/mlflow/users/create"
    headers = {"Content-Type": "application/json"}
    auth = (ADMIN_USERNAME, ADMIN_PASSWORD)

    # Literal null body.
    resp = requests.post(url, data="null", headers=headers, auth=auth)
    assert resp.status_code == 400

    # Empty body.
    resp = requests.post(url, data="", headers=headers, auth=auth)
    assert resp.status_code == 400


def test_self_service_password_change_with_null_body_returns_400(client, monkeypatch):
    # Defensive: self-service password changes used to crash with
    # ``request.json.get(...)`` when the body was literal ``null`` (raises
    # AttributeError -> 500). Make sure it's the standard 400 instead.
    username = random_str()
    password = random_str()
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.create_user(username, password)

    url = f"{client.tracking_uri}/api/2.0/mlflow/users/update-password"
    resp = requests.patch(
        url,
        data="null",
        headers={"Content-Type": "application/json"},
        auth=(username, password),
    )
    assert resp.status_code == 400


def test_update_user_admin(client, monkeypatch):
    username = random_str()
    password = random_str()
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.create_user(username, password)

    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.update_user_admin(username, True)
        user = client.get_user(username)
        assert user.is_admin is True

    with assert_unauthenticated():
        client.update_user_admin(username, True)

    username2 = random_str()
    password2 = random_str()
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.create_user(username2, password2)
    with User(username2, password2, monkeypatch), assert_unauthorized():
        client.update_user_admin(username, True)


def test_delete_user(client, monkeypatch):
    username = random_str()
    password = random_str()
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.create_user(username, password)

    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.update_user_admin(username, True)
        client.delete_user(username)
        with pytest.raises(
            MlflowException,
            match=rf"User with username={username} not found",
        ) as exception_context:
            client.get_user(username)
        assert exception_context.value.error_code == ErrorCode.Name(RESOURCE_DOES_NOT_EXIST)

    with assert_unauthenticated():
        client.delete_user(username)

    username2 = random_str()
    password2 = random_str()
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.create_user(username2, password2)
    with User(username2, password2, monkeypatch), assert_unauthorized():
        client.delete_user(username)


def test_delete_user_rejects_self_delete(client, monkeypatch):
    with (
        User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch),
        pytest.raises(MlflowException, match=r"cannot delete their own account") as exc,
    ):
        client.delete_user(ADMIN_USERNAME)
    assert exc.value.error_code == ErrorCode.Name(BAD_REQUEST)
    # Admin account still exists and is still usable.
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        assert client.get_user(ADMIN_USERNAME).username == ADMIN_USERNAME


# ---- Unified per-user permission convenience APIs ----


def _create_experiment(tracking_uri: str, monkeypatch, name: str) -> str:
    """Create an experiment as admin so ``get_user_permission`` workspace
    lookup succeeds; without a real row the resolver falls through to default.
    """
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        return MlflowClient(tracking_uri).create_experiment(name)


def _new_user(client, monkeypatch):
    """Create a fresh user as admin and return (username, password)."""
    username = random_str()
    password = random_str()
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.create_user(username, password)
    return username, password


def test_grant_user_permission_roundtrip(client, monkeypatch):
    # Admin grants READ on an experiment to a new user, then the user's
    # synthetic role surfaces the row via list_user_roles.
    username, _ = _new_user(client, monkeypatch)
    exp_id = _create_experiment(client.tracking_uri, monkeypatch, f"grant-{random_str()}")
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.grant_user_permission(username, "experiment", exp_id, "READ")
        roles = client.list_user_roles(username)

    synthetic = [r for r in roles if r.name.startswith("__user_")]
    assert len(synthetic) == 1
    grants = [(p.resource_type, p.resource_pattern, p.permission) for p in synthetic[0].permissions]
    assert ("experiment", exp_id, "READ") in grants


def test_grant_user_permission_duplicate_raises(client, monkeypatch):
    username, _ = _new_user(client, monkeypatch)
    exp_id = _create_experiment(client.tracking_uri, monkeypatch, f"dup-{random_str()}")
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.grant_user_permission(username, "experiment", exp_id, "READ")
        with pytest.raises(MlflowException, match="already exists") as exc:
            client.grant_user_permission(username, "experiment", exp_id, "EDIT")
    from mlflow.protos.databricks_pb2 import RESOURCE_ALREADY_EXISTS

    assert exc.value.error_code == ErrorCode.Name(RESOURCE_ALREADY_EXISTS)


def test_grant_user_permission_unknown_user_raises(client, monkeypatch):
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        with pytest.raises(MlflowException, match="not found") as exc:
            client.grant_user_permission(
                "no-such-user-" + random_str(), "experiment", "exp-1", "READ"
            )
    assert exc.value.error_code == ErrorCode.Name(RESOURCE_DOES_NOT_EXIST)


def test_grant_user_permission_invalid_resource_type(client, monkeypatch):
    username, _ = _new_user(client, monkeypatch)
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        with pytest.raises(MlflowException, match="Invalid resource type"):
            client.grant_user_permission(username, "bogus", "x", "READ")


def test_grant_user_permission_malformed_scorer_id_returns_clean_4xx(client, monkeypatch):
    # Malformed scorer ``resource_id`` raises inside ``validate_can_manage_resource``
    # (a before-request validator). Pin that the wrapping ``catch_mlflow_exception``
    # turns it into a 400, not a 500.
    target_username, _ = _new_user(client, monkeypatch)
    requester_username, requester_password = _new_user(client, monkeypatch)
    with User(requester_username, requester_password, monkeypatch):
        with pytest.raises(MlflowException, match="Invalid scorer resource_id") as exc:
            client.grant_user_permission(target_username, "scorer", "no-slash-here", "READ")
    assert exc.value.error_code == ErrorCode.Name(INVALID_PARAMETER_VALUE)


@pytest.mark.parametrize(
    ("api_method", "args"),
    [
        ("grant_user_permission", ("workspace", "*", "USE")),
        ("revoke_user_permission", ("workspace", "*")),
    ],
    ids=["grant", "revoke"],
)
def test_admin_cannot_target_workspace_resource_type(client, monkeypatch, api_method, args):
    # Super admins skip ``validate_can_manage_resource`` via ``sender_is_admin``.
    # Store-level rejection is the only defense — must fire for the admin path.
    username, _ = _new_user(client, monkeypatch)
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        with pytest.raises(MlflowException, match="not supported by the per-user"):
            getattr(client, api_method)(username, *args)


def test_grant_user_permission_invalid_permission_for_resource_type(client, monkeypatch):
    # ``NO_PERMISSIONS`` is intentionally disallowed at the resource scope —
    # absence of a grant + ``default_permission`` already expresses "no access".
    username, _ = _new_user(client, monkeypatch)
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        with pytest.raises(MlflowException, match="Invalid permission"):
            client.grant_user_permission(username, "experiment", "exp-1", "NO_PERMISSIONS")


def test_revoke_user_permission_roundtrip(client, monkeypatch):
    username, _ = _new_user(client, monkeypatch)
    exp_id = _create_experiment(client.tracking_uri, monkeypatch, f"revoke-{random_str()}")
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.grant_user_permission(username, "experiment", exp_id, "READ")
        roles_before = client.list_user_roles(username)
        synthetic_before = next(r for r in roles_before if r.name.startswith("__user_"))
        assert any(p.resource_pattern == exp_id for p in synthetic_before.permissions)

        client.revoke_user_permission(username, "experiment", exp_id)
        roles_after = client.list_user_roles(username)
        synthetic_after = next((r for r in roles_after if r.name.startswith("__user_")), None)
        remaining = synthetic_after.permissions if synthetic_after else []
        assert not any(p.resource_pattern == exp_id for p in remaining)


def test_revoke_user_permission_missing_row_raises(client, monkeypatch):
    username, _ = _new_user(client, monkeypatch)
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        with pytest.raises(MlflowException, match="not found") as exc:
            client.revoke_user_permission(username, "experiment", "exp-not-granted")
    assert exc.value.error_code == ErrorCode.Name(RESOURCE_DOES_NOT_EXIST)


def test_get_user_permission_self_check(client, monkeypatch):
    # A non-admin user can check their own permissions on any resource.
    username, password = _new_user(client, monkeypatch)
    exp_id = _create_experiment(client.tracking_uri, monkeypatch, f"self-{random_str()}")
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.grant_user_permission(username, "experiment", exp_id, "READ")

    with User(username, password, monkeypatch):
        result = client.get_user_permission(username, "experiment", exp_id)

    # ``allowed`` mirrors ``Permission.can_use``. READ is defined with
    # ``can_use=False`` in ``mlflow.server.auth.permissions`` (see ``READ =``),
    # so a READ grant resolves to ``allowed=False``.
    assert result.permission == "READ"
    assert result.allowed is False


def test_get_user_permission_returns_max_grant(client, monkeypatch):
    # An explicit MANAGE grant yields allowed=True (MANAGE.can_use=True).
    username, password = _new_user(client, monkeypatch)
    exp_id = _create_experiment(client.tracking_uri, monkeypatch, f"mgr-{random_str()}")
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.grant_user_permission(username, "experiment", exp_id, "MANAGE")

    with User(username, password, monkeypatch):
        result = client.get_user_permission(username, "experiment", exp_id)

    assert result.permission == "MANAGE"
    assert result.allowed is True


def test_get_user_permission_cross_user_requires_admin(client, monkeypatch):
    # A plain user cannot check another user's permissions.
    target, _ = _new_user(client, monkeypatch)
    requester, requester_pw = _new_user(client, monkeypatch)

    with User(requester, requester_pw, monkeypatch), assert_unauthorized():
        client.get_user_permission(target, "experiment", "exp-x")


def test_get_user_permission_admin_can_check_any_user(client, monkeypatch):
    username, _ = _new_user(client, monkeypatch)
    exp_id = _create_experiment(client.tracking_uri, monkeypatch, f"admin-{random_str()}")
    with User(ADMIN_USERNAME, ADMIN_PASSWORD, monkeypatch):
        client.grant_user_permission(username, "experiment", exp_id, "EDIT")
        result = client.get_user_permission(username, "experiment", exp_id)

    assert result.permission == "EDIT"
    assert result.allowed is True


def test_grant_user_permission_requires_authentication(client):
    with assert_unauthenticated():
        client.grant_user_permission("alice", "experiment", "exp-1", "READ")


def test_grant_user_permission_non_admin_without_manage_rejected(client, monkeypatch):
    # A plain user without per-resource MANAGE cannot grant permissions.
    requester, requester_pw = _new_user(client, monkeypatch)
    target, _ = _new_user(client, monkeypatch)

    with User(requester, requester_pw, monkeypatch), assert_unauthorized():
        client.grant_user_permission(target, "experiment", "exp-1", "READ")


@pytest.mark.parametrize("api_prefix", ["api", "ajax-api"])
@pytest.mark.parametrize(
    ("endpoint", "method"),
    [
        ("/3.0/mlflow/users/permissions/grant", "POST"),
        ("/3.0/mlflow/users/permissions/revoke", "POST"),
        ("/3.0/mlflow/users/permissions/get", "GET"),
    ],
)
def test_unified_permission_endpoints_reachable_at_both_path_prefixes(
    client, api_prefix, endpoint, method
):
    # The MLflow frontend hits /ajax-api/ paths; the Python client hits /api/ paths.
    # Every unified permission route must be reachable at both — a 404 here would
    # silently break the admin UI without surfacing as a permission-system failure.
    resp = requests.request(
        method,
        f"{client.tracking_uri}/{api_prefix}{endpoint}",
        auth=(ADMIN_USERNAME, ADMIN_PASSWORD),
    )
    assert resp.status_code != 404, f"{method} /{api_prefix}{endpoint} unexpectedly returned 404"


# Workspace permission methods that the RBAC migration removed from
# ``AuthServiceClient``. Their replacement is the unified
# ``grant_user_permission`` / ``revoke_user_permission`` surface. This list
# guards against accidental re-addition during refactors or merges.
_REMOVED_WORKSPACE_PERMISSION_METHODS = [
    "set_workspace_permission",
    "list_workspace_permissions",
    "delete_workspace_permission",
    "list_user_workspace_permissions",
]


@pytest.mark.parametrize("method_name", _REMOVED_WORKSPACE_PERMISSION_METHODS)
def test_removed_workspace_permission_methods_absent(method_name):
    fake_client = AuthServiceClient("http://127.0.0.1:1")
    assert not hasattr(fake_client, method_name), (
        f"{method_name} was removed in the RBAC migration but is still defined on AuthServiceClient"
    )


# Wire endpoints that the RBAC migration removed. Hitting these paths should
# not match any registered handler; Flask returns 404 for unregistered paths.
# Guards against accidental re-registration in ``create_app``.
@pytest.mark.parametrize(
    ("path", "method"),
    [
        ("/api/2.0/mlflow/workspaces/team-a/permissions/get", "GET"),
        ("/api/2.0/mlflow/workspaces/team-a/permissions/create", "POST"),
        ("/api/2.0/mlflow/workspaces/team-a/permissions/update", "PATCH"),
        ("/api/2.0/mlflow/workspaces/team-a/permissions/delete", "DELETE"),
        ("/api/3.0/mlflow/workspaces/team-a/permissions/get", "GET"),
        ("/api/3.0/mlflow/workspaces/team-a/permissions/create", "POST"),
        ("/api/3.0/mlflow/workspaces/team-a/permissions/update", "PATCH"),
        ("/api/3.0/mlflow/workspaces/team-a/permissions/delete", "DELETE"),
    ],
)
def test_removed_workspace_permission_endpoints_return_404(client, path, method):
    resp = requests.request(
        method, client.tracking_uri + path, auth=(ADMIN_USERNAME, ADMIN_PASSWORD)
    )
    assert resp.status_code == 404, (
        f"{method} {path} was removed in the RBAC migration but returned "
        f"{resp.status_code} (expected 404)"
    )