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("", 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 = '' rewritten = _rewrite_html(html, prefix) self.assertIn(f'', rewritten) self.assertIn(f'src="{prefix}/assets/app.js"', rewritten) self.assertIn("https://", _rewrite_html('x', 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( '', 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()