"""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_tourney_default_output_uses_platform_temp_directory(monkeypatch, tmp_path): command = Mock() monkeypatch.setattr(cli, "_cmd_tourney", command) monkeypatch.setattr(cli.tempfile, "gettempdir", lambda: str(tmp_path)) cli.main(["tourney", "local/model"]) assert command.call_args.args[0].output_dir == str(tmp_path / "obliteratus_tourney") def test_refusal_max_tokens_cli_default_and_positive_override(monkeypatch): command = Mock() monkeypatch.setattr(cli, "_cmd_abliterate", command) cli.main(["abliterate", "local/model"]) assert command.call_args.args[0].refusal_max_tokens is None command.reset_mock() cli.main(["abliterate", "local/model", "--refusal-max-tokens", "512"]) assert command.call_args.args[0].refusal_max_tokens == 512 @pytest.mark.parametrize("invalid", ["0", "-1", "not-an-integer"]) def test_refusal_max_tokens_cli_rejects_invalid_values(invalid): with pytest.raises(SystemExit) as exc: cli.main(["abliterate", "local/model", "--refusal-max-tokens", invalid]) assert exc.value.code == 2 def test_gpu_memory_utilization_cli_default_and_explicit_override(monkeypatch): command = Mock() monkeypatch.setattr(cli, "_cmd_abliterate", command) cli.main(["obliterate", "local/model"]) assert command.call_args.args[0].gpu_memory_utilization is None command.reset_mock() cli.main(["obliterate", "local/model", "--gpu-memory-utilization", "0.95"]) assert command.call_args.args[0].gpu_memory_utilization == 0.95 @pytest.mark.parametrize("invalid", ["0", "-0.1", "1.1", "nan", "inf", "not-a-number"]) def test_gpu_memory_utilization_cli_rejects_invalid_values(invalid): with pytest.raises(SystemExit) as exc: cli.main(["obliterate", "local/model", "--gpu-memory-utilization", invalid]) assert exc.value.code == 2 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", known_hosts_file="known_hosts", install_timeout=120, python="python", sync_results=True, gpus="0", install_source="obliteratus==0.1.2", ) 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, "refusal_max_tokens": 512, "gpu_memory_utilization": 0.95, "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 obliteratus.remote.RemoteConfig.from_cli_args.assert_called_once_with( "user@host", port=22, ssh_key=None, remote_dir="/work", python="python3", sync_results=True, gpus="0", ) 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 assert runner.run_obliterate.call_args.kwargs["refusal_max_tokens"] == 512 assert runner.run_obliterate.call_args.kwargs["gpu_memory_utilization"] == 0.95 args.refusal_max_tokens = None args.gpu_memory_utilization = None runner.run_obliterate.reset_mock() cli._cmd_remote_abliterate(args) assert "refusal_max_tokens" not in runner.run_obliterate.call_args.kwargs assert "gpu_memory_utilization" not in runner.run_obliterate.call_args.kwargs for name in ( "quantization", "n_directions", "direction_method", "regularization", "refinement_passes", "min_layer_fraction", "max_layer_fraction", "harmless_pc_count", "shield_concept_count", "shield_ridge", "shield_residualize", "shield_layer_penalty", "projection_target", "projection_row_fraction", "verify_sample_size", "refusal_max_tokens", ): setattr(args, name, None) args.large_model = False runner.run_obliterate.reset_mock() cli._cmd_remote_abliterate(args) assert runner.run_obliterate.call_args.kwargs == { "model": "model", "local_output_dir": "out", "method": "advanced", "device": "cuda", "dtype": "float16", } 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) @pytest.mark.parametrize("port", ["0", "65536", "not-a-port"]) def test_remote_cli_rejects_invalid_ssh_ports_before_dispatch(port): with pytest.raises(SystemExit) as exc: cli.main(["run", "config.yml", "--remote", "host", "--ssh-port", port]) assert exc.value.code == 2 def test_remote_cli_accepts_valid_ssh_port_and_dispatches(monkeypatch): dispatch = Mock() monkeypatch.setattr(cli, "_cmd_remote_run", dispatch) cli.main(["run", "config.yml", "--remote", "host", "--ssh-port", "2222"]) assert dispatch.call_args.args[0].ssh_port == 2222 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, gpu_memory_utilization=0.95, large_model=False, verify_sample_size=3, refusal_max_tokens=512, residue_file=["audit.json"], dataset="builtin", residue_weight=2, residue_max=4, contribute=True, contribute_notes="fixture", ) cli._cmd_abliterate(args) assert factory.call_args.kwargs["refusal_max_tokens"] == 512 assert factory.call_args.kwargs["gpu_memory_utilization"] == 0.95 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()