pyr0ball
dc508d7197
Some checks failed
CI / test (push) Failing after 28s
- task_scheduler: extend LocalScheduler (concrete class), not TaskScheduler (Protocol); remove unsupported VRAM kwargs from super().__init__() - dev-api: lazy import db_migrate inside _startup() to avoid worktree scripts cache issue in test_dev_api_settings.py - test_task_scheduler: update VRAM-attribute tests to match LocalScheduler (no _available_vram/_reserved_vram); drop deepest-queue VRAM-gating ordering assertion (LocalScheduler is FIFO, not priority-gated); suppress PytestUnhandledThreadExceptionWarning on crash test; fix budget assertion to not depend on shared pytest tmp dir state - test_dev_api_settings: patch path functions (_resume_path, _search_prefs_path, _license_path, _tokens_path, _config_dir) instead of removed module-level constants; mock _TRAINING_JSONL for finetune status idle test - test_wizard_tiers: Vue SPA is free tier (issue #20), assert True - test_wizard_api: patch _search_prefs_path() function, not SEARCH_PREFS_PATH - test_ui_switcher: free-tier vue preference no longer downgrades to streamlit
368 lines
18 KiB
Python
368 lines
18 KiB
Python
"""Tests for wizard API endpoints (GET/POST /api/wizard/*)."""
|
|
import os
|
|
import sys
|
|
import yaml
|
|
import pytest
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
from fastapi.testclient import TestClient
|
|
|
|
# ── Path bootstrap ────────────────────────────────────────────────────────────
|
|
_REPO = Path(__file__).parent.parent
|
|
if str(_REPO) not in sys.path:
|
|
sys.path.insert(0, str(_REPO))
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def client():
|
|
from dev_api import app
|
|
return TestClient(app)
|
|
|
|
|
|
# ── Helpers ───────────────────────────────────────────────────────────────────
|
|
|
|
def _write_user_yaml(path: Path, data: dict | None = None) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
payload = data if data is not None else {}
|
|
path.write_text(yaml.dump(payload, allow_unicode=True, default_flow_style=False))
|
|
|
|
|
|
def _read_user_yaml(path: Path) -> dict:
|
|
if not path.exists():
|
|
return {}
|
|
return yaml.safe_load(path.read_text()) or {}
|
|
|
|
|
|
# ── GET /api/config/app — wizardComplete + isDemo ─────────────────────────────
|
|
|
|
class TestAppConfigWizardFields:
|
|
def test_wizard_complete_false_when_missing(self, client, tmp_path):
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
# user.yaml does not exist yet
|
|
with patch("dev_api._user_yaml_path", return_value=str(yaml_path)):
|
|
r = client.get("/api/config/app")
|
|
assert r.status_code == 200
|
|
assert r.json()["wizardComplete"] is False
|
|
|
|
def test_wizard_complete_true_when_set(self, client, tmp_path):
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
_write_user_yaml(yaml_path, {"wizard_complete": True})
|
|
with patch("dev_api._user_yaml_path", return_value=str(yaml_path)):
|
|
r = client.get("/api/config/app")
|
|
assert r.json()["wizardComplete"] is True
|
|
|
|
def test_is_demo_false_by_default(self, client, tmp_path):
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
_write_user_yaml(yaml_path, {"wizard_complete": True})
|
|
with patch("dev_api._user_yaml_path", return_value=str(yaml_path)):
|
|
with patch.dict(os.environ, {"DEMO_MODE": ""}, clear=False):
|
|
r = client.get("/api/config/app")
|
|
assert r.json()["isDemo"] is False
|
|
|
|
def test_is_demo_true_when_env_set(self, client, tmp_path):
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
_write_user_yaml(yaml_path, {"wizard_complete": True})
|
|
with patch("dev_api._user_yaml_path", return_value=str(yaml_path)):
|
|
with patch.dict(os.environ, {"DEMO_MODE": "true"}, clear=False):
|
|
r = client.get("/api/config/app")
|
|
assert r.json()["isDemo"] is True
|
|
|
|
|
|
# ── GET /api/wizard/status ────────────────────────────────────────────────────
|
|
|
|
class TestWizardStatus:
|
|
def test_returns_not_complete_when_no_yaml(self, client, tmp_path):
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
with patch("dev_api._wizard_yaml_path", return_value=str(yaml_path)):
|
|
r = client.get("/api/wizard/status")
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["wizard_complete"] is False
|
|
assert body["wizard_step"] == 0
|
|
|
|
def test_returns_saved_step(self, client, tmp_path):
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
_write_user_yaml(yaml_path, {"wizard_step": 3, "name": "Alex"})
|
|
with patch("dev_api._wizard_yaml_path", return_value=str(yaml_path)):
|
|
r = client.get("/api/wizard/status")
|
|
body = r.json()
|
|
assert body["wizard_step"] == 3
|
|
assert body["saved_data"]["name"] == "Alex"
|
|
|
|
def test_returns_complete_true(self, client, tmp_path):
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
_write_user_yaml(yaml_path, {"wizard_complete": True})
|
|
with patch("dev_api._wizard_yaml_path", return_value=str(yaml_path)):
|
|
r = client.get("/api/wizard/status")
|
|
assert r.json()["wizard_complete"] is True
|
|
|
|
|
|
# ── GET /api/wizard/hardware ──────────────────────────────────────────────────
|
|
|
|
class TestWizardHardware:
|
|
def test_returns_profiles_list(self, client):
|
|
r = client.get("/api/wizard/hardware")
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert set(body["profiles"]) == {"remote", "cpu", "single-gpu", "dual-gpu"}
|
|
assert "gpus" in body
|
|
assert "suggested_profile" in body
|
|
|
|
def test_gpu_from_env_var(self, client):
|
|
with patch.dict(os.environ, {"PEREGRINE_GPU_NAMES": "RTX 4090,RTX 3080"}, clear=False):
|
|
r = client.get("/api/wizard/hardware")
|
|
body = r.json()
|
|
assert body["gpus"] == ["RTX 4090", "RTX 3080"]
|
|
assert body["suggested_profile"] == "dual-gpu"
|
|
|
|
def test_single_gpu_suggests_single(self, client):
|
|
with patch.dict(os.environ, {"PEREGRINE_GPU_NAMES": "RTX 4090"}, clear=False):
|
|
with patch.dict(os.environ, {"RECOMMENDED_PROFILE": ""}, clear=False):
|
|
r = client.get("/api/wizard/hardware")
|
|
assert r.json()["suggested_profile"] == "single-gpu"
|
|
|
|
def test_no_gpus_suggests_remote(self, client):
|
|
with patch.dict(os.environ, {"PEREGRINE_GPU_NAMES": ""}, clear=False):
|
|
with patch.dict(os.environ, {"RECOMMENDED_PROFILE": ""}, clear=False):
|
|
with patch("subprocess.check_output", side_effect=FileNotFoundError):
|
|
r = client.get("/api/wizard/hardware")
|
|
assert r.json()["suggested_profile"] == "remote"
|
|
assert r.json()["gpus"] == []
|
|
|
|
def test_recommended_profile_env_takes_priority(self, client):
|
|
with patch.dict(os.environ,
|
|
{"PEREGRINE_GPU_NAMES": "RTX 4090", "RECOMMENDED_PROFILE": "cpu"},
|
|
clear=False):
|
|
r = client.get("/api/wizard/hardware")
|
|
assert r.json()["suggested_profile"] == "cpu"
|
|
|
|
|
|
# ── POST /api/wizard/step ─────────────────────────────────────────────────────
|
|
|
|
class TestWizardStep:
|
|
def test_step1_saves_inference_profile(self, client, tmp_path):
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
_write_user_yaml(yaml_path, {})
|
|
with patch("dev_api._wizard_yaml_path", return_value=str(yaml_path)):
|
|
r = client.post("/api/wizard/step",
|
|
json={"step": 1, "data": {"inference_profile": "single-gpu"}})
|
|
assert r.status_code == 200
|
|
saved = _read_user_yaml(yaml_path)
|
|
assert saved["inference_profile"] == "single-gpu"
|
|
assert saved["wizard_step"] == 1
|
|
|
|
def test_step1_rejects_unknown_profile(self, client, tmp_path):
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
_write_user_yaml(yaml_path, {})
|
|
with patch("dev_api._wizard_yaml_path", return_value=str(yaml_path)):
|
|
r = client.post("/api/wizard/step",
|
|
json={"step": 1, "data": {"inference_profile": "turbo-gpu"}})
|
|
assert r.status_code == 400
|
|
|
|
def test_step2_saves_tier(self, client, tmp_path):
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
_write_user_yaml(yaml_path, {})
|
|
with patch("dev_api._wizard_yaml_path", return_value=str(yaml_path)):
|
|
r = client.post("/api/wizard/step",
|
|
json={"step": 2, "data": {"tier": "paid"}})
|
|
assert r.status_code == 200
|
|
assert _read_user_yaml(yaml_path)["tier"] == "paid"
|
|
|
|
def test_step2_rejects_unknown_tier(self, client, tmp_path):
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
_write_user_yaml(yaml_path, {})
|
|
with patch("dev_api._wizard_yaml_path", return_value=str(yaml_path)):
|
|
r = client.post("/api/wizard/step",
|
|
json={"step": 2, "data": {"tier": "enterprise"}})
|
|
assert r.status_code == 400
|
|
|
|
def test_step3_writes_resume_yaml(self, client, tmp_path):
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
_write_user_yaml(yaml_path, {})
|
|
resume = {"experience": [{"title": "Engineer", "company": "Acme"}]}
|
|
with patch("dev_api._wizard_yaml_path", return_value=str(yaml_path)):
|
|
r = client.post("/api/wizard/step",
|
|
json={"step": 3, "data": {"resume": resume}})
|
|
assert r.status_code == 200
|
|
resume_path = yaml_path.parent / "plain_text_resume.yaml"
|
|
assert resume_path.exists()
|
|
saved_resume = yaml.safe_load(resume_path.read_text())
|
|
assert saved_resume["experience"][0]["title"] == "Engineer"
|
|
|
|
def test_step4_saves_identity_fields(self, client, tmp_path):
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
_write_user_yaml(yaml_path, {})
|
|
identity = {
|
|
"name": "Alex Rivera",
|
|
"email": "alex@example.com",
|
|
"phone": "555-1234",
|
|
"linkedin": "https://linkedin.com/in/alex",
|
|
"career_summary": "Experienced engineer.",
|
|
}
|
|
with patch("dev_api._wizard_yaml_path", return_value=str(yaml_path)):
|
|
r = client.post("/api/wizard/step", json={"step": 4, "data": identity})
|
|
assert r.status_code == 200
|
|
saved = _read_user_yaml(yaml_path)
|
|
assert saved["name"] == "Alex Rivera"
|
|
assert saved["career_summary"] == "Experienced engineer."
|
|
assert saved["wizard_step"] == 4
|
|
|
|
def test_step5_writes_env_keys(self, client, tmp_path):
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
env_path = tmp_path / ".env"
|
|
env_path.write_text("SOME_KEY=existing\n")
|
|
_write_user_yaml(yaml_path, {})
|
|
# Patch both _wizard_yaml_path and the Path resolution inside wizard_save_step
|
|
with patch("dev_api._wizard_yaml_path", return_value=str(yaml_path)):
|
|
with patch("dev_api.Path") as mock_path_cls:
|
|
# Only intercept the .env path construction; let other Path() calls pass through
|
|
real_path = Path
|
|
def path_side_effect(*args):
|
|
result = real_path(*args)
|
|
return result
|
|
mock_path_cls.side_effect = path_side_effect
|
|
|
|
# Direct approach: monkeypatch the env path
|
|
import dev_api as _dev_api
|
|
original_fn = _dev_api.wizard_save_step
|
|
|
|
# Simpler: just test via the real endpoint, verify env not written if no key given
|
|
r = client.post("/api/wizard/step",
|
|
json={"step": 5, "data": {"services": {"ollama_host": "localhost"}}})
|
|
assert r.status_code == 200
|
|
|
|
def test_step6_writes_search_profiles(self, client, tmp_path):
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
search_path = tmp_path / "config" / "search_profiles.yaml"
|
|
_write_user_yaml(yaml_path, {})
|
|
with patch("dev_api._wizard_yaml_path", return_value=str(yaml_path)):
|
|
with patch("dev_api._search_prefs_path", return_value=search_path):
|
|
r = client.post("/api/wizard/step",
|
|
json={"step": 6, "data": {
|
|
"titles": ["Software Engineer", "Backend Developer"],
|
|
"locations": ["Remote", "Austin, TX"],
|
|
}})
|
|
assert r.status_code == 200
|
|
assert search_path.exists()
|
|
prefs = yaml.safe_load(search_path.read_text())
|
|
assert prefs["default"]["job_titles"] == ["Software Engineer", "Backend Developer"]
|
|
assert "Remote" in prefs["default"]["location"]
|
|
|
|
def test_step7_only_advances_counter(self, client, tmp_path):
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
_write_user_yaml(yaml_path, {})
|
|
with patch("dev_api._wizard_yaml_path", return_value=str(yaml_path)):
|
|
r = client.post("/api/wizard/step", json={"step": 7, "data": {}})
|
|
assert r.status_code == 200
|
|
assert _read_user_yaml(yaml_path)["wizard_step"] == 7
|
|
|
|
def test_invalid_step_number(self, client, tmp_path):
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
_write_user_yaml(yaml_path, {})
|
|
with patch("dev_api._wizard_yaml_path", return_value=str(yaml_path)):
|
|
r = client.post("/api/wizard/step", json={"step": 99, "data": {}})
|
|
assert r.status_code == 400
|
|
|
|
def test_crash_recovery_round_trip(self, client, tmp_path):
|
|
"""Save steps 1-4 sequentially, then verify status reflects step 4."""
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
_write_user_yaml(yaml_path, {})
|
|
steps = [
|
|
(1, {"inference_profile": "cpu"}),
|
|
(2, {"tier": "free"}),
|
|
(4, {"name": "Alex", "email": "a@b.com", "career_summary": "Eng."}),
|
|
]
|
|
with patch("dev_api._wizard_yaml_path", return_value=str(yaml_path)):
|
|
for step, data in steps:
|
|
r = client.post("/api/wizard/step", json={"step": step, "data": data})
|
|
assert r.status_code == 200
|
|
|
|
r = client.get("/api/wizard/status")
|
|
|
|
body = r.json()
|
|
assert body["wizard_step"] == 4
|
|
assert body["saved_data"]["name"] == "Alex"
|
|
assert body["saved_data"]["inference_profile"] == "cpu"
|
|
|
|
|
|
# ── POST /api/wizard/inference/test ──────────────────────────────────────────
|
|
|
|
class TestWizardInferenceTest:
|
|
def test_local_profile_ollama_running(self, client):
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 200
|
|
with patch("dev_api.requests.get", return_value=mock_resp):
|
|
r = client.post("/api/wizard/inference/test",
|
|
json={"profile": "cpu", "ollama_host": "localhost",
|
|
"ollama_port": 11434})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["ok"] is True
|
|
assert "Ollama" in body["message"]
|
|
|
|
def test_local_profile_ollama_down_soft_fail(self, client):
|
|
import requests as _req
|
|
with patch("dev_api.requests.get", side_effect=_req.exceptions.ConnectionError):
|
|
r = client.post("/api/wizard/inference/test",
|
|
json={"profile": "single-gpu"})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["ok"] is False
|
|
assert "configure" in body["message"].lower()
|
|
|
|
def test_remote_profile_llm_responding(self, client):
|
|
# LLMRouter is imported inside wizard_test_inference — patch the source module
|
|
with patch("scripts.llm_router.LLMRouter") as mock_cls:
|
|
mock_cls.return_value.complete.return_value = "OK"
|
|
r = client.post("/api/wizard/inference/test",
|
|
json={"profile": "remote", "anthropic_key": "sk-ant-test"})
|
|
assert r.status_code == 200
|
|
assert r.json()["ok"] is True
|
|
|
|
def test_remote_profile_llm_error(self, client):
|
|
with patch("scripts.llm_router.LLMRouter") as mock_cls:
|
|
mock_cls.return_value.complete.side_effect = RuntimeError("no key")
|
|
r = client.post("/api/wizard/inference/test",
|
|
json={"profile": "remote"})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["ok"] is False
|
|
assert "failed" in body["message"].lower()
|
|
|
|
|
|
# ── POST /api/wizard/complete ─────────────────────────────────────────────────
|
|
|
|
class TestWizardComplete:
|
|
def test_sets_wizard_complete_true(self, client, tmp_path):
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
_write_user_yaml(yaml_path, {"wizard_step": 6, "name": "Alex"})
|
|
# apply_service_urls is a local import inside wizard_complete — patch source module
|
|
with patch("dev_api._wizard_yaml_path", return_value=str(yaml_path)):
|
|
with patch("scripts.generate_llm_config.apply_service_urls",
|
|
side_effect=Exception("no llm.yaml")):
|
|
r = client.post("/api/wizard/complete")
|
|
assert r.status_code == 200
|
|
assert r.json()["ok"] is True
|
|
saved = _read_user_yaml(yaml_path)
|
|
assert saved["wizard_complete"] is True
|
|
assert "wizard_step" not in saved
|
|
assert saved["name"] == "Alex" # other fields preserved
|
|
|
|
def test_complete_removes_wizard_step(self, client, tmp_path):
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
_write_user_yaml(yaml_path, {"wizard_step": 7, "tier": "paid"})
|
|
with patch("dev_api._wizard_yaml_path", return_value=str(yaml_path)):
|
|
with patch("scripts.generate_llm_config.apply_service_urls", return_value=None):
|
|
client.post("/api/wizard/complete")
|
|
saved = _read_user_yaml(yaml_path)
|
|
assert "wizard_step" not in saved
|
|
assert saved["tier"] == "paid"
|
|
|
|
def test_complete_tolerates_missing_llm_yaml(self, client, tmp_path):
|
|
yaml_path = tmp_path / "config" / "user.yaml"
|
|
_write_user_yaml(yaml_path, {})
|
|
with patch("dev_api._wizard_yaml_path", return_value=str(yaml_path)):
|
|
# llm.yaml doesn't exist → apply_service_urls is never called, no error
|
|
r = client.post("/api/wizard/complete")
|
|
assert r.status_code == 200
|
|
assert r.json()["ok"] is True
|