cesnimda
801 lines
30 KiB
Python
801 lines
30 KiB
Python
import importlib
|
|
import io
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import textwrap
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
from docx import Document
|
|
from fastapi.testclient import TestClient
|
|
from PIL import Image
|
|
from pypdf import PdfWriter
|
|
|
|
|
|
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 payload["parser_isolated"] is True
|
|
assert payload["parser_concurrency"] == 1
|
|
assert payload["parser_timeout_seconds"] == 25
|
|
assert payload["parser_memory_mib"] == 512
|
|
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_extract_text_rejects_oversized_upload_before_parsing(monkeypatch):
|
|
module = load_app_module(monkeypatch)
|
|
monkeypatch.setattr(module, "MAX_EXTRACT_FILE_BYTES", 32)
|
|
monkeypatch.setattr(module, "_run_parser_child", lambda extension, data: (_ for _ in ()).throw(AssertionError("parser must not run")))
|
|
client = TestClient(module.app)
|
|
|
|
response = client.post("/extract-text", files={"file": ("large.txt", b"x" * 64, "text/plain")})
|
|
|
|
assert response.status_code == 400
|
|
assert "too large" in response.json()["detail"].lower()
|
|
|
|
|
|
def test_extract_text_rejects_extension_signature_mismatch(monkeypatch):
|
|
module = load_app_module(monkeypatch)
|
|
client = TestClient(module.app)
|
|
|
|
response = client.post("/extract-text", files={"file": ("resume.pdf", b"not a pdf", "application/pdf")})
|
|
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"] == "The uploaded file does not match its PDF extension."
|
|
|
|
|
|
def test_extract_text_rejects_binary_plain_text(monkeypatch):
|
|
module = load_app_module(monkeypatch)
|
|
client = TestClient(module.app)
|
|
|
|
response = client.post("/extract-text", files={"file": ("resume.txt", b"Name\x00binary", "text/plain")})
|
|
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"] == "The uploaded text file contains binary data."
|
|
|
|
|
|
def test_extract_text_rejects_pdf_over_page_limit(monkeypatch):
|
|
load_app_module(monkeypatch)
|
|
import parser_child
|
|
|
|
writer = PdfWriter()
|
|
writer.add_blank_page(width=100, height=100)
|
|
writer.add_blank_page(width=100, height=100)
|
|
payload = io.BytesIO()
|
|
writer.write(payload)
|
|
try:
|
|
parser_child.extract_pdf_text(payload.getvalue(), page_limit=1)
|
|
except parser_child.ParserError as exc:
|
|
assert exc.code == "pdf_page_limit"
|
|
else:
|
|
raise AssertionError("Expected the PDF page limit to reject the document")
|
|
|
|
|
|
def test_extract_text_rejects_image_pixel_limit_before_ocr(monkeypatch):
|
|
load_app_module(monkeypatch)
|
|
import parser_child
|
|
|
|
monkeypatch.setattr(parser_child, "_ocr_image", lambda image: (_ for _ in ()).throw(AssertionError("OCR must not run")))
|
|
payload = io.BytesIO()
|
|
Image.new("RGB", (10, 10), "white").save(payload, format="PNG")
|
|
|
|
try:
|
|
parser_child.extract_image_text(payload.getvalue(), pixel_limit=50)
|
|
except parser_child.ParserError as exc:
|
|
assert exc.code == "image_pixel_limit"
|
|
else:
|
|
raise AssertionError("Expected the image pixel limit to reject the document")
|
|
|
|
|
|
def test_docx_container_rejects_excessive_entries(monkeypatch):
|
|
module = load_app_module(monkeypatch)
|
|
monkeypatch.setattr(module, "MAX_DOCX_ENTRIES", 2)
|
|
payload = io.BytesIO()
|
|
with zipfile.ZipFile(payload, "w") as archive:
|
|
archive.writestr("[Content_Types].xml", "types")
|
|
archive.writestr("word/document.xml", "document")
|
|
archive.writestr("word/styles.xml", "styles")
|
|
|
|
try:
|
|
module._validate_docx_container(payload.getvalue())
|
|
except module.HTTPException as exc:
|
|
assert exc.status_code == 422
|
|
assert exc.detail == "The DOCX contains too many files."
|
|
else:
|
|
raise AssertionError("Expected the DOCX entry limit to reject the container")
|
|
|
|
|
|
def test_extract_text_does_not_leak_parser_exception(monkeypatch):
|
|
module = load_app_module(monkeypatch)
|
|
monkeypatch.setattr(module, "_run_parser_child", lambda extension, data: module._raise_parser_error("parser_failed"))
|
|
client = TestClient(module.app)
|
|
|
|
response = client.post("/extract-text", files={"file": ("resume.pdf", b"%PDF-invalid", "application/pdf")})
|
|
|
|
assert response.status_code == 422
|
|
assert response.json()["detail"] == "Document extraction failed."
|
|
assert "secret" not in response.text
|
|
|
|
|
|
def test_extraction_normalization_preserves_sections_and_bullets(monkeypatch):
|
|
load_app_module(monkeypatch)
|
|
import parser_child
|
|
|
|
normalized = parser_child.normalize_text("PROFILE\n\nEngineer\n• Built APIs\n• Shipped services")
|
|
|
|
assert normalized.splitlines() == ["PROFILE", "", "Engineer", "• Built APIs", "• Shipped services"]
|
|
|
|
|
|
def test_docx_extraction_preserves_paragraph_and_table_boundaries(monkeypatch):
|
|
load_app_module(monkeypatch)
|
|
import parser_child
|
|
|
|
document = Document()
|
|
document.add_heading("Technical Skills", level=1)
|
|
table = document.add_table(rows=2, cols=2)
|
|
table.cell(0, 0).text = "Backend"
|
|
table.cell(0, 1).text = "C#, .NET"
|
|
table.cell(1, 0).text = "DevOps"
|
|
table.cell(1, 1).text = "Docker, Linux"
|
|
payload = io.BytesIO()
|
|
document.save(payload)
|
|
|
|
extracted = parser_child.extract_docx_text(payload.getvalue())
|
|
|
|
assert "Technical Skills" in extracted
|
|
assert "Backend | C#, .NET" in extracted
|
|
assert "DevOps | Docker, Linux" in extracted
|
|
|
|
|
|
def test_extract_text_runs_in_isolated_child(monkeypatch):
|
|
module = load_app_module(monkeypatch)
|
|
client = TestClient(module.app)
|
|
|
|
response = client.post(
|
|
"/extract-text",
|
|
files={"file": ("resume.md", b"# Ada Lovelace\n\n## Skills\nC#", "text/markdown")},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["text"] == "# Ada Lovelace\n\n## Skills\nC#"
|
|
|
|
|
|
def test_parser_child_receives_no_provider_secrets(monkeypatch, tmp_path):
|
|
module = load_app_module(monkeypatch)
|
|
monkeypatch.setenv("GEMINI_API_KEY", "must-not-leak")
|
|
monkeypatch.setenv("GROQ_API_KEY", "must-not-leak")
|
|
child = tmp_path / "environment_probe.py"
|
|
child.write_text(textwrap.dedent("""
|
|
import argparse, json, os
|
|
from pathlib import Path
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--input')
|
|
parser.add_argument('--output')
|
|
parser.add_argument('--extension')
|
|
parser.add_argument('--cpu-seconds')
|
|
parser.add_argument('--memory-bytes')
|
|
args = parser.parse_args()
|
|
leaked = any(os.environ.get(key) for key in ('GEMINI_API_KEY', 'GROQ_API_KEY', 'AI_SERVICE_TOKEN'))
|
|
Path(args.output).write_text(json.dumps({'ok': not leaked, 'text': 'safe', 'ocr_used': False, 'page_count': None} if not leaked else {'ok': False, 'code': 'parser_failed'}), encoding='utf-8')
|
|
"""), encoding="utf-8")
|
|
monkeypatch.setattr(module, "PARSER_CHILD_PATH", child)
|
|
|
|
assert module._run_parser_child(".txt", b"safe")["text"] == "safe"
|
|
|
|
|
|
def test_parser_child_timeout_is_stable_and_releases_capacity(monkeypatch, tmp_path):
|
|
module = load_app_module(monkeypatch)
|
|
child = tmp_path / "sleeping_parser.py"
|
|
child.write_text("import time\ntime.sleep(30)\n", encoding="utf-8")
|
|
monkeypatch.setattr(module, "PARSER_CHILD_PATH", child)
|
|
monkeypatch.setattr(module, "PARSER_TIMEOUT_SECONDS", 0.1)
|
|
|
|
try:
|
|
module._run_parser_child(".txt", b"safe")
|
|
except module.HTTPException as exc:
|
|
assert exc.status_code == 504
|
|
assert exc.detail == "Document extraction timed out."
|
|
else:
|
|
raise AssertionError("Expected the parser deadline to terminate the child")
|
|
|
|
assert module._parser_capacity.acquire(blocking=False)
|
|
module._parser_capacity.release()
|
|
|
|
|
|
def test_parser_timeout_terminates_descendants(monkeypatch, tmp_path):
|
|
module = load_app_module(monkeypatch)
|
|
pid_path = tmp_path / "descendant.pid"
|
|
child = tmp_path / "parser_with_descendant.py"
|
|
child.write_text(textwrap.dedent(f"""
|
|
import subprocess, sys, time
|
|
from pathlib import Path
|
|
descendant = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(30)'])
|
|
Path({str(pid_path)!r}).write_text(str(descendant.pid), encoding='ascii')
|
|
time.sleep(30)
|
|
"""), encoding="utf-8")
|
|
monkeypatch.setattr(module, "PARSER_CHILD_PATH", child)
|
|
monkeypatch.setattr(module, "PARSER_TIMEOUT_SECONDS", 0.2)
|
|
|
|
try:
|
|
module._run_parser_child(".txt", b"safe")
|
|
except module.HTTPException as exc:
|
|
assert exc.status_code == 504
|
|
else:
|
|
raise AssertionError("Expected the parser deadline to terminate the process tree")
|
|
|
|
descendant_pid = int(pid_path.read_text(encoding="ascii"))
|
|
if os.name == "nt":
|
|
listing = subprocess.run(
|
|
["tasklist", "/FI", f"PID eq {descendant_pid}", "/FO", "CSV", "/NH"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
).stdout
|
|
assert f'"{descendant_pid}"' not in listing
|
|
else:
|
|
try:
|
|
os.kill(descendant_pid, 0)
|
|
except ProcessLookupError:
|
|
pass
|
|
else:
|
|
# Minimal CI containers may not run an init process that promptly
|
|
# reaps orphaned children. A zombie has already been terminated and
|
|
# cannot execute or retain parser resources, even though kill(pid, 0)
|
|
# continues to see its process-table entry.
|
|
stat_path = Path(f"/proc/{descendant_pid}/stat")
|
|
try:
|
|
state = stat_path.read_text(encoding="ascii").split()[2]
|
|
except (FileNotFoundError, IndexError, OSError):
|
|
state = None
|
|
if state == "Z":
|
|
return
|
|
os.kill(descendant_pid, 9)
|
|
raise AssertionError("Parser descendant survived the parent deadline")
|
|
|
|
|
|
def test_parser_temp_cleanup_removes_only_stale_work_directories(monkeypatch, tmp_path):
|
|
module = load_app_module(monkeypatch)
|
|
parser_root = tmp_path / "parser-root"
|
|
stale = parser_root / "work-stale"
|
|
recent = parser_root / "work-recent"
|
|
unrelated = parser_root / "keep-me"
|
|
stale.mkdir(parents=True)
|
|
recent.mkdir()
|
|
unrelated.mkdir()
|
|
old = 1_000_000
|
|
os.utime(stale, (old, old))
|
|
monkeypatch.setattr(module, "PARSER_TEMP_ROOT", parser_root.resolve())
|
|
monkeypatch.setattr(module, "PARSER_STALE_WORK_SECONDS", 300)
|
|
monkeypatch.setattr(module, "_parser_temp_ready", False)
|
|
|
|
assert module._prepare_parser_temp_root() == parser_root.resolve()
|
|
assert not stale.exists()
|
|
assert recent.exists()
|
|
assert unrelated.exists()
|
|
|
|
|
|
def test_cache_purge_requires_service_token_and_clears_content(monkeypatch):
|
|
module = load_app_module(monkeypatch, service_token="s3cret")
|
|
module.cache["synthetic-key"] = "synthetic-summary"
|
|
client = TestClient(module.app)
|
|
|
|
assert client.delete("/maintenance/cache").status_code == 401
|
|
response = client.delete(
|
|
"/maintenance/cache",
|
|
headers={"X-Ai-Service-Token": "s3cret"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"cleared": 1}
|
|
assert len(module.cache) == 0
|
|
|
|
|
|
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
|