126 lines
3.9 KiB
Python
126 lines
3.9 KiB
Python
"""Tests for the authorization takeover configuration endpoint."""
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from database import SessionLocal
|
|
from main import app
|
|
from models import Authorization
|
|
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
def test_update_takeover_accepts_camel_case_and_persists(authorization_context):
|
|
context = authorization_context
|
|
response = client.put(
|
|
f"/api/avatar/{context['avatar'].id}/authorizations/takeover",
|
|
headers=context["owner_headers"],
|
|
json={
|
|
"authorizationId": context["authorization"].id,
|
|
"takeoverEnabled": True,
|
|
"takeoverMode": "delayed",
|
|
"takeoverDelaySeconds": 60,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["code"] == 200
|
|
assert payload["data"]["takeoverEnabled"] is True
|
|
assert payload["data"]["takeoverMode"] == "delayed"
|
|
assert payload["data"]["takeoverDelaySeconds"] == 60
|
|
assert "takeover" in payload["data"]["permissions"]
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
stored = db.query(Authorization).filter(
|
|
Authorization.id == context["authorization"].id
|
|
).first()
|
|
assert stored.takeover_enabled is True
|
|
assert stored.takeover_mode == "delayed"
|
|
assert stored.takeover_delay_seconds == 60
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_disabling_authorization_also_disables_takeover(authorization_context):
|
|
context = authorization_context
|
|
endpoint = f"/api/avatar/{context['avatar'].id}/authorizations/takeover"
|
|
client.put(
|
|
endpoint,
|
|
headers=context["owner_headers"],
|
|
json={
|
|
"authorizationId": context["authorization"].id,
|
|
"takeoverEnabled": True,
|
|
},
|
|
)
|
|
|
|
updated = client.put(
|
|
f"/api/avatar/{context['avatar'].id}/authorizations",
|
|
headers=context["owner_headers"],
|
|
json={"id": context["authorization"].id, "status": "inactive"},
|
|
).json()
|
|
assert updated["code"] == 200
|
|
assert updated["data"]["status"] == "inactive"
|
|
assert updated["data"]["takeoverEnabled"] is False
|
|
assert "takeover" not in updated["data"]["permissions"]
|
|
|
|
|
|
def test_takeover_rejects_invalid_values_and_cross_avatar_access(authorization_context):
|
|
context = authorization_context
|
|
endpoint = f"/api/avatar/{context['avatar'].id}/authorizations/takeover"
|
|
|
|
invalid_mode = client.put(
|
|
endpoint,
|
|
headers=context["owner_headers"],
|
|
json={
|
|
"authorization_id": context["authorization"].id,
|
|
"takeover_mode": "invalid",
|
|
},
|
|
).json()
|
|
assert invalid_mode["code"] == 400
|
|
|
|
invalid_delay = client.put(
|
|
endpoint,
|
|
headers=context["owner_headers"],
|
|
json={
|
|
"authorization_id": context["authorization"].id,
|
|
"takeover_delay_seconds": 2,
|
|
},
|
|
).json()
|
|
assert invalid_delay["code"] == 400
|
|
|
|
forbidden = client.put(
|
|
endpoint,
|
|
headers=context["other_headers"],
|
|
json={
|
|
"authorizationId": context["authorization"].id,
|
|
"takeoverEnabled": True,
|
|
},
|
|
)
|
|
assert forbidden.status_code == 403
|
|
|
|
|
|
def test_takeover_is_limited_to_active_user_authorizations(authorization_context):
|
|
context = authorization_context
|
|
avatar_id = context["avatar"].id
|
|
created = client.post(
|
|
f"/api/avatar/{avatar_id}/authorizations",
|
|
headers=context["owner_headers"],
|
|
json={
|
|
"targetType": "organization",
|
|
"targetId": f"org-{context['suffix']}",
|
|
"targetName": "测试组织",
|
|
"permissions": ["chat"],
|
|
},
|
|
).json()
|
|
response = client.put(
|
|
f"/api/avatar/{avatar_id}/authorizations/takeover",
|
|
headers=context["owner_headers"],
|
|
json={
|
|
"authorizationId": created["data"]["id"],
|
|
"takeoverEnabled": True,
|
|
},
|
|
).json()
|
|
assert response["code"] == 400
|
|
assert "单聊接管" in response["message"]
|