awoooi/apps/api/tests/test_telegram_group_chat_provider_router.py

from __future__ import annotations

from unittest.mock import AsyncMock

import pytest

from src.services import chat_manager as chat_module
from src.services import ai_rate_limiter as limiter_module
from src.services import telegram_gateway as gateway_module
from src.services.chat_manager import ChatManager
from src.services.telegram_gateway import TelegramGateway


class _ReceiptBackedCostLimiter:
    async def get_usage_stats(self, provider: str) -> dict:
        return {
            "provider": provider,
            "readback_status": "verified",
            "total_cost_usd": {
                "current": 0.011802 if provider == "claude" else 0.000297,
            },
            "accounting": {
                "complete": True,
                "status": "receipt_backed",
                "degraded": False,
            },
        }


@pytest.mark.asyncio
async def test_nemotron_group_mention_uses_structured_provider_status_router(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Telegram mention must not invent a Gemini lock or call a provider."""

    readback = {
        "schema_version": "sre_chat_provider_runtime_readback_v1",
        "policy_order": [
            "ollama_gcp_a",
            "ollama_gcp_b",
            "ollama_local",
            "claude",
            "gemini",
        ],
        "ollama_health": {
            "ollama_gcp_a": "offline",
            "ollama_gcp_b": "healthy",
            "ollama_local": "not_checked",
        },
        "paid": {
            "claude": {
                "configured": True,
                "authenticated": True,
                "authentication_verified": True,
                "verification_status": "authenticated_verified_by_generation",
                "execution_mode": "canary",
                "canary_percent": 5,
                "route_lease_state": "inactive_disabled_after_canary",
                "runtime_state": (
                    "configured_authenticated_canary_waiting_for_run_owned_lease"
                ),
            },
            "gemini": {
                "configured": True,
                "authenticated": False,
                "authentication_verified": False,
                "verification_status": "authentication_rejected",
                "execution_mode": "canary",
                "canary_percent": 5,
                "route_lease_state": "inactive_disabled_after_canary",
                "runtime_state": "configured_canary_authentication_rejected",
            },
        },
        "provider_route_switch_performed": False,
        "infrastructure_remediation_write_allowed": False,
        "cross_domain_execution_allowed": False,
    }
    manager = ChatManager()
    monkeypatch.setattr(
        manager,
        "get_system_context",
        AsyncMock(return_value="bounded system context"),
    )
    monkeypatch.setattr(
        manager,
        "get_provider_runtime_readback",
        AsyncMock(return_value=readback),
    )
    monkeypatch.setattr(chat_module, "get_chat_manager", lambda: manager)

    def _provider_call_forbidden():
        raise AssertionError("provider executor must not run for a status readback")

    monkeypatch.setattr(chat_module, "get_ai_executor", _provider_call_forbidden)
    monkeypatch.setattr(gateway_module.settings, "HERMES_NL_ENABLED", False)

    gateway = object.__new__(TelegramGateway)
    send_as_nemotron = AsyncMock(return_value={"ok": True})
    monkeypatch.setattr(gateway, "send_as_nemotron", send_as_nemotron)

    await gateway._handle_group_message(
        "@NemoTronAwoooI_Bot Gemini API 現在是硬鎖定狀態嗎？",
        user_id=99,
        username="operator",
        chat_id=-100123,
        message_id=456,
    )

    send_as_nemotron.assert_awaited_once()
    outbound = send_as_nemotron.await_args.kwargs["text"]
    assert "硬鎖" not in outbound
    assert "locked" not in outbound.casefold()
    assert "無法正常回應" not in outbound
    assert "Claude：configured=true｜authenticated=true" in outbound
    assert "Gemini：configured=true｜authenticated=false" in outbound
    assert "verification=authentication_rejected" in outbound
    assert "canary=canary 5%" in outbound
    assert "route_lease=inactive_disabled_after_canary" in outbound
    assert "provider=runtime_readback" in outbound
    assert "model=deterministic_policy" in outbound
    assert "fallback=provider_status_request_no_generation" in outbound
    assert "cost_receipt=none" in outbound
    assert send_as_nemotron.await_args.kwargs == {
        "text": outbound,
        "reply_to_message_id": 456,
        "inbound_chat_id": -100123,
    }


@pytest.mark.asyncio
async def test_group_cost_uses_receipt_backed_accounting_not_legacy_keys(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(
        limiter_module,
        "get_ai_rate_limiter",
        lambda: _ReceiptBackedCostLimiter(),
    )
    gateway = object.__new__(TelegramGateway)
    send_as_openclaw = AsyncMock(return_value={"ok": True})
    monkeypatch.setattr(gateway, "send_as_openclaw", send_as_openclaw)

    await gateway._handle_group_command(
        "/cost",
        -100123,
        789,
        full_text="/cost",
    )

    send_as_openclaw.assert_awaited_once()
    outbound = send_as_openclaw.await_args.kwargs["text"]
    assert "Claude: <b>$0.011802</b> | receipt=verified" in outbound
    assert "Gemini: <b>$0.000297</b> | receipt=verified" in outbound
    assert "合計: <b>$0.012099</b>" in outbound
    assert "receipt-backed" in outbound
    assert "Flash-Lite" not in outbound
    assert "Haiku" not in outbound