PMFlow/backend/tests/test_feishu_new_project_signal_sources.py

from __future__ import annotations

import importlib
import os
from datetime import datetime

import pytest
import pytest_asyncio
from fastapi import HTTPException
from sqlalchemy import JSON, select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

os.environ["DATABASE_URL"] = "postgresql+asyncpg://tester:tester@localhost:5432/pmflow_test"

from app.models.base import Base
from app.models.project import Project
from app.models.project_signal_link import ProjectSignalLink
from app.models.signal import Signal
from app.models.signal_source_registry import SignalSourceRegistry
from app.models.user import User
from app.models.user_project import UserProject
from app.services.signal.signal_capture_service import ProactiveSignalCaptureService


def _load_signal_source_helper():
    return importlib.import_module("app.api.signal_sources_api.helpers")


def _use_sqlite_json_types() -> None:
    for table in (User.__table__, SignalSourceRegistry.__table__):
        for column in table.columns:
            if column.name in {"assigned_domains", "config", "raw_context"}:
                column.type = JSON()


@pytest_asyncio.fixture
async def db_session():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    _use_sqlite_json_types()
    async with engine.begin() as conn:
        await conn.run_sync(
            lambda sync_conn: Base.metadata.create_all(
                sync_conn,
                tables=[
                    User.__table__,
                    Project.__table__,
                    Signal.__table__,
                    ProjectSignalLink.__table__,
                    SignalSourceRegistry.__table__,
                    UserProject.__table__,
                ],
            )
        )
    maker = async_sessionmaker(engine, expire_on_commit=False)
    async with maker() as session:
        yield session
    await engine.dispose()


@pytest.mark.asyncio
async def test_draft_new_project_brief_raises_when_llm_fails(db_session, monkeypatch) -> None:
    service = ProactiveSignalCaptureService(db_session)

    async def fake_generate(prompt: str, **kwargs):  # noqa: ANN001
        _ = (prompt, kwargs)
        raise RuntimeError("llm boom")

    monkeypatch.setattr(service.llm, "generate", fake_generate)

    with pytest.raises(RuntimeError, match="llm boom"):
        await service._draft_new_project_brief(
            message="我们要新建一个 AI 客服项目",
            tenant_id=1,
            sender="Alice",
            requested_at=datetime(2026, 3, 26, 12, 0, 0),
        )


@pytest.mark.asyncio
async def test_auto_create_project_from_new_signal_persists_signal_sources(db_session, monkeypatch) -> None:
    owner = User(
        tenant_id=1,
        username="owner",
        email="owner@example.com",
        password_hash="hash",
        name="负责人",
        role="REVIEWER",
        status="active",
        is_active=True,
    )
    signal = Signal(
        tenant_id=1,
        source="Feishu",
        content="请帮我新建项目",
        priority="medium",
        processed=False,
    )
    db_session.add_all([owner, signal])
    await db_session.commit()

    service = ProactiveSignalCaptureService(db_session)

    async def fake_brief(**kwargs):  # noqa: ANN001
        _ = kwargs
        return {
            "title": "AI 客服平台",
            "description": "飞书消息里提到的新项目。",
            "primary_role": "backend",
            "requirement_items": [
                {
                    "section": "核心痛点",
                    "title": "内容本地化",
                    "description": "翻译公司贵、慢、不地道",
                },
                {
                    "section": "MVP 功能",
                    "title": "社媒运营",
                    "description": "自动生成文案并排期发布",
                },
            ],
        }

    monkeypatch.setattr(service, "_draft_new_project_brief", fake_brief)

    project_id = await service._auto_create_project_from_new_signal(
        signal_id=int(signal.id),
        message="我们要做一个 AI 客服",
        sender="Alice",
        tenant_id=1,
        requested_at=datetime(2026, 3, 26, 12, 0, 0),
    )
    await db_session.commit()

    assert project_id is not None

    project = (
        await db_session.execute(
            select(Project).where(Project.id == int(project_id), Project.tenant_id == 1)
        )
    ).scalar_one()
    assert project.title == "AI 客服平台"

    signal_row = (
        await db_session.execute(select(Signal).where(Signal.id == int(signal.id), Signal.tenant_id == 1))
    ).scalar_one()
    await db_session.refresh(signal_row)
    assert signal_row.processed is True

    registries = (
        await db_session.execute(
            select(SignalSourceRegistry).where(
                SignalSourceRegistry.project_id == int(project_id),
                SignalSourceRegistry.tenant_id == 1,
            )
        )
    ).scalars().all()
    assert len(registries) == 2
    assert all(registry.signal_id is not None for registry in registries)
    assert registries[0].raw_context["origin"] == "feishu_new_project"
    assert registries[0].source_kind == "feishu"

    generated_signals = (
        await db_session.execute(
            select(Signal).where(
                Signal.id != int(signal.id),
                Signal.tenant_id == 1,
            )
        )
    ).scalars().all()
    assert len(generated_signals) == 2

    links = (
        await db_session.execute(
            select(ProjectSignalLink).where(
                ProjectSignalLink.project_id == int(project_id),
                ProjectSignalLink.tenant_id == 1,
            )
        )
    ).scalars().all()
    assert len(links) == 3


@pytest.mark.asyncio
async def test_signal_source_helper_no_longer_uses_recipe_keyword_project_guess(db_session) -> None:
    helper = _load_signal_source_helper()
    db_session.add_all(
        [
            Project(tenant_id=1, title="Recipe Studio", name="Recipe Studio", status="active"),
            Project(tenant_id=1, title="Growth Console", name="Growth Console", status="active"),
        ]
    )
    await db_session.commit()

    with pytest.raises(HTTPException, match="显式传入 project_id"):
        await helper.resolve_target_project_id(
            db=db_session,
            tenant_id=1,
            project_id=None,
        )