cesnimda
5eb9b3cb96
Keep external providers behind server consent, task, and prompt-cost gates while persisting actual provider provenance.
521 lines
20 KiB
Python
521 lines
20 KiB
Python
import importlib
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
|
|
def load_app_module(
|
|
monkeypatch,
|
|
*,
|
|
skip_model_load=True,
|
|
ollama_model=None,
|
|
service_token=None,
|
|
external_ai_enabled=False,
|
|
routing_mode="local_first",
|
|
circuit_threshold=3,
|
|
external_prompt_limit=24000,
|
|
):
|
|
if skip_model_load:
|
|
monkeypatch.setenv("AI_SERVICE_SKIP_MODEL_LOAD", "1")
|
|
else:
|
|
monkeypatch.delenv("AI_SERVICE_SKIP_MODEL_LOAD", raising=False)
|
|
monkeypatch.delenv("AI_SERVICE_EAGER_MODEL_LOAD", raising=False)
|
|
# Default to keyless so the existing suite is unaffected by a token in the dev shell.
|
|
if service_token is None:
|
|
monkeypatch.delenv("AI_SERVICE_TOKEN", raising=False)
|
|
else:
|
|
monkeypatch.setenv("AI_SERVICE_TOKEN", service_token)
|
|
if ollama_model is None:
|
|
monkeypatch.delenv("OLLAMA_MODEL", raising=False)
|
|
else:
|
|
monkeypatch.setenv("OLLAMA_MODEL", ollama_model)
|
|
if external_ai_enabled:
|
|
monkeypatch.setenv("EXTERNAL_AI_ENABLED", "true")
|
|
else:
|
|
monkeypatch.delenv("EXTERNAL_AI_ENABLED", raising=False)
|
|
monkeypatch.setenv("AI_ROUTING_MODE", routing_mode)
|
|
monkeypatch.setenv("LOCAL_AI_CIRCUIT_FAILURE_THRESHOLD", str(circuit_threshold))
|
|
monkeypatch.setenv("EXTERNAL_AI_MAX_PROMPT_CHARS", str(external_prompt_limit))
|
|
monkeypatch.delenv("EXTERNAL_AI_ALLOWED_TASKS", raising=False)
|
|
if "app" in sys.modules:
|
|
del sys.modules["app"]
|
|
module = importlib.import_module("app")
|
|
return importlib.reload(module)
|
|
|
|
|
|
def test_health_reports_runtime_without_ollama_and_without_forcing_model_load(monkeypatch):
|
|
module = load_app_module(monkeypatch)
|
|
client = TestClient(module.app)
|
|
|
|
response = client.get("/health")
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["ok"] is True
|
|
assert payload["device"] == "cpu"
|
|
assert payload["model_loaded"] is False
|
|
assert payload["model_disabled"] is True
|
|
assert payload["summarize_available"] is False
|
|
assert "disabled" in payload["model_load_error"].lower()
|
|
assert payload["ollama_configured"] is False
|
|
assert payload["ollama_model"] is None
|
|
assert payload["ollama_installed_models"] == []
|
|
assert payload["ollama_loaded_models"] == []
|
|
|
|
|
|
def test_summarize_returns_503_with_explicit_reason_when_model_loading_is_disabled(monkeypatch):
|
|
module = load_app_module(monkeypatch)
|
|
client = TestClient(module.app)
|
|
|
|
response = client.post("/summarize", json={"text": "Platform engineering role with APIs and Python experience."})
|
|
|
|
assert response.status_code == 503
|
|
payload = response.json()
|
|
assert "disabled" in payload["detail"].lower()
|
|
|
|
|
|
def test_health_reports_ollama_unreachable_when_configured_but_not_available(monkeypatch):
|
|
module = load_app_module(monkeypatch, ollama_model="qwen2.5:7b")
|
|
|
|
def boom(path: str):
|
|
raise OSError("connection refused")
|
|
|
|
monkeypatch.setattr(module, "_ollama_json", boom)
|
|
client = TestClient(module.app)
|
|
|
|
response = client.get("/health")
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["ollama_configured"] is True
|
|
assert payload["ollama_reachable"] is False
|
|
assert payload["ollama_model"] == "qwen2.5:7b"
|
|
assert payload["ollama_model_available"] is False
|
|
|
|
|
|
def test_rewrite_cv_returns_plain_rewritten_text(monkeypatch):
|
|
module = load_app_module(monkeypatch, ollama_model="qwen2.5:7b")
|
|
monkeypatch.setattr(module, "_ollama_generate_text", lambda prompt: "# Professional Summary\nBuilt resilient backend systems.\n\n# Skills\n- C#\n- .NET")
|
|
client = TestClient(module.app)
|
|
|
|
response = client.post("/cv/rewrite", json={
|
|
"instruction": "Rewrite this CV into a cleaner master CV.",
|
|
"text": "Professional Summary\nBuilt backend systems.",
|
|
"max_length": 220,
|
|
"min_length": 80,
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["rewritten_text"].startswith("# Professional Summary")
|
|
assert "Role summary:" not in payload["rewritten_text"]
|
|
|
|
|
|
def test_classify_block_returns_structured_json(monkeypatch):
|
|
module = load_app_module(monkeypatch)
|
|
|
|
def fake_generate_json(prompt: str):
|
|
assert "Senior Platform Engineer" in prompt
|
|
return {
|
|
"section": "Work Experience",
|
|
"confidence": 0.91,
|
|
"reason": "job block",
|
|
"title": "Senior Platform Engineer",
|
|
"company": "Atlas Systems",
|
|
"location": "Oslo",
|
|
"start": "2019",
|
|
"end": "Present",
|
|
"bullets": ["Built event-driven APIs and migration tooling."],
|
|
"summary": [],
|
|
"skills": ["Python", "SQL"],
|
|
}
|
|
|
|
monkeypatch.setattr(module, "_ollama_generate_json", fake_generate_json)
|
|
client = TestClient(module.app)
|
|
|
|
response = client.post("/cv/classify-block", json={"block": "Senior Platform Engineer at Atlas Systems, Oslo, 2019 - Present. Built event-driven APIs and migration tooling."})
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["section"] == "Work Experience"
|
|
assert payload["title"] == "Senior Platform Engineer"
|
|
assert payload["company"] == "Atlas Systems"
|
|
assert payload["bullets"] == ["Built event-driven APIs and migration tooling."]
|
|
assert payload["summary"] == []
|
|
assert payload["skills"] == ["Python", "SQL"]
|
|
|
|
|
|
def test_classify_block_supports_projects_section(monkeypatch):
|
|
# Phase 2.1-b: Projects and Certifications are now valid classified sections so project blocks
|
|
# (e.g. the benchmark CV's JobTrack/InboxIntel) are no longer dropped into "Other".
|
|
module = load_app_module(monkeypatch)
|
|
|
|
def fake_generate_json(prompt: str):
|
|
assert "Projects" in prompt # the enum now advertises Projects to the model
|
|
return {
|
|
"section": "Projects",
|
|
"confidence": 0.83,
|
|
"reason": "project block",
|
|
"title": "JobTrack",
|
|
"company": None,
|
|
"location": None,
|
|
"start": None,
|
|
"end": None,
|
|
"bullets": ["Full-stack job-application tracker (React, ASP.NET Core, SQLite, Docker)."],
|
|
"summary": [],
|
|
"skills": [],
|
|
}
|
|
|
|
monkeypatch.setattr(module, "_ollama_generate_json", fake_generate_json)
|
|
client = TestClient(module.app)
|
|
|
|
response = client.post("/cv/classify-block", json={"block": "JobTrack - Full-stack job-application tracker (React, ASP.NET Core, SQLite, Docker)."})
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["section"] == "Projects"
|
|
assert payload["title"] == "JobTrack"
|
|
|
|
|
|
def test_classify_block_defaults_missing_section_to_other(monkeypatch):
|
|
module = load_app_module(monkeypatch)
|
|
monkeypatch.setattr(module, "_ollama_generate_json", lambda prompt: {"bullets": []})
|
|
client = TestClient(module.app)
|
|
|
|
response = client.post("/cv/classify-block", json={"block": "Miscellaneous profile text"})
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["section"] == "Other"
|
|
assert payload["bullets"] == []
|
|
assert payload["summary"] == []
|
|
assert payload["skills"] == []
|
|
|
|
|
|
# --- AI provider router -------------------------------------------------------
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, payload):
|
|
self._data = json.dumps(payload).encode("utf-8")
|
|
|
|
def read(self):
|
|
return self._data
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
|
|
def _install_fake_urlopen(monkeypatch, module, response_payload, captured):
|
|
def fake_urlopen(req, timeout=None):
|
|
captured["url"] = req.full_url
|
|
captured["headers"] = {k.lower(): v for k, v in req.header_items()}
|
|
captured["body"] = json.loads(req.data.decode("utf-8"))
|
|
return _FakeResponse(response_payload)
|
|
|
|
monkeypatch.setattr(module.urllib_request, "urlopen", fake_urlopen)
|
|
|
|
|
|
def test_provider_defaults_to_ollama_and_is_unchanged(monkeypatch):
|
|
monkeypatch.delenv("AI_PROVIDER", raising=False)
|
|
monkeypatch.setenv("OLLAMA_BASE_URL", "http://ollama-host:11434")
|
|
module = load_app_module(monkeypatch, ollama_model="qwen2.5:7b")
|
|
assert module.AI_PROVIDER == "ollama"
|
|
|
|
captured = {}
|
|
_install_fake_urlopen(monkeypatch, module, {"response": '{"score": 7}'}, captured)
|
|
|
|
assert module._ollama_generate_json("hi") == {"score": 7}
|
|
assert captured["url"] == "http://ollama-host:11434/api/generate"
|
|
assert captured["body"]["model"] == "qwen2.5:7b"
|
|
assert captured["body"]["format"] == "json"
|
|
assert captured["body"]["options"]["temperature"] == 0.1
|
|
|
|
|
|
def test_local_success_wins_even_when_external_fallback_is_permitted(monkeypatch):
|
|
monkeypatch.setenv("AI_PROVIDER", "gemini")
|
|
monkeypatch.setenv("GEMINI_API_KEY", "test-key")
|
|
monkeypatch.setenv("GEMINI_MODEL", "gemini-2.0-flash")
|
|
module = load_app_module(monkeypatch, ollama_model="qwen2.5:7b", external_ai_enabled=True)
|
|
module._external_ai_allowed.set(True)
|
|
module._ai_task_type.set("cv-normalize")
|
|
|
|
calls = []
|
|
monkeypatch.setattr(module, "_ollama_generate", lambda *args, **kwargs: calls.append("ollama") or '{"score": 7}')
|
|
monkeypatch.setattr(module, "_gemini_generate", lambda *args, **kwargs: calls.append("gemini") or '{"score": 9}')
|
|
|
|
assert module._ollama_generate_json("hi") == {"score": 7}
|
|
assert calls == ["ollama"]
|
|
|
|
|
|
def test_local_failure_uses_permitted_groq_fallback_sequentially(monkeypatch):
|
|
monkeypatch.setenv("AI_PROVIDER", "groq")
|
|
monkeypatch.setenv("GROQ_API_KEY", "test-key")
|
|
module = load_app_module(monkeypatch, ollama_model="qwen2.5:7b", external_ai_enabled=True)
|
|
module._external_ai_allowed.set(True)
|
|
module._ai_task_type.set("cv-rewrite")
|
|
|
|
calls = []
|
|
|
|
def local_failure(*args, **kwargs):
|
|
calls.append("ollama")
|
|
raise module.URLError("synthetic local outage")
|
|
|
|
monkeypatch.setattr(module, "_ollama_generate", local_failure)
|
|
monkeypatch.setattr(module, "_groq_generate", lambda *args, **kwargs: calls.append("groq") or "rewritten CV text")
|
|
|
|
assert module._ollama_generate_text("rewrite this") == "rewritten CV text"
|
|
assert calls == ["ollama", "groq"]
|
|
assert module._route_state.get()["provider"] == "groq"
|
|
assert module._route_state.get()["fallback_reason"] == "local_provider_unavailable"
|
|
|
|
|
|
def test_missing_external_key_never_bypasses_local_failure(monkeypatch):
|
|
monkeypatch.setenv("AI_PROVIDER", "gemini")
|
|
monkeypatch.delenv("GEMINI_API_KEY", raising=False)
|
|
module = load_app_module(monkeypatch, ollama_model="qwen2.5:7b", external_ai_enabled=True)
|
|
module._external_ai_allowed.set(True)
|
|
module._ai_task_type.set("cv-normalize")
|
|
monkeypatch.setattr(module, "_ollama_generate", lambda *args, **kwargs: (_ for _ in ()).throw(module.URLError("synthetic outage")))
|
|
monkeypatch.setattr(module, "_gemini_generate", lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("must not call Gemini")))
|
|
|
|
from fastapi import HTTPException
|
|
|
|
try:
|
|
module._ollama_generate_json("hi")
|
|
except module.HTTPException as ex:
|
|
assert ex.status_code == 503
|
|
assert "Ollama" in ex.detail
|
|
else:
|
|
raise AssertionError("expected the local failure")
|
|
|
|
|
|
def test_health_reports_active_provider(monkeypatch):
|
|
monkeypatch.setenv("AI_PROVIDER", "gemini")
|
|
monkeypatch.setenv("GEMINI_API_KEY", "test-key")
|
|
module = load_app_module(monkeypatch)
|
|
client = TestClient(module.app)
|
|
|
|
payload = client.get("/health").json()
|
|
|
|
assert payload["ai_provider"] == "gemini"
|
|
assert payload["ai_provider_configured"] is True
|
|
assert payload["ai_routing_mode"] == "local_first"
|
|
assert payload["local_circuit_open"] is False
|
|
|
|
|
|
def test_external_fallback_requires_admin_gate_and_backend_consent_header(monkeypatch):
|
|
monkeypatch.setenv("AI_PROVIDER", "gemini")
|
|
monkeypatch.setenv("GEMINI_API_KEY", "test-key")
|
|
module = load_app_module(monkeypatch, ollama_model="qwen2.5:7b", external_ai_enabled=True)
|
|
calls = []
|
|
|
|
def fake_generate(provider, *args, **kwargs):
|
|
calls.append(provider)
|
|
if provider == "ollama":
|
|
raise module._ProviderFailure("ollama", "provider_unavailable", 503)
|
|
return "rewritten externally"
|
|
|
|
monkeypatch.setattr(module, "_generate_from_provider", fake_generate)
|
|
client = TestClient(module.app)
|
|
|
|
local_response = client.post("/cv/rewrite", json={"instruction": "Rewrite", "text": "Synthetic CV"})
|
|
assert local_response.status_code == 503
|
|
assert calls == ["ollama"]
|
|
assert local_response.headers["X-Ai-Provider"] == "ollama"
|
|
|
|
calls.clear()
|
|
external_response = client.post(
|
|
"/cv/rewrite",
|
|
json={"instruction": "Rewrite", "text": "Synthetic CV"},
|
|
headers={"X-Ai-External-Allowed": "true"},
|
|
)
|
|
assert external_response.status_code == 200
|
|
assert calls == ["ollama", "gemini"]
|
|
assert external_response.headers["X-Ai-Provider"] == "gemini"
|
|
assert external_response.headers["X-Ai-Model"] == "gemini-2.0-flash"
|
|
assert external_response.headers["X-Ai-Fallback-Reason"] == "local_provider_unavailable"
|
|
assert external_response.headers["X-Ai-Route-Reason"] == "external_fallback"
|
|
|
|
|
|
def test_invalid_local_json_can_fallback_but_prompt_cost_cap_cannot(monkeypatch):
|
|
monkeypatch.setenv("AI_PROVIDER", "gemini")
|
|
monkeypatch.setenv("GEMINI_API_KEY", "test-key")
|
|
module = load_app_module(
|
|
monkeypatch,
|
|
ollama_model="qwen2.5:7b",
|
|
external_ai_enabled=True,
|
|
external_prompt_limit=1000,
|
|
)
|
|
module._external_ai_allowed.set(True)
|
|
module._ai_task_type.set("cv-normalize")
|
|
calls = []
|
|
monkeypatch.setattr(module, "_ollama_generate", lambda *args, **kwargs: calls.append("ollama") or "not json")
|
|
monkeypatch.setattr(module, "_gemini_generate", lambda *args, **kwargs: calls.append("gemini") or '{"score": 9}')
|
|
|
|
assert module._ollama_generate_json("short") == {"score": 9}
|
|
assert calls == ["ollama", "gemini"]
|
|
assert module._route_state.get()["fallback_reason"] == "local_schema_invalid"
|
|
|
|
calls.clear()
|
|
try:
|
|
module._ollama_generate_json("x" * 1001)
|
|
except module.HTTPException as ex:
|
|
assert ex.status_code == 502
|
|
else:
|
|
raise AssertionError("expected local schema failure above the external prompt cap")
|
|
assert calls == ["ollama"]
|
|
assert "external_prompt_limit" in module._route_state.get()["route_reason"]
|
|
|
|
|
|
def test_open_local_circuit_skips_local_only_when_fallback_is_permitted(monkeypatch):
|
|
monkeypatch.setenv("AI_PROVIDER", "groq")
|
|
monkeypatch.setenv("GROQ_API_KEY", "test-key")
|
|
module = load_app_module(
|
|
monkeypatch,
|
|
ollama_model="qwen2.5:7b",
|
|
external_ai_enabled=True,
|
|
circuit_threshold=1,
|
|
)
|
|
module._external_ai_allowed.set(True)
|
|
module._ai_task_type.set("cv-rewrite")
|
|
calls = []
|
|
|
|
def local_failure(*args, **kwargs):
|
|
calls.append("ollama")
|
|
raise module.URLError("synthetic local outage")
|
|
|
|
monkeypatch.setattr(module, "_ollama_generate", local_failure)
|
|
monkeypatch.setattr(module, "_groq_generate", lambda *args, **kwargs: calls.append("groq") or "external")
|
|
|
|
assert module._ollama_generate_text("first") == "external"
|
|
assert module._ollama_generate_text("second") == "external"
|
|
assert calls == ["ollama", "groq", "groq"]
|
|
assert module._route_state.get()["fallback_reason"] == "local_circuit_open"
|
|
|
|
|
|
def test_external_failure_is_clear_and_unapproved_background_task_stays_local(monkeypatch):
|
|
monkeypatch.setenv("AI_PROVIDER", "gemini")
|
|
monkeypatch.setenv("GEMINI_API_KEY", "test-key")
|
|
module = load_app_module(monkeypatch, ollama_model="qwen2.5:7b", external_ai_enabled=True)
|
|
module._external_ai_allowed.set(True)
|
|
calls = []
|
|
|
|
def provider_failure(provider, *args, **kwargs):
|
|
calls.append(provider)
|
|
raise module._ProviderFailure(provider, "provider_unavailable", 503)
|
|
|
|
monkeypatch.setattr(module, "_generate_from_provider", provider_failure)
|
|
module._ai_task_type.set("cv-rewrite")
|
|
try:
|
|
module._ollama_generate_text("allowed")
|
|
except module.HTTPException as ex:
|
|
assert ex.status_code == 503
|
|
assert "Gemini" in ex.detail
|
|
else:
|
|
raise AssertionError("expected external provider failure")
|
|
assert calls == ["ollama", "gemini"]
|
|
assert module._route_state.get()["fallback_reason"] == "local_provider_unavailable"
|
|
|
|
calls.clear()
|
|
module._ai_task_type.set("strategy.snapshot")
|
|
try:
|
|
module._ollama_generate_text("not task-approved")
|
|
except module.HTTPException as ex:
|
|
assert ex.status_code == 503
|
|
assert "Ollama" in ex.detail
|
|
else:
|
|
raise AssertionError("expected local provider failure")
|
|
assert calls == ["ollama"]
|
|
assert "task_not_allowed_external" in module._route_state.get()["route_reason"]
|
|
|
|
|
|
def test_external_only_mode_still_requires_explicit_permission_and_task_allowlist(monkeypatch):
|
|
monkeypatch.setenv("AI_PROVIDER", "groq")
|
|
monkeypatch.setenv("GROQ_API_KEY", "test-key")
|
|
module = load_app_module(
|
|
monkeypatch,
|
|
ollama_model="qwen2.5:7b",
|
|
external_ai_enabled=True,
|
|
routing_mode="external_only",
|
|
)
|
|
module._ai_task_type.set("cv-rewrite")
|
|
calls = []
|
|
monkeypatch.setattr(module, "_ollama_generate", lambda *args, **kwargs: calls.append("ollama") or "local")
|
|
monkeypatch.setattr(module, "_groq_generate", lambda *args, **kwargs: calls.append("groq") or "external")
|
|
|
|
try:
|
|
module._ollama_generate_text("without consent")
|
|
except module.HTTPException as ex:
|
|
assert ex.status_code == 403
|
|
else:
|
|
raise AssertionError("expected external permission denial")
|
|
assert calls == []
|
|
|
|
module._external_ai_allowed.set(True)
|
|
assert module._ollama_generate_text("with consent") == "external"
|
|
assert calls == ["groq"]
|
|
|
|
|
|
def test_service_token_rejects_calls_without_the_header(monkeypatch):
|
|
module = load_app_module(monkeypatch, service_token="s3cret")
|
|
client = TestClient(module.app)
|
|
|
|
response = client.post("/summarize", json={"text": "Platform engineering role."})
|
|
|
|
assert response.status_code == 401
|
|
assert "token" in response.json()["detail"].lower()
|
|
|
|
|
|
def test_service_token_rejects_a_wrong_header(monkeypatch):
|
|
module = load_app_module(monkeypatch, service_token="s3cret")
|
|
client = TestClient(module.app)
|
|
|
|
response = client.post(
|
|
"/summarize",
|
|
json={"text": "Platform engineering role."},
|
|
headers={"X-Ai-Service-Token": "wrong"},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_service_token_allows_the_correct_header(monkeypatch):
|
|
module = load_app_module(monkeypatch, service_token="s3cret")
|
|
client = TestClient(module.app)
|
|
|
|
response = client.post(
|
|
"/summarize",
|
|
json={"text": "Platform engineering role."},
|
|
headers={"X-Ai-Service-Token": "s3cret"},
|
|
)
|
|
|
|
# 503 = passed the token gate and reached the handler, which is model-disabled here.
|
|
assert response.status_code == 503
|
|
|
|
|
|
def test_health_stays_open_so_probes_and_healthchecks_work(monkeypatch):
|
|
module = load_app_module(monkeypatch, service_token="s3cret")
|
|
client = TestClient(module.app)
|
|
|
|
assert client.get("/health").status_code == 200
|
|
|
|
|
|
def test_endpoints_stay_open_when_no_token_is_configured(monkeypatch):
|
|
module = load_app_module(monkeypatch)
|
|
client = TestClient(module.app)
|
|
|
|
# Keyless local dev: reaches the handler (503 model-disabled), not a 401.
|
|
response = client.post("/summarize", json={"text": "Platform engineering role."})
|
|
|
|
assert response.status_code == 503
|