mirror of
https://github.com/Routstr/routstr-core.git
synced 2026-10-05 12:28:22 +00:00
fix: type certification regression fixtures for strict mypy
This commit is contained in:
@@ -273,8 +273,12 @@ def test_create_model_mappings_applies_custom_provider_fees_before_advertising(
|
||||
}
|
||||
|
||||
def fake_row_to_model(
|
||||
row, *, apply_provider_fee, provider_fee, provider_type
|
||||
) -> Model: # type: ignore[no-untyped-def]
|
||||
row: SimpleNamespace,
|
||||
*,
|
||||
apply_provider_fee: bool,
|
||||
provider_fee: float,
|
||||
provider_type: str,
|
||||
) -> Model:
|
||||
assert provider_type == providers[row.upstream_provider_id - 1].provider_type
|
||||
assert apply_provider_fee is True
|
||||
return create_test_model(
|
||||
|
||||
@@ -192,10 +192,14 @@ def test_no_evaluated_models_warns_all_pricing_rows() -> None:
|
||||
_report_row_served_matches_configured,
|
||||
)
|
||||
]
|
||||
assert all(
|
||||
row["status"] == "warn" and row["evidence"]["checked"] == 0 for row in rows
|
||||
)
|
||||
assert all("Not evaluated" in row["detail"] for row in rows)
|
||||
for row in rows:
|
||||
assert row["status"] == "warn"
|
||||
evidence = row["evidence"]
|
||||
assert isinstance(evidence, dict)
|
||||
assert evidence["checked"] == 0
|
||||
detail = row["detail"]
|
||||
assert isinstance(detail, str)
|
||||
assert "Not evaluated" in detail
|
||||
assert (
|
||||
next(c for c in build_checklist(rows) if c["goal"] == "pricing_v1_models")[
|
||||
"status"
|
||||
|
||||
@@ -1,12 +1,16 @@
|
||||
import json
|
||||
from collections.abc import AsyncIterator
|
||||
from contextlib import asynccontextmanager
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from sqlmodel.ext.asyncio.session import AsyncSession
|
||||
|
||||
from routstr.core.db import ModelRow
|
||||
from routstr.payment.models import Pricing, _row_to_model, list_models
|
||||
from routstr.upstream.base import BaseUpstreamProvider
|
||||
from routstr.upstream.helpers import get_all_models_with_overrides
|
||||
from routstr.upstream.model_paths import _price_in_sats
|
||||
|
||||
@@ -34,24 +38,25 @@ def _row(cache_rate: float = 0) -> ModelRow:
|
||||
)
|
||||
|
||||
|
||||
def _session(row: ModelRow, provider_type: str):
|
||||
def _session(row: ModelRow, provider_type: str) -> AsyncSession:
|
||||
provider = SimpleNamespace(
|
||||
id=1, enabled=True, provider_fee=1.1, provider_type=provider_type
|
||||
)
|
||||
return SimpleNamespace(
|
||||
return MagicMock(
|
||||
spec=AsyncSession,
|
||||
exec=AsyncMock(
|
||||
side_effect=[
|
||||
SimpleNamespace(all=lambda: [row]),
|
||||
SimpleNamespace(all=lambda: [provider]),
|
||||
]
|
||||
)
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("provider_type", ["ppqai", "venice"])
|
||||
@pytest.mark.parametrize("cache_rate", [0, 7e-7])
|
||||
def test_db_conversion_preserves_native_or_explicit_cache_prices(
|
||||
provider_type, cache_rate
|
||||
provider_type: str, cache_rate: float
|
||||
) -> None:
|
||||
row = _row(cache_rate)
|
||||
stored = row.pricing
|
||||
@@ -71,7 +76,7 @@ def test_db_conversion_preserves_native_or_explicit_cache_prices(
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("provider_type", ["ppqai", "venice"])
|
||||
async def test_admin_listing_passes_provider_policy_to_database_conversion(
|
||||
provider_type,
|
||||
provider_type: str,
|
||||
) -> None:
|
||||
with (
|
||||
patch("routstr.payment.models.backfill_cache_pricing") as backfill,
|
||||
@@ -86,16 +91,17 @@ async def test_admin_listing_passes_provider_policy_to_database_conversion(
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("provider_type", ["ppqai", "venice"])
|
||||
async def test_runtime_overrides_do_not_reintroduce_generic_cache_prices(
|
||||
provider_type,
|
||||
provider_type: str,
|
||||
) -> None:
|
||||
row = _row()
|
||||
session = _session(row, provider_type)
|
||||
|
||||
@asynccontextmanager
|
||||
async def create_session():
|
||||
async def create_session() -> AsyncIterator[AsyncSession]:
|
||||
yield session
|
||||
|
||||
upstream = SimpleNamespace(
|
||||
upstream = MagicMock(
|
||||
spec=BaseUpstreamProvider,
|
||||
db_id=1,
|
||||
provider_type=provider_type,
|
||||
base_url="https://example.invalid",
|
||||
@@ -115,8 +121,8 @@ async def test_runtime_overrides_do_not_reintroduce_generic_cache_prices(
|
||||
|
||||
|
||||
@pytest.mark.parametrize("provider_type", ["ppqai", "venice"])
|
||||
def test_path_sats_conversion_respects_native_cache_policy(provider_type) -> None:
|
||||
model = {
|
||||
def test_path_sats_conversion_respects_native_cache_policy(provider_type: str) -> None:
|
||||
model: dict[str, Any] = {
|
||||
"id": "vendor/model",
|
||||
"pricing": {"prompt": 4e-6, "completion": 8e-6},
|
||||
"context_length": 8192,
|
||||
|
||||
@@ -21,6 +21,7 @@ def _model() -> Model:
|
||||
input_modalities=["text"],
|
||||
output_modalities=["text"],
|
||||
tokenizer="Unknown",
|
||||
instruct_type=None,
|
||||
),
|
||||
pricing=Pricing(
|
||||
prompt=1e-6,
|
||||
@@ -89,7 +90,7 @@ async def test_ppq_alias_matches_do_not_share_mutated_prices() -> None:
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("metadata", [None, []])
|
||||
async def test_ppq_partial_api_prices_fall_back_per_field_preserving_zero(
|
||||
metadata,
|
||||
metadata: list[dict] | None,
|
||||
) -> None:
|
||||
(model,) = await _fetch(
|
||||
[
|
||||
@@ -111,7 +112,7 @@ async def test_ppq_partial_api_prices_fall_back_per_field_preserving_zero(
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("rate", [None, -1, float("inf"), float("nan")])
|
||||
async def test_ppq_unpriced_or_invalid_native_rate_is_not_replaced_by_openrouter(
|
||||
rate,
|
||||
rate: float | None,
|
||||
) -> None:
|
||||
assert (
|
||||
await _fetch(
|
||||
@@ -126,7 +127,9 @@ async def test_ppq_unpriced_or_invalid_native_rate_is_not_replaced_by_openrouter
|
||||
|
||||
|
||||
@pytest.mark.parametrize("provider", [PPQAIUpstreamProvider, VeniceUpstreamProvider])
|
||||
def test_native_catalog_providers_do_not_backfill_generic_cache_rates(provider) -> None:
|
||||
def test_native_catalog_providers_do_not_backfill_generic_cache_rates(
|
||||
provider: type[PPQAIUpstreamProvider | VeniceUpstreamProvider],
|
||||
) -> None:
|
||||
model = _model()
|
||||
model.pricing = Pricing(prompt=4e-6, completion=8e-6)
|
||||
with patch("routstr.upstream.base.backfill_cache_pricing") as backfill:
|
||||
|
||||
Reference in New Issue
Block a user