235 lines
7.4 KiB
Python
235 lines
7.4 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import JSON, select
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
os.environ["DATABASE_URL"] = "postgresql+asyncpg://pmflow:pmflow_pass@127.0.0.1:5435/pmflow"
|
|
|
|
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.services.signal.signal_source_repair import SignalSourceRepairService
|
|
|
|
|
|
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__,
|
|
],
|
|
)
|
|
)
|
|
maker = async_sessionmaker(engine, expire_on_commit=False)
|
|
async with maker() as session:
|
|
yield session
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_repair_registries_materializes_formal_signal_without_capture(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,
|
|
)
|
|
project = Project(
|
|
tenant_id=1,
|
|
name="cashflow",
|
|
title="现金流项目",
|
|
status="active",
|
|
assigned_role="backend",
|
|
)
|
|
db_session.add_all([owner, project])
|
|
await db_session.flush()
|
|
|
|
registry = SignalSourceRegistry(
|
|
tenant_id=1,
|
|
project_id=int(project.id),
|
|
created_by_user_id=int(owner.id),
|
|
source_kind="github",
|
|
source_name="现金流日报仓库",
|
|
source_identifier="daily_cashflow_analysis",
|
|
description="每日现金流分析脚本仓库",
|
|
requested_via="ai_steward",
|
|
status="pending",
|
|
onboarding_status="pending",
|
|
config={},
|
|
raw_context={},
|
|
)
|
|
db_session.add(registry)
|
|
await db_session.commit()
|
|
|
|
service = SignalSourceRepairService(db_session)
|
|
|
|
async def _fail_if_capture_called(**kwargs): # noqa: ANN001
|
|
_ = kwargs
|
|
raise AssertionError("repair should materialize Signal directly instead of calling capture flow")
|
|
|
|
async def _noop_emit_visibility_records(**kwargs): # noqa: ANN001
|
|
_ = kwargs
|
|
|
|
monkeypatch.setattr(service, "_capture_registry", _fail_if_capture_called)
|
|
monkeypatch.setattr(service, "_emit_visibility_records", _noop_emit_visibility_records)
|
|
|
|
result = await service.repair_registries(
|
|
tenant_id=1,
|
|
project_id=int(project.id),
|
|
registry_ids=[int(registry.id)],
|
|
sender="PMFlow修复器",
|
|
)
|
|
|
|
assert len(result) == 1
|
|
assert result[0]["capture_status"] == "materialized"
|
|
assert result[0]["signal_id"] is not None
|
|
|
|
refreshed_registry = (
|
|
await db_session.execute(select(SignalSourceRegistry).where(SignalSourceRegistry.id == int(registry.id)))
|
|
).scalar_one()
|
|
assert refreshed_registry.signal_id == int(result[0]["signal_id"])
|
|
|
|
signal = (
|
|
await db_session.execute(select(Signal).where(Signal.id == int(result[0]["signal_id"])))
|
|
).scalar_one()
|
|
assert signal.title == "现金流日报仓库"
|
|
assert signal.matched_project_id == int(project.id)
|
|
assert signal.source == "daily_cashflow_analysis"
|
|
|
|
link = (
|
|
await db_session.execute(
|
|
select(ProjectSignalLink).where(
|
|
ProjectSignalLink.project_id == int(project.id),
|
|
ProjectSignalLink.signal_id == int(signal.id),
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
assert link is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_repair_registries_relinks_existing_project_signal_before_materializing(db_session, monkeypatch) -> None:
|
|
owner = User(
|
|
tenant_id=1,
|
|
username="owner2",
|
|
email="owner2@example.com",
|
|
password_hash="hash",
|
|
name="负责人2",
|
|
role="REVIEWER",
|
|
status="active",
|
|
is_active=True,
|
|
)
|
|
project = Project(
|
|
tenant_id=1,
|
|
name="cashflow-existing",
|
|
title="现金流项目-已有信号",
|
|
status="active",
|
|
assigned_role="backend",
|
|
)
|
|
db_session.add_all([owner, project])
|
|
await db_session.flush()
|
|
|
|
signal = Signal(
|
|
tenant_id=1,
|
|
source="cashflow_alert_subscription",
|
|
content="已有现金流预警信号",
|
|
impact_score=0.7,
|
|
priority="high",
|
|
processed=False,
|
|
strategic_analysis="{}",
|
|
tasks_generated="[]",
|
|
matched_project_id=None,
|
|
title="现金流预警系统需求",
|
|
signal_type="新需求",
|
|
summary="现金流健康评分与预警系统",
|
|
reasoning="[]",
|
|
confidence=0.9,
|
|
recommended_action="enter_only",
|
|
status="待处理",
|
|
)
|
|
db_session.add(signal)
|
|
await db_session.flush()
|
|
db_session.add(
|
|
ProjectSignalLink(
|
|
tenant_id=1,
|
|
project_id=int(project.id),
|
|
signal_id=int(signal.id),
|
|
link_type="related",
|
|
relevance_score=1.0,
|
|
)
|
|
)
|
|
|
|
registry = SignalSourceRegistry(
|
|
tenant_id=1,
|
|
project_id=int(project.id),
|
|
created_by_user_id=int(owner.id),
|
|
source_kind="webhook",
|
|
source_name="现金流预警订阅",
|
|
source_identifier="cashflow_alert_subscription",
|
|
description="企业订阅现金流预警，当健康评分低于阈值时推送通知",
|
|
requested_via="ai_steward",
|
|
status="pending",
|
|
onboarding_status="pending",
|
|
config={},
|
|
raw_context={},
|
|
)
|
|
db_session.add(registry)
|
|
await db_session.commit()
|
|
|
|
service = SignalSourceRepairService(db_session)
|
|
|
|
async def _noop_emit_visibility_records(**kwargs): # noqa: ANN001
|
|
_ = kwargs
|
|
|
|
monkeypatch.setattr(service, "_emit_visibility_records", _noop_emit_visibility_records)
|
|
|
|
result = await service.repair_registries(
|
|
tenant_id=1,
|
|
project_id=int(project.id),
|
|
registry_ids=[int(registry.id)],
|
|
sender="PMFlow修复器",
|
|
)
|
|
|
|
assert len(result) == 1
|
|
assert result[0]["capture_status"] == "relinked"
|
|
assert result[0]["signal_id"] == int(signal.id)
|
|
|
|
refreshed_registry = (
|
|
await db_session.execute(select(SignalSourceRegistry).where(SignalSourceRegistry.id == int(registry.id)))
|
|
).scalar_one()
|
|
assert refreshed_registry.signal_id == int(signal.id)
|
|
|
|
refreshed_signal = (
|
|
await db_session.execute(select(Signal).where(Signal.id == int(signal.id)))
|
|
).scalar_one()
|
|
assert refreshed_signal.matched_project_id == int(project.id)
|
|
|
|
signals = (await db_session.execute(select(Signal).order_by(Signal.id.asc()))).scalars().all()
|
|
assert len(signals) == 1
|