309 lines
10 KiB
Python
309 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import time
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from memrelay.models import (
|
|
CurationDependencyState,
|
|
CurationJob,
|
|
CurationModelConfig,
|
|
CurationSettings,
|
|
)
|
|
from tests.conftest import csrf_headers
|
|
from tests.fakes import FakeBasicMemory
|
|
|
|
|
|
def _runtime_model() -> tuple[str, str, str]:
|
|
base_url = os.getenv("MEMRELAY_TEST_MODEL_BASE_URL")
|
|
token = os.getenv("MEMRELAY_TEST_MODEL_TOKEN")
|
|
model = os.getenv("MEMRELAY_TEST_MODEL_NAME", "gpt-5.6-sol")
|
|
if not base_url or not token:
|
|
pytest.skip("真实整理测试只在运行时注入模型凭证后执行")
|
|
return base_url, token, model
|
|
|
|
|
|
def _save_memory(
|
|
client: TestClient,
|
|
*,
|
|
request_id: str,
|
|
scope: str,
|
|
title: str,
|
|
content: str,
|
|
project_id: str | None = None,
|
|
) -> dict:
|
|
response = client.post(
|
|
"/api/v1/memories",
|
|
json={
|
|
"request_id": request_id,
|
|
"scope": scope,
|
|
"project_id": project_id,
|
|
"memory_type": "fact",
|
|
"title": title,
|
|
"content": content,
|
|
"tags": ["live-curation"],
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert response.status_code == 201, response.text
|
|
return response.json()
|
|
|
|
|
|
def _run_job(client: TestClient, payload: dict, *, recover_transient: bool = True) -> dict:
|
|
queued = client.post(
|
|
"/api/v1/curation/jobs",
|
|
json=payload,
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert queued.status_code == 200, queued.text
|
|
assert asyncio.run(client.app.state.curation.process_once()) is True
|
|
job_id = queued.json()["id"]
|
|
detail = client.get(f"/api/v1/curation/status?job_id={job_id}")
|
|
assert detail.status_code == 200, detail.text
|
|
result = detail.json()
|
|
for _ in range(3):
|
|
if not recover_transient or result["status"] != "waiting_dependency":
|
|
break
|
|
capabilities = asyncio.run(client.app.state.curation.test_model_connection(test_both=False))
|
|
if not capabilities["available"]:
|
|
break
|
|
assert asyncio.run(client.app.state.curation.process_once()) is True
|
|
detail = client.get(f"/api/v1/curation/status?job_id={job_id}")
|
|
assert detail.status_code == 200, detail.text
|
|
result = detail.json()
|
|
return result
|
|
|
|
|
|
def test_real_curation_end_to_end_when_runtime_credentials_are_supplied(
|
|
initialized_client: TestClient,
|
|
) -> None:
|
|
base_url, token, model = _runtime_model()
|
|
client = initialized_client
|
|
fake = FakeBasicMemory()
|
|
client.app.state.basic_memory = fake
|
|
manager = client.app.state.curation
|
|
manager.basic_memory = fake
|
|
|
|
project_response = client.post(
|
|
"/api/v1/projects",
|
|
json={
|
|
"name": "Live curation workspace",
|
|
"workspace_type": "development",
|
|
"curation_enabled": True,
|
|
"source_strategy": "mcp",
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert project_response.status_code == 201, project_response.text
|
|
project_id = project_response.json()["id"]
|
|
_save_memory(
|
|
client,
|
|
request_id="live-global-source",
|
|
scope="global",
|
|
title="Global collaboration preference",
|
|
content="The user prefers concise Chinese communication and concrete verification results.",
|
|
)
|
|
_save_memory(
|
|
client,
|
|
request_id="live-project-source-one",
|
|
scope="project",
|
|
project_id=project_id,
|
|
title="Workspace purpose",
|
|
content="This workspace validates unattended AI curation over the Responses API.",
|
|
)
|
|
|
|
with client.app.state.session_factory() as db:
|
|
settings = db.get(CurationSettings, 1)
|
|
dependency = db.get(CurationDependencyState, 1)
|
|
assert settings is not None and dependency is not None
|
|
config = CurationModelConfig(
|
|
revision=1,
|
|
name="runtime-live-curation",
|
|
base_url=base_url,
|
|
encrypted_token=manager.secret_box.encrypt(token, "curation-model-token"),
|
|
text_model=model,
|
|
protocol="responses",
|
|
effective_protocol="responses",
|
|
stream=False,
|
|
timeout_seconds=300,
|
|
capability_json=json.dumps(
|
|
{
|
|
"responses": {"status": "supported"},
|
|
"non_stream": {"status": "supported"},
|
|
"stream": {"status": "supported"},
|
|
"native_structured_output": {"status": "supported"},
|
|
}
|
|
),
|
|
active=True,
|
|
)
|
|
db.add(config)
|
|
db.flush()
|
|
settings.enabled = True
|
|
dependency.status = "available"
|
|
db.commit()
|
|
working_config_id = config.id
|
|
|
|
global_job = _run_job(
|
|
client,
|
|
{
|
|
"scope": "global",
|
|
"mode": "incremental",
|
|
"source_strategy": "mcp",
|
|
"document_types": ["profile"],
|
|
},
|
|
)
|
|
assert global_job["status"] == "completed", (
|
|
global_job["status"],
|
|
global_job["error_code"],
|
|
global_job["error_message"],
|
|
global_job["model_calls"],
|
|
)
|
|
|
|
first_project_job = _run_job(
|
|
client,
|
|
{
|
|
"scope": "project",
|
|
"project_id": project_id,
|
|
"mode": "incremental",
|
|
"source_strategy": "mcp",
|
|
"document_types": ["overview"],
|
|
},
|
|
)
|
|
assert first_project_job["status"] == "completed", (
|
|
first_project_job["status"],
|
|
first_project_job["error_code"],
|
|
first_project_job["error_message"],
|
|
first_project_job["model_calls"],
|
|
)
|
|
|
|
_save_memory(
|
|
client,
|
|
request_id="live-project-source-two",
|
|
scope="project",
|
|
project_id=project_id,
|
|
title="Workspace progress",
|
|
content="The live global and initial project curation checks completed successfully.",
|
|
)
|
|
second_project_job = _run_job(
|
|
client,
|
|
{
|
|
"scope": "project",
|
|
"project_id": project_id,
|
|
"mode": "incremental",
|
|
"source_strategy": "mcp",
|
|
"document_types": ["overview"],
|
|
"pending_policy": "replace",
|
|
},
|
|
)
|
|
assert second_project_job["status"] == "completed", (
|
|
second_project_job["status"],
|
|
second_project_job["error_code"],
|
|
second_project_job["error_message"],
|
|
second_project_job["model_calls"],
|
|
)
|
|
|
|
documents = client.get(
|
|
"/api/v1/curation/documents",
|
|
params={"scope": "project", "project_id": project_id},
|
|
)
|
|
assert documents.status_code == 200, documents.text
|
|
overview = next(item for item in documents.json() if item["document_type"] == "overview")
|
|
history = client.get(f"/api/v1/curation/documents/{overview['id']}/history")
|
|
assert history.status_code == 200, history.text
|
|
assert len(history.json()) >= 2
|
|
first_revision = min(item["revision"] for item in history.json())
|
|
latest_revision = max(item["revision"] for item in history.json())
|
|
difference = client.get(
|
|
f"/api/v1/curation/documents/{overview['id']}/diff",
|
|
params={"from_revision": first_revision, "to_revision": latest_revision},
|
|
)
|
|
assert difference.status_code == 200, difference.text
|
|
assert difference.json()["from_revision"] == first_revision
|
|
reverted = client.post(
|
|
f"/api/v1/curation/documents/{overview['id']}/revert",
|
|
params={"revision": first_revision},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert reverted.status_code == 200, reverted.text
|
|
assert reverted.json()["revision"] == latest_revision + 1
|
|
|
|
_save_memory(
|
|
client,
|
|
request_id="live-project-source-three",
|
|
scope="project",
|
|
project_id=project_id,
|
|
title="Recovery source",
|
|
content="A temporarily unavailable model task must resume after connectivity returns.",
|
|
)
|
|
with client.app.state.session_factory() as db:
|
|
working = db.get(CurationModelConfig, working_config_id)
|
|
assert working is not None
|
|
working.active = False
|
|
outage = CurationModelConfig(
|
|
revision=2,
|
|
name="runtime-network-outage",
|
|
base_url="http://127.0.0.1:9/v1",
|
|
encrypted_token=manager.secret_box.encrypt(
|
|
"temporary-test-token", "curation-model-token"
|
|
),
|
|
text_model=model,
|
|
protocol="responses",
|
|
effective_protocol="responses",
|
|
timeout_seconds=5,
|
|
active=True,
|
|
)
|
|
db.add(outage)
|
|
db.commit()
|
|
|
|
waiting = _run_job(
|
|
client,
|
|
{
|
|
"scope": "project",
|
|
"project_id": project_id,
|
|
"mode": "incremental",
|
|
"source_strategy": "mcp",
|
|
"document_types": ["overview"],
|
|
"pending_policy": "replace",
|
|
},
|
|
recover_transient=False,
|
|
)
|
|
assert waiting["status"] == "waiting_dependency", waiting
|
|
assert waiting["error_code"] == "MODEL_UNAVAILABLE"
|
|
|
|
with client.app.state.session_factory() as db:
|
|
working = db.get(CurationModelConfig, working_config_id)
|
|
outage = db.get(CurationModelConfig, waiting["model_config_id"])
|
|
assert working is not None and outage is not None
|
|
outage.active = False
|
|
working.active = True
|
|
db.commit()
|
|
capabilities = asyncio.run(
|
|
manager.test_model_connection(config_id=working_config_id, test_both=False)
|
|
)
|
|
assert capabilities["available"] is True
|
|
with client.app.state.session_factory() as db:
|
|
resumed = db.get(CurationJob, waiting["id"])
|
|
assert resumed is not None
|
|
resumed_status = resumed.status
|
|
if resumed_status != "completed":
|
|
processed = asyncio.run(manager.process_once())
|
|
assert processed is True or resumed_status in {"collecting", "running", "applying"}
|
|
for _ in range(120):
|
|
with client.app.state.session_factory() as db:
|
|
resumed = db.get(CurationJob, waiting["id"])
|
|
assert resumed is not None
|
|
if resumed.status in {"completed", "failed", "waiting_dependency", "cancelled"}:
|
|
break
|
|
time.sleep(0.25)
|
|
with client.app.state.session_factory() as db:
|
|
resumed = db.get(CurationJob, waiting["id"])
|
|
assert resumed is not None and resumed.status == "completed", (
|
|
resumed.status if resumed else None,
|
|
resumed.error_code if resumed else None,
|
|
resumed.error_message if resumed else None,
|
|
)
|