"""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()