generated from kgod/ai-review-template
258 lines
8.9 KiB
Python
258 lines
8.9 KiB
Python
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]
|