mirror of
https://github.com/elder-plinius/OBLITERATUS.git
synced 2026-08-17 16:37:30 +02:00
461 lines
20 KiB
Python
461 lines
20 KiB
Python
"""Behavioral CLI boundaries without network, accelerators, or model downloads."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, Mock
|
|
|
|
import pytest
|
|
|
|
from obliteratus import cli
|
|
|
|
|
|
def ns(**values):
|
|
return SimpleNamespace(**values)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("argv", "target"),
|
|
[
|
|
(["gpu-calc", "--params", "1"], "_cmd_gpu_calc"),
|
|
(["run", "config.yml"], "_cmd_run"),
|
|
(["interactive"], "_cmd_interactive"),
|
|
(["models"], "_cmd_models"),
|
|
(["presets"], "_cmd_presets"),
|
|
(["info", "local/model"], "_cmd_info"),
|
|
(["strategies"], "_cmd_strategies"),
|
|
(["report", "results.json"], "_cmd_report"),
|
|
(["aggregate"], "_cmd_aggregate"),
|
|
(["ui"], "_cmd_ui"),
|
|
(["recommend", "local/model"], "_cmd_recommend"),
|
|
(["tourney", "local/model"], "_cmd_tourney"),
|
|
(["self-improve", "local/model", "--audit", "audit.json", "--output-dir", "out"], "_cmd_self_improve"),
|
|
(["abliterate", "local/model"], "_cmd_abliterate"),
|
|
],
|
|
)
|
|
def test_main_routes_local_commands(monkeypatch, argv, target):
|
|
command = Mock()
|
|
monkeypatch.setattr(cli, target, command)
|
|
cli.main(argv)
|
|
command.assert_called_once()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("argv", "target"),
|
|
[
|
|
(["run", "config.yml", "--remote", "gpu"], "_cmd_remote_run"),
|
|
(["tourney", "model", "--remote", "gpu"], "_cmd_remote_tourney"),
|
|
(["obliterate", "model", "--remote", "gpu"], "_cmd_remote_abliterate"),
|
|
],
|
|
)
|
|
def test_main_routes_remote_commands(monkeypatch, argv, target):
|
|
command = Mock()
|
|
monkeypatch.setattr(cli, target, command)
|
|
cli.main(argv)
|
|
command.assert_called_once()
|
|
|
|
|
|
def test_version_is_stable_and_does_not_dispatch(capsys):
|
|
from obliteratus import __version__
|
|
|
|
with pytest.raises(SystemExit) as exc:
|
|
cli.main(["--version"])
|
|
assert exc.value.code == 0
|
|
assert capsys.readouterr().out.endswith(f"obliteratus {__version__}\n")
|
|
|
|
|
|
def test_gpu_selection_contract(monkeypatch):
|
|
monkeypatch.delenv("CUDA_VISIBLE_DEVICES", raising=False)
|
|
cli._apply_gpu_selection(ns(gpus=None, remote=None))
|
|
cli._apply_gpu_selection(ns(gpus="all", remote=None))
|
|
cli._apply_gpu_selection(ns(gpus="9", remote="host"))
|
|
assert "CUDA_VISIBLE_DEVICES" not in os.environ
|
|
|
|
cli._apply_gpu_selection(ns(gpus=" 2, 0 ", remote=None))
|
|
assert os.environ["CUDA_VISIBLE_DEVICES"] == "2,0"
|
|
with pytest.raises(SystemExit) as exc:
|
|
cli._apply_gpu_selection(ns(gpus="gpu0", remote=None))
|
|
assert exc.value.code == 1
|
|
|
|
|
|
def test_ui_and_interactive_forwarding(monkeypatch):
|
|
import obliteratus.interactive
|
|
import obliteratus.local_ui
|
|
|
|
launch = Mock()
|
|
interactive = Mock()
|
|
monkeypatch.setattr(obliteratus.local_ui, "launch_local_ui", launch)
|
|
monkeypatch.setattr(obliteratus.interactive, "run_interactive", interactive)
|
|
cli._cmd_ui(ns(auth="user:pass", host="127.0.0.1", port=9000, share=True, no_browser=True, quiet=True))
|
|
launch.assert_called_once_with(
|
|
host="127.0.0.1",
|
|
port=9000,
|
|
share=True,
|
|
open_browser=False,
|
|
auth=("user", "pass"),
|
|
quiet=True,
|
|
)
|
|
cli._cmd_interactive()
|
|
interactive.assert_called_once_with()
|
|
|
|
|
|
def test_models_presets_and_strategies_render(monkeypatch):
|
|
import obliteratus.presets
|
|
import obliteratus.strategies
|
|
import obliteratus.study_presets
|
|
|
|
model = ns(
|
|
name="Tiny",
|
|
hf_id="local/tiny",
|
|
params="1B",
|
|
tier="tiny",
|
|
recommended_dtype="float32",
|
|
recommended_quantization=None,
|
|
description="fixture",
|
|
)
|
|
preset = ns(key="quick", name="Quick", strategies=[{"name": "heads"}], max_samples=2, description="fixture")
|
|
strategy = type("Strategy", (), {"__doc__": "First line.\nSecond line."})
|
|
monkeypatch.setattr(obliteratus.presets, "list_all_presets", lambda: [model])
|
|
monkeypatch.setattr(obliteratus.presets, "get_presets_by_tier", lambda _tier: [model])
|
|
monkeypatch.setattr(obliteratus.study_presets, "list_study_presets", lambda: [preset])
|
|
monkeypatch.setattr(obliteratus.strategies, "STRATEGY_REGISTRY", {"fixture": strategy})
|
|
console = Mock()
|
|
monkeypatch.setattr(cli, "console", console)
|
|
cli._cmd_models(ns(tier=None))
|
|
cli._cmd_models(ns(tier="tiny"))
|
|
cli._cmd_presets()
|
|
cli._cmd_strategies()
|
|
assert console.print.call_count >= 8
|
|
|
|
|
|
def test_run_local_overrides_and_remote_config(monkeypatch, tmp_path):
|
|
import obliteratus.config
|
|
import obliteratus.remote
|
|
import obliteratus.runner
|
|
|
|
config_path = tmp_path / "study.yml"
|
|
config_path.write_text("model: {name: fixture}\n")
|
|
config = ns(remote=None, output_dir="old")
|
|
monkeypatch.setattr(obliteratus.config.StudyConfig, "from_yaml", Mock(return_value=config))
|
|
monkeypatch.setattr(obliteratus.config.StudyConfig, "from_dict", Mock(return_value=config))
|
|
run = Mock()
|
|
monkeypatch.setattr(obliteratus.runner, "run_study", run)
|
|
cli._cmd_run(ns(config=str(config_path), preset="quick", output_dir="new"))
|
|
assert config.output_dir == "new"
|
|
run.assert_called_once_with(config)
|
|
|
|
config.remote = ns(
|
|
host="host", user="user", port=2200, ssh_key="key", remote_dir="/work",
|
|
python="python", sync_results=True, gpus="0",
|
|
)
|
|
runner = MagicMock()
|
|
runner.run_config.return_value = "/local/results"
|
|
monkeypatch.setattr(obliteratus.remote, "RemoteRunner", Mock(return_value=runner))
|
|
cli._cmd_run(ns(config=str(config_path), preset=None, output_dir=None))
|
|
runner.run_config.assert_called_once()
|
|
runner.run_config.return_value = None
|
|
with pytest.raises(SystemExit) as exc:
|
|
cli._cmd_run(ns(config=str(config_path), preset=None, output_dir=None))
|
|
assert exc.value.code == 1
|
|
|
|
|
|
def test_info_prints_stable_summary(monkeypatch):
|
|
import obliteratus.models.loader
|
|
|
|
handle = ns(summary=lambda: {"total_params": 1234, "architecture": "fixture"})
|
|
load = Mock(return_value=handle)
|
|
monkeypatch.setattr(obliteratus.models.loader, "load_model", load)
|
|
console = Mock()
|
|
monkeypatch.setattr(cli, "console", console)
|
|
cli._cmd_info(ns(model="local/model", task="causal_lm", device="cpu", dtype="float32"))
|
|
load.assert_called_once_with(model_name="local/model", task="causal_lm", device="cpu", dtype="float32")
|
|
assert console.print.call_count == 3
|
|
|
|
|
|
def test_report_success_and_plot_failure(monkeypatch, tmp_path):
|
|
import obliteratus.reporting.report
|
|
|
|
path = tmp_path / "results.json"
|
|
path.write_text(json.dumps({
|
|
"model_name": "fixture",
|
|
"baseline_metrics": {"score": 1.0},
|
|
"results": [{
|
|
"strategy": "heads", "component": "0", "description": "fixture",
|
|
"metrics": {"score": 0.5}, "metadata": {"source": "test"},
|
|
}],
|
|
}))
|
|
report = MagicMock()
|
|
monkeypatch.setattr(obliteratus.reporting.report, "AblationReport", Mock(return_value=report))
|
|
cli._cmd_report(ns(results_json=str(path), output_dir=str(tmp_path / "plots")))
|
|
report.add_baseline.assert_called_once_with({"score": 1.0})
|
|
report.add_result.assert_called_once()
|
|
report.plot_impact.side_effect = RuntimeError("no renderer")
|
|
cli._cmd_report(ns(results_json=str(path), output_dir=None))
|
|
|
|
|
|
def test_aggregate_formats_filters_and_empty_result(monkeypatch):
|
|
import obliteratus.community
|
|
|
|
records = [object()]
|
|
aggregated = {
|
|
"org/model": {
|
|
"keep": {"n_runs": 2, "refusal_rate": {"mean": 0.1}, "perplexity": {"mean": 4.5}},
|
|
"drop": {"n_runs": 1},
|
|
},
|
|
}
|
|
monkeypatch.setattr(obliteratus.community, "load_contributions", lambda _dir: records)
|
|
monkeypatch.setattr(obliteratus.community, "aggregate_results", lambda _records: json.loads(json.dumps(aggregated)))
|
|
latex = Mock(return_value="TABLE")
|
|
monkeypatch.setattr(obliteratus.community, "generate_latex_table", latex)
|
|
cli._cmd_aggregate(ns(dir="results", min_runs=2, format="summary", metric="refusal_rate"))
|
|
cli._cmd_aggregate(ns(dir="results", min_runs=1, format="latex", metric="score"))
|
|
latex.assert_called_once()
|
|
|
|
monkeypatch.setattr(obliteratus.community, "aggregate_results", lambda _records: {"model": {"one": {"n_runs": 1}}})
|
|
cli._cmd_aggregate(ns(dir="results", min_runs=3, format="summary", metric="score"))
|
|
|
|
|
|
def test_recommend_handles_config_fallback_telemetry_and_insights(monkeypatch):
|
|
import obliteratus.adaptive_defaults
|
|
import obliteratus.architecture_profiles
|
|
import transformers
|
|
|
|
profile = ns(
|
|
profile_label="Fixture", arch_class=ns(value="dense"), reasoning_class=ns(value="general"),
|
|
total_params_b=1.0, num_layers=2, hidden_size=8, recommended_method="advanced",
|
|
method_overrides={"n_directions": 2},
|
|
)
|
|
monkeypatch.setattr(transformers.AutoConfig, "from_pretrained", Mock(return_value=ns(num_hidden_layers=2, hidden_size=8)))
|
|
detect = Mock(return_value=profile)
|
|
monkeypatch.setattr(obliteratus.architecture_profiles, "detect_architecture", detect)
|
|
monkeypatch.setattr(obliteratus.architecture_profiles, "enhance_profile_with_telemetry", lambda p: (p, {"method": "advanced"}))
|
|
monkeypatch.setattr(obliteratus.adaptive_defaults, "format_recommendation", lambda _rec: "recommendation")
|
|
monkeypatch.setattr(obliteratus.adaptive_defaults, "get_global_insights", lambda: {
|
|
"total_records": 3,
|
|
"overall_best_methods": [{"method": "advanced", "mean_score": 0.8, "n_runs": 3}],
|
|
"architecture_breakdown": {"dense": {"best_method": "advanced", "best_score": 0.8, "n_methods_tested": 2, "total_runs": 3}},
|
|
})
|
|
cli._cmd_recommend(ns(model="local/model", insights=True))
|
|
detect.assert_called_once()
|
|
|
|
transformers.AutoConfig.from_pretrained.side_effect = OSError("offline")
|
|
monkeypatch.setattr(obliteratus.architecture_profiles, "enhance_profile_with_telemetry", lambda p: (p, None))
|
|
cli._cmd_recommend(ns(model="local/model", insights=False))
|
|
|
|
|
|
def test_tourney_callbacks_and_winner(monkeypatch):
|
|
import obliteratus.tourney
|
|
|
|
result = ns(
|
|
winner=ns(method="advanced", score=0.9, metrics={"refusal_rate": 0.1, "coherence": 0.8}),
|
|
hub_repo="org/winner",
|
|
)
|
|
runner = MagicMock()
|
|
runner.run.return_value = result
|
|
factory = Mock(return_value=runner)
|
|
monkeypatch.setattr(obliteratus.tourney, "TourneyRunner", factory)
|
|
cli._cmd_tourney(ns(
|
|
model="model", hub_org="org", hub_repo=None, device="cpu", dtype="float32",
|
|
dataset="builtin", quantization=None, methods=["advanced"], output_dir="out",
|
|
))
|
|
callbacks = factory.call_args.kwargs
|
|
callbacks["on_log"]("message")
|
|
callbacks["on_round"](ns(round_num=1, advanced_to=[1], eliminated=[2]))
|
|
|
|
|
|
def test_gpu_calculator_validation_profile_and_moe(monkeypatch):
|
|
import obliteratus.model_profile
|
|
|
|
console = Mock()
|
|
monkeypatch.setattr(cli, "console", console)
|
|
cli._cmd_gpu_calc(ns(model=None, params=10.0, active_params=2.0, dtype="float16", gpu_mem=24.0))
|
|
profile = ns(total_params_b=7.0, active_params_b=3.0)
|
|
monkeypatch.setattr(obliteratus.model_profile, "profile_model", Mock(return_value=profile))
|
|
cli._cmd_gpu_calc(ns(model="local/model", params=None, active_params=None, dtype="int8", gpu_mem=16.0))
|
|
|
|
with pytest.raises(SystemExit):
|
|
cli._cmd_gpu_calc(ns(model=None, params=None, active_params=None, dtype="float16", gpu_mem=16.0))
|
|
with pytest.raises(SystemExit):
|
|
cli._cmd_gpu_calc(ns(model=None, params=1.0, active_params=None, dtype="float16", gpu_mem=1.0))
|
|
monkeypatch.setattr(obliteratus.model_profile, "profile_model", Mock(side_effect=OSError("offline")))
|
|
with pytest.raises(SystemExit):
|
|
cli._cmd_gpu_calc(ns(model="x", params=None, active_params=None, dtype="float16", gpu_mem=16.0))
|
|
|
|
|
|
def test_parameter_estimators_cover_dense_moe_and_invalid():
|
|
assert cli._estimate_total_params_b(ns(num_parameters=2_000_000_000)) == 2.0
|
|
dense = ns(hidden_size=128, num_hidden_layers=2, vocab_size=1000, intermediate_size=512)
|
|
total = cli._estimate_total_params_b(dense)
|
|
assert total > 0
|
|
assert cli._estimate_active_params_b(dense, total) == total
|
|
moe = ns(
|
|
hidden_size=4096, num_hidden_layers=32, vocab_size=100_000, intermediate_size=14_336,
|
|
moe_intermediate_size=2048, num_local_experts=8, num_experts_per_tok=2,
|
|
)
|
|
moe_total = cli._estimate_total_params_b(moe)
|
|
assert 0.1 <= cli._estimate_active_params_b(moe, moe_total) < moe_total
|
|
with pytest.raises(SystemExit):
|
|
cli._estimate_total_params_b(ns(hidden_size=0, num_hidden_layers=0, vocab_size=0))
|
|
|
|
|
|
def _remote_args(**overrides):
|
|
values = {
|
|
"remote": "user@host", "ssh_port": 22, "ssh_key": None, "remote_dir": "/work",
|
|
"remote_python": "python3", "no_sync": False, "gpus": "0", "model": "model",
|
|
"output_dir": "out", "method": "advanced", "device": "cuda", "dtype": "float16",
|
|
"quantization": "4bit", "n_directions": 2, "direction_method": "svd",
|
|
"regularization": 0.2, "refinement_passes": 2, "min_layer_fraction": 0.1,
|
|
"max_layer_fraction": 0.9, "harmless_pc_count": 1, "shield_concept_count": 2,
|
|
"shield_ridge": 0.1, "shield_residualize": True, "shield_layer_penalty": 0.2,
|
|
"projection_target": "all", "projection_row_fraction": 0.5, "large_model": True,
|
|
"verify_sample_size": 5, "config": "config.yml", "preset": "quick", "methods": ["advanced"],
|
|
"hub_org": "org", "hub_repo": None, "dataset": "builtin",
|
|
}
|
|
values.update(overrides)
|
|
return ns(**values)
|
|
|
|
|
|
def test_remote_runner_factory_and_commands(monkeypatch):
|
|
import obliteratus.remote
|
|
|
|
config = object()
|
|
runner = MagicMock()
|
|
monkeypatch.setattr(obliteratus.remote.RemoteConfig, "from_cli_args", Mock(return_value=config))
|
|
monkeypatch.setattr(obliteratus.remote, "RemoteRunner", Mock(return_value=runner))
|
|
args = _remote_args()
|
|
assert cli._make_remote_runner(args) is runner
|
|
|
|
monkeypatch.setattr(cli, "_make_remote_runner", lambda _args: runner)
|
|
runner.run_obliterate.return_value = "results"
|
|
cli._cmd_remote_abliterate(args)
|
|
assert runner.run_obliterate.call_args.kwargs["projection_row_fraction"] == 0.5
|
|
runner.run_config.return_value = "results"
|
|
cli._cmd_remote_run(args)
|
|
runner.run_tourney.return_value = "results"
|
|
cli._cmd_remote_tourney(args)
|
|
|
|
runner.run_obliterate.return_value = None
|
|
with pytest.raises(SystemExit):
|
|
cli._cmd_remote_abliterate(args)
|
|
runner.run_config.return_value = None
|
|
with pytest.raises(SystemExit):
|
|
cli._cmd_remote_run(args)
|
|
runner.run_tourney.return_value = None
|
|
with pytest.raises(SystemExit):
|
|
cli._cmd_remote_tourney(args)
|
|
|
|
|
|
def test_abliterate_pipeline_callbacks_residue_and_contribution(monkeypatch, tmp_path):
|
|
import obliteratus.abliterate
|
|
import obliteratus.community
|
|
import obliteratus.hard_negative
|
|
import obliteratus.telemetry
|
|
import rich.live
|
|
|
|
stages = [ns(key=f"s{i}", name=f"Stage {i}") for i in range(6)]
|
|
monkeypatch.setattr(obliteratus.abliterate, "STAGES", stages)
|
|
monkeypatch.setattr(obliteratus.abliterate, "METHODS", {"advanced": {"label": "Advanced"}})
|
|
monkeypatch.setattr(
|
|
obliteratus.hard_negative,
|
|
"build_weighted_prompt_pairs",
|
|
lambda **_kwargs: (["harm"], ["safe"], {"residue_examples": 1, "residue_added_pairs": 2, "total_pairs": 3}),
|
|
)
|
|
monkeypatch.setattr(obliteratus.community, "save_contribution", lambda *_args, **_kwargs: "contribution.json")
|
|
telemetry = Mock()
|
|
monkeypatch.setattr(obliteratus.telemetry, "maybe_send_pipeline_report", telemetry)
|
|
|
|
result_path = tmp_path / "result"
|
|
result_path.mkdir()
|
|
pipeline = MagicMock()
|
|
|
|
def run():
|
|
kwargs = factory.call_args.kwargs
|
|
kwargs["on_log"]("working")
|
|
kwargs["on_stage"](ns(stage="s0", status="running", message="work"))
|
|
kwargs["on_stage"](ns(stage="s0", status="done", message="done"))
|
|
return str(result_path)
|
|
|
|
pipeline.run.side_effect = run
|
|
factory = Mock(return_value=pipeline)
|
|
monkeypatch.setattr(obliteratus.abliterate, "AbliterationPipeline", factory)
|
|
|
|
class FakeLive:
|
|
def __init__(self, *_args, **_kwargs):
|
|
self.update = Mock()
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
monkeypatch.setattr(rich.live, "Live", FakeLive)
|
|
args = ns(
|
|
model="org/model", output_dir=None, device="cpu", dtype="float32", method="advanced",
|
|
n_directions=2, direction_method="svd", regularization=0.2, refinement_passes=1,
|
|
min_layer_fraction=0.1, max_layer_fraction=0.9, harmless_pc_count=1,
|
|
shield_concept_count=2, shield_ridge=0.1, shield_residualize=True,
|
|
shield_layer_penalty=0.2, projection_target="all", projection_row_fraction=0.5,
|
|
quantization=None, large_model=False, verify_sample_size=3,
|
|
residue_file=["audit.json"], dataset="builtin", residue_weight=2, residue_max=4,
|
|
contribute=True, contribute_notes="fixture",
|
|
)
|
|
cli._cmd_abliterate(args)
|
|
assert (result_path / "hard_negative_residue.json").is_file()
|
|
telemetry.assert_called_once_with(pipeline)
|
|
|
|
|
|
def test_self_improve_dry_run_and_pipeline(monkeypatch, tmp_path):
|
|
import obliteratus.abliterate
|
|
import obliteratus.hard_negative
|
|
import obliteratus.model_profile
|
|
|
|
profile = ns(
|
|
total_params_b=1.0, total_params=1_000_000_000, active_params_b=1.0,
|
|
source="fixture", num_layers=2, hidden_size=8,
|
|
to_json=lambda: {
|
|
"model_id": "model", "total_params": 1_000_000_000, "total_params_b": 1.0,
|
|
"active_params_b": 1.0, "num_layers": 2, "hidden_size": 8, "source": "fixture",
|
|
},
|
|
)
|
|
monkeypatch.setattr(obliteratus.model_profile, "profile_model", lambda *_args, **_kwargs: profile)
|
|
monkeypatch.setattr(obliteratus.model_profile, "default_self_improve_params", lambda _profile: {
|
|
"n_directions": 2, "regularization": 0.2, "refinement_passes": 1,
|
|
"residue_weight": 3, "verify_sample_size": 5, "note": "fixture",
|
|
})
|
|
monkeypatch.setattr(obliteratus.hard_negative, "load_residue_file", lambda _path: ["residue"])
|
|
def save_residue(_items, path):
|
|
path = Path(path)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("[]")
|
|
|
|
monkeypatch.setattr(obliteratus.hard_negative, "save_residue_file", save_residue)
|
|
monkeypatch.setattr(obliteratus.hard_negative, "build_weighted_prompt_pairs", lambda **_kwargs: (
|
|
["harm"], ["safe"], {"residue_examples": 1, "residue_added_pairs": 3, "total_pairs": 4},
|
|
))
|
|
output = tmp_path / "out"
|
|
args = ns(
|
|
model="model", dtype="float32", params_b=None, no_param_auto_scale=False,
|
|
n_directions=None, regularization=None, refinement_passes=None, residue_weight=None,
|
|
verify_sample_size=None, audit=["audit.json"], residue_out=None, output_dir=str(output),
|
|
dataset="builtin", residue_max=None, projection_row_fraction=None, method="advanced",
|
|
direction_method="svd", min_layer_fraction=None, max_layer_fraction=None,
|
|
harmless_pc_count=None, shield_concept_count=None, shield_ridge=None,
|
|
shield_residualize=None, shield_layer_penalty=None, projection_target=None,
|
|
device="cpu", dry_run=True,
|
|
)
|
|
cli._cmd_self_improve(args)
|
|
assert (output / "self_improve_plan.json").is_file()
|
|
|
|
result = tmp_path / "candidate"
|
|
result.mkdir()
|
|
pipeline = MagicMock()
|
|
pipeline.run.return_value = str(result)
|
|
monkeypatch.setattr(obliteratus.abliterate, "AbliterationPipeline", Mock(return_value=pipeline))
|
|
args.dry_run = False
|
|
cli._cmd_self_improve(args)
|
|
assert (result / "hard_negative_residue.json").is_file()
|