254 lines
8.3 KiB
Python
254 lines
8.3 KiB
Python
"""Tests for takeover configuration and BOXIM connection status."""
|
|
|
|
from datetime import datetime, timedelta
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from database import SessionLocal
|
|
from main import app
|
|
from models import Authorization, Avatar, TakeoverCursor, TakeoverReplyTask, User
|
|
|
|
|
|
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"]
|
|
|
|
|
|
def test_takeover_status_reports_disabled_and_requires_owner_login(authorization_context):
|
|
context = authorization_context
|
|
endpoint = f"/api/avatar/{context['avatar'].id}/takeover/status"
|
|
|
|
disabled = client.get(endpoint, headers=context["owner_headers"])
|
|
assert disabled.status_code == 200
|
|
assert disabled.json()["data"]["status"] == "disabled"
|
|
|
|
client.put(
|
|
f"/api/avatar/{context['avatar'].id}/permission-settings",
|
|
headers=context["owner_headers"],
|
|
json={"permissions": ["chat", "takeover"]},
|
|
)
|
|
needs_login = client.get(endpoint, headers=context["owner_headers"]).json()["data"]
|
|
assert needs_login["enabled"] is True
|
|
assert needs_login["status"] == "needs_login"
|
|
assert "BOXIM" in needs_login["message"]
|
|
|
|
assert client.get(endpoint).status_code == 401
|
|
assert client.get(endpoint, headers=context["other_headers"]).status_code == 403
|
|
|
|
|
|
def test_takeover_status_reports_ready_pending_count_and_errors(authorization_context):
|
|
context = authorization_context
|
|
avatar_id = context["avatar"].id
|
|
endpoint = f"/api/avatar/{avatar_id}/takeover/status"
|
|
client.put(
|
|
f"/api/avatar/{avatar_id}/permission-settings",
|
|
headers=context["owner_headers"],
|
|
json={"permissions": ["chat", "takeover"]},
|
|
)
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
owner = db.query(User).filter(User.id == context["owner"].id).one()
|
|
owner.huihui_token = "production-login-token"
|
|
cursor = TakeoverCursor(
|
|
avatar_id=avatar_id,
|
|
owner_id=owner.huihui_user_id,
|
|
boxim_owner_id="100",
|
|
last_message_id="10",
|
|
initialized=True,
|
|
last_polled_at=datetime.utcnow(),
|
|
)
|
|
task = TakeoverReplyTask(
|
|
avatar_id=avatar_id,
|
|
owner_id=owner.huihui_user_id,
|
|
peer_id="200",
|
|
trigger_message_id="11",
|
|
source_message_ids=["11"],
|
|
prompt="你好",
|
|
status="pending",
|
|
scheduled_at=datetime.utcnow(),
|
|
boxim_local_id="123",
|
|
)
|
|
db.add_all([cursor, task])
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
ready = client.get(endpoint, headers=context["owner_headers"]).json()["data"]
|
|
assert ready["status"] == "ready"
|
|
assert ready["pendingCount"] == 1
|
|
assert ready["lastPolledAt"]
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
cursor = db.query(TakeoverCursor).filter(TakeoverCursor.avatar_id == avatar_id).one()
|
|
cursor.last_polled_at = datetime.utcnow() - timedelta(seconds=30)
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
long_polling = client.get(endpoint, headers=context["owner_headers"]).json()["data"]
|
|
assert long_polling["status"] == "ready"
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
cursor = db.query(TakeoverCursor).filter(TakeoverCursor.avatar_id == avatar_id).one()
|
|
cursor.last_polled_at = datetime.utcnow() - timedelta(seconds=61)
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
stale = client.get(endpoint, headers=context["owner_headers"]).json()["data"]
|
|
assert stale["status"] == "connecting"
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
cursor = db.query(TakeoverCursor).filter(TakeoverCursor.avatar_id == avatar_id).one()
|
|
cursor.last_error = "BOXIM 暂时不可用"
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
failed = client.get(endpoint, headers=context["owner_headers"]).json()["data"]
|
|
assert failed["status"] == "error"
|
|
assert failed["message"] == "BOXIM 暂时不可用"
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
avatar = db.query(Avatar).filter(Avatar.id == avatar_id).one()
|
|
avatar.config = {"authorizationPermissions": ["chat"]}
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
auto_disabled = client.get(endpoint, headers=context["owner_headers"]).json()["data"]
|
|
assert auto_disabled["enabled"] is False
|
|
assert auto_disabled["status"] == "error"
|
|
|
|
client.put(
|
|
f"/api/avatar/{avatar_id}/permission-settings",
|
|
headers=context["owner_headers"],
|
|
json={"permissions": ["chat", "takeover"]},
|
|
)
|
|
db = SessionLocal()
|
|
try:
|
|
cursor = db.query(TakeoverCursor).filter(TakeoverCursor.avatar_id == avatar_id).one()
|
|
assert cursor.initialized is False
|
|
assert cursor.last_message_id == "0"
|
|
assert cursor.last_error == ""
|
|
finally:
|
|
db.close()
|