56 lines
1.7 KiB
Python
56 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
from src.services.ollama_endpoint_resolver import (
|
|
resolve_ollama_endpoint,
|
|
resolve_ollama_selection,
|
|
)
|
|
|
|
|
|
def _settings(
|
|
*,
|
|
primary: str = "http://192.168.0.110:11435",
|
|
secondary: str = "http://192.168.0.110:11436",
|
|
fallback: str = "http://192.168.0.110:11437",
|
|
) -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
OLLAMA_URL=primary,
|
|
OLLAMA_SECONDARY_URL=secondary,
|
|
OLLAMA_FALLBACK_URL=fallback,
|
|
)
|
|
|
|
|
|
def test_batch_workloads_prefer_gcp_b() -> None:
|
|
cfg = _settings()
|
|
|
|
for workload in ("batch", "embedding", "rag", "code_review", "shadow", "canary"):
|
|
selection = resolve_ollama_selection(workload, config=cfg)
|
|
assert selection.url == "http://192.168.0.110:11436"
|
|
assert selection.provider_name == "ollama_gcp_b"
|
|
assert selection.reason == "gcp_b_batch_lane"
|
|
|
|
|
|
def test_interactive_workloads_stay_on_gcp_a() -> None:
|
|
cfg = _settings()
|
|
|
|
for workload in ("interactive", "healthcheck"):
|
|
selection = resolve_ollama_selection(workload, config=cfg)
|
|
assert selection.url == "http://192.168.0.110:11435"
|
|
assert selection.provider_name == "ollama_gcp_a"
|
|
|
|
|
|
def test_local_required_workloads_use_local_lane() -> None:
|
|
cfg = _settings()
|
|
|
|
for workload in ("local_required", "privacy_sensitive", "dr"):
|
|
selection = resolve_ollama_selection(workload, config=cfg)
|
|
assert selection.url == "http://192.168.0.110:11437"
|
|
assert selection.provider_name == "ollama_local"
|
|
|
|
|
|
def test_batch_workloads_fall_back_to_primary_when_secondary_missing() -> None:
|
|
cfg = _settings(secondary="")
|
|
|
|
assert resolve_ollama_endpoint("embedding", config=cfg) == "http://192.168.0.110:11435"
|