from __future__ import annotations import os from pathlib import Path from types import SimpleNamespace import pytest from fastapi import HTTPException from app.gateway import deps from app.gateway.routers import artifacts 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, ) @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") monkeypatch.setattr(deps, "get_thread_store", lambda _request: store) monkeypatch.setattr(deps, "get_agent_store", lambda _request: _FakeAgentStore()) artifacts._artifact_library_cache.clear() raw_endpoint = artifacts.list_artifact_library.__wrapped__ response = await raw_endpoint( request=SimpleNamespace(), query="研判对话", file_kind="text_data", source_type="general_qa", offset=0, limit=1, refresh=True, ) 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=SimpleNamespace(), 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=SimpleNamespace(), 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" @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") monkeypatch.setattr(deps, "get_thread_store", lambda _request: thread_store) monkeypatch.setattr(deps, "get_agent_store", lambda _request: _FakeAgentStore()) artifacts._artifact_library_cache.clear() 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=True, ) 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=True, ) 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, }, { "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_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")