260 lines
10 KiB
Python
260 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import zipfile
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from app.gateway.routers import artifacts
|
|
from deerflow.html_runtime import offline_bundle
|
|
from deerflow.html_runtime.offline_bundle import build_html_offline_bundle
|
|
from deerflow.uploads.policy import default_upload_policy, set_runtime_upload_policy
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _restore_upload_policy():
|
|
set_runtime_upload_policy(default_upload_policy())
|
|
yield
|
|
set_runtime_upload_policy(default_upload_policy())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_artifact_preview_reads_small_text_without_unbounded_artifact_response(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
source = tmp_path / "notes.txt"
|
|
source.write_text("第一行\n第二行", encoding="utf-8")
|
|
|
|
async def resolve(_thread_id: str, _path: str) -> Path:
|
|
return source
|
|
|
|
monkeypatch.setattr(artifacts, "aresolve_thread_virtual_path", resolve)
|
|
response = await artifacts.preview_artifact.__wrapped__(
|
|
thread_id="thread-1",
|
|
request=SimpleNamespace(),
|
|
path="mnt/user-data/uploads/notes.txt",
|
|
max_chars=1024,
|
|
)
|
|
|
|
assert response.content.splitlines() == ["第一行", "第二行"]
|
|
assert response.truncated is False
|
|
assert response.strategy == "complete-text"
|
|
assert response.total_bytes == source.stat().st_size
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_artifact_preview_samples_large_text_with_full_file_statistics(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
source = tmp_path / "words.txt"
|
|
source.write_text("".join(f"word-{index:04d}\n" for index in range(1, 1001)), encoding="utf-8")
|
|
policy = default_upload_policy()
|
|
policy["large_file"] = {
|
|
"threshold_bytes": 100,
|
|
"max_result_chars": 4096,
|
|
"hard_max_result_chars": 4096,
|
|
}
|
|
set_runtime_upload_policy(policy)
|
|
|
|
async def resolve(_thread_id: str, _path: str) -> Path:
|
|
return source
|
|
|
|
monkeypatch.setattr(artifacts, "aresolve_thread_virtual_path", resolve)
|
|
response = await artifacts.preview_artifact.__wrapped__(
|
|
thread_id="thread-1",
|
|
request=SimpleNamespace(),
|
|
path="mnt/user-data/uploads/words.txt",
|
|
max_chars=4096,
|
|
)
|
|
|
|
assert response.truncated is True
|
|
assert response.stats is not None
|
|
assert response.stats["lines"] == 1000
|
|
assert "流式扫描" in response.content
|
|
assert "word-0001" in response.content
|
|
assert len(response.content) <= 4096
|
|
assert "sampl" in response.strategy
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_artifact_preview_rejects_binary_content(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
source = tmp_path / "payload.bin"
|
|
source.write_bytes(b"abc\x00def")
|
|
|
|
async def resolve(_thread_id: str, _path: str) -> Path:
|
|
return source
|
|
|
|
monkeypatch.setattr(artifacts, "aresolve_thread_virtual_path", resolve)
|
|
with pytest.raises(artifacts.HTTPException) as raised:
|
|
await artifacts.preview_artifact.__wrapped__(
|
|
thread_id="thread-1",
|
|
request=SimpleNamespace(),
|
|
path="mnt/user-data/uploads/payload.bin",
|
|
max_chars=1024,
|
|
)
|
|
assert raised.value.status_code == 415
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_html_artifact_offline_bundle_uses_zip_export(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
source = tmp_path / "page.html"
|
|
source.write_text("<!doctype html><title>portable</title>", encoding="utf-8")
|
|
calls: list[Path] = []
|
|
|
|
async def resolve(_thread_id: str, _path: str) -> Path:
|
|
return source
|
|
|
|
async def bundle_response(path: Path):
|
|
calls.append(path)
|
|
return artifacts.Response(content=b"zip-content", media_type="application/zip")
|
|
|
|
monkeypatch.setattr(artifacts, "aresolve_thread_virtual_path", resolve)
|
|
monkeypatch.setattr(artifacts, "build_html_offline_bundle_response", bundle_response)
|
|
response = await artifacts.get_artifact.__wrapped__(
|
|
thread_id="thread-1",
|
|
path="mnt/user-data/outputs/page.html",
|
|
request=SimpleNamespace(),
|
|
download=True,
|
|
offline_bundle=True,
|
|
)
|
|
|
|
assert response.media_type == "application/zip"
|
|
assert calls == [source]
|
|
|
|
|
|
def _write_offline_fixture(path: Path, content: str | bytes) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
if isinstance(content, bytes):
|
|
path.write_bytes(content)
|
|
else:
|
|
path.write_text(content, encoding="utf-8")
|
|
|
|
|
|
def _offline_runtime_fixture(root: Path) -> Path:
|
|
runtime = root / "runtime"
|
|
_write_offline_fixture(runtime / "echarts.min.js", "window.echarts={};")
|
|
_write_offline_fixture(runtime / "deerflow-runtime.js", "window.__runtimeLoaded=true;")
|
|
_write_offline_fixture(runtime / "brand-logo-dark.png", b"fallback-logo")
|
|
_write_offline_fixture(runtime / "LICENSE.echarts.txt", "ECharts license")
|
|
for name in (
|
|
"china-provinces.geojson",
|
|
"japan-prefectures.geojson",
|
|
"taiwan-counties.geojson",
|
|
"us-states.geojson",
|
|
):
|
|
_write_offline_fixture(runtime / "maps" / name, '{"type":"FeatureCollection","features":[]}')
|
|
return runtime
|
|
|
|
|
|
def test_html_offline_bundle_rewrites_and_packages_gateway_runtime(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
runtime = _offline_runtime_fixture(tmp_path)
|
|
logo = tmp_path / "branding" / "logo.webp"
|
|
_write_offline_fixture(logo, b"configured-logo")
|
|
uploaded_image = tmp_path / "uploads" / "background.png"
|
|
_write_offline_fixture(uploaded_image, b"uploaded-image")
|
|
monkeypatch.setattr(
|
|
offline_bundle,
|
|
"verify_html_image_token",
|
|
lambda _token: ("thread-1", "uploads", "background.png"),
|
|
)
|
|
monkeypatch.setattr(
|
|
offline_bundle,
|
|
"resolve_html_image_path",
|
|
lambda _thread_id, _kind, _filename: uploaded_image,
|
|
)
|
|
source = tmp_path / "态势报告.html"
|
|
source.write_text(
|
|
"""<!doctype html><html><head>
|
|
<script src="http://gateway.local/prefix/api/public/html-runtime/echarts.min.js"></script>
|
|
<script src="http://gateway.local/prefix/api/public/html-runtime/deerflow-runtime.js"></script>
|
|
<style>.brand{background:url('http://gateway.local/prefix/api/public/html-runtime/generated-page-logo?v=2')}</style>
|
|
</head><body><script type="application/json" data-df-geo="map">{"map":"china-provinces"}</script>
|
|
<script type="application/json" data-df-map="tile-map">{"markers":[]}</script>
|
|
<img src="http://gateway.local/prefix/api/public/html-images/signed.token">
|
|
<a href="https://example.com/source">来源</a></body></html>""",
|
|
encoding="utf-8",
|
|
)
|
|
destination = tmp_path / "bundle.zip"
|
|
|
|
result = build_html_offline_bundle(
|
|
source,
|
|
destination,
|
|
runtime_base_url="http://gateway.local/prefix/api/public/html-runtime",
|
|
map_tile_url="https://tiles.example/{z}/{x}/{y}.png",
|
|
generated_logo_path=logo,
|
|
runtime_dir=runtime,
|
|
)
|
|
|
|
assert destination.is_file()
|
|
assert "assets/html-runtime/echarts.min.js" in result.bundled_assets
|
|
assert "https://example.com/source" in result.external_urls
|
|
assert "https://tiles.example/{z}/{x}/{y}.png" in result.external_urls
|
|
with zipfile.ZipFile(destination) as archive:
|
|
names = set(archive.namelist())
|
|
assert "index.html" in names
|
|
assert "README.txt" in names
|
|
assert "manifest.json" in names
|
|
assert "assets/html-runtime/echarts.min.js" in names
|
|
assert "assets/html-runtime/deerflow-runtime.js" in names
|
|
assert "assets/html-runtime/generated-page-logo.webp" in names
|
|
assert "assets/html-runtime/maps/china-provinces.geojson" in names
|
|
assert any(name.startswith("assets/html-images/") for name in names)
|
|
assert "THIRD_PARTY_LICENSES/LICENSE.echarts.txt" in names
|
|
|
|
html = archive.read("index.html").decode("utf-8")
|
|
assert "gateway.local" not in html
|
|
assert 'src="assets/html-runtime/echarts.min.js"' in html
|
|
assert "assets/html-runtime/generated-page-logo.webp" in html
|
|
assert "/api/public/html-images/" not in html
|
|
|
|
runtime_script = archive.read("assets/html-runtime/deerflow-runtime.js").decode("utf-8")
|
|
assert "DEERFLOW_HTML_RUNTIME_CONFIG" in runtime_script
|
|
assert "assets/html-runtime/maps/china-provinces.geojson" in runtime_script
|
|
assert "window.__runtimeLoaded=true" in runtime_script
|
|
|
|
manifest = json.loads(archive.read("manifest.json"))
|
|
assert manifest["entry"] == "index.html"
|
|
assert manifest["source_name"] == "态势报告.html"
|
|
assert manifest["external_urls"] == [
|
|
"https://example.com/source",
|
|
"https://tiles.example/{z}/{x}/{y}.png",
|
|
]
|
|
|
|
|
|
def test_html_offline_bundle_copies_cognitive_runtime_tree(tmp_path: Path) -> None:
|
|
runtime = _offline_runtime_fixture(tmp_path)
|
|
_write_offline_fixture(runtime / "cognitive-situation-map" / "app.js", "window.cognitive=true;")
|
|
_write_offline_fixture(runtime / "cognitive-situation-map" / "styles.css", "body{color:white}")
|
|
_write_offline_fixture(runtime / "cognitive-situation-map" / "assets" / "maps" / "taiwan.png", b"map")
|
|
source = tmp_path / "认知态势图.html"
|
|
source.write_text(
|
|
"""<script>window.__DF_SKILL_BASE__="http://gw/api/public/html-runtime/cognitive-situation-map";</script>
|
|
<link rel="stylesheet" href="http://gw/api/public/html-runtime/cognitive-situation-map/styles.css">
|
|
<script src="http://gw/api/public/html-runtime/cognitive-situation-map/app.js"></script>""",
|
|
encoding="utf-8",
|
|
)
|
|
destination = tmp_path / "cognitive.zip"
|
|
|
|
build_html_offline_bundle(
|
|
source,
|
|
destination,
|
|
runtime_base_url="http://gw/api/public/html-runtime",
|
|
map_tile_url="https://tiles.example/{z}/{x}/{y}.png",
|
|
runtime_dir=runtime,
|
|
)
|
|
|
|
with zipfile.ZipFile(destination) as archive:
|
|
html = archive.read("index.html").decode("utf-8")
|
|
assert "http://gw" not in html
|
|
assert "assets/html-runtime/cognitive-situation-map" in html
|
|
assert "assets/html-runtime/cognitive-situation-map/app.js" in archive.namelist()
|
|
assert "assets/html-runtime/cognitive-situation-map/assets/maps/taiwan.png" in archive.namelist()
|