Spaces:
Running on Zero
Running on Zero
| """候选环境启动方式、缓存指纹和失败回退回归。""" | |
| from pathlib import Path | |
| from typing import Any | |
| import pytest | |
| from mineru_space import bootstrap | |
| def test_validate_uses_module_from_space_root(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: | |
| """预检作为包模块启动，使 spawn worker 能从任意调用目录导入适配源码。""" | |
| calls = [] | |
| def run(command: list[str], **kwargs: Any) -> None: | |
| """记录命令参数而不启动或安装环境。""" | |
| calls.append((command, kwargs)) | |
| monkeypatch.setattr(bootstrap, "run", run) | |
| bootstrap.validate(tmp_path, tmp_path / "venv", "commit") | |
| assert calls == [ | |
| ( | |
| [str(tmp_path / "venv/bin/python"), "-m", "mineru_space.runtime_preflight", "commit"], | |
| {"timeout": 90, "cwd": tmp_path}, | |
| ) | |
| ] | |
| def test_render_adapter_changes_invalidate_runtime(tmp_path: Path) -> None: | |
| """适配源码变化必须改变指纹，不能沿用此前只有签名预检的成功标记。""" | |
| root = Path(__file__).resolve().parents[1] | |
| target = tmp_path / "mineru_space" | |
| target.mkdir() | |
| for source in (root / "mineru_space").iterdir(): | |
| if source.is_file(): | |
| (target / source.name).write_bytes(source.read_bytes()) | |
| before = bootstrap.runtime_fingerprint(tmp_path) | |
| adapter = target / "space_runtime.py" | |
| adapter.write_bytes(adapter.read_bytes() + b"\n# updated adapter\n") | |
| assert bootstrap.runtime_fingerprint(tmp_path) != before | |
| def test_candidate_failure_revalidates_previous_environment(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: | |
| """新候选失败时先验证旧环境，只有仍兼容当前适配代码的旧环境可以继续启动。""" | |
| previous = tmp_path / "previous" | |
| state = {"commit": "previous-sha", "fingerprint": "old"} | |
| validations = [] | |
| def fail(*_args: object) -> None: | |
| """模拟候选渲染预检失败。""" | |
| raise RuntimeError("incompatible render interface") | |
| def validate(root: Path, environment: Path, commit: str) -> None: | |
| """记录回退环境重新验证的真实参数。""" | |
| validations.append((root, environment, commit)) | |
| monkeypatch.setattr(bootstrap, "active_environment", lambda cache: (previous, state)) | |
| monkeypatch.setattr(bootstrap, "dev_head", lambda: "new-sha") | |
| monkeypatch.setattr(bootstrap, "build_candidate", fail) | |
| monkeypatch.setattr(bootstrap, "validate", validate) | |
| root = Path(__file__).resolve().parents[1] | |
| assert bootstrap.select_environment(root, tmp_path) == previous | |
| assert validations == [(root, previous, "previous-sha")] | |
| monkeypatch.setattr(bootstrap, "validate", fail) | |
| with pytest.raises(RuntimeError, match="incompatible render interface"): | |
| bootstrap.select_environment(root, tmp_path) | |
| def test_gradio_preflight_builds_embedded_client_without_workers( | |
| monkeypatch: pytest.MonkeyPatch, | |
| ) -> None: | |
| """预检使用真实客户端和界面，但不能启动模型、工作进程或 HTTP 服务。""" | |
| import gradio as gr | |
| from mineru.model import download as model_download | |
| from mineru_space import runtime_preflight, space_app | |
| from mineru.kit import gradio as gradio_kit | |
| def forbidden(*_args: object, **_kwargs: object) -> None: | |
| """任何昂贵启动副作用都必须使该预检测试失败。""" | |
| raise AssertionError("preflight must not start runtime workers") | |
| monkeypatch.setattr(space_app, "sidecar", forbidden) | |
| monkeypatch.setattr(space_app, "preload_torch_models", forbidden) | |
| monkeypatch.setattr(space_app, "_GPU_TASK", forbidden) | |
| monkeypatch.setattr(model_download, "download_model_repo", forbidden) | |
| monkeypatch.setattr(gr.Blocks, "launch", forbidden) | |
| original = gradio_kit.build_gradio_app | |
| constructed = [] | |
| def build(client: Any, capabilities: Any, **kwargs: Any) -> Any: | |
| """记录真实能力声明，并禁止生成的界面启动 HTTP 监听。""" | |
| assert isinstance(client, space_app.InProcessArtifactClient) | |
| assert not hasattr(capabilities, "model_output") | |
| assert kwargs["enable_example"] is False | |
| assert kwargs["enable_api"] is True | |
| assert kwargs["max_pages"] == space_app.MAX_PAGES | |
| assert isinstance(kwargs["document_io"], space_app.SpaceDocumentIO) | |
| demo = original(client, capabilities, **kwargs) | |
| constructed.append((demo, kwargs["output_root"])) | |
| return demo | |
| monkeypatch.setattr(gradio_kit, "build_gradio_app", build) | |
| runtime_preflight.check_gradio_adapter() | |
| assert len(constructed) == 1 | |
| assert not constructed[0][1].exists() | |
| def test_gradio_preflight_rejects_incompatible_adapter(monkeypatch: pytest.MonkeyPatch, failure: str) -> None: | |
| """构造参数和界面签名不兼容必须传播到候选环境验证，不能跳过错误继续启动。""" | |
| from mineru_space import runtime_preflight, space_app | |
| from mineru.kit import gradio as gradio_kit | |
| def incompatible(*_args: object, **_kwargs: object) -> None: | |
| """模拟此次能力字段漂移或未来界面参数变化。""" | |
| raise TypeError("incompatible Gradio adapter") | |
| if failure == "client": | |
| monkeypatch.setattr(space_app, "InProcessArtifactClient", incompatible) | |
| elif failure == "interface": | |
| monkeypatch.setattr(gradio_kit, "build_gradio_app", incompatible) | |
| elif failure == "protocol": | |
| monkeypatch.delattr(gradio_kit, "GradioDocumentIO") | |
| else: | |
| def old_build( | |
| client: object, | |
| capabilities: object, | |
| *, | |
| output_root: Path, | |
| enable_example: bool, | |
| enable_api: bool, | |
| max_pages: int, | |
| ) -> None: | |
| """模拟尚无显式文档接口的旧版本，参数绑定必须先于函数体拒绝调用。""" | |
| raise AssertionError("old Gradio builder must not run") | |
| monkeypatch.setattr(gradio_kit, "build_gradio_app", old_build) | |
| error = ImportError if failure == "protocol" else TypeError | |
| message = { | |
| "client": "incompatible Gradio adapter", | |
| "interface": "incompatible Gradio adapter", | |
| "document_io": "document_io", | |
| "protocol": "GradioDocumentIO", | |
| }[failure] | |
| with pytest.raises(error, match=message): | |
| runtime_preflight.check_gradio_adapter() | |