from __future__ import annotations import json from pathlib import Path from typing import Any import httpx from fastapi.testclient import TestClient from app.main import create_app from app.offerpai_auth import ( OfferPaiAuthClient, OfferPaiAuthError, OfferPaiIdentity, ) from app.settings import Settings from test_api import BASE, active_component, event TOKEN = "header.payload.signature-value" def test_offerpai_client_uses_cookie_for_both_get_requests() -> None: requests: list[httpx.Request] = [] def handler(request: httpx.Request) -> httpx.Response: requests.append(request) assert request.headers.get("cookie") == f"Token={TOKEN}" if request.url.path == "/api/public/checkLogin": return httpx.Response(200, json={"code": "0", "data": True}) if request.url.path == "/api/user/manage/info": return httpx.Response( 200, json={ "code": "0", "data": { "id": "2081575100391407617", "mobileNumber": "13421012384", "nick": "用户2384", "inviteCode": "2GOTI0H0X7", "createTime": 1785121152000, }, }, ) return httpx.Response(404) with httpx.Client( base_url="https://test.offerpai.com.cn", transport=httpx.MockTransport(handler), ) as http_client: identity = OfferPaiAuthClient( "https://test.offerpai.com.cn", client=http_client ).authenticate(TOKEN) assert [request.method for request in requests] == ["GET", "GET"] assert [request.url.path for request in requests] == [ "/api/public/checkLogin", "/api/user/manage/info", ] assert identity.user_id == "2081575100391407617" assert identity.mobile_number == "13421012384" assert identity.nick == "用户2384" def test_offerpai_client_rejects_failed_login_without_profile_request() -> None: requests: list[httpx.Request] = [] def handler(request: httpx.Request) -> httpx.Response: requests.append(request) return httpx.Response(200, json={"code": "0", "data": False}) with httpx.Client( base_url="https://test.offerpai.com.cn", transport=httpx.MockTransport(handler), ) as http_client: provider = OfferPaiAuthClient("https://test.offerpai.com.cn", client=http_client) try: provider.authenticate(TOKEN) except OfferPaiAuthError as exc: assert exc.code == "external_auth_invalid" assert exc.status_code == 401 else: raise AssertionError("Expected invalid external authentication") assert len(requests) == 1 class FakeIdentityProvider: def __init__(self, error: OfferPaiAuthError | None = None) -> None: self.error = error self.tokens: list[str] = [] def authenticate(self, token: str) -> OfferPaiIdentity: self.tokens.append(token) if self.error is not None: raise self.error return OfferPaiIdentity( user_id="2081575100391407617", mobile_number="13421012384", nick="用户2384", invite_code="2GOTI0H0X7", create_time=1785121152000, ) def auth_client( tmp_path: Path, provider: Any, *, auth_required: bool = True ) -> tuple[Any, TestClient]: application = create_app( database_path=tmp_path / "offerpai-auth.db", cors_origins=["http://localhost:5173"], settings=Settings( llm_provider="rule", offerpai_auth_required=auth_required ), offerpai_identity_provider=provider, ) return application, TestClient(application) def test_session_creation_authenticates_and_defaults_account_phone(tmp_path: Path) -> None: provider = FakeIdentityProvider() application, client = auth_client(tmp_path, provider) with client: response = client.post( f"{BASE}/sessions", json={}, headers={"Authorization": f"Bearer {TOKEN}"}, ) assert response.status_code == 201, response.text created = response.json() session_id = created["session_id"] with application.state.database.transaction() as connection: session = application.state.database.fetch_session(connection, session_id) assert session is not None profile = session["profile"] assert profile["account_phone"] == "13421012384" assert profile["external_account"] == { "provider": "offerpai", "user_id": "2081575100391407617", "mobile_number": "13421012384", "nick": "用户2384", "invite_code": "2GOTI0H0X7", "create_time": 1785121152000, } assert TOKEN not in json.dumps(profile, ensure_ascii=False) auth_headers = {"Authorization": f"Bearer {TOKEN}"} phone_selector = event( client, session_id, created, "accept", {"accepted": True}, headers=auth_headers, ).json() data = active_component(phone_selector)["data"] assert data["has_account_phone"] is True assert data["masked_phone"] == "134****2384" assert data["default_value"] == "account" personal = event( client, session_id, phone_selector, "select", {"source": "account"}, headers=auth_headers, ) assert personal.status_code == 200, personal.text with application.state.database.transaction() as connection: updated = application.state.database.fetch_session(connection, session_id) assert updated is not None assert updated["profile"]["phone"] == "13421012384" assert updated["profile"]["phone_source"] == "account" assert provider.tokens == [TOKEN, TOKEN, TOKEN] def test_invalid_external_token_does_not_create_session(tmp_path: Path) -> None: provider = FakeIdentityProvider( OfferPaiAuthError( "external_auth_invalid", "登录凭证无效或已过期，请重新从 OfferPai 进入。", status_code=401, ) ) application, client = auth_client(tmp_path, provider) with client: response = client.post( f"{BASE}/sessions", json={}, headers={"Authorization": f"Bearer {TOKEN}"}, ) assert response.status_code == 401 assert response.json()["error"]["code"] == "external_auth_invalid" with application.state.database.transaction() as connection: count = connection.execute("SELECT COUNT(*) FROM sessions").fetchone()[0] assert count == 0 def test_optional_anonymous_mode_does_not_call_external_provider(tmp_path: Path) -> None: provider = FakeIdentityProvider() _application, client = auth_client(tmp_path, provider, auth_required=False) with client: response = client.post(f"{BASE}/sessions", json={}) assert response.status_code == 201 assert provider.tokens == [] def test_optional_mode_still_protects_sessions_bound_to_offerpai( tmp_path: Path, ) -> None: provider = FakeIdentityProvider() _application, client = auth_client(tmp_path, provider, auth_required=False) with client: created = client.post( f"{BASE}/sessions", json={}, headers={"Authorization": f"Bearer {TOKEN}"}, ) assert created.status_code == 201, created.text session_id = created.json()["session_id"] missing = client.get(f"{BASE}/sessions/{session_id}/timeline") assert missing.status_code == 401 assert missing.json()["error"]["code"] == "external_auth_required" def test_authentication_is_required_when_enabled(tmp_path: Path) -> None: provider = FakeIdentityProvider() _application, client = auth_client(tmp_path, provider) with client: response = client.post(f"{BASE}/sessions", json={}) assert response.status_code == 401 assert response.json()["error"]["code"] == "external_auth_required" assert provider.tokens == [] def test_authenticated_start_restores_latest_user_session(tmp_path: Path) -> None: provider = FakeIdentityProvider() _application, client = auth_client(tmp_path, provider, auth_required=True) with client: first = client.post( f"{BASE}/sessions", json={}, headers={"Authorization": f"Bearer {TOKEN}"}, ) second = client.post( f"{BASE}/sessions", json={}, headers={"Authorization": f"Bearer {TOKEN}"}, ) assert first.status_code == 201 assert second.status_code == 201 assert second.json()["session_id"] == first.json()["session_id"] assert provider.tokens == [TOKEN, TOKEN]