diff --git a/backend/test/integration/api/test_evaluation_router.py b/backend/test/integration/api/test_evaluation_router.py index 92b223ac..1f0e7016 100644 --- a/backend/test/integration/api/test_evaluation_router.py +++ b/backend/test/integration/api/test_evaluation_router.py @@ -11,38 +11,38 @@ import pytest pytestmark = [pytest.mark.asyncio, pytest.mark.integration] -async def _upload_test_benchmark(test_client, admin_headers: dict[str, str], db_id: str) -> tuple[str, str]: - benchmark_name = f"pytest_benchmark_{uuid.uuid4().hex[:8]}" +async def _upload_test_dataset(test_client, admin_headers: dict[str, str], db_id: str) -> tuple[str, str]: + dataset_name = f"pytest_dataset_{uuid.uuid4().hex[:8]}" line = '{"query":"什么是单元测试？","gold_answer":"用于验证代码行为的自动化测试"}\n' response = await test_client.post( - f"/api/evaluation/databases/{db_id}/benchmarks/upload", - data={"name": benchmark_name, "description": "pytest benchmark for download"}, - files={"file": ("pytest_benchmark.jsonl", line.encode("utf-8"), "application/x-ndjson")}, + f"/api/evaluation/databases/{db_id}/datasets/upload", + data={"name": dataset_name, "description": "pytest dataset for download"}, + files={"file": ("pytest_dataset.jsonl", line.encode("utf-8"), "application/x-ndjson")}, headers=admin_headers, ) assert response.status_code == 200, response.text payload = response.json() assert payload.get("message") == "success" - benchmark_id = payload.get("data", {}).get("benchmark_id") - assert benchmark_id - return benchmark_id, line + dataset_id = payload.get("data", {}).get("dataset_id") + assert dataset_id + return dataset_id, line -async def test_download_benchmark_requires_admin(test_client, standard_user): +async def test_download_dataset_requires_admin(test_client, standard_user): response = await test_client.get( - "/api/evaluation/benchmarks/benchmark_fake/download", + "/api/evaluation/datasets/dataset_fake/download", headers=standard_user["headers"], ) assert response.status_code == 403 -async def test_admin_can_download_benchmark(test_client, admin_headers, knowledge_database): - benchmark_id, expected_line = await _upload_test_benchmark(test_client, admin_headers, knowledge_database["db_id"]) +async def test_admin_can_download_dataset(test_client, admin_headers, knowledge_database): + dataset_id, expected_line = await _upload_test_dataset(test_client, admin_headers, knowledge_database["db_id"]) response = await test_client.get( - f"/api/evaluation/benchmarks/{benchmark_id}/download", + f"/api/evaluation/datasets/{dataset_id}/download", headers=admin_headers, ) assert response.status_code == 200, response.text @@ -53,9 +53,9 @@ async def test_admin_can_download_benchmark(test_client, admin_headers, knowledg assert expected_line.strip() in content -async def test_download_benchmark_not_found(test_client, admin_headers): +async def test_download_dataset_not_found(test_client, admin_headers): response = await test_client.get( - f"/api/evaluation/benchmarks/benchmark_not_found_{uuid.uuid4().hex[:8]}/download", + f"/api/evaluation/datasets/dataset_not_found_{uuid.uuid4().hex[:8]}/download", headers=admin_headers, ) assert response.status_code == 404, response.text diff --git a/backend/test/integration/conftest.py b/backend/test/integration/conftest.py index fbaa0c47..52bd497b 100644 --- a/backend/test/integration/conftest.py +++ b/backend/test/integration/conftest.py @@ -34,6 +34,22 @@ HTTP_TIMEOUT = httpx.Timeout(60.0, connect=5.0) SANDBOX_CONTAINER_PREFIX = os.getenv("YUXI_SANDBOX_CONTAINER_PREFIX", "yuxi-sandbox") +@pytest.fixture(scope="session", autouse=True) +def ensure_live_api_schema(): + if not ADMIN_LOGIN or not ADMIN_PASSWORD: + return + + async def run_schema_setup() -> None: + from yuxi.storage.postgres.manager import pg_manager + + pg_manager.initialize() + await pg_manager.create_tables() + await pg_manager.ensure_business_schema() + await pg_manager.ensure_knowledge_schema() + + anyio.run(run_schema_setup) + + def _require_admin_credentials() -> tuple[str, str]: if not ADMIN_LOGIN or not ADMIN_PASSWORD: pytest.skip("Integration credentials are not configured via TEST_USERNAME / TEST_PASSWORD.") diff --git a/backend/test/unit/knowledge/eval/test_benchmark_generation.py b/backend/test/unit/knowledge/eval/test_benchmark_generation.py index 3c230763..d8936e40 100644 --- a/backend/test/unit/knowledge/eval/test_benchmark_generation.py +++ b/backend/test/unit/knowledge/eval/test_benchmark_generation.py @@ -1,3 +1,4 @@ +import asyncio import os from types import SimpleNamespace @@ -11,6 +12,7 @@ from yuxi.knowledge.eval.benchmark_generation import ( clamp_neighbors_count, collect_kb_chunks, iter_generated_benchmark_items, + normalize_generation_concurrency_count, select_neighbor_chunks_by_kb_query, ) @@ -65,12 +67,40 @@ class NoQueryKnowledgeBase(FakeGenerationKnowledgeBase): raise AssertionError("neighbors_count=1 时不应调用 aquery") +class TrackingLlm: + def __init__(self, content=None, delay=0): + self.content = content or '{"query":"问题","gold_answer":"答案","gold_chunk_ids":["anchor_chunk"]}' + self.delay = delay + self.active_calls = 0 + self.max_active_calls = 0 + self.calls = 0 + + async def call(self, prompt, stream): + self.calls += 1 + self.active_calls += 1 + self.max_active_calls = max(self.max_active_calls, self.active_calls) + try: + if self.delay: + await asyncio.sleep(self.delay) + return SimpleNamespace(content=self.content) + finally: + self.active_calls -= 1 + + def test_clamp_neighbors_count(): assert clamp_neighbors_count(-1) == 0 assert clamp_neighbors_count(3) == 3 assert clamp_neighbors_count(11) == 10 +def test_normalize_generation_concurrency_count(): + assert normalize_generation_concurrency_count(None) == 10 + assert normalize_generation_concurrency_count("") == 10 + assert normalize_generation_concurrency_count(0) == 1 + assert normalize_generation_concurrency_count(-5) == 1 + assert normalize_generation_concurrency_count(10000) == 20 + + def test_build_benchmark_generation_prompt_contains_required_schema(): prompt = build_benchmark_generation_prompt([("chunk_1", "片段内容")]) @@ -188,3 +218,65 @@ async def test_iter_generated_benchmark_items_falls_back_to_anchor_when_query_em assert items == [{"query": "问题", "gold_chunk_ids": ["anchor_chunk"], "gold_answer": "答案"}] assert "片段ID=anchor_chunk" in fake_llm.prompts[0] + + +@pytest.mark.asyncio +async def test_iter_generated_benchmark_items_respects_concurrency_count(monkeypatch): + fake_llm = TrackingLlm(delay=0.01) + monkeypatch.setattr(benchmark_generation, "select_model", lambda model_spec: fake_llm) + + items = [ + item + async for item in iter_generated_benchmark_items( + kb_instance=NoQueryKnowledgeBase(), + db_id="db_1", + count=4, + neighbors_count=1, + concurrency_count=2, + llm_model_spec="test-provider:test-model", + ) + ] + + assert len(items) == 4 + assert fake_llm.max_active_calls == 2 + + +@pytest.mark.asyncio +async def test_iter_generated_benchmark_items_returns_at_most_count(monkeypatch): + fake_llm = TrackingLlm(delay=0.01) + monkeypatch.setattr(benchmark_generation, "select_model", lambda model_spec: fake_llm) + + items = [ + item + async for item in iter_generated_benchmark_items( + kb_instance=NoQueryKnowledgeBase(), + db_id="db_1", + count=3, + neighbors_count=1, + concurrency_count=10, + llm_model_spec="test-provider:test-model", + ) + ] + + assert len(items) == 3 + + +@pytest.mark.asyncio +async def test_iter_generated_benchmark_items_stops_at_max_attempts(monkeypatch): + fake_llm = TrackingLlm(content='{"query":"","gold_answer":"答案","gold_chunk_ids":["anchor_chunk"]}') + monkeypatch.setattr(benchmark_generation, "select_model", lambda model_spec: fake_llm) + + items = [ + item + async for item in iter_generated_benchmark_items( + kb_instance=NoQueryKnowledgeBase(), + db_id="db_1", + count=2, + neighbors_count=1, + concurrency_count=10, + llm_model_spec="test-provider:test-model", + ) + ] + + assert items == [] + assert fake_llm.calls == 50