230 lines
9.1 KiB
Python
230 lines
9.1 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.main import create_app
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
PORTAL = ROOT / "app" / "static" / "portal"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_portal_root_redirects_to_public_home() -> None:
|
|
transport = httpx.ASGITransport(app=create_app())
|
|
async with httpx.AsyncClient(
|
|
transport=transport, base_url="http://test", follow_redirects=False
|
|
) as client:
|
|
response = await client.get("/")
|
|
assert response.status_code in {302, 307}
|
|
assert response.headers["location"] == "/portal/guest/home/"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"path",
|
|
[
|
|
"/portal/guest/home/",
|
|
"/portal/guest/products/",
|
|
"/portal/guest/product-detail/?code=510300",
|
|
"/portal/customer/login/",
|
|
"/portal/customer/dashboard/",
|
|
"/portal/customer/holdings/",
|
|
"/portal/customer/profit-loss/",
|
|
"/portal/customer/orders/",
|
|
"/portal/customer/transactions/",
|
|
"/portal/customer/cash-ledger/",
|
|
"/portal/customer/risk-questionnaire/",
|
|
"/portal/employee-console/login/",
|
|
"/portal/employee-console/workspace/",
|
|
"/portal/employee-risk/dashboard/",
|
|
],
|
|
)
|
|
async def test_public_portal_pages_are_served(path: str) -> None:
|
|
transport = httpx.ASGITransport(app=create_app())
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get(path)
|
|
assert response.status_code == 200
|
|
assert 'lang="zh-CN"' in response.text
|
|
assert '<meta charset="UTF-8">' in response.text
|
|
|
|
|
|
def test_every_portal_page_has_local_js_and_css_entry() -> None:
|
|
pages = list(PORTAL.glob("*/*/index.html"))
|
|
assert pages
|
|
for page in pages:
|
|
page_name = page.parent.name
|
|
assert (page.parent / f"{page_name}.js").is_file(), page
|
|
assert (page.parent / f"{page_name}.css").is_file(), page
|
|
|
|
|
|
def test_business_pages_do_not_call_fetch_directly() -> None:
|
|
direct_fetch_files = [
|
|
path.relative_to(PORTAL).as_posix()
|
|
for path in PORTAL.rglob("*.js")
|
|
if "fetch(" in path.read_text(encoding="utf-8")
|
|
]
|
|
assert direct_fetch_files == ["common/api-client.js"]
|
|
|
|
|
|
def test_api_client_registers_all_trading_endpoint_ids() -> None:
|
|
source = (PORTAL / "common" / "api-client.js").read_text(encoding="utf-8")
|
|
for endpoint_id in ("T001", "T002", "T003", "T004", "T005", "T006", "T007", "T008", "T009"):
|
|
assert f"{endpoint_id}:" in source
|
|
|
|
|
|
def test_product_detail_preserves_customer_session_for_trade_entry() -> None:
|
|
html = (PORTAL / "guest" / "product-detail" / "index.html").read_text(encoding="utf-8")
|
|
source = (PORTAL / "guest" / "product-detail" / "product-detail.js").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
dashboard = (PORTAL / "customer" / "dashboard" / "dashboard.js").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
assert 'data-trade-action' in html
|
|
assert "getAuthContext" in source
|
|
assert "textContent = '进入交易'" in source
|
|
assert "action=trade" in source
|
|
assert "productInput.value = productCode" in dashboard
|
|
|
|
|
|
def test_api_client_registers_onboarding_risk_and_admin_endpoints() -> None:
|
|
source = (PORTAL / "common" / "api-client.js").read_text(encoding="utf-8")
|
|
for endpoint_id in (
|
|
"ONB001", "ONB002", "RK001", "RK002", "RK003", "RK004", "RK005",
|
|
"RK006", "RK007", "RK008", "RK009", "RK010", "RK011", "RK012",
|
|
"RK013", "RK014", "RK015", "A002", "A003", "A004", "A005", "A006",
|
|
"A012", "A033", "A035", "A036", "A037", "A038", "A039", "A040",
|
|
):
|
|
assert f"{endpoint_id}:" in source
|
|
|
|
|
|
def test_advisor_workspace_registers_documented_operation_endpoints() -> None:
|
|
source = (PORTAL / "common" / "api-client.js").read_text(encoding="utf-8")
|
|
dashboard = (PORTAL / "employee-advisor" / "dashboard" / "index.html").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
for endpoint_id in (
|
|
"ADVISOR_PUBLISHED", "ADVISOR_GOAL", "ADVISOR_ANALYSIS",
|
|
"ADVISOR_ALLOCATION", "ADVISOR_RECOMMEND", "ADVISOR_CREATE_GOAL",
|
|
):
|
|
assert f"{endpoint_id}:" in source
|
|
for label in ("组合分析", "资产配置", "生成推荐草案", "录入客户目标"):
|
|
assert label in dashboard
|
|
|
|
|
|
def test_risk_scan_endpoint_uses_extended_timeout() -> None:
|
|
source = (PORTAL / "common" / "api-client.js").read_text(encoding="utf-8")
|
|
assert (
|
|
"RK006: { method: 'POST', path: '/api/v1/risk/alerts/scan', "
|
|
"idempotent: true, timeout: 60000 }" in source
|
|
)
|
|
assert "options.timeout || endpoint.timeout || 8000" in source
|
|
|
|
|
|
def test_customer_questionnaire_uses_server_contract() -> None:
|
|
source = (PORTAL / "customer" / "risk-questionnaire" / "risk-questionnaire.js").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
assert "ONB001" in source
|
|
assert "ONB002" in source
|
|
assert "declaration_accepted: true" in source
|
|
assert "total_score" not in source
|
|
|
|
|
|
def test_questionnaire_is_customer_only_and_auth_context_matches_token() -> None:
|
|
auth = (PORTAL / "common" / "auth.js").read_text(encoding="utf-8")
|
|
login = (PORTAL / "common" / "login-controller.js").read_text(encoding="utf-8")
|
|
questionnaire = (
|
|
PORTAL / "customer" / "risk-questionnaire" / "risk-questionnaire.js"
|
|
).read_text(encoding="utf-8")
|
|
assert "export function requireCustomerOnly()" in auth
|
|
assert "IDENTITY_COOKIE = 'portal_auth_user'" in auth
|
|
assert "readCookie(IDENTITY_COOKIE) === String(context.userId)" in auth
|
|
assert "requireCustomerOnly()" in questionnaire
|
|
assert "? ['customer']" in login
|
|
|
|
|
|
def test_portal_auth_supports_cross_tab_logout_and_account_switching() -> None:
|
|
auth = (PORTAL / "common" / "auth.js").read_text(encoding="utf-8")
|
|
shell = (PORTAL / "common" / "layout" / "app-shell.js").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
login = (PORTAL / "common" / "login-controller.js").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
assert "new BroadcastChannel(AUTH_CHANNEL_NAME)" in auth
|
|
assert "CONTEXT_COOKIE = 'portal_auth_context'" in auth
|
|
assert "readSessionContext() || readSharedContext()" in auth
|
|
assert "signed-in" in auth
|
|
assert "signed-out" in auth
|
|
assert "account-switched" in auth
|
|
assert "postMessage({ type })" in auth
|
|
assert "type === 'signed-in'" in auth
|
|
assert "hasMatchingIdentity(readCookie('auth_token'), context)" in auth
|
|
broadcast_line = next(line for line in auth.splitlines() if "postMessage" in line)
|
|
assert "access_token" not in broadcast_line
|
|
assert "data-switch-account" in shell
|
|
assert "data-logout" in shell
|
|
assert "切换账号" in shell
|
|
assert "退出登录" in shell
|
|
assert "REASON_MESSAGES" in login
|
|
|
|
|
|
def test_risk_workspace_covers_documented_modules() -> None:
|
|
html = (PORTAL / "employee-risk" / "dashboard" / "index.html").read_text(encoding="utf-8")
|
|
source = (PORTAL / "employee-risk" / "dashboard" / "dashboard.js").read_text(encoding="utf-8")
|
|
permissions = (PORTAL / "common" / "permission-codes.js").read_text(encoding="utf-8")
|
|
for label in ("预警队列", "证据查询", "通知记录", "风控助手", "风险日报"):
|
|
assert label in html
|
|
combined = html + source + permissions
|
|
for permission in (
|
|
"risk:alert:read",
|
|
"risk:alert:write",
|
|
"risk:alert:scan",
|
|
"risk:report:mail",
|
|
):
|
|
assert permission in combined
|
|
|
|
|
|
def test_admin_workspace_is_not_an_identity_placeholder() -> None:
|
|
html = (PORTAL / "employee-console" / "workspace" / "index.html").read_text(encoding="utf-8")
|
|
assert "只展示服务端确认的身份边界" not in html
|
|
for label in ("角色与权限", "配置与模型", "审计记录", "转人工工单", "画像候选"):
|
|
assert label in html
|
|
|
|
|
|
def test_frontend_has_no_remote_scripts_or_token_local_storage() -> None:
|
|
sources = "\n".join(
|
|
path.read_text(encoding="utf-8")
|
|
for path in PORTAL.rglob("*")
|
|
if path.is_file() and path.suffix in {".html", ".js", ".css"}
|
|
)
|
|
assert '<script src="http' not in sources
|
|
assert "localStorage.setItem('token'" not in sources
|
|
assert "console.log" not in sources
|
|
|
|
|
|
def test_shared_state_view_has_distinct_authentication_and_permission_states() -> None:
|
|
source = (PORTAL / "common" / "state-view.js").read_text(encoding="utf-8")
|
|
assert "ERR_CODES.AUTHENTICATION_REQUIRED" in source
|
|
assert "ERR_CODES.AGENT_PERMISSION_DENIED" in source
|
|
assert "登录状态已失效" in source
|
|
assert "当前账户暂不可访问" in source
|
|
|
|
|
|
def test_protected_api_unauthorized_response_clears_shared_session() -> None:
|
|
source = (PORTAL / "common" / "api-client.js").read_text(encoding="utf-8")
|
|
assert "clearAuthSession, getAccessToken" in source
|
|
assert "response.status === 401 && endpoint.auth !== false" in source
|
|
|
|
|
|
def test_public_home_uses_a_local_hero_image() -> None:
|
|
home = (PORTAL / "guest" / "home" / "index.html").read_text(encoding="utf-8")
|
|
image = PORTAL / "guest" / "home" / "assets" / "wealth_architecture_hero.jpg"
|
|
assert "/static/portal/guest/home/assets/wealth_architecture_hero.jpg" in home
|
|
assert image.is_file()
|
|
assert image.stat().st_size > 100_000
|