750 lines
25 KiB
Python
750 lines
25 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from app.gateway.routers import artifacts
|
|
from deerflow.persistence.artifact_library import MemoryArtifactLibraryStore
|
|
|
|
|
|
class _FakeThreadStore:
|
|
def __init__(self, rows: list[dict]) -> None:
|
|
self.rows = rows
|
|
self.calls: list[dict] = []
|
|
|
|
async def search(self, **kwargs) -> list[dict]:
|
|
self.calls.append(kwargs)
|
|
offset = int(kwargs.get("offset", 0))
|
|
limit = int(kwargs.get("limit", 100))
|
|
return self.rows[offset : offset + limit]
|
|
|
|
|
|
class _FakeAgentStore:
|
|
async def list_by_ids(self, agent_ids: list[str], user_id: str) -> list[dict]:
|
|
assert user_id == "owner"
|
|
agents = {
|
|
"analysis-agent": {
|
|
"id": "analysis-agent",
|
|
"name": "舆情分析智能体",
|
|
}
|
|
}
|
|
return [agents[agent_id] for agent_id in agent_ids if agent_id in agents]
|
|
|
|
|
|
def _strategy_task(
|
|
project_id: str,
|
|
*,
|
|
name: str,
|
|
synthesis: str,
|
|
status: str = "done",
|
|
stale: bool = False,
|
|
updated_at: str = "2026-09-22T08:00:00+00:00",
|
|
) -> dict:
|
|
return {
|
|
"id": project_id,
|
|
"title": name,
|
|
"created_at": updated_at,
|
|
"updated_at": updated_at,
|
|
"messages": [
|
|
{
|
|
"kind": "strategy-selection-project",
|
|
"version": 1,
|
|
"snapshot": {
|
|
"version": 1,
|
|
"taskDraft": {"name": name},
|
|
"synthesis": synthesis,
|
|
"synthesisStatus": status,
|
|
"workflowStale": stale,
|
|
},
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
class _FakeStrategyTaskStore:
|
|
def __init__(self, rows_by_user: dict[str, list[dict]]) -> None:
|
|
self.rows_by_user = rows_by_user
|
|
self.list_calls: list[dict] = []
|
|
self.get_calls: list[tuple[str, str]] = []
|
|
|
|
async def list_tasks(
|
|
self,
|
|
user_id: str,
|
|
*,
|
|
limit: int = 200,
|
|
offset: int = 0,
|
|
id_prefix: str | None = None,
|
|
include_messages: bool = False,
|
|
) -> list[dict]:
|
|
self.list_calls.append(
|
|
{
|
|
"user_id": user_id,
|
|
"limit": limit,
|
|
"offset": offset,
|
|
"id_prefix": id_prefix,
|
|
"include_messages": include_messages,
|
|
}
|
|
)
|
|
rows = self.rows_by_user.get(user_id, [])
|
|
if id_prefix:
|
|
rows = [row for row in rows if str(row.get("id") or "").startswith(id_prefix)]
|
|
rows = rows[offset : offset + limit]
|
|
if include_messages:
|
|
return rows
|
|
return [{key: value for key, value in row.items() if key != "messages"} for row in rows]
|
|
|
|
async def get_task(self, task_id: str, user_id: str) -> dict | None:
|
|
self.get_calls.append((task_id, user_id))
|
|
return next(
|
|
(row for row in self.rows_by_user.get(user_id, []) if row.get("id") == task_id),
|
|
None,
|
|
)
|
|
|
|
|
|
def _library_item(*, thread_id: str, path: str, name: str, modified_at: str) -> dict:
|
|
return {
|
|
"thread_id": thread_id,
|
|
"thread_title": "资料来源",
|
|
"agent_id": None,
|
|
"agent_name": None,
|
|
"thread_type": None,
|
|
"source_type": "general_qa",
|
|
"strategy_project_id": None,
|
|
"strategy_name": None,
|
|
"strategy_stale": False,
|
|
"strategy_source_kind": None,
|
|
"download_url": None,
|
|
"path": path,
|
|
"name": name,
|
|
"size_bytes": 10,
|
|
"mime_type": "text/markdown",
|
|
"modified_at": modified_at,
|
|
"file_kind": "text_data",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_artifact_library_aggregates_generated_non_media_files(tmp_path: Path, monkeypatch):
|
|
thread_one = tmp_path / "thread-one" / "outputs"
|
|
thread_two = tmp_path / "thread-two" / "outputs"
|
|
thread_one.mkdir(parents=True)
|
|
thread_two.mkdir(parents=True)
|
|
|
|
report = thread_one / "研判报告.docx"
|
|
table = thread_one / "数据汇总.xlsx"
|
|
image = thread_one / "配图.png"
|
|
audio = thread_two / "录音.mp3"
|
|
notes = thread_two / "结论.md"
|
|
archive = thread_two / "证据.zip"
|
|
for path, content in (
|
|
(report, b"doc"),
|
|
(table, b"sheet"),
|
|
(image, b"image"),
|
|
(audio, b"audio"),
|
|
(notes, b"# notes"),
|
|
(archive, b"zip"),
|
|
):
|
|
path.write_bytes(content)
|
|
|
|
os.utime(report, (100, 100))
|
|
os.utime(table, (200, 200))
|
|
os.utime(notes, (300, 300))
|
|
os.utime(archive, (400, 400))
|
|
|
|
rows = [
|
|
{
|
|
"thread_id": "thread-one",
|
|
"display_name": "战场环境综合研判",
|
|
"metadata": {"agent_id": "analysis-agent"},
|
|
},
|
|
{
|
|
"thread_id": "thread-two",
|
|
"display_name": "生成发展报告",
|
|
"metadata": {},
|
|
},
|
|
]
|
|
store = _FakeThreadStore(rows)
|
|
|
|
class _Paths:
|
|
def sandbox_outputs_dir(self, thread_id: str, *, user_id: str) -> Path:
|
|
assert user_id == "owner"
|
|
return tmp_path / thread_id / "outputs"
|
|
|
|
async def _resolve_owner(_thread_id: str) -> str:
|
|
return "owner"
|
|
|
|
monkeypatch.setattr(artifacts, "get_paths", lambda: _Paths())
|
|
monkeypatch.setattr(artifacts, "aresolve_path_user_id", _resolve_owner)
|
|
|
|
items = await artifacts._scan_artifact_library(store)
|
|
|
|
assert [item["name"] for item in items] == ["证据.zip", "结论.md", "数据汇总.xlsx", "研判报告.docx"]
|
|
assert [item["file_kind"] for item in items] == ["archive", "text_data", "spreadsheet", "document"]
|
|
assert [item["source_type"] for item in items] == ["general_qa", "general_qa", "agent", "agent"]
|
|
assert items[-1]["thread_title"] == "战场环境综合研判"
|
|
assert items[-1]["agent_id"] == "analysis-agent"
|
|
assert all(item["name"] not in {"配图.png", "录音.mp3"} for item in items)
|
|
assert store.calls[0]["exclude_system"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_artifact_library_endpoint_filters_searches_and_pages(tmp_path: Path, monkeypatch):
|
|
output_dir = tmp_path / "thread-one" / "outputs"
|
|
agent_output_dir = tmp_path / "thread-agent" / "outputs"
|
|
output_dir.mkdir(parents=True)
|
|
agent_output_dir.mkdir(parents=True)
|
|
(output_dir / "任务报告.md").write_text("report", encoding="utf-8")
|
|
(output_dir / "任务数据.csv").write_text("a,b", encoding="utf-8")
|
|
(agent_output_dir / "智能体结论.md").write_text("agent report", encoding="utf-8")
|
|
|
|
store = _FakeThreadStore(
|
|
[
|
|
{
|
|
"thread_id": "thread-one",
|
|
"display_name": "任务研判对话",
|
|
"metadata": {},
|
|
},
|
|
{
|
|
"thread_id": "thread-agent",
|
|
"display_name": "智能体研判对话",
|
|
"metadata": {"agent_id": "analysis-agent"},
|
|
},
|
|
]
|
|
)
|
|
|
|
class _Paths:
|
|
def sandbox_outputs_dir(self, thread_id: str, *, user_id: str) -> Path:
|
|
return tmp_path / thread_id / "outputs"
|
|
|
|
async def _resolve_owner(_thread_id: str) -> str:
|
|
return "owner"
|
|
|
|
monkeypatch.setattr(artifacts, "get_paths", lambda: _Paths())
|
|
monkeypatch.setattr(artifacts, "aresolve_path_user_id", _resolve_owner)
|
|
monkeypatch.setattr(artifacts, "get_effective_user_id", lambda: "owner")
|
|
index = MemoryArtifactLibraryStore()
|
|
generation = await index.claim_sync("owner", force=True)
|
|
assert generation is not None
|
|
indexed_items = await artifacts._scan_artifact_library(store, user_id="owner")
|
|
indexed_items = await artifacts._resolve_artifact_library_agent_names(
|
|
indexed_items,
|
|
_FakeAgentStore(),
|
|
user_id="owner",
|
|
)
|
|
assert await index.finish_sync("owner", generation, indexed_items)
|
|
request = SimpleNamespace(
|
|
app=SimpleNamespace(
|
|
state=SimpleNamespace(
|
|
artifact_library_store=index,
|
|
artifact_library_sync_tasks={},
|
|
artifact_library_sync_semaphore=asyncio.Semaphore(1),
|
|
thread_store=store,
|
|
agent_store=_FakeAgentStore(),
|
|
parallel_agent_task_store=None,
|
|
)
|
|
)
|
|
)
|
|
|
|
raw_endpoint = artifacts.list_artifact_library.__wrapped__
|
|
response = await raw_endpoint(
|
|
request=request,
|
|
query="研判对话",
|
|
file_kind="text_data",
|
|
source_type="general_qa",
|
|
offset=0,
|
|
limit=1,
|
|
refresh=False,
|
|
)
|
|
|
|
assert response.total == 1
|
|
assert response.offset == 0
|
|
assert response.limit == 1
|
|
assert [item.name for item in response.items] == ["任务报告.md"]
|
|
|
|
spreadsheet_response = await raw_endpoint(
|
|
request=request,
|
|
query="任务数据",
|
|
file_kind="spreadsheet",
|
|
source_type="all",
|
|
offset=0,
|
|
limit=50,
|
|
refresh=False,
|
|
)
|
|
assert spreadsheet_response.total == 1
|
|
assert spreadsheet_response.items[0].name == "任务数据.csv"
|
|
|
|
agent_response = await raw_endpoint(
|
|
request=request,
|
|
query=None,
|
|
file_kind="all",
|
|
source_type="agent",
|
|
offset=0,
|
|
limit=50,
|
|
refresh=False,
|
|
)
|
|
assert agent_response.total == 1
|
|
assert agent_response.items[0].source_type == "agent"
|
|
assert agent_response.items[0].agent_name == "舆情分析智能体"
|
|
assert agent_response.items[0].name == "智能体结论.md"
|
|
|
|
first_page = await raw_endpoint(
|
|
request=request,
|
|
query=None,
|
|
file_kind="all",
|
|
source_type="all",
|
|
offset=0,
|
|
limit=1,
|
|
refresh=False,
|
|
)
|
|
second_page = await raw_endpoint(
|
|
request=request,
|
|
query=None,
|
|
file_kind="all",
|
|
source_type="all",
|
|
offset=1,
|
|
limit=1,
|
|
refresh=False,
|
|
)
|
|
assert first_page.total == 3
|
|
assert second_page.total == 3
|
|
assert first_page.items[0].path != second_page.items[0].path
|
|
assert second_page.offset == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_artifact_library_marks_analysis_project_strategy_files(tmp_path: Path, monkeypatch):
|
|
output_dir = tmp_path / "strategy-thread" / "outputs"
|
|
output_dir.mkdir(parents=True)
|
|
(output_dir / "策略简报.md").write_text("# strategy", encoding="utf-8")
|
|
store = _FakeThreadStore(
|
|
[
|
|
{
|
|
"thread_id": "strategy-thread",
|
|
"display_name": "策略项目对话",
|
|
"metadata": {
|
|
"analysis_project_id": "analysis-strategy-1",
|
|
"analysis_project_mode": "strategy",
|
|
"analysis_project_name": "海外市场策略",
|
|
"agent_id": "analysis-agent",
|
|
},
|
|
}
|
|
]
|
|
)
|
|
|
|
class _Paths:
|
|
def sandbox_outputs_dir(self, thread_id: str, *, user_id: str) -> Path:
|
|
assert thread_id == "strategy-thread"
|
|
assert user_id == "owner"
|
|
return output_dir
|
|
|
|
async def _resolve_owner(_thread_id: str) -> str:
|
|
return "owner"
|
|
|
|
monkeypatch.setattr(artifacts, "get_paths", lambda: _Paths())
|
|
monkeypatch.setattr(artifacts, "aresolve_path_user_id", _resolve_owner)
|
|
|
|
items = await artifacts._scan_artifact_library(store)
|
|
|
|
assert len(items) == 1
|
|
assert items[0]["source_type"] == "strategy"
|
|
assert items[0]["strategy_source_kind"] == "analysis_project"
|
|
assert items[0]["strategy_project_id"] == "analysis-strategy-1"
|
|
assert items[0]["strategy_name"] == "海外市场策略"
|
|
assert items[0]["agent_id"] == "analysis-agent"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_artifact_library_projects_completed_strategy_snapshots_and_filters(monkeypatch):
|
|
thread_store = _FakeThreadStore([])
|
|
strategy_store = _FakeStrategyTaskStore(
|
|
{
|
|
"owner": [
|
|
_strategy_task(
|
|
"strategy-selection-ready",
|
|
name="产业韧性策略",
|
|
synthesis="# 产业韧性方案\n\n正文",
|
|
stale=True,
|
|
),
|
|
_strategy_task(
|
|
"strategy-selection-running",
|
|
name="尚未完成",
|
|
synthesis="draft",
|
|
status="running",
|
|
),
|
|
_strategy_task(
|
|
"strategy-selection-empty",
|
|
name="空终稿",
|
|
synthesis=" ",
|
|
),
|
|
{
|
|
**_strategy_task(
|
|
"strategy-selection-future",
|
|
name="未来版本",
|
|
synthesis="future",
|
|
),
|
|
"messages": [
|
|
{
|
|
"kind": "strategy-selection-project",
|
|
"version": 1,
|
|
"snapshot": {
|
|
"version": 2,
|
|
"taskDraft": {"name": "未来版本"},
|
|
"synthesis": "future",
|
|
"synthesisStatus": "done",
|
|
"workflowStale": False,
|
|
},
|
|
}
|
|
],
|
|
},
|
|
{
|
|
**_strategy_task(
|
|
"parallel-task-lookalike",
|
|
name="非策略任务",
|
|
synthesis="must stay hidden",
|
|
),
|
|
"id": "parallel-task-lookalike",
|
|
},
|
|
],
|
|
"other": [
|
|
_strategy_task(
|
|
"strategy-selection-other-user",
|
|
name="他人策略",
|
|
synthesis="private",
|
|
)
|
|
],
|
|
}
|
|
)
|
|
request = SimpleNamespace(
|
|
app=SimpleNamespace(
|
|
state=SimpleNamespace(parallel_agent_task_store=strategy_store)
|
|
)
|
|
)
|
|
monkeypatch.setattr(artifacts, "get_effective_user_id", lambda: "owner")
|
|
index = MemoryArtifactLibraryStore()
|
|
generation = await index.claim_sync("owner", force=True)
|
|
assert generation is not None
|
|
indexed_items = await artifacts._scan_artifact_library(
|
|
thread_store,
|
|
strategy_store,
|
|
user_id="owner",
|
|
)
|
|
assert await index.finish_sync("owner", generation, indexed_items)
|
|
request.app.state.artifact_library_store = index
|
|
request.app.state.artifact_library_sync_tasks = {}
|
|
request.app.state.artifact_library_sync_semaphore = asyncio.Semaphore(1)
|
|
request.app.state.thread_store = thread_store
|
|
request.app.state.agent_store = _FakeAgentStore()
|
|
|
|
raw_endpoint = artifacts.list_artifact_library.__wrapped__
|
|
all_strategy_items = await raw_endpoint(
|
|
request=request,
|
|
query=None,
|
|
file_kind="all",
|
|
source_type="strategy",
|
|
offset=0,
|
|
limit=50,
|
|
refresh=False,
|
|
)
|
|
assert [item.strategy_project_id for item in all_strategy_items.items] == [
|
|
"strategy-selection-ready"
|
|
]
|
|
|
|
response = await raw_endpoint(
|
|
request=request,
|
|
query="韧性",
|
|
file_kind="text_data",
|
|
source_type="strategy",
|
|
offset=0,
|
|
limit=50,
|
|
refresh=False,
|
|
)
|
|
|
|
assert response.total == 1
|
|
item = response.items[0]
|
|
assert item.source_type == "strategy"
|
|
assert item.strategy_source_kind == "strategy_selection"
|
|
assert item.strategy_project_id == "strategy-selection-ready"
|
|
assert item.strategy_name == "产业韧性策略"
|
|
assert item.strategy_stale is True
|
|
assert item.name == "产业韧性策略.md"
|
|
assert item.size_bytes == len("# 产业韧性方案\n\n正文".encode())
|
|
assert item.download_url == "/api/strategy-project-artifacts/strategy-selection-ready/download"
|
|
assert strategy_store.list_calls == [
|
|
{
|
|
"user_id": "owner",
|
|
"limit": 200,
|
|
"offset": 0,
|
|
"id_prefix": "strategy-selection-",
|
|
"include_messages": True,
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_strategy_snapshot_scan_pages_the_strategy_namespace(monkeypatch):
|
|
strategy_store = _FakeStrategyTaskStore(
|
|
{
|
|
"owner": [
|
|
_strategy_task(
|
|
"strategy-selection-newer",
|
|
name="较新策略",
|
|
synthesis="newer",
|
|
),
|
|
_strategy_task(
|
|
"strategy-selection-older",
|
|
name="较早策略",
|
|
synthesis="older",
|
|
),
|
|
_strategy_task(
|
|
"parallel-unrelated",
|
|
name="普通并行任务",
|
|
synthesis="not a strategy project",
|
|
),
|
|
]
|
|
}
|
|
)
|
|
monkeypatch.setattr(artifacts, "get_effective_user_id", lambda: "owner")
|
|
monkeypatch.setattr(artifacts, "_STRATEGY_PROJECT_BATCH_SIZE", 1)
|
|
|
|
items = await artifacts._scan_strategy_project_artifacts(strategy_store)
|
|
|
|
assert [item["strategy_project_id"] for item in items] == [
|
|
"strategy-selection-newer",
|
|
"strategy-selection-older",
|
|
]
|
|
assert [call["offset"] for call in strategy_store.list_calls] == [0, 1, 2]
|
|
assert all(
|
|
call["id_prefix"] == "strategy-selection-"
|
|
and call["include_messages"] is True
|
|
for call in strategy_store.list_calls
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_strategy_snapshot_scan_strict_mode_rejects_partial_source():
|
|
class _FailingStrategyTaskStore:
|
|
async def list_tasks(self, *_args, **_kwargs):
|
|
raise OSError("temporary strategy store failure")
|
|
|
|
assert await artifacts._scan_strategy_project_artifacts(
|
|
_FailingStrategyTaskStore(),
|
|
user_id="owner",
|
|
) == []
|
|
with pytest.raises(artifacts.ArtifactLibraryScanIncomplete):
|
|
await artifacts._scan_strategy_project_artifacts(
|
|
_FailingStrategyTaskStore(),
|
|
user_id="owner",
|
|
strict=True,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_strategy_project_artifact_download_is_owner_scoped(monkeypatch):
|
|
content = "# 当前用户策略方案\n\n仅当前用户可下载。"
|
|
strategy_store = _FakeStrategyTaskStore(
|
|
{
|
|
"owner": [
|
|
_strategy_task(
|
|
"strategy-selection-中文",
|
|
name='A/B:策略',
|
|
synthesis=content,
|
|
stale=True,
|
|
),
|
|
_strategy_task(
|
|
"strategy-selection-pending",
|
|
name="未完成策略",
|
|
synthesis="draft",
|
|
status="running",
|
|
),
|
|
],
|
|
"other": [
|
|
_strategy_task(
|
|
"strategy-selection-private",
|
|
name="他人策略",
|
|
synthesis="private",
|
|
)
|
|
],
|
|
}
|
|
)
|
|
request = SimpleNamespace(
|
|
state=SimpleNamespace(user=SimpleNamespace(id="owner")),
|
|
app=SimpleNamespace(
|
|
state=SimpleNamespace(parallel_agent_task_store=strategy_store)
|
|
),
|
|
)
|
|
monkeypatch.setattr(artifacts, "get_effective_user_id", lambda: "fallback")
|
|
raw_endpoint = artifacts.download_strategy_project_artifact.__wrapped__
|
|
|
|
response = await raw_endpoint("strategy-selection-中文", request)
|
|
|
|
assert response.body.decode("utf-8") == content
|
|
assert response.media_type == "text/markdown; charset=utf-8"
|
|
assert "attachment" in response.headers["content-disposition"]
|
|
assert "A_B%3A" not in response.headers["content-disposition"]
|
|
assert "A_B_" in response.headers["content-disposition"]
|
|
assert strategy_store.get_calls[-1] == ("strategy-selection-中文", "owner")
|
|
|
|
with pytest.raises(HTTPException) as error:
|
|
await raw_endpoint("strategy-selection-private", request)
|
|
assert error.value.status_code == 404
|
|
assert strategy_store.get_calls[-1] == ("strategy-selection-private", "owner")
|
|
|
|
with pytest.raises(HTTPException) as error:
|
|
await raw_endpoint("strategy-selection-pending", request)
|
|
assert error.value.status_code == 404
|
|
assert strategy_store.get_calls[-1] == ("strategy-selection-pending", "owner")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_artifact_library_refresh_serves_snapshot_while_sync_runs(monkeypatch):
|
|
index = MemoryArtifactLibraryStore()
|
|
initial_generation = await index.claim_sync("owner", force=True)
|
|
assert initial_generation is not None
|
|
assert await index.finish_sync(
|
|
"owner",
|
|
initial_generation,
|
|
[
|
|
_library_item(
|
|
thread_id="old-thread",
|
|
path="mnt/user-data/outputs/old.md",
|
|
name="旧报告.md",
|
|
modified_at="2026-09-01T00:00:00+00:00",
|
|
)
|
|
],
|
|
)
|
|
|
|
scan_started = asyncio.Event()
|
|
release_scan = asyncio.Event()
|
|
|
|
async def _delayed_scan(*_args, **_kwargs):
|
|
scan_started.set()
|
|
await release_scan.wait()
|
|
return [
|
|
_library_item(
|
|
thread_id="new-thread",
|
|
path="mnt/user-data/outputs/new.md",
|
|
name="新报告.md",
|
|
modified_at="2026-09-02T00:00:00+00:00",
|
|
)
|
|
]
|
|
|
|
monkeypatch.setattr(artifacts, "_scan_artifact_library", _delayed_scan)
|
|
monkeypatch.setattr(artifacts, "get_effective_user_id", lambda: "owner")
|
|
request = SimpleNamespace(
|
|
app=SimpleNamespace(
|
|
state=SimpleNamespace(
|
|
artifact_library_store=index,
|
|
artifact_library_sync_tasks={},
|
|
artifact_library_sync_semaphore=asyncio.Semaphore(1),
|
|
thread_store=_FakeThreadStore([]),
|
|
agent_store=_FakeAgentStore(),
|
|
parallel_agent_task_store=None,
|
|
)
|
|
)
|
|
)
|
|
raw_endpoint = artifacts.list_artifact_library.__wrapped__
|
|
|
|
response = await raw_endpoint(
|
|
request=request,
|
|
query=None,
|
|
file_kind="all",
|
|
source_type="all",
|
|
offset=0,
|
|
limit=50,
|
|
refresh=True,
|
|
)
|
|
|
|
assert response.syncing is True
|
|
assert response.sync_status == "syncing"
|
|
assert [item.name for item in response.items] == ["旧报告.md"]
|
|
await asyncio.wait_for(scan_started.wait(), timeout=1)
|
|
task = request.app.state.artifact_library_sync_tasks["owner"]
|
|
release_scan.set()
|
|
await asyncio.wait_for(task, timeout=1)
|
|
|
|
refreshed = await raw_endpoint(
|
|
request=request,
|
|
query=None,
|
|
file_kind="all",
|
|
source_type="all",
|
|
offset=0,
|
|
limit=50,
|
|
refresh=False,
|
|
)
|
|
assert refreshed.syncing is False
|
|
assert refreshed.sync_status == "ready"
|
|
assert [item.name for item in refreshed.items] == ["新报告.md"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_incomplete_background_scan_keeps_last_complete_snapshot(monkeypatch):
|
|
index = MemoryArtifactLibraryStore()
|
|
initial_generation = await index.claim_sync("owner", force=True)
|
|
assert initial_generation is not None
|
|
assert await index.finish_sync(
|
|
"owner",
|
|
initial_generation,
|
|
[
|
|
_library_item(
|
|
thread_id="stable-thread",
|
|
path="mnt/user-data/outputs/stable.md",
|
|
name="稳定报告.md",
|
|
modified_at="2026-09-01T00:00:00+00:00",
|
|
)
|
|
],
|
|
)
|
|
|
|
async def _failed_scan(*_args, **_kwargs):
|
|
raise artifacts.ArtifactLibraryScanIncomplete(
|
|
r"failed at D:\private\sandbox\outputs"
|
|
)
|
|
|
|
monkeypatch.setattr(artifacts, "_scan_artifact_library", _failed_scan)
|
|
monkeypatch.setattr(artifacts, "get_effective_user_id", lambda: "owner")
|
|
request = SimpleNamespace(
|
|
app=SimpleNamespace(
|
|
state=SimpleNamespace(
|
|
artifact_library_store=index,
|
|
artifact_library_sync_tasks={},
|
|
artifact_library_sync_semaphore=asyncio.Semaphore(1),
|
|
thread_store=_FakeThreadStore([]),
|
|
agent_store=_FakeAgentStore(),
|
|
parallel_agent_task_store=None,
|
|
)
|
|
)
|
|
)
|
|
raw_endpoint = artifacts.list_artifact_library.__wrapped__
|
|
|
|
first = await raw_endpoint(
|
|
request=request,
|
|
query=None,
|
|
file_kind="all",
|
|
source_type="all",
|
|
offset=0,
|
|
limit=50,
|
|
refresh=True,
|
|
)
|
|
assert first.syncing is True
|
|
task = request.app.state.artifact_library_sync_tasks["owner"]
|
|
await asyncio.wait_for(task, timeout=1)
|
|
|
|
after_failure = await raw_endpoint(
|
|
request=request,
|
|
query=None,
|
|
file_kind="all",
|
|
source_type="all",
|
|
offset=0,
|
|
limit=50,
|
|
refresh=False,
|
|
)
|
|
assert after_failure.sync_status == "error"
|
|
assert after_failure.syncing is False
|
|
assert after_failure.sync_error == "资料索引同步失败，请稍后重试"
|
|
assert "private" not in (after_failure.sync_error or "").lower()
|
|
assert [item.name for item in after_failure.items] == ["稳定报告.md"]
|