207 lines
9.1 KiB
Python
207 lines
9.1 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
from uuid import uuid4
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from core.web_previews import (
|
|
MAX_FILE_BYTES,
|
|
WebPreviewError,
|
|
inspect_preview_tree,
|
|
resolve_preview_root,
|
|
)
|
|
from tools.publish_web_preview import PublishWebPreviewTool
|
|
from web.routers.web_previews import (
|
|
_mint_preview_token,
|
|
_rewrite_css,
|
|
_rewrite_html,
|
|
register_web_preview_routes,
|
|
)
|
|
|
|
|
|
class WebPreviewPathTests(unittest.TestCase):
|
|
def test_static_tree_accepts_entry_and_counts_files(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
working = Path(tmp)
|
|
dist = working / "dist"
|
|
(dist / "assets").mkdir(parents=True)
|
|
(dist / "index.html").write_text("<script src='/assets/app.js'></script>", encoding="utf-8")
|
|
(dist / "assets" / "app.js").write_text("console.log('ok')", encoding="utf-8")
|
|
|
|
root, rel = resolve_preview_root(working, "dist")
|
|
entry, count, size = inspect_preview_tree(root, "index.html")
|
|
|
|
self.assertEqual(rel, "dist")
|
|
self.assertEqual(entry, "index.html")
|
|
self.assertEqual(count, 2)
|
|
self.assertGreater(size, 0)
|
|
|
|
def test_directory_escape_and_symlink_are_rejected(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
working = Path(tmp) / "task"
|
|
working.mkdir()
|
|
with self.assertRaises(WebPreviewError):
|
|
resolve_preview_root(working, "../outside")
|
|
dist = working / "dist"
|
|
dist.mkdir()
|
|
(dist / "index.html").write_text("ok", encoding="utf-8")
|
|
try:
|
|
(dist / "outside-link").symlink_to(Path(tmp))
|
|
except OSError:
|
|
self.skipTest("symlinks are unavailable on this Windows test host")
|
|
with self.assertRaisesRegex(WebPreviewError, "symbolic links"):
|
|
inspect_preview_tree(dist, "index.html")
|
|
|
|
def test_large_file_is_rejected_without_reading_it(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / "index.html").write_text("ok", encoding="utf-8")
|
|
large = root / "large.bin"
|
|
with large.open("wb") as handle:
|
|
handle.truncate(MAX_FILE_BYTES + 1)
|
|
with self.assertRaisesRegex(WebPreviewError, "50 MiB"):
|
|
inspect_preview_tree(root, "index.html")
|
|
|
|
def test_root_urls_are_scoped_to_signed_preview_prefix(self) -> None:
|
|
prefix = "/web-preview/pid/token"
|
|
html = '<html><head></head><body><script src="/assets/app.js"></script></body></html>'
|
|
rewritten = _rewrite_html(html, prefix)
|
|
self.assertIn(f'<base href="{prefix}/">', rewritten)
|
|
self.assertIn(f'src="{prefix}/assets/app.js"', rewritten)
|
|
self.assertIn("https://", _rewrite_html('<a href="https://example.com">x</a>', prefix))
|
|
self.assertIn(f"url({prefix}/img/a.png)", _rewrite_css("a{background:url(/img/a.png)}", prefix))
|
|
|
|
|
|
class PublishWebPreviewToolTests(unittest.TestCase):
|
|
def test_tool_returns_persistable_preview_marker(self) -> None:
|
|
preview_id = uuid4()
|
|
preview = {"type": "web_preview", "preview_id": str(preview_id), "status": "active"}
|
|
with tempfile.TemporaryDirectory() as tmp, patch(
|
|
"tools.publish_web_preview.create_web_preview", return_value=preview
|
|
) as create:
|
|
tool = PublishWebPreviewTool(
|
|
uuid4(), uuid4(), working_dir=Path(tmp), base_dir=Path(tmp), user_root=Path(tmp)
|
|
)
|
|
result = tool.execute("dist", name="演示项目")
|
|
self.assertTrue(result.startswith("[WebPreview] "))
|
|
self.assertEqual(json.loads(result.removeprefix("[WebPreview] "))["preview_id"], str(preview_id))
|
|
self.assertEqual(create.call_args.kwargs["directory"], "dist")
|
|
|
|
def test_registry_exposes_preview_only_to_interactive_runs(self) -> None:
|
|
from core.tool_registry import ToolContext, build_tools
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
common = {
|
|
"tool_base": root,
|
|
"ur_path": root,
|
|
"working_dir_path": root,
|
|
"task_id": uuid4(),
|
|
"uid": uuid4(),
|
|
"cfg": {},
|
|
"caps": SimpleNamespace(enable_run_python=False),
|
|
"skills": SimpleNamespace(skills={}),
|
|
"cancel_check": None,
|
|
"deferred_actions": SimpleNamespace(),
|
|
"ark_cfg": None,
|
|
"img_provider": "",
|
|
"img_key": "",
|
|
"img_cfg": None,
|
|
"img_provider_cfg": None,
|
|
"video_variant": "",
|
|
"office_to_pdf_available": False,
|
|
}
|
|
with (
|
|
patch.dict(os.environ, {"DOCUMENT_SEARCH_API_KEY": "", "MP_API_KEY": ""}),
|
|
patch("core.tool_registry._external_system_status_available", return_value=False),
|
|
patch("core.tool_registry._external_systems_available", return_value=False),
|
|
patch("core.tool_registry.smtp_configured", return_value=False),
|
|
patch("core.tool_registry.wechat_push_available", return_value=False),
|
|
patch("core.tool_registry.lfasr_configured", return_value=False),
|
|
patch("core.tool_registry.BochaConfig.load", return_value=None),
|
|
):
|
|
interactive = build_tools(ToolContext(scheduled_run=False, **common))
|
|
scheduled = build_tools(ToolContext(scheduled_run=True, **common))
|
|
self.assertIn("publish_web_preview", interactive)
|
|
self.assertNotIn("publish_web_preview", scheduled)
|
|
|
|
|
|
class WebPreviewRouteTests(unittest.TestCase):
|
|
def test_signed_gateway_serves_rewritten_html_and_assets(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
working = Path(tmp)
|
|
dist = working / "dist"
|
|
(dist / "assets").mkdir(parents=True)
|
|
(dist / "index.html").write_text(
|
|
'<!doctype html><script type="module" src="/assets/app.js"></script>',
|
|
encoding="utf-8",
|
|
)
|
|
(dist / "assets" / "app.js").write_text("export default 1", encoding="utf-8")
|
|
preview_id = uuid4()
|
|
row = SimpleNamespace(
|
|
preview_id=preview_id,
|
|
user_id=uuid4(),
|
|
task_id=uuid4(),
|
|
root_path="dist",
|
|
entry_path="index.html",
|
|
name="测试项目",
|
|
spa_fallback=True,
|
|
status="active",
|
|
file_count=2,
|
|
size_bytes=100,
|
|
expires_at=datetime.now(timezone.utc) + timedelta(hours=1),
|
|
)
|
|
cfg = SimpleNamespace(jwt_secret="preview-test-secret-at-least-32-bytes")
|
|
token = _mint_preview_token(cfg, preview_id, row.expires_at)
|
|
app = FastAPI()
|
|
register_web_preview_routes(app, require_user=lambda: row.user_id, auth_cfg=cfg)
|
|
client = TestClient(app)
|
|
with (
|
|
patch("web.routers.web_previews.get_web_preview", return_value=(row, "ignored")),
|
|
patch("web.routers.web_previews.from_db_path", return_value=working),
|
|
):
|
|
response = client.get(f"/web-preview/{preview_id}/{token}/")
|
|
asset = client.get(f"/web-preview/{preview_id}/{token}/assets/app.js")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn(f"/web-preview/{preview_id}/{token}/assets/app.js", response.text)
|
|
self.assertIn("sandbox", response.headers["content-security-policy"])
|
|
self.assertNotIn("allow-same-origin", response.headers["content-security-policy"])
|
|
self.assertEqual(response.headers["referrer-policy"], "no-referrer")
|
|
self.assertEqual(asset.status_code, 200)
|
|
self.assertIn("javascript", asset.headers["content-type"])
|
|
|
|
def test_invalid_preview_token_is_not_distinguishable_from_missing(self) -> None:
|
|
app = FastAPI()
|
|
cfg = SimpleNamespace(jwt_secret="preview-test-secret-at-least-32-bytes")
|
|
register_web_preview_routes(app, require_user=lambda: uuid4(), auth_cfg=cfg)
|
|
response = TestClient(app).get(f"/web-preview/{uuid4()}/not-a-token/")
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
|
|
class WebPreviewMigrationTests(unittest.TestCase):
|
|
def test_0036_adds_preview_table_and_indexes(self) -> None:
|
|
module = importlib.import_module(
|
|
"db.migrations.versions.20260821_0900_0036_web_previews"
|
|
)
|
|
self.assertEqual(module.revision, "0036")
|
|
self.assertEqual(module.down_revision, "0035")
|
|
source = Path(module.__file__).read_text(encoding="utf-8")
|
|
self.assertIn('"web_previews"', source)
|
|
self.assertIn("ix_web_previews_user_status", source)
|
|
self.assertIn("ondelete=\"CASCADE\"", source)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|