71 lines
2.2 KiB
Python
71 lines
2.2 KiB
Python
"""Tests for /api/training/runs and registry endpoints."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from forja_core.api import deps
|
|
from forja_core.main import create_app
|
|
|
|
FIXTURES = Path(__file__).parent.parent / "fixtures"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def patch_settings(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("DATA_DIR", str(tmp_path))
|
|
monkeypatch.setenv("AGENTS_DIR", str(FIXTURES / "agents"))
|
|
monkeypatch.setenv("POLICIES_DIR", str(FIXTURES / "policies"))
|
|
monkeypatch.setenv("DATASETS_DIR", str(FIXTURES / "datasets"))
|
|
monkeypatch.setenv("MODELS_DIR", str(FIXTURES / "models"))
|
|
monkeypatch.setenv("TRAINING_BACKEND", "mock")
|
|
for fn in (
|
|
deps.get_settings,
|
|
deps.get_registry,
|
|
deps.get_policy_store,
|
|
deps.get_dataset_store,
|
|
deps.get_model_store,
|
|
deps.get_training_backend,
|
|
deps.get_orchestrator,
|
|
deps.get_llm_provider,
|
|
deps.get_guardrail_engine,
|
|
):
|
|
fn.cache_clear()
|
|
|
|
|
|
def test_list_datasets() -> None:
|
|
r = TestClient(create_app()).get("/api/datasets")
|
|
assert r.status_code == 200
|
|
assert any(d["name"] == "incident_sft" for d in r.json())
|
|
|
|
|
|
def test_create_training_run_mock_backend(tmp_path: Path) -> None:
|
|
client = TestClient(create_app())
|
|
r = client.post(
|
|
"/api/training/runs",
|
|
json={
|
|
"dataset_name": "incident_sft",
|
|
"model_name": "gpt4o_lora_base",
|
|
},
|
|
)
|
|
assert r.status_code == 201
|
|
body = r.json()
|
|
assert body["status"] == "succeeded"
|
|
assert body["backend"] == "mock"
|
|
assert body["external_job_id"].startswith("mock:")
|
|
assert len(body["artifact_refs"]) >= 1
|
|
run_id = body["id"]
|
|
assert (tmp_path / "training_runs" / f"{run_id}.json").exists()
|
|
|
|
r2 = client.get(f"/api/training/runs/{run_id}")
|
|
assert r2.status_code == 200
|
|
assert r2.json()["id"] == run_id
|
|
|
|
|
|
def test_create_training_run_unknown_dataset_404() -> None:
|
|
r = TestClient(create_app()).post(
|
|
"/api/training/runs",
|
|
json={"dataset_name": "missing", "model_name": "gpt4o_lora_base"},
|
|
)
|
|
assert r.status_code == 404
|