#!/usr/bin/env python3
"""Budget gate boundary tests — TokenHub 企业级三原则落地（2026-08-06 吸收）。

TokenHub 知识核心：「企业级不是功能列表更长，而是关键行为可验证、故障可复现、
修改不容易引入回归。边界问题（长上下文/流式中断/配额耗尽）必须固化为自动化测试。」

本文件把 Hermes 预算模块（app.py _check_budget/_record_budget_spend）的
「配额耗尽」边界固化为自动化测试，防止未来改动（如 L2 加 depth_tiers /
success_weight / 缓存感知定价）引入回归。

覆盖边界：
- 配额耗尽（BUDGET_EXHAUSTED）→ 精确拦截，不产生超额调用
- 预警阈值（BUDGET_WARNING）→ 放行但打 trace
- 正常路径 → 放行
- 未知 job_type → 不受限
- 月度自动重置 → 跨月隔离
- 记账累加 → cost_per_call 单调累计
"""
import json
from datetime import datetime as _dt

import pytest

import app


def _cur_month() -> str:
    """预算模块用当前月份作 state key——测试必须用同一 key。"""
    return _dt.now().strftime("%Y-%m")


@pytest.fixture(autouse=True)
def _isolated_budget_file(tmp_path, monkeypatch):
    """每个测试用独立预算文件，互不污染；测试后还原真实文件路径。"""
    monkeypatch.setattr(app, "BUDGET_FILE", tmp_path / "hermes_budget.json")
    yield


# =============================================================================
# 配额耗尽边界（TokenHub 三类边界问题之③）
# =============================================================================


class TestBudgetExhausted:
    """配额耗尽时精确拦截——不产生超额调用。"""

    def test_exhausted_blocks_with_reason_code(self, monkeypatch):
        """已达上限 → (False, BUDGET_EXHAUSTED)，reason_code 可归因。"""
        monkeypatch.setattr(
            app, "_load_budget_state",
            lambda: {_cur_month(): {"learn_absorb": 20.0}},  # >= monthly_limit
        )
        allowed, rc = app._check_budget("learn_absorb")
        assert allowed is False
        assert rc == app.VerifyReason.BUDGET_EXHAUSTED

    def test_exhausted_returns_gate_action_block(self, monkeypatch):
        """BUDGET_EXHAUSTED 必须映射到 BLOCK 动作（spawn 层拒绝启动）。"""
        monkeypatch.setattr(
            app, "_load_budget_state",
            lambda: {_cur_month(): {"learn_absorb": 99.0}},
        )
        allowed, rc = app._check_budget("learn_absorb")
        assert rc == app.VerifyReason.BUDGET_EXHAUSTED
        assert app.GateAction.BLOCK is not None  # BLOCK 动作常量存在

    def test_exactly_at_limit_blocks(self, monkeypatch):
        """恰好等于上限也拦截（>= 语义，不产生超额调用）。"""
        monkeypatch.setattr(
            app, "_load_budget_state",
            lambda: {_cur_month(): {"learn_absorb": 20.0}},
        )
        allowed, _ = app._check_budget("learn_absorb")
        assert allowed is False

    def test_below_limit_allowed(self, monkeypatch):
        """未达上限 → 放行。"""
        monkeypatch.setattr(
            app, "_load_budget_state",
            lambda: {_cur_month(): {"learn_absorb": 10.0}},
        )
        allowed, rc = app._check_budget("learn_absorb")
        assert allowed is True
        assert rc is None


# =============================================================================
# 预警阈值边界
# =============================================================================


class TestBudgetWarning:
    """达到预警阈值（默认 80%）→ 放行但返回 BUDGET_WARNING reason_code。"""

    def test_at_warn_threshold_warns(self, monkeypatch):
        monkeypatch.setattr(
            app, "_load_budget_state",
            lambda: {_cur_month(): {"learn_absorb": 16.0}},  # 80% of 20.0
        )
        allowed, rc = app._check_budget("learn_absorb")
        assert allowed is True
        assert rc == app.VerifyReason.BUDGET_WARNING

    def test_below_warn_threshold_no_reason(self, monkeypatch):
        monkeypatch.setattr(
            app, "_load_budget_state",
            lambda: {_cur_month(): {"learn_absorb": 15.9}},
        )
        allowed, rc = app._check_budget("learn_absorb")
        assert allowed is True
        assert rc is None


# =============================================================================
# 正常路径 + 未知 job_type
# =============================================================================


class TestBudgetNormalPath:
    def test_no_state_file_allowed(self):
        """预算文件不存在（首次运行）→ 放行，不崩溃。"""
        allowed, rc = app._check_budget("learn_absorb")
        assert allowed is True
        assert rc is None

    def test_unknown_job_type_unrestricted(self):
        """无配置的 job_type 不受预算限制。"""
        allowed, rc = app._check_budget("no_such_job")
        assert allowed is True
        assert rc is None

    def test_zero_cost_job_never_exhausts(self, monkeypatch):
        """cost_per_call=0 的任务即使有记录也不触发耗尽（除以 0 保护）。"""
        monkeypatch.setattr(
            app, "_load_budget_state",
            lambda: {"2099-01": {"learn_absorb": 0.0}},
        )
        allowed, _ = app._check_budget("learn_absorb")
        assert allowed is True


# =============================================================================
# 月度自动重置（跨月隔离）
# =============================================================================


class TestBudgetMonthlyReset:
    def test_previous_month_spend_ignored(self, monkeypatch):
        """上月已耗尽 → 本月重新计数，放行。"""
        monkeypatch.setattr(
            app, "_load_budget_state",
            lambda: {"2000-01": {"learn_absorb": 999.0}},  # 远古月份
        )
        allowed, rc = app._check_budget("learn_absorb")
        assert allowed is True
        assert rc is None

    def test_record_spend_writes_current_month(self, tmp_path):
        """记账写入当前月份 key（YYYY-MM），不写入旧月份。"""
        app._record_budget_spend("learn_absorb")
        state = json.loads((tmp_path / "hermes_budget.json").read_text())
        keys = list(state.keys())
        assert len(keys) == 1
        import datetime as _dt
        assert keys[0] == _dt.datetime.now().strftime("%Y-%m")


# =============================================================================
# 记账累加
# =============================================================================


class TestBudgetRecordSpend:
    def test_spend_accumulates(self, tmp_path):
        """两次调用 → 2× cost_per_call，单调累计。"""
        app._record_budget_spend("learn_absorb")
        app._record_budget_spend("learn_absorb")
        state = json.loads((tmp_path / "hermes_budget.json").read_text())
        month = list(state.keys())[0]
        assert state[month]["learn_absorb"] == pytest.approx(2 * 0.50)

    def test_spend_isolated_per_job_type(self, tmp_path):
        """不同 job_type 独立记账。"""
        app._record_budget_spend("chat_reply")
        app._record_budget_spend("learn_absorb")
        state = json.loads((tmp_path / "hermes_budget.json").read_text())
        month = list(state.keys())[0]
        assert state[month]["chat_reply"] == pytest.approx(0.05)
        assert state[month]["learn_absorb"] == pytest.approx(0.50)

    def test_unknown_job_type_no_record(self, tmp_path):
        """未知 job_type 不写预算文件。"""
        app._record_budget_spend("no_such_job")
        assert not (tmp_path / "hermes_budget.json").exists()


# =============================================================================
# End-to-end：耗尽 → BLOCK（spawn gate 四元组 trace）
# =============================================================================


class TestBudgetGateTrace:
    def test_exhausted_writes_block_trace(self, monkeypatch, tmp_path):
        """模拟 spawn gate：耗尽时 trace 含 node/attempt/gate/action 四元组。"""
        monkeypatch.setattr(
            app, "_load_budget_state",
            lambda: {"2099-01": {"learn_absorb": 20.0}},
        )
        allowed, rc = app._check_budget("learn_absorb")
        if not allowed:
            trace = (f"[node=spawn,attempt=1,gate={rc},"
                     f"action={app.GateAction.BLOCK}]")
            assert "node=spawn" in trace
            assert "attempt=1" in trace
            assert f"gate={app.VerifyReason.BUDGET_EXHAUSTED}" in trace
            assert "action=BLOCK" in trace