Spaces:
Sleeping
Sleeping
| """CLI compatibility, lazy imports, routing, report output and cleanup contracts.""" | |
| import json | |
| from pathlib import Path | |
| import runpy | |
| import subprocess | |
| import sys | |
| from types import SimpleNamespace | |
| import pytest | |
| import torch | |
| ROOT = Path(__file__).resolve().parents[1] | |
| WRAPPERS = { | |
| "bol_weight_fingerprint.py": "fingerprint", | |
| "bol_silhouette.py": "silhouette", | |
| "bol_semantic_probe.py": "semantic", | |
| "bol_cka.py": "cka", | |
| "bol_attention_map.py": "attention", | |
| "bol_gradient_attribution.py": "gradient", | |
| "bol_layer_sweep.py": "layer_sweep", | |
| "bol_component_ablation.py": "component", | |
| "bol_trisect.py": "trisect", | |
| "bol_wormhole.py": "wormhole", | |
| "bol_compression_map.py": "noise", | |
| } | |
| def test_retained_scripts_import_and_show_help_without_heavy_dependencies(help_mode): | |
| # Blocking imports proves these entry points cannot reach checkpoint loading. | |
| script = """ | |
| import contextlib, importlib.abc, io, runpy, sys | |
| class BlockHeavyImports(importlib.abc.MetaPathFinder): | |
| def find_spec(self, fullname, path=None, target=None): | |
| if fullname.split('.')[0] in {'torch', 'transformers', 'huggingface_hub', 'gradio', 'spaces', 'pandas', 'plotly'}: | |
| raise RuntimeError('Unexpected heavyweight import: ' + fullname) | |
| sys.meta_path.insert(0, BlockHeavyImports()) | |
| """ | |
| script += f"paths = {list(WRAPPERS)!r}\nhelp_mode = {help_mode!r}\n" | |
| script += """ | |
| for path in paths: | |
| output = io.StringIO() | |
| sys.argv = [path, '--help'] if help_mode else [path] | |
| with contextlib.redirect_stdout(output): | |
| try: | |
| runpy.run_path(path, run_name='__main__' if help_mode else 'import_probe') | |
| except SystemExit as exc: | |
| assert help_mode and exc.code == 0 | |
| if help_mode: | |
| assert '--max-tokens' in output.getvalue() | |
| assert '--output-dir' in output.getvalue() | |
| else: | |
| assert not output.getvalue() | |
| """ | |
| result = subprocess.run([sys.executable, "-c", script], cwd=ROOT, capture_output=True, text=True) | |
| assert result.returncode == 0, result.stderr | |
| def cli_environment(monkeypatch, tmp_path): | |
| model = torch.nn.Linear(2, 2) | |
| model.weight.grad = torch.ones_like(model.weight) | |
| ticket = {"repo_id": "local/tiny", "revision": "abc123", "architecture": "tiny", | |
| "dtype": "bfloat16", "path": "/private/checkpoint-cache"} | |
| calls = [] | |
| def prepare_checkpoint(repo_id, revision="main", access_token=""): | |
| calls.append((repo_id, revision, access_token)) | |
| return dict(ticket, repo_id=repo_id) | |
| def analysis(engine): | |
| def run(model, tokenizer, tool, options): | |
| return {"title": f"{engine}:{tool}", "summary": f"Measured with {engine}", | |
| "rows": [{"loss": 1.2}], "plot": {}, "details": {}, "warnings": []} | |
| return run | |
| def create_report(result): | |
| directory = tmp_path / "generated" / result["run"]["id"] | |
| directory.mkdir(parents=True) | |
| json_path, html_path = directory / "experiment.json", directory / "experiment.html" | |
| json_path.write_text(json.dumps(result), encoding="utf-8") | |
| html_path.write_text(f"<html>{result['title']}</html>", encoding="utf-8") | |
| (directory / "experiment.txt").write_text(f"Printout: {result['title']}", encoding="utf-8") | |
| return str(json_path), str(html_path) | |
| monkeypatch.setitem(sys.modules, "bol_space.checkpoints", SimpleNamespace( | |
| DEFAULT_MODEL="HuggingFaceTB/SmolLM2-1.7B-Instruct", prepare_checkpoint=prepare_checkpoint, | |
| load_checkpoint=lambda ticket: (model, object()), | |
| )) | |
| monkeypatch.setitem(sys.modules, "bol_space.inspection", SimpleNamespace(run_inspection=analysis("inspection"))) | |
| monkeypatch.setitem(sys.modules, "bol_space.surgery", SimpleNamespace(run_surgery=analysis("surgery"))) | |
| monkeypatch.setitem(sys.modules, "bol_space.presentation", SimpleNamespace(create_report=create_report)) | |
| monkeypatch.setattr(torch.cuda, "is_available", lambda: False) | |
| monkeypatch.delenv("BOL_MODEL_TOKEN", raising=False) | |
| monkeypatch.delenv("HF_TOKEN", raising=False) | |
| return SimpleNamespace(model=model, calls=calls, output=tmp_path / "results") | |
| def test_cli_routes_tools_and_writes_complete_reports(tool, engine, cli_environment): | |
| from bol_space.cli import main | |
| env = cli_environment | |
| directory = main(tool, ["--model", "local/tiny", "--revision", "dev", | |
| "--max-tokens", "64", "--sample-layers", "3", | |
| "--output-dir", str(env.output)]) | |
| report = json.loads((directory / "experiment.json").read_text()) | |
| assert report["title"] == f"{engine}:{tool}" | |
| assert (directory / "experiment.html").read_text() == f"<html>{engine}:{tool}</html>" | |
| assert (directory / "experiment.txt").read_text() == f"Printout: {engine}:{tool}" | |
| assert not list((env.output.parent / "generated").iterdir()) | |
| assert report["model"] == {"repo_id": "local/tiny", "revision": "abc123", "architecture": "tiny", | |
| "dtype": "bfloat16", "parameters": 6} | |
| assert set(report["run"]) == {"id", "created_at", "tool", "seconds", "peak_vram_gb", "gpu", "options"} | |
| assert report["run"]["options"]["max_tokens"] == 64 | |
| assert report["run"]["options"]["sample_layers"] == 3 | |
| assert report["run"]["gpu"] == "CPU" | |
| assert report["run"]["seconds"] >= 0 | |
| assert all(parameter.grad is None for parameter in env.model.parameters()) | |
| assert env.calls == [("local/tiny", "dev", "")] | |
| def test_cli_uses_default_model_and_unique_run_folders(cli_environment): | |
| from bol_space.cli import main | |
| env = cli_environment | |
| first = main("fingerprint", ["--output-dir", str(env.output)]) | |
| second = main("fingerprint", ["--output-dir", str(env.output)]) | |
| assert first != second | |
| assert (first / "experiment.json").exists() | |
| assert (second / "experiment.json").exists() | |
| assert env.calls[0][0] == "HuggingFaceTB/SmolLM2-1.7B-Instruct" | |
| def test_cli_authentication_never_enters_reports_or_console(cli_environment, monkeypatch, capsys): | |
| from bol_space.cli import main | |
| marker = "synthetic-test-token-not-a-credential" | |
| monkeypatch.setenv("BOL_MODEL_TOKEN", marker) | |
| monkeypatch.setenv("HF_TOKEN", "unused-fallback") | |
| directory = main("fingerprint", ["--output-dir", str(cli_environment.output)]) | |
| assert cli_environment.calls[0][2] == marker | |
| assert marker not in (directory / "experiment.json").read_text() | |
| assert marker not in capsys.readouterr().out | |
| def test_cli_cleans_gradients_when_analysis_fails(cli_environment, monkeypatch): | |
| from bol_space.cli import main | |
| def fail(*args): | |
| raise RuntimeError("injected analysis failure") | |
| monkeypatch.setitem(sys.modules, "bol_space.inspection", SimpleNamespace(run_inspection=fail)) | |
| with pytest.raises(RuntimeError, match="injected analysis failure"): | |
| main("gradient", ["--output-dir", str(cli_environment.output)]) | |
| assert all(parameter.grad is None for parameter in cli_environment.model.parameters()) | |
| assert not cli_environment.output.exists() | |
| def test_invalid_options_fail_before_checkpoint_access(cli_environment): | |
| from bol_space.cli import main | |
| with pytest.raises(SystemExit): | |
| main("noise", ["--noise-scale", "nan"]) | |
| assert not cli_environment.calls | |
| def test_text_free_probes_accept_blank_text(cli_environment, tool): | |
| from bol_space.cli import main | |
| directory = main(tool, ["--text", "", "--output-dir", str(cli_environment.output)]) | |
| report = json.loads((directory / "experiment.json").read_text()) | |
| assert report["run"]["options"]["text"] == "" | |
| def test_text_experiments_reject_blank_text(cli_environment, tool): | |
| from bol_space.cli import main | |
| with pytest.raises(SystemExit): | |
| main(tool, ["--text", " ", "--output-dir", str(cli_environment.output)]) | |
| assert not cli_environment.calls | |
| def test_each_retained_filename_routes_to_its_corresponding_tool(monkeypatch): | |
| called = [] | |
| monkeypatch.setitem(sys.modules, "bol_space.cli", SimpleNamespace(main=called.append)) | |
| for filename, expected_tool in WRAPPERS.items(): | |
| runpy.run_path(str(ROOT / filename), run_name="__main__") | |
| assert called[-1] == expected_tool | |
| def test_compatibility_config_never_imports_checkpoint_libraries(): | |
| script = """ | |
| import importlib.abc, sys | |
| class BlockCheckpoints(importlib.abc.MetaPathFinder): | |
| def find_spec(self, fullname, path=None, target=None): | |
| if fullname.split('.')[0] in {'transformers', 'huggingface_hub'}: | |
| raise RuntimeError('Configuration tried to import a checkpoint library') | |
| sys.meta_path.insert(0, BlockCheckpoints()) | |
| import bol_config | |
| assert bol_config.TOKENIZER == bol_config.MODEL | |
| assert str(bol_config.DTYPE) == 'torch.bfloat16' | |
| """ | |
| result = subprocess.run([sys.executable, "-c", script], cwd=ROOT, capture_output=True, text=True) | |
| assert result.returncode == 0, result.stderr | |
| assert not result.stdout | |