devplacepy/tests/api/admin/gateway/models.py

# retoor <retoor@molodetz.nl>

import requests

from tests.conftest import BASE_URL
from tests.api.admin.gateway.index import (
    JSON_gateway,
    admin_session,
    member_key,
    _unique_gateway,
)


def test_models_require_admin(seeded_db):
    assert (
        requests.get(
            f"{BASE_URL}/admin/gateway/models",
            headers=JSON_gateway,
            allow_redirects=False,
        ).status_code
        == 401
    )
    key = member_key()
    assert (
        requests.get(
            f"{BASE_URL}/admin/gateway/models",
            headers={**JSON_gateway, "X-API-KEY": key},
            allow_redirects=False,
        ).status_code
        == 403
    )
    assert (
        admin_session(seeded_db).get(f"{BASE_URL}/admin/gateway/models").status_code
        == 200
    )


def test_model_route_create_and_economy(seeded_db):
    admin = admin_session(seeded_db)
    source = _unique_gateway("src")

    created = admin.post(
        f"{BASE_URL}/admin/gateway/models",
        json={
            "source_model": source,
            "provider": "",
            "target_model": "vendor/y",
            "kind": "chat",
            "price_output_per_m": 5.0,
            "price_cache_miss_per_m": 2.0,
            "context_window": 32000,
        },
    )
    assert created.status_code == 200, created.text[:300]
    assert created.json()["model"]["target_model"] == "vendor/y"

    listed = admin.get(f"{BASE_URL}/admin/gateway/models").json()
    row = next(m for m in listed["models"] if m["source_model"] == source)
    assert row["kind"] == "chat"
    assert row["price_output_per_m"] == 5.0
    assert row["context_window"] == 32000

    deleted = admin.delete(f"{BASE_URL}/admin/gateway/models/{source}")
    assert deleted.status_code == 200
    assert admin.delete(f"{BASE_URL}/admin/gateway/models/{source}").status_code == 404


def test_model_route_tiered_and_off_peak_pricing_round_trip(seeded_db):
    admin = admin_session(seeded_db)
    source = _unique_gateway("tiered")

    created = admin.post(
        f"{BASE_URL}/admin/gateway/models",
        json={
            "source_model": source,
            "target_model": "vendor/tiered",
            "kind": "chat",
            "price_cache_hit_per_m": 0.0028,
            "price_cache_miss_per_m": 0.14,
            "price_output_per_m": 0.28,
            "context_tier_threshold_tokens": 128000,
            "price_output_per_m_tier2": 0.56,
            "off_peak_start_minute": 990,
            "off_peak_end_minute": 30,
            "off_peak_discount_pct": 25.0,
        },
    )
    assert created.status_code == 200, created.text[:300]
    model = created.json()["model"]
    assert model["context_tier_threshold_tokens"] == 128000
    assert model["price_output_per_m_tier2"] == 0.56
    assert model["price_cache_hit_per_m_tier2"] is None
    assert model["off_peak_start_minute"] == 990
    assert model["off_peak_end_minute"] == 30
    assert model["off_peak_discount_pct"] == 25.0

    listed = admin.get(f"{BASE_URL}/admin/gateway/models").json()
    row = next(m for m in listed["models"] if m["source_model"] == source)
    assert row["context_tier_threshold_tokens"] == 128000
    assert row["price_output_per_m_tier2"] == 0.56
    assert row["off_peak_start_minute"] == 990
    assert row["off_peak_end_minute"] == 30
    assert row["off_peak_discount_pct"] == 25.0

    deleted = admin.delete(f"{BASE_URL}/admin/gateway/models/{source}")
    assert deleted.status_code == 200


def test_model_route_off_peak_requires_both_start_and_end(seeded_db):
    admin = admin_session(seeded_db)
    source = _unique_gateway("badwindow")

    only_start = admin.post(
        f"{BASE_URL}/admin/gateway/models",
        json={
            "source_model": source,
            "target_model": "vendor/w",
            "kind": "chat",
            "off_peak_start_minute": 60,
        },
    )
    assert only_start.status_code == 400
    assert only_start.json()["ok"] is False

    only_end = admin.post(
        f"{BASE_URL}/admin/gateway/models",
        json={
            "source_model": source,
            "target_model": "vendor/w",
            "kind": "chat",
            "off_peak_end_minute": 120,
        },
    )
    assert only_end.status_code == 400
    assert only_end.json()["ok"] is False


def test_model_route_fallback_round_trip(seeded_db):
    admin = admin_session(seeded_db)
    backup = _unique_gateway("fb-backup")
    primary = _unique_gateway("fb-primary")

    created_backup = admin.post(
        f"{BASE_URL}/admin/gateway/models",
        json={"source_model": backup, "target_model": "vendor/backup", "kind": "chat"},
    )
    assert created_backup.status_code == 200, created_backup.text[:300]

    created_primary = admin.post(
        f"{BASE_URL}/admin/gateway/models",
        json={
            "source_model": primary,
            "target_model": "vendor/primary",
            "kind": "chat",
            "fallback_model": backup,
        },
    )
    assert created_primary.status_code == 200, created_primary.text[:300]
    assert created_primary.json()["model"]["fallback_model"] == backup

    listed = admin.get(f"{BASE_URL}/admin/gateway/models").json()
    row = next(m for m in listed["models"] if m["source_model"] == primary)
    assert row["fallback_model"] == backup

    admin.delete(f"{BASE_URL}/admin/gateway/models/{primary}")
    admin.delete(f"{BASE_URL}/admin/gateway/models/{backup}")


def test_model_route_fallback_must_reference_an_existing_route(seeded_db):
    admin = admin_session(seeded_db)
    source = _unique_gateway("fb-ghost")

    response = admin.post(
        f"{BASE_URL}/admin/gateway/models",
        json={
            "source_model": source,
            "target_model": "vendor/z",
            "kind": "chat",
            "fallback_model": _unique_gateway("does-not-exist"),
        },
    )
    assert response.status_code == 400
    assert response.json()["ok"] is False


def test_model_route_fallback_must_be_the_same_kind(seeded_db):
    admin = admin_session(seeded_db)
    embed_route = _unique_gateway("fb-embed")
    chat_route = _unique_gateway("fb-chat")

    admin.post(
        f"{BASE_URL}/admin/gateway/models",
        json={"source_model": embed_route, "target_model": "vendor/e", "kind": "embed"},
    )
    response = admin.post(
        f"{BASE_URL}/admin/gateway/models",
        json={
            "source_model": chat_route,
            "target_model": "vendor/c",
            "kind": "chat",
            "fallback_model": embed_route,
        },
    )
    assert response.status_code == 400
    assert response.json()["ok"] is False

    admin.delete(f"{BASE_URL}/admin/gateway/models/{embed_route}")


def test_model_route_fallback_rejects_self_reference(seeded_db):
    admin = admin_session(seeded_db)
    source = _unique_gateway("fb-self")

    response = admin.post(
        f"{BASE_URL}/admin/gateway/models",
        json={
            "source_model": source,
            "target_model": "vendor/z",
            "kind": "chat",
            "fallback_model": source,
        },
    )
    assert response.status_code == 400
    assert response.json()["ok"] is False


def test_model_form_pages_require_admin(seeded_db):
    assert (
        requests.get(
            f"{BASE_URL}/admin/gateway/models/new",
            headers=JSON_gateway,
            allow_redirects=False,
        ).status_code
        == 401
    )
    key = member_key()
    assert (
        requests.get(
            f"{BASE_URL}/admin/gateway/models/new",
            headers={**JSON_gateway, "X-API-KEY": key},
            allow_redirects=False,
        ).status_code
        == 403
    )


def test_model_add_edit_delete_via_page(seeded_db):
    admin = admin_session(seeded_db)
    source = _unique_gateway("modelpage")

    new_page = admin.get(f"{BASE_URL}/admin/gateway/models/new")
    assert new_page.status_code == 200
    assert new_page.json()["is_edit"] is False

    created = admin.post(
        f"{BASE_URL}/admin/gateway/models/new",
        data={
            "source_model": source,
            "target_model": "vendor/page",
            "kind": "chat",
            "price_output_per_m": "1.5",
            "off_peak_start": "22:30",
            "off_peak_end": "06:00",
            "off_peak_discount_pct": "10",
            "is_active": "1",
        },
        allow_redirects=False,
    )
    assert created.status_code == 302
    assert created.headers["location"] == "/admin/gateway?tab=models"

    edit_page = admin.get(f"{BASE_URL}/admin/gateway/models/{source}/edit")
    assert edit_page.status_code == 200
    form = edit_page.json()["form"]
    assert form["target_model"] == "vendor/page"
    assert form["off_peak_start"] == "22:30"
    assert form["off_peak_end"] == "06:00"

    edit_html = admin.get(f"{BASE_URL}/admin/gateway/models/{source}/edit", headers={"Accept": "text/html"})
    assert f'value="{source}"' in edit_html.text
    assert "readonly" in edit_html.text

    updated = admin.post(
        f"{BASE_URL}/admin/gateway/models/{source}/edit",
        data={"target_model": "vendor/page2", "kind": "chat", "is_active": "1"},
        allow_redirects=False,
    )
    assert updated.status_code == 302

    relisted = admin.get(f"{BASE_URL}/admin/gateway/models").json()
    row = next(m for m in relisted["models"] if m["source_model"] == source)
    assert row["target_model"] == "vendor/page2"

    deleted = admin.post(
        f"{BASE_URL}/admin/gateway/models/{source}/delete", allow_redirects=False
    )
    assert deleted.status_code == 302
    assert admin.get(f"{BASE_URL}/admin/gateway/models/{source}/edit").status_code == 404


def test_model_page_fallback_must_be_the_same_kind(seeded_db):
    admin = admin_session(seeded_db)
    embed_route = _unique_gateway("fbpage-embed")
    chat_route = _unique_gateway("fbpage-chat")

    admin.post(
        f"{BASE_URL}/admin/gateway/models",
        json={"source_model": embed_route, "target_model": "vendor/e", "kind": "embed"},
    )
    response = admin.post(
        f"{BASE_URL}/admin/gateway/models/new",
        data={
            "source_model": chat_route,
            "target_model": "vendor/c",
            "kind": "chat",
            "fallback_model": embed_route,
        },
    )
    assert response.status_code == 400
    assert "message" in response.json()["error"]

    admin.delete(f"{BASE_URL}/admin/gateway/models/{embed_route}")


def test_model_edit_page_404_for_missing_model(seeded_db):
    admin = admin_session(seeded_db)
    missing = _unique_gateway("ghostmodel")
    assert admin.get(f"{BASE_URL}/admin/gateway/models/{missing}/edit").status_code == 404
    assert (
        admin.post(f"{BASE_URL}/admin/gateway/models/{missing}/delete").status_code == 404
    )


def test_model_route_validation(seeded_db):
    admin = admin_session(seeded_db)
    missing_target = admin.post(
        f"{BASE_URL}/admin/gateway/models",
        json={"source_model": _unique_gateway("src")},
    )
    assert missing_target.status_code == 400
    assert missing_target.json()["ok"] is False

    bad_kind = admin.post(
        f"{BASE_URL}/admin/gateway/models",
        json={
            "source_model": _unique_gateway("src"),
            "target_model": "vendor/z",
            "kind": "bogus",
        },
    )
    assert bad_kind.status_code == 400