75 lines
2.9 KiB
Python
75 lines
2.9 KiB
Python
|
"""首页采集状态来自真实批次，关闭读取不抹掉历史成功记录。"""
|
||
|
|
|
||
|
|
from datetime import datetime
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from sqlalchemy import Column, DateTime, Integer, MetaData, String, Table, create_engine
|
||
|
|
from sqlalchemy.orm import Session
|
||
|
|
|
||
|
|
from app.core.config import Settings
|
||
|
|
from app.services import dashboard_service
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def sync_database(monkeypatch):
|
||
|
|
engine = create_engine("sqlite://")
|
||
|
|
metadata = MetaData()
|
||
|
|
# 只建 V14 已有列，证明读取状态不要求业务库先升级 V20。
|
||
|
|
table = Table(
|
||
|
|
"tplus_sync_batches", metadata,
|
||
|
|
Column("id", Integer, primary_key=True),
|
||
|
|
Column("sync_mode", String(16)),
|
||
|
|
Column("status", String(16)),
|
||
|
|
Column("finished_at", DateTime),
|
||
|
|
)
|
||
|
|
metadata.create_all(engine)
|
||
|
|
settings = Settings(_env_file=None, app_secret_key="x" * 32)
|
||
|
|
monkeypatch.setattr(dashboard_service, "get_settings", lambda: settings, raising=False)
|
||
|
|
with Session(engine) as session:
|
||
|
|
yield session, table, settings
|
||
|
|
engine.dispose()
|
||
|
|
|
||
|
|
|
||
|
|
def test_关闭采集仍显示历史成功且不将旧失败冒充当前失败(sync_database):
|
||
|
|
session, table, _ = sync_database
|
||
|
|
finished = datetime(2026, 10, 30, 16, 52, 3)
|
||
|
|
session.execute(table.insert(), [
|
||
|
|
{"id": 1, "sync_mode": "FULL", "status": "FAILED", "finished_at": finished},
|
||
|
|
{"id": 2, "sync_mode": "FULL", "status": "SUCCEEDED", "finished_at": finished},
|
||
|
|
{"id": 3, "sync_mode": "AUDIT", "status": "FAILED", "finished_at": finished},
|
||
|
|
])
|
||
|
|
result = dashboard_service._sync_status(session)
|
||
|
|
assert result["available"] is True
|
||
|
|
assert result["enabled"] is False and result["automation_enabled"] is False
|
||
|
|
assert result["latest_status"] == "SUCCEEDED"
|
||
|
|
assert result["last_success_at"] == finished
|
||
|
|
assert result["historical_failure_total"] == 1
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("status", ["FAILED", "PARTIAL", "QUEUED", "RUNNING"])
|
||
|
|
def test_首页如实返回最近采集状态(sync_database, status):
|
||
|
|
session, table, settings = sync_database
|
||
|
|
settings.t_plus_enabled = True
|
||
|
|
session.execute(table.insert(), {"id": 1, "sync_mode": "FULL", "status": status})
|
||
|
|
result = dashboard_service._sync_status(session)
|
||
|
|
assert result["latest_status"] == status and result["enabled"] is True
|
||
|
|
assert result["last_success_at"] is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_无记录不显示已采集(sync_database):
|
||
|
|
session, _, _ = sync_database
|
||
|
|
result = dashboard_service._sync_status(session)
|
||
|
|
assert result["available"] is True
|
||
|
|
assert result["latest_status"] is None and result["last_success_at"] is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_未建同步表时明确返回不可查询():
|
||
|
|
engine = create_engine("sqlite://")
|
||
|
|
try:
|
||
|
|
with Session(engine) as session:
|
||
|
|
result = dashboard_service._sync_status(session)
|
||
|
|
assert result["available"] is False
|
||
|
|
assert result["last_success_at"] is None
|
||
|
|
finally:
|
||
|
|
engine.dispose()
|