"""D11：/html/{filename} 报告代理路由。
解析链：RustFS 报告键（新产物裸文件）→ HTMLFile 遗留记录（JSON 包装）
→ 节点本地 html_output（存量兜底）→ 404；路径穿越一律 404。
"""
import json
from types import SimpleNamespace
import pytest
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from moldinsight.api import html_report_router
from moldinsight.storage.rustfs_storage import rustfs_manager
@pytest.fixture
def app():
application = FastAPI()
application.include_router(html_report_router.router)
return application
@pytest.fixture
async def client(app):
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as c:
yield c
@pytest.fixture(autouse=True)
def _isolate_external_sources(monkeypatch, tmp_path):
"""默认隔离：RustFS 未连接、本地兜底目录指向空临时目录，由各测试按需打开。"""
monkeypatch.setattr(rustfs_manager, "is_connected", False)
monkeypatch.setattr(html_report_router, "LOCAL_HTML_DIR", tmp_path / "html_output")
# ── 链路 1：RustFS 报告键（新产物） ──────────────────────────────
async def test_report_served_from_rustfs_report_key(client, monkeypatch):
async def fake_download_report(filename):
assert filename == "mold_demo_20260918.html"
return b"fresh"
monkeypatch.setattr(rustfs_manager, "is_connected", True)
monkeypatch.setattr(rustfs_manager, "download_report_artifact", fake_download_report)
resp = await client.get("/html/mold_demo_20260918.html")
assert resp.status_code == 200
assert resp.content == b"fresh"
assert "text/html" in resp.headers["content-type"]
# ── 链路 2：HTMLFile 遗留记录（html/{hash}.json JSON 包装） ──────
class _FakeScalars:
def __init__(self, record):
self._record = record
def first(self):
return self._record
class _FakeResult:
def __init__(self, record):
self._record = record
def scalars(self):
return _FakeScalars(self._record)
class _FakeSession:
def __init__(self, record):
self._record = record
async def __aenter__(self):
return self
async def __aexit__(self, *exc):
return False
async def execute(self, stmt):
return _FakeResult(self._record)
class _FakeDBManager:
def __init__(self, record):
self._record = record
def session(self):
return _FakeSession(self._record)
async def test_falls_back_to_legacy_json_wrapper(client, monkeypatch):
async def fake_download_report(filename):
raise RuntimeError("报告键未命中")
async def fake_download_file(file_type, object_key):
assert file_type == "html_files"
assert object_key == "html/abc123.json"
return json.dumps(
{"content": "legacy", "filename": "mold_old.html"}
).encode("utf-8")
record = SimpleNamespace(object_key="html/abc123.json", filename="mold_old.html")
monkeypatch.setattr(rustfs_manager, "is_connected", True)
monkeypatch.setattr(rustfs_manager, "download_report_artifact", fake_download_report)
monkeypatch.setattr(rustfs_manager, "download_file", fake_download_file)
monkeypatch.setattr(html_report_router, "db_manager", _FakeDBManager(record))
resp = await client.get("/html/mold_old.html")
assert resp.status_code == 200
assert resp.content == b"legacy"
async def test_legacy_record_with_report_key_returns_raw(client, monkeypatch):
"""新格式记录（键在报告前缀下）必须按裸文件返回，不能当 JSON 包装解析。"""
async def fake_download_report(filename):
raise RuntimeError("报告键直取瞬时失败")
async def fake_download_file(file_type, object_key):
assert object_key == "html/reports/mold_new.html"
return b"raw-by-report-key"
record = SimpleNamespace(object_key="html/reports/mold_new.html", filename="mold_new.html")
monkeypatch.setattr(rustfs_manager, "is_connected", True)
monkeypatch.setattr(rustfs_manager, "download_report_artifact", fake_download_report)
monkeypatch.setattr(rustfs_manager, "download_file", fake_download_file)
monkeypatch.setattr(html_report_router, "db_manager", _FakeDBManager(record))
resp = await client.get("/html/mold_new.html")
assert resp.status_code == 200
assert resp.content == b"raw-by-report-key"
# ── 链路 3：节点本地 html_output（存量兜底） ─────────────────────
async def test_falls_back_to_local_dir(client, tmp_path):
local_dir = tmp_path / "html_output"
local_dir.mkdir()
(local_dir / "mold_local.html").write_text("local", encoding="utf-8")
resp = await client.get("/html/mold_local.html")
assert resp.status_code == 200
assert resp.content == b"local"
async def test_json_media_type_from_extension(client, tmp_path):
local_dir = tmp_path / "html_output"
local_dir.mkdir()
(local_dir / "mold_a_data.json").write_bytes(b"{}")
resp = await client.get("/html/mold_a_data.json")
assert resp.status_code == 200
assert "application/json" in resp.headers["content-type"]
# ── 未命中与防护 ────────────────────────────────────────────────
async def test_404_when_all_sources_miss(client):
resp = await client.get("/html/missing.html")
assert resp.status_code == 404
async def test_path_traversal_rejected(client):
resp = await client.get("/html/%2e%2e/secret.txt")
assert resp.status_code == 404
async def test_hidden_filename_rejected(client):
resp = await client.get("/html/%2e%2eetc%2fpasswd")
assert resp.status_code == 404