huihuiSquare/digital-avatar-app/backend/tests/test_authorizations_api.py

from fastapi.testclient import TestClient

from main import app


client = TestClient(app)


def test_authorization_list_is_scoped_to_owned_avatar(authorization_context):
    context = authorization_context
    response = client.get(
        f"/api/avatar/{context['avatar'].id}/authorizations",
        headers=context["owner_headers"],
    )
    assert response.status_code == 200
    payload = response.json()
    assert payload["code"] == 200
    assert [item["id"] for item in payload["data"]] == [context["authorization"].id]

    forbidden = client.get(
        f"/api/avatar/{context['other_avatar'].id}/authorizations",
        headers=context["owner_headers"],
    )
    assert forbidden.status_code == 403


def test_create_update_and_delete_authorization(authorization_context):
    context = authorization_context
    avatar_id = context["avatar"].id
    target_id = f"new-contact-{context['suffix']}"
    created = client.post(
        f"/api/avatar/{avatar_id}/authorizations",
        headers=context["owner_headers"],
        json={
            "targetType": "user",
            "targetId": target_id,
            "targetName": "新联系人",
            "permissions": ["friend", "chat", "browse"],
        },
    ).json()
    assert created["code"] == 200
    authorization_id = created["data"]["id"]
    assert created["data"]["permissions"] == ["friend", "chat", "browse"]

    duplicate = client.post(
        f"/api/avatar/{avatar_id}/authorizations",
        headers=context["owner_headers"],
        json={
            "targetType": "user",
            "targetId": target_id,
            "targetName": "重复联系人",
            "permissions": ["chat"],
        },
    ).json()
    assert duplicate["code"] == 409

    updated = client.put(
        f"/api/avatar/{avatar_id}/authorizations",
        headers=context["owner_headers"],
        json={
            "id": authorization_id,
            "targetName": "联系人新名称",
            "permissions": ["interact", "publish"],
        },
    ).json()
    assert updated["code"] == 200
    assert updated["data"]["targetName"] == "联系人新名称"
    assert updated["data"]["permissions"] == ["publish", "interact"]

    deleted = client.delete(
        f"/api/avatar/{avatar_id}/authorizations/{authorization_id}",
        headers=context["owner_headers"],
    ).json()
    assert deleted["code"] == 200
    assert deleted["data"]["id"] == authorization_id


def test_authorization_requires_login_and_rejects_unknown_permissions(authorization_context):
    context = authorization_context
    avatar_id = context["avatar"].id
    no_session = client.get(f"/api/avatar/{avatar_id}/authorizations")
    assert no_session.status_code == 401

    invalid = client.post(
        f"/api/avatar/{avatar_id}/authorizations",
        headers=context["owner_headers"],
        json={
            "targetType": "user",
            "targetId": "invalid-target",
            "targetName": "无效权限",
            "permissions": ["admin"],
        },
    ).json()
    assert invalid["code"] == 400


def test_avatar_permission_settings_default_and_persist(authorization_context):
    context = authorization_context
    endpoint = f"/api/avatar/{context['avatar'].id}/permission-settings"

    initial = client.get(endpoint, headers=context["owner_headers"]).json()
    assert initial["code"] == 200
    assert initial["data"] == {
        "avatarId": context["avatar"].id,
        "permissions": ["friend", "chat"],
        "takeoverReplyDelaySeconds": 180,
    }

    updated = client.put(
        endpoint,
        headers=context["owner_headers"],
        json={"permissions": ["interact", "takeover", "publish", "friend", "friend"]},
    ).json()
    assert updated["code"] == 200
    assert updated["data"]["permissions"] == ["friend", "publish", "interact", "takeover"]

    reloaded = client.get(endpoint, headers=context["owner_headers"]).json()
    assert reloaded["data"]["permissions"] == ["friend", "publish", "interact", "takeover"]
    assert reloaded["data"]["takeoverReplyDelaySeconds"] == 180


def test_avatar_permission_settings_allow_all_disabled(authorization_context):
    context = authorization_context
    endpoint = f"/api/avatar/{context['avatar'].id}/permission-settings"

    response = client.put(
        endpoint,
        headers=context["owner_headers"],
        json={"permissions": []},
    ).json()
    assert response["code"] == 200
    assert response["data"]["permissions"] == []


def test_avatar_permission_settings_validate_owner_and_permissions(authorization_context):
    context = authorization_context
    endpoint = f"/api/avatar/{context['avatar'].id}/permission-settings"

    invalid = client.put(
        endpoint,
        headers=context["owner_headers"],
        json={"permissions": ["admin"]},
    ).json()
    assert invalid["code"] == 400

    missing = client.put(
        endpoint,
        headers=context["owner_headers"],
        json={},
    ).json()
    assert missing["code"] == 400

    forbidden = client.get(
        f"/api/avatar/{context['other_avatar'].id}/permission-settings",
        headers=context["owner_headers"],
    )
    assert forbidden.status_code == 403

    unauthenticated = client.get(endpoint)
    assert unauthenticated.status_code == 401


def test_takeover_delay_minimum_and_single_active_avatar_per_owner(authorization_context):
    from database import SessionLocal
    from models import Avatar

    context = authorization_context
    endpoint = f"/api/avatar/{context['avatar'].id}/permission-settings"
    invalid = client.put(
        endpoint,
        headers=context["owner_headers"],
        json={"permissions": ["chat"], "takeoverReplyDelaySeconds": 2},
    ).json()
    assert invalid["code"] == 400

    second_avatar_id = f"second-{context['suffix']}"
    db = SessionLocal()
    try:
        db.add(
            Avatar(
                id=second_avatar_id,
                owner_id=context["owner"].huihui_user_id,
                name="第二个分身",
                status="active",
                config={"authorizationPermissions": ["chat", "takeover"]},
            )
        )
        db.commit()
    finally:
        db.close()

    try:
        updated = client.put(
            endpoint,
            headers=context["owner_headers"],
            json={"permissions": ["chat", "takeover"], "takeoverReplyDelaySeconds": 3},
        ).json()
        assert updated["code"] == 200
        assert updated["data"]["takeoverReplyDelaySeconds"] == 3
        assert updated["data"]["disabledAvatarIds"] == [second_avatar_id]

        db = SessionLocal()
        try:
            second = db.query(Avatar).filter(Avatar.id == second_avatar_id).one()
            assert "takeover" not in second.config["authorizationPermissions"]
        finally:
            db.close()
    finally:
        db = SessionLocal()
        try:
            db.query(Avatar).filter(Avatar.id == second_avatar_id).delete()
            db.commit()
        finally:
            db.close()