Codex
277 lines
10 KiB
Python
277 lines
10 KiB
Python
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.core.config import get_settings
|
|
from app.db.models import AgentRun, Base, Signal
|
|
from app.db.session import get_db
|
|
from app.main import create_app
|
|
|
|
|
|
def test_health_endpoint_returns_app_identity(monkeypatch) -> None:
|
|
# 健康检查只暴露服务身份，不触发Agent运行或数据库写入。
|
|
monkeypatch.setenv("SIGNALSCOUT_ENV", "test")
|
|
get_settings.cache_clear()
|
|
with TestClient(create_app()) as client:
|
|
response = client.get("/health")
|
|
get_settings.cache_clear()
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "ok"
|
|
assert response.json()["app"] == "SignalScout"
|
|
|
|
|
|
def test_home_page_returns_dashboard_shell(monkeypatch) -> None:
|
|
# 根路径应该展示业务工作台，而不是返回接口404。
|
|
monkeypatch.setenv("SIGNALSCOUT_ENV", "test")
|
|
get_settings.cache_clear()
|
|
with TestClient(create_app()) as client:
|
|
response = client.get("/")
|
|
get_settings.cache_clear()
|
|
|
|
assert response.status_code == 200
|
|
assert "SignalScout" in response.text
|
|
assert "今日 AI 信号" in response.text
|
|
assert "手动搜索" not in response.text
|
|
assert "renderMarkdownReport" in response.text
|
|
assert "自动记录" not in response.text
|
|
assert "采集今日情报" not in response.text
|
|
|
|
|
|
def test_manual_agent_run_endpoint_starts_background_job_when_enabled(monkeypatch) -> None:
|
|
# 启用采集后手动入口复用调度任务，方便验证每轮候选和入库数量。
|
|
monkeypatch.setenv("SIGNALSCOUT_ENV", "test")
|
|
monkeypatch.setenv("SIGNALSCOUT_AGENT_ENABLED", "true")
|
|
get_settings.cache_clear()
|
|
started = []
|
|
engine = create_engine(
|
|
"sqlite:///:memory:",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
future=True,
|
|
)
|
|
TestingSessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
|
|
Base.metadata.create_all(engine)
|
|
|
|
def fake_run_scheduled_job(settings):
|
|
started.append(settings.env)
|
|
|
|
def override_get_db():
|
|
# 手动搜索测试只验证接口调度行为，运行中状态查询使用内存库。
|
|
db = TestingSessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
monkeypatch.setattr("app.api.routes.run_scheduled_job", fake_run_scheduled_job)
|
|
app = create_app()
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
with TestClient(app) as client:
|
|
response = client.post("/agent/runs/daily")
|
|
get_settings.cache_clear()
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "started"
|
|
assert started == ["test"]
|
|
|
|
|
|
def test_manual_agent_run_endpoint_uses_scheduler_when_available(monkeypatch) -> None:
|
|
# 生产环境手动搜索投递到常驻调度器，避免HTTP请求承载长时间联网搜索。
|
|
monkeypatch.setenv("SIGNALSCOUT_ENV", "test")
|
|
monkeypatch.setenv("SIGNALSCOUT_AGENT_ENABLED", "true")
|
|
get_settings.cache_clear()
|
|
engine = create_engine(
|
|
"sqlite:///:memory:",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
future=True,
|
|
)
|
|
TestingSessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
|
|
Base.metadata.create_all(engine)
|
|
scheduled = []
|
|
|
|
class FakeScheduler:
|
|
running = True
|
|
timezone = ZoneInfo("Asia/Shanghai")
|
|
|
|
def add_job(self, *args, **kwargs):
|
|
scheduled.append(kwargs)
|
|
|
|
def override_get_db():
|
|
# 调度器路径同样使用内存库检查运行中状态。
|
|
db = TestingSessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
app = create_app()
|
|
app.state.scheduler = FakeScheduler()
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
with TestClient(app) as client:
|
|
response = client.post("/agent/runs/daily")
|
|
get_settings.cache_clear()
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "started"
|
|
assert scheduled[0]["trigger"] == "date"
|
|
assert scheduled[0]["run_date"].tzinfo is not None
|
|
|
|
|
|
def test_manual_agent_run_endpoint_rejects_requests_when_disabled(monkeypatch) -> None:
|
|
# 采集暂停时接口拒绝请求，避免其他调用方绕过页面入口消耗模型额度。
|
|
monkeypatch.setenv("SIGNALSCOUT_ENV", "test")
|
|
monkeypatch.setenv("SIGNALSCOUT_AGENT_ENABLED", "false")
|
|
get_settings.cache_clear()
|
|
with TestClient(create_app()) as client:
|
|
response = client.post("/agent/runs/daily")
|
|
get_settings.cache_clear()
|
|
|
|
assert response.status_code == 503
|
|
assert response.json()["detail"] == "情报采集当前已暂停"
|
|
|
|
|
|
def test_dashboard_returns_news_and_github_signals_separately(monkeypatch) -> None:
|
|
# 仪表盘接口直接返回分组后的业务数据，前端不再用标题或来源名自行猜测分类。
|
|
monkeypatch.setenv("SIGNALSCOUT_ENV", "test")
|
|
get_settings.cache_clear()
|
|
engine = create_engine(
|
|
"sqlite:///:memory:",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
future=True,
|
|
)
|
|
TestingSessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
|
|
Base.metadata.create_all(engine)
|
|
|
|
with TestingSessionLocal() as session:
|
|
run = AgentRun(status="completed", started_at=datetime(2026, 7, 8, 8, 0), request_payload={})
|
|
session.add(run)
|
|
session.flush()
|
|
session.add_all(
|
|
[
|
|
Signal(
|
|
run_id=run.id,
|
|
source_type="news",
|
|
topic="AI 新闻",
|
|
title="New model release",
|
|
summary="模型发布动态。",
|
|
source_url="https://example.com/news",
|
|
source_url_hash="news-hash",
|
|
source_name="Example News",
|
|
published_at=datetime(2026, 7, 8, 9, 0),
|
|
importance=4,
|
|
entities=["Example"],
|
|
),
|
|
Signal(
|
|
run_id=run.id,
|
|
source_type="github",
|
|
topic="GitHub 热门项目",
|
|
title="example/agent-kit",
|
|
summary="开源项目动态。",
|
|
source_url="https://github.com/example/agent-kit",
|
|
source_url_hash="github-hash",
|
|
source_name="GitHub",
|
|
published_at=datetime(2026, 7, 8, 10, 0),
|
|
importance=5,
|
|
entities=["example/agent-kit"],
|
|
),
|
|
]
|
|
)
|
|
session.commit()
|
|
|
|
def override_get_db():
|
|
# 测试依赖把接口请求限制在内存库里，避免触碰开发数据库。
|
|
db = TestingSessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
app = create_app()
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
with TestClient(app) as client:
|
|
dashboard_response = client.get("/dashboard")
|
|
github_response = client.get("/signals", params={"source_type": "github"})
|
|
get_settings.cache_clear()
|
|
|
|
dashboard = dashboard_response.json()
|
|
assert dashboard_response.status_code == 200
|
|
assert [item["source_type"] for item in dashboard["news_signals"]] == ["news"]
|
|
assert [item["source_type"] for item in dashboard["github_signals"]] == ["github"]
|
|
assert github_response.status_code == 200
|
|
assert [item["title"] for item in github_response.json()] == ["example/agent-kit"]
|
|
|
|
|
|
def test_dashboard_uses_latest_completed_run_signals(monkeypatch) -> None:
|
|
# 情报页展示最新完成运行的快照，历史运行只保留在记录和明细接口里。
|
|
monkeypatch.setenv("SIGNALSCOUT_ENV", "test")
|
|
get_settings.cache_clear()
|
|
engine = create_engine(
|
|
"sqlite:///:memory:",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
future=True,
|
|
)
|
|
TestingSessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
|
|
Base.metadata.create_all(engine)
|
|
|
|
with TestingSessionLocal() as session:
|
|
old_run = AgentRun(status="completed", started_at=datetime(2026, 7, 8, 8, 0), request_payload={})
|
|
new_run = AgentRun(status="completed", started_at=datetime(2026, 7, 8, 9, 0), request_payload={})
|
|
session.add_all([old_run, new_run])
|
|
session.flush()
|
|
session.add_all(
|
|
[
|
|
Signal(
|
|
run_id=old_run.id,
|
|
source_type="news",
|
|
topic="AI 新闻",
|
|
title="Old signal",
|
|
summary="旧运行信号。",
|
|
source_url="https://example.com/old",
|
|
source_url_hash="old-hash",
|
|
source_name="Example",
|
|
published_at=datetime(2026, 7, 8, 8, 0),
|
|
importance=3,
|
|
entities=[],
|
|
),
|
|
Signal(
|
|
run_id=new_run.id,
|
|
source_type="news",
|
|
topic="AI 新闻",
|
|
title="New signal",
|
|
summary="新运行信号。",
|
|
source_url="https://example.com/new",
|
|
source_url_hash="new-hash",
|
|
source_name="Example",
|
|
published_at=datetime(2026, 7, 8, 9, 0),
|
|
importance=4,
|
|
entities=[],
|
|
),
|
|
]
|
|
)
|
|
session.commit()
|
|
|
|
def override_get_db():
|
|
# 仪表盘快照测试使用内存库构造多运行场景。
|
|
db = TestingSessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
app = create_app()
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
with TestClient(app) as client:
|
|
response = client.get("/dashboard")
|
|
get_settings.cache_clear()
|
|
|
|
assert response.status_code == 200
|
|
assert [item["title"] for item in response.json()["news_signals"]] == ["New signal"]
|