geMoldInsight/tests/test_status_endpoint_auth.py

"""批次 0 安全修复回归测试（TECH_DEBT D5 等）。

覆盖：
- /api/status/{task_id} 鉴权与任务归属校验（无 token 401 / 他人任务 403 / 不存在 404 / 所有者 200）
- bcrypt 72 字节上限：超长密码显式拒绝而非静默截断
- SECRET_KEY 惰性校验：未配置时给出明确错误
"""
import pytest
from fastapi import FastAPI
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import async_sessionmaker, AsyncSession

from moldinsight.api.task_router import router as task_router
from shared.database.database import get_db_session
from shared.models.identity import User
from moldinsight.models import STPFile, ProcessingTask
from shared.services.auth_service import (
    get_current_active_user,
    get_password_hash,
    verify_password,
    create_access_token,
)
from shared.config.settings import settings


@pytest.fixture(scope="function")
async def status_client(async_engine, seeded_db):
    """带 task_router 的测试应用：播种两个任务（owner=user 1 / user 999）。"""
    session_factory = async_sessionmaker(async_engine, class_=AsyncSession, expire_on_commit=False)

    async with session_factory() as session:
        stp_own = STPFile(
            id=9001, user_id=1, object_key="test/own.step", storage_bucket="moldinsight",
            original_filename="own.step", file_size=128, status="completed",
        )
        task_own = ProcessingTask(
            task_id="task-owned", stp_file_id=9001,
            task_type="stp_parsing", status="completed", parameters={},
        )
        stp_other = STPFile(
            id=9002, user_id=999, object_key="test/other.step", storage_bucket="moldinsight",
            original_filename="other.step", file_size=128, status="completed",
        )
        task_other = ProcessingTask(
            task_id="task-foreign", stp_file_id=9002,
            task_type="stp_parsing", status="completed", parameters={},
        )
        session.add_all([stp_own, task_own, stp_other, task_other])
        await session.commit()

    test_app = FastAPI()
    # 生产环境中 /api 前缀由入口层挂载时添加，测试中需显式指定才能对齐真实路径
    test_app.include_router(task_router, prefix="/api")

    async def override_get_db_session():
        async with session_factory() as session:
            yield session

    test_app.dependency_overrides[get_db_session] = override_get_db_session

    transport = ASGITransport(app=test_app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac, test_app

    test_app.dependency_overrides.clear()


@pytest.mark.asyncio
async def test_status_without_token_is_401(status_client):
    """未携带 token 访问 /api/status 必须拒绝（此前该端点完全未鉴权）。"""
    ac, _ = status_client
    resp = await ac.post("/api/status/task-owned")
    assert resp.status_code == 401


@pytest.mark.asyncio
async def test_status_owner_can_view(status_client):
    ac, app = status_client
    app.dependency_overrides[get_current_active_user] = lambda: User(id=1, username="tester")

    resp = await ac.post("/api/status/task-owned")
    assert resp.status_code == 200
    body = resp.json()
    assert body["task_id"] == "task-owned"
    assert body["status"] == "completed"


@pytest.mark.asyncio
async def test_status_non_owner_is_403(status_client):
    """他人任务（含无主任务）必须 403，不允许凭任务号枚举。"""
    ac, app = status_client
    app.dependency_overrides[get_current_active_user] = lambda: User(id=2, username="intruder")

    resp = await ac.post("/api/status/task-owned")
    assert resp.status_code == 403

    resp = await ac.post("/api/status/task-foreign")
    assert resp.status_code == 403


@pytest.mark.asyncio
async def test_status_unknown_task_is_404(status_client):
    ac, app = status_client
    app.dependency_overrides[get_current_active_user] = lambda: User(id=1, username="tester")

    resp = await ac.post("/api/status/task-missing")
    assert resp.status_code == 404


def test_password_hash_rejects_over_72_bytes():
    """bcrypt 72 字节上限：必须显式报错，不能静默截断。"""
    with pytest.raises(ValueError, match="72"):
        get_password_hash("a" * 73)
    # 多字节字符按字节数计：25 个汉字 = 75 字节
    with pytest.raises(ValueError, match="72"):
        get_password_hash("模" * 25)


def test_password_hash_roundtrip_at_limit():
    password = "a" * 72
    hashed = get_password_hash(password)
    assert verify_password(password, hashed)
    # 历史口令按 bcrypt 语义截断比较：73 字节输入截断后与 72 字节口令匹配（兼容旧数据），
    # 但新口令在 get_password_hash 处已被显式拒绝，不会再产生这类哈希
    assert verify_password(password + "x", hashed)
    assert not verify_password("b" * 72, hashed)


def test_verify_password_over_72_bytes_returns_false_not_raise():
    """超长密码登录不得抛 ValueError（否则登录接口 500），应返回 False 走正常失败路径。"""
    hashed = get_password_hash("short-password")
    assert not verify_password("长" * 40, hashed)  # 120 字节
    assert not verify_password("a" * 73, hashed)


def test_create_token_without_secret_key_is_explicit_error(monkeypatch):
    """SECRET_KEY 未配置时给出可读错误，而不是 jwt.encode 的晦涩 TypeError。"""
    monkeypatch.setattr(settings, "SECRET_KEY", None)
    with pytest.raises(RuntimeError, match="SECRET_KEY"):
        create_access_token({"sub": "tester"})