vps-manager/tests/test_stats_overview.py

"""资产总览统计测试（get_overview）

覆盖的核心问题：「本年支出」是只把填了费用的资产加起来的和。cost 默认 0，
所以没填费用和真的免费在数据上无法区分——合计因此天然是个下限，而不是准数。
前端必须能拿到缺口大小（year_cost_missing）才能如实说明，否则用户会把下限
当成实际支出。

顺带回归到期口径：已过期（overdue）与 30 天内到期（expiring_30）分开计数，
已注销资产两边都不进。

运行：.venv/bin/pytest tests/test_stats_overview.py -v
"""

from datetime import date, timedelta

import pytest
from sqlmodel import Session, SQLModel, create_engine

import app.models.asset  # noqa: F401
import app.models.credential  # noqa: F401
import app.models.monitor  # noqa: F401
import app.models.provider  # noqa: F401
import app.models.ssl  # noqa: F401
from app.models.asset import Asset, AssetStatus, AssetType
from app.services.asset_service import get_overview


@pytest.fixture()
def engine(tmp_path):
    eng = create_engine(
        f"sqlite:///{tmp_path / 'stats.db'}", connect_args={"check_same_thread": False}
    )
    SQLModel.metadata.create_all(eng)
    yield eng
    eng.dispose()


def _add(engine, name, *, expiry=None, cost=0.0, status=AssetStatus.ACTIVE,
         asset_type=AssetType.VPS):
    with Session(engine) as s:
        s.add(Asset(
            name=name, asset_type=asset_type, provider="vultr",
            expiry_date=expiry, cost=cost, status=status,
        ))
        s.commit()


def _overview(engine):
    with Session(engine) as s:
        return get_overview(s)


def test_empty_db_returns_all_contract_keys(engine):
    """空库也必须给出完整结构：/api/stats/overview 是前端总览页的数据源，
    缺键会让 dashboard.js 渲染出 undefined"""
    got = _overview(engine)
    assert got == {
        "total": 0,
        "by_type": {},
        "by_status": {},
        "abnormal_count": 0,
        "expiring_30": 0,
        "overdue_count": 0,
        "year_cost": 0,
        "month_cost": 0,
        "year_cost_missing": 0,
    }


def test_year_cost_sums_only_filled_costs(engine):
    today = date.today()
    _add(engine, "paid", expiry=today, cost=12.5)
    _add(engine, "also-paid", expiry=today + timedelta(days=3), cost=7.25)
    got = _overview(engine)
    assert got["year_cost"] == 19.75
    assert got["year_cost_missing"] == 0


def test_unfilled_cost_is_counted_as_gap_not_zero_spend(engine):
    """cost 默认 0，没填的资产既不进合计、也不能悄悄消失"""
    today = date.today()
    _add(engine, "paid", expiry=today, cost=10.0)
    _add(engine, "unfilled-a", expiry=today + timedelta(days=1))
    _add(engine, "unfilled-b", expiry=today + timedelta(days=2))
    got = _overview(engine)
    assert got["year_cost"] == 10.0
    assert got["year_cost_missing"] == 2
    assert got["total"] == 3


def test_assets_without_expiry_do_not_enter_cost_window(engine):
    """没有到期日的资产不属于任何年度窗口：既不该加进合计，也不该算作缺口，
    否则「另有 N 项未填费用」会把一批跟今年无关的资产算进去"""
    _add(engine, "no-expiry-unfilled")
    _add(engine, "no-expiry-paid", cost=99.0)
    got = _overview(engine)
    assert got["year_cost"] == 0
    assert got["year_cost_missing"] == 0
    assert got["total"] == 2


def test_other_year_expiry_excluded_from_cost(engine):
    """跨年资产不进本年合计，也不进本年缺口"""
    this_year = date.today()
    other_year = date(this_year.year + 1, 6, 15)
    _add(engine, "this-year", expiry=this_year, cost=5.0)
    _add(engine, "next-year-unfilled", expiry=other_year)
    got = _overview(engine)
    assert got["year_cost"] == 5.0
    assert got["year_cost_missing"] == 0


def test_month_cost_only_current_month(engine):
    today = date.today()
    # 取本月内的另一天：若今天是 1 号则往后挪，避免算出上/下月的日期
    same_month = today.replace(day=1) if today.day > 1 else today
    next_month = (today.replace(day=1) + timedelta(days=32)).replace(day=2)
    _add(engine, "this-month", expiry=same_month, cost=20.0)
    _add(engine, "next-month", expiry=next_month, cost=30.0)
    got = _overview(engine)
    assert got["month_cost"] == 20.0
    # 下月那笔仍属于本年
    assert got["year_cost"] == 50.0


def test_year_cost_rounded_to_two_decimals(engine):
    """币种最小单位是分，浮点累加的 0.1+0.2 尾巴不能漏到界面上"""
    today = date.today()
    _add(engine, "a", expiry=today, cost=0.1)
    _add(engine, "b", expiry=today, cost=0.2)
    assert _overview(engine)["year_cost"] == 0.3


def test_overdue_and_expiring_counted_separately(engine):
    """已过期比 30 天内到期更该被看见，不能因为不在未来窗口里就从统计消失"""
    today = date.today()
    _add(engine, "overdue", expiry=today - timedelta(days=1))
    _add(engine, "soon", expiry=today + timedelta(days=10))
    _add(engine, "far", expiry=today + timedelta(days=90))
    got = _overview(engine)
    assert got["overdue_count"] == 1
    assert got["expiring_30"] == 1
    assert got["total"] == 3


def test_today_counts_as_expiring_not_overdue(engine):
    """到期日就是今天：还能续费，属于倒计时而不是已错过"""
    _add(engine, "due-today", expiry=date.today())
    got = _overview(engine)
    assert got["expiring_30"] == 1
    assert got["overdue_count"] == 0


def test_cancelled_assets_excluded_from_expiry_counts(engine):
    """已注销资产不该再触发续费提醒"""
    today = date.today()
    _add(engine, "cancelled-overdue", expiry=today - timedelta(days=5),
         status=AssetStatus.CANCELLED)
    _add(engine, "cancelled-soon", expiry=today + timedelta(days=5),
         status=AssetStatus.CANCELLED)
    got = _overview(engine)
    assert got["overdue_count"] == 0
    assert got["expiring_30"] == 0
    # 但仍计入总数：台账要能查到已注销的历史资产
    assert got["total"] == 2


def test_abnormal_counts_stopped_and_expired(engine):
    _add(engine, "stopped", status=AssetStatus.STOPPED)
    _add(engine, "expired", status=AssetStatus.EXPIRED)
    _add(engine, "active", status=AssetStatus.ACTIVE)
    got = _overview(engine)
    assert got["abnormal_count"] == 2
    assert got["by_status"] == {"stopped": 1, "expired": 1, "active": 1}


def test_by_type_uses_enum_values_not_names(engine):
    """前端按 'vps'/'domain' 匹配图标与文案，给出 'VPS' 会全部落到兜底分支"""
    _add(engine, "v1", asset_type=AssetType.VPS)
    _add(engine, "d1", asset_type=AssetType.DOMAIN)
    _add(engine, "d2", asset_type=AssetType.DOMAIN)
    got = _overview(engine)
    assert got["by_type"] == {"vps": 1, "domain": 2}