496 lines
17 KiB
Python
496 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
BACKEND_ROOT = Path(__file__).resolve().parents[1]
|
|
SEED_SKILLS_ROOT = (
|
|
BACKEND_ROOT / "app" / "gateway" / "routers" / "_agentfx_seed_assets" / "skills"
|
|
)
|
|
LIB_PATH = SEED_SKILLS_ROOT / "task-report-lib" / "convert_lib.py"
|
|
|
|
|
|
def _load_convert_lib():
|
|
spec = importlib.util.spec_from_file_location("task_report_convert_lib", LIB_PATH)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def convert_lib():
|
|
return _load_convert_lib()
|
|
|
|
|
|
def _write_hits(path: Path, *, marker: str = "甲") -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"query": f"{marker} 地区 舆情",
|
|
"results": [
|
|
{
|
|
"title": f"{marker}事件进展",
|
|
"snippet": f"{marker}地区发布最新通报。",
|
|
"url": f"https://example.test/{marker}",
|
|
"published_at": "2026-09-24",
|
|
},
|
|
{
|
|
"name": f"{marker}背景资料",
|
|
"content": f"有关{marker}地区的人文与信息环境资料。",
|
|
"link": f"https://example.test/{marker}/background",
|
|
},
|
|
],
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def _records(path: Path) -> list[dict[str, object]]:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
assert isinstance(payload, list)
|
|
return payload
|
|
|
|
|
|
def _content_by_category(path: Path) -> dict[str, object]:
|
|
return {
|
|
str(record["categoryType"]): json.loads(str(record["contentJson"]))
|
|
for record in _records(path)
|
|
}
|
|
|
|
|
|
def _assert_dashboard_shapes(contents: dict[str, object], evidence_marker: str) -> None:
|
|
targets = contents["敌情目标 JSON"]
|
|
assert isinstance(targets, dict)
|
|
assert set(targets) >= {"organization", "human"}
|
|
assert isinstance(targets["organization"], list) and targets["organization"]
|
|
assert set(targets["organization"][0]) >= {"name", "text", "etype", "reason"}
|
|
|
|
events = contents["敌情关键事件 JSON"]
|
|
assert isinstance(events, list) and events
|
|
assert set(events[0]) >= {"topic", "count", "events"}
|
|
assert isinstance(events[0]["events"], list) and events[0]["events"]
|
|
assert set(events[0]["events"][0]) >= {
|
|
"recUuid",
|
|
"date",
|
|
"publish_time",
|
|
"link",
|
|
"title",
|
|
"content",
|
|
}
|
|
|
|
enemy_narratives = contents["敌情对手叙事 JSON"]
|
|
assert isinstance(enemy_narratives, list) and enemy_narratives
|
|
assert set(enemy_narratives[0]) >= {
|
|
"topic",
|
|
"narrator",
|
|
"intention",
|
|
"topicType",
|
|
"summary",
|
|
}
|
|
|
|
speeches = contents["我情口径 JSON"]
|
|
assert isinstance(speeches, dict)
|
|
assert set(speeches) >= {"selfSpeech", "selfConsistents"}
|
|
assert isinstance(speeches["selfSpeech"], list) and speeches["selfSpeech"]
|
|
assert set(speeches["selfSpeech"][0]) >= {
|
|
"trendSpeechTitle",
|
|
"trendSpeechLike",
|
|
"trendSpeechContent",
|
|
"narrativeDirection",
|
|
"trendSpeechTime",
|
|
}
|
|
|
|
ours_narratives = contents["我情叙事 JSON"]
|
|
assert isinstance(ours_narratives, list) and ours_narratives
|
|
assert set(ours_narratives[0]) >= {"标题", "意图", "标签", "内容"}
|
|
assert isinstance(contents["我情上级要求 JSON"], str)
|
|
|
|
for category in ("信息环境 JSON", "人文环境 JSON"):
|
|
section_map = contents[category]
|
|
assert isinstance(section_map, dict)
|
|
sections = [value for key, value in section_map.items() if key != "type"]
|
|
assert sections and all(isinstance(value, list) and value for value in sections)
|
|
|
|
strengths = contents["优劣势分析 JSON"]
|
|
assert isinstance(strengths, dict)
|
|
assert set(strengths) >= {"优势", "劣势"}
|
|
assert all(set(items[0]) >= {"title", "content"} for items in strengths.values())
|
|
|
|
constraints = contents["条件与限制 JSON"]
|
|
assert isinstance(constraints, dict) and constraints
|
|
assert all(isinstance(value, list) and value for value in constraints.values())
|
|
|
|
risks = contents["风险及应对 JSON"]
|
|
assert isinstance(risks, dict)
|
|
assert set(risks) >= {"风险分析", "应对方案"}
|
|
assert all(isinstance(value, list) and value for value in risks.values())
|
|
|
|
decision = contents["决心建议 JSON"]
|
|
assert isinstance(decision, dict)
|
|
assert set(decision) >= {"决心建议", "实施要点", "决心建议总结"}
|
|
decision_core = decision["决心建议"]
|
|
assert isinstance(decision_core, dict)
|
|
assert set(decision_core) >= {"任务标题", "任务简介", "作战目的", "打击目标"}
|
|
assert decision_core["打击目标"]
|
|
assert set(decision_core["打击目标"][0]) >= {
|
|
"目标名称",
|
|
"目标描述",
|
|
"选择理由",
|
|
"潜在影响",
|
|
"推荐依据",
|
|
}
|
|
|
|
disposal = contents["处置建议 JSON"]
|
|
assert isinstance(disposal, dict)
|
|
assert set(disposal) >= {"目标", "处置建议总结", "处置建议"}
|
|
assert disposal["处置建议"]
|
|
assert set(disposal["处置建议"][0]) >= {
|
|
"建议内容",
|
|
"处置方式与步骤",
|
|
"口径基调",
|
|
"预期效果",
|
|
"处置主体",
|
|
"立场与原则",
|
|
}
|
|
assert set(disposal["处置建议"][0]["处置主体"]) >= {"主责单位", "配合单位"}
|
|
|
|
plans = contents["行动方案 JSON"]
|
|
assert isinstance(plans, list) and plans
|
|
assert set(plans[0]) >= {
|
|
"trendPlanDesc",
|
|
"createTime",
|
|
"esId",
|
|
"trendPlanName",
|
|
"updateTime",
|
|
"id",
|
|
"trendPlanGoal",
|
|
"trendPlanAction",
|
|
"taskId",
|
|
}
|
|
|
|
for category, content in contents.items():
|
|
serialized = json.dumps(content, ensure_ascii=False)
|
|
assert evidence_marker in serialized, category
|
|
|
|
|
|
def test_resolve_hit_files_maps_virtual_user_data_to_workspace(tmp_path: Path, convert_lib) -> None:
|
|
user_data = tmp_path / "thread" / "user-data"
|
|
workspace = user_data / "workspace"
|
|
output = user_data / "outputs"
|
|
workspace.mkdir(parents=True)
|
|
output.mkdir()
|
|
hit_file = workspace / "hits" / "q01.json"
|
|
_write_hits(hit_file)
|
|
|
|
files, warnings = convert_lib.resolve_hit_files(
|
|
"/mnt/user-data/workspace/hits",
|
|
cwd=workspace,
|
|
)
|
|
|
|
assert files == [hit_file.resolve()]
|
|
assert warnings == []
|
|
|
|
|
|
def test_all_group_writes_fourteen_unicode_records(tmp_path: Path, convert_lib) -> None:
|
|
workspace = tmp_path / "user-data" / "workspace"
|
|
(tmp_path / "user-data" / "outputs").mkdir(parents=True)
|
|
_write_hits(workspace / "hits" / "q01.json", marker="海峡")
|
|
|
|
result = convert_lib.convert_reports(
|
|
group="all",
|
|
task_id="T-001",
|
|
task_name="海峡态势研判",
|
|
task_content="分析公开信息与风险信号",
|
|
hits_dir="/mnt/user-data/workspace/hits",
|
|
output="/mnt/user-data/outputs/task-reports.json",
|
|
cwd=workspace,
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["hits"] == 2
|
|
output_path = Path(str(result["output"]))
|
|
records = _records(output_path)
|
|
assert [record["categoryType"] for record in records] == convert_lib.CATEGORY_TYPES
|
|
assert len(records) == 14
|
|
assert all(record["taskId"] == "T-001" for record in records)
|
|
assert "海峡" in output_path.read_text(encoding="utf-8")
|
|
assert "\\u6d77" not in output_path.read_text(encoding="utf-8")
|
|
_assert_dashboard_shapes(_content_by_category(output_path), convert_lib.EVIDENCE_MARKER)
|
|
|
|
|
|
def test_page_group_replaces_only_its_categories(tmp_path: Path, convert_lib) -> None:
|
|
workspace = tmp_path / "user-data" / "workspace"
|
|
outputs = tmp_path / "user-data" / "outputs"
|
|
outputs.mkdir(parents=True)
|
|
hit_file = workspace / "hits" / "q01.json"
|
|
_write_hits(hit_file, marker="旧材料")
|
|
output = outputs / "task-reports.json"
|
|
|
|
first = convert_lib.convert_reports(
|
|
group="all",
|
|
task_id="T-002",
|
|
task_name="测试任务",
|
|
task_content="测试内容",
|
|
hits_dir=str(hit_file.parent),
|
|
output=str(output),
|
|
cwd=workspace,
|
|
)
|
|
assert first["success"] is True
|
|
before = {record["categoryType"]: record for record in _records(output)}
|
|
|
|
_write_hits(hit_file, marker="新敌情")
|
|
second = convert_lib.convert_reports(
|
|
group="enemy",
|
|
task_id="T-002",
|
|
task_name="测试任务",
|
|
task_content="测试内容",
|
|
hits=[str(hit_file)],
|
|
output=str(output),
|
|
cwd=workspace,
|
|
)
|
|
assert second["success"] is True
|
|
after = {record["categoryType"]: record for record in _records(output)}
|
|
|
|
assert set(after) == set(convert_lib.CATEGORY_TYPES)
|
|
assert after["信息环境 JSON"] == before["信息环境 JSON"]
|
|
assert "新敌情" in str(after["敌情关键事件 JSON"]["contentJson"])
|
|
assert "旧材料" not in str(after["敌情关键事件 JSON"]["contentJson"])
|
|
|
|
|
|
def test_page_group_rejects_existing_records_from_another_task(tmp_path: Path, convert_lib) -> None:
|
|
workspace = tmp_path / "user-data" / "workspace"
|
|
outputs = tmp_path / "user-data" / "outputs"
|
|
outputs.mkdir(parents=True)
|
|
hit_file = workspace / "hits" / "q01.json"
|
|
_write_hits(hit_file, marker="原任务")
|
|
output = outputs / "task-reports.json"
|
|
initial = convert_lib.convert_reports(
|
|
group="all",
|
|
task_id="TASK-A",
|
|
hits=[str(hit_file)],
|
|
output=str(output),
|
|
cwd=workspace,
|
|
)
|
|
assert initial["success"] is True
|
|
original = output.read_bytes()
|
|
|
|
_write_hits(hit_file, marker="另一任务")
|
|
result = convert_lib.convert_reports(
|
|
group="enemy",
|
|
task_id="TASK-B",
|
|
hits=[str(hit_file)],
|
|
output=str(output),
|
|
cwd=workspace,
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert any("TASK-A" in warning and "TASK-B" in warning for warning in result["warnings"])
|
|
assert output.read_bytes() == original
|
|
|
|
|
|
def test_zero_hits_fails_without_overwriting_existing_output(tmp_path: Path, convert_lib) -> None:
|
|
workspace = tmp_path / "user-data" / "workspace"
|
|
outputs = tmp_path / "user-data" / "outputs"
|
|
workspace.mkdir(parents=True)
|
|
outputs.mkdir()
|
|
hits_dir = workspace / "hits"
|
|
hits_dir.mkdir()
|
|
(hits_dir / "q01.json").write_text(
|
|
'{"query":"没有结果的检索","results":[]}',
|
|
encoding="utf-8",
|
|
)
|
|
output = outputs / "task-reports.json"
|
|
original = '[{"sentinel": true}]\n'
|
|
output.write_text(original, encoding="utf-8")
|
|
|
|
result = convert_lib.convert_reports(
|
|
group="all",
|
|
task_id="T-003",
|
|
task_name="无命中任务",
|
|
task_content="不能编造证据",
|
|
hits_dir="/mnt/user-data/workspace/hits",
|
|
output="/mnt/user-data/outputs/task-reports.json",
|
|
cwd=workspace,
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["hits"] == 0
|
|
assert result["warnings"]
|
|
assert output.read_text(encoding="utf-8") == original
|
|
|
|
|
|
@pytest.mark.parametrize("unsafe_kind", ["hits", "output"])
|
|
def test_paths_outside_user_data_are_rejected(
|
|
tmp_path: Path,
|
|
convert_lib,
|
|
unsafe_kind: str,
|
|
) -> None:
|
|
workspace = tmp_path / "thread" / "user-data" / "workspace"
|
|
outputs = workspace.parent / "outputs"
|
|
outputs.mkdir(parents=True)
|
|
safe_hit = workspace / "hits" / "q01.json"
|
|
_write_hits(safe_hit)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
outside_hit = outside / "q01.json"
|
|
_write_hits(outside_hit, marker="越界")
|
|
outside_output = outside / "task-reports.json"
|
|
|
|
result = convert_lib.convert_reports(
|
|
group="all",
|
|
task_id="T-SAFE",
|
|
hits=[str(outside_hit if unsafe_kind == "hits" else safe_hit)],
|
|
output=str(outside_output if unsafe_kind == "output" else outputs / "task-reports.json"),
|
|
cwd=workspace,
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert any("user-data" in warning for warning in result["warnings"])
|
|
assert not outside_output.exists()
|
|
|
|
|
|
def test_symlink_escape_is_rejected(tmp_path: Path, convert_lib) -> None:
|
|
workspace = tmp_path / "thread" / "user-data" / "workspace"
|
|
(workspace.parent / "outputs").mkdir(parents=True)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
_write_hits(outside / "q01.json", marker="链接越界")
|
|
link = workspace / "hits"
|
|
workspace.mkdir(exist_ok=True)
|
|
try:
|
|
link.symlink_to(outside, target_is_directory=True)
|
|
except OSError as exc:
|
|
pytest.skip(f"symlink creation is unavailable: {exc}")
|
|
|
|
result = convert_lib.convert_reports(
|
|
group="all",
|
|
task_id="T-LINK",
|
|
hits_dir=str(link),
|
|
output="/mnt/user-data/outputs/task-reports.json",
|
|
cwd=workspace,
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert any("user-data" in warning for warning in result["warnings"])
|
|
|
|
|
|
def test_thin_cli_prints_one_json_line(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "user-data" / "workspace"
|
|
outputs = tmp_path / "user-data" / "outputs"
|
|
outputs.mkdir(parents=True)
|
|
hit_file = workspace / "hits" / "q01.json"
|
|
_write_hits(hit_file, marker="命令行")
|
|
script = SEED_SKILLS_ROOT / "task-report-env" / "scripts" / "convert.py"
|
|
output = outputs / "task-reports.json"
|
|
|
|
completed = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(script),
|
|
"--task-id",
|
|
"T-CLI",
|
|
"--task-name",
|
|
"命令行测试",
|
|
"--task-content",
|
|
"验证标准输出",
|
|
"--hits",
|
|
str(hit_file),
|
|
"--output",
|
|
str(output),
|
|
],
|
|
cwd=workspace,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
check=False,
|
|
)
|
|
|
|
assert completed.returncode == 0, completed.stderr
|
|
lines = completed.stdout.splitlines()
|
|
assert len(lines) == 1
|
|
result = json.loads(lines[0])
|
|
assert result["success"] is True
|
|
assert result["hits"] == 2
|
|
assert [record["categoryType"] for record in _records(output)] == convert_lib_env_categories()
|
|
|
|
|
|
def test_task_context_cli_treats_shell_metacharacters_as_data(tmp_path: Path, convert_lib) -> None:
|
|
workspace = tmp_path / "user-data" / "workspace"
|
|
outputs = tmp_path / "user-data" / "outputs"
|
|
outputs.mkdir(parents=True)
|
|
_write_hits(workspace / "hits" / "q01.json", marker="上下文")
|
|
task_name = '名称"; $(New-Item PWNED); `whoami` $env:PATH'
|
|
task_content = "内容'; rm -rf ignored; && echo not-executed"
|
|
(workspace / "task-context.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"taskId": "T-CONTEXT",
|
|
"taskName": task_name,
|
|
"taskContent": task_content,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
script = SEED_SKILLS_ROOT / "task-report-build" / "scripts" / "convert.py"
|
|
|
|
completed = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(script),
|
|
"--task-context",
|
|
"/mnt/user-data/workspace/task-context.json",
|
|
],
|
|
cwd=workspace,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
check=False,
|
|
)
|
|
|
|
assert completed.returncode == 0, completed.stderr
|
|
assert len(completed.stdout.splitlines()) == 1
|
|
result = json.loads(completed.stdout)
|
|
output = Path(result["output"])
|
|
decision = _content_by_category(output)["决心建议 JSON"]
|
|
assert decision["决心建议"]["任务标题"] == task_name
|
|
assert decision["决心建议"]["任务简介"] == task_content
|
|
assert not (tmp_path / "PWNED").exists()
|
|
_assert_dashboard_shapes(_content_by_category(output), convert_lib.EVIDENCE_MARKER)
|
|
|
|
|
|
def test_task_context_path_is_confined(tmp_path: Path, convert_lib) -> None:
|
|
workspace = tmp_path / "user-data" / "workspace"
|
|
(workspace.parent / "outputs").mkdir(parents=True)
|
|
outside = tmp_path / "task-context.json"
|
|
outside.write_text('{"taskId":"OUTSIDE"}', encoding="utf-8")
|
|
|
|
with pytest.raises(convert_lib.PathConfinementError):
|
|
convert_lib.load_task_context(outside, cwd=workspace)
|
|
|
|
|
|
def test_random_workspace_directory_is_not_a_trusted_user_data_root(
|
|
tmp_path: Path,
|
|
convert_lib,
|
|
) -> None:
|
|
fake_workspace = tmp_path / "workspace"
|
|
fake_workspace.mkdir()
|
|
|
|
assert convert_lib.discover_user_data_root(fake_workspace) is None
|
|
with pytest.raises(convert_lib.PathConfinementError):
|
|
convert_lib.resolve_sandbox_path("/mnt/user-data/outputs/task-reports.json", cwd=fake_workspace)
|
|
|
|
|
|
def convert_lib_env_categories() -> list[str]:
|
|
return ["信息环境 JSON", "人文环境 JSON"]
|