134 lines
4.9 KiB
Python
134 lines
4.9 KiB
Python
# ============================================================
|
||
# 对话：会话未入库时自动 ensure，避免误报 404
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sqlite3
|
||
import uuid
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from server.auth.context import IdentityContext, bind_identity, reset_identity
|
||
from server.gateway.app import ChatRequest, create_app
|
||
from server.state.projects import PERSONAL_PROJECT_ID, get_project_store
|
||
|
||
|
||
_REPO_ROOT = Path(__file__).resolve().parents[2]
|
||
_LIVE_MASTER_DB = (_REPO_ROOT / "server" / "data" / "master.db").resolve()
|
||
|
||
|
||
def _live_session_exists(session_id: str) -> bool:
|
||
"""只读检查现场库；测试不得在该库中留下自己的随机会话。"""
|
||
if not _LIVE_MASTER_DB.exists():
|
||
return False
|
||
uri = f"file:{_LIVE_MASTER_DB.as_posix()}?mode=ro"
|
||
with sqlite3.connect(uri, uri=True) as connection:
|
||
table = connection.execute(
|
||
"SELECT 1 FROM sqlite_master WHERE type='table' AND name='aps_chat_sessions'"
|
||
).fetchone()
|
||
if table is None:
|
||
return False
|
||
return connection.execute(
|
||
"SELECT 1 FROM aps_chat_sessions WHERE id = ? LIMIT 1", (session_id,)
|
||
).fetchone() is not None
|
||
|
||
|
||
def test_chat_request_carries_project_scope():
|
||
req = ChatRequest(sessionId="sess_local", projectId="proj_current", text="分析一下文件夹")
|
||
assert req.projectId == "proj_current"
|
||
|
||
|
||
def test_explicit_missing_project_never_falls_back_to_personal():
|
||
identity = IdentityContext(
|
||
user_id=9002,
|
||
username="desktop-project-test",
|
||
fullname="项目归属测试用户",
|
||
tenant_uuid=f"tenant-project-{uuid.uuid4().hex[:12]}",
|
||
roles=("desktop", "planner"),
|
||
auth_kind="license",
|
||
)
|
||
token = bind_identity(identity)
|
||
sid = f"sess_project_{uuid.uuid4().hex[:12]}"
|
||
try:
|
||
with pytest.raises(PermissionError):
|
||
get_project_store().ensure_session(
|
||
sid,
|
||
project_id="proj_not_created",
|
||
title="分析一下文件夹",
|
||
)
|
||
with pytest.raises(ValueError):
|
||
get_project_store().get_messages(sid)
|
||
finally:
|
||
reset_identity(token)
|
||
|
||
|
||
def test_chat_auto_ensures_missing_session():
|
||
isolated_db = Path(os.environ["APS_DB_PATH"]).resolve()
|
||
assert isolated_db != _LIVE_MASTER_DB, "pytest 必须在写入前拒绝现场 master.db"
|
||
assert not os.environ.get("APS_DATABASE_URL"), "外部数据库 URL 会绕过 APS_DB_PATH 隔离"
|
||
|
||
identity = IdentityContext(
|
||
user_id=9001,
|
||
username="desktop-test",
|
||
fullname="本机授权用户",
|
||
tenant_uuid="tenant-chat-ensure",
|
||
roles=("desktop", "planner"),
|
||
auth_kind="license",
|
||
)
|
||
token = bind_identity(identity)
|
||
app = create_app()
|
||
client = TestClient(app)
|
||
# 绕过 HTTP 鉴权：直接在请求上下文用 middleware 已绑 identity 较难，
|
||
# 这里用 TestClient + 依赖覆盖较重；改为直接测 store + 端点内逻辑的 store 层。
|
||
try:
|
||
sid = f"sess_orphan_{uuid.uuid4().hex[:12]}"
|
||
assert not _live_session_exists(sid)
|
||
# 确认尚不存在
|
||
try:
|
||
get_project_store().get_messages(sid)
|
||
exists = True
|
||
except (ValueError, PermissionError):
|
||
exists = False
|
||
assert not exists
|
||
|
||
created = get_project_store().ensure_session(sid, project_id=PERSONAL_PROJECT_ID, title="分析一下文件夹")
|
||
assert created["id"] == sid
|
||
assert get_project_store().get_messages(sid) == []
|
||
finally:
|
||
reset_identity(token)
|
||
# 避免泄漏 TestClient 引用
|
||
del client
|
||
assert not _live_session_exists(sid), "chat ensure 测试写入了现场 master.db"
|
||
|
||
|
||
def test_existing_personal_session_rebinds_to_explicit_project():
|
||
identity = IdentityContext(
|
||
user_id=9003,
|
||
username="desktop-rebind-test",
|
||
fullname="会话重绑测试用户",
|
||
tenant_uuid=f"tenant-rebind-{uuid.uuid4().hex[:12]}",
|
||
roles=("desktop", "planner"),
|
||
auth_kind="license",
|
||
)
|
||
token = bind_identity(identity)
|
||
sid = f"sess_rebind_{uuid.uuid4().hex[:12]}"
|
||
pid = f"proj_rebind_{uuid.uuid4().hex[:12]}"
|
||
try:
|
||
store = get_project_store()
|
||
personal = store.ensure_session(sid, project_id=PERSONAL_PROJECT_ID)
|
||
assert personal["projectId"] == PERSONAL_PROJECT_ID
|
||
|
||
store.create_project("锐扬项目", work_dir="demo-data/ruiyang", project_id=pid)
|
||
rebound = store.ensure_session(sid, project_id=pid)
|
||
assert rebound["projectId"] == pid
|
||
assert rebound["scope"] == "project"
|
||
|
||
snap = store.snapshot(include_messages=False)
|
||
assert snap["activeProjectId"] == pid
|
||
assert snap["activeSessionId"] == sid
|
||
finally:
|
||
reset_identity(token)
|