aps-agent/tests/golden/test_chat_ensure_session.py

# ============================================================
# 对话：会话未入库时自动 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=r"D:\ItemSpace\14.工业智核", 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)