mirror of
https://github.com/elder-plinius/OBLITERATUS.git
synced 2026-08-30 14:40:38 +02:00
feat: harden local runtime and verification
This commit is contained in:
@@ -18,6 +18,7 @@ from obliteratus.abliterate import (
|
||||
AbliterationPipeline,
|
||||
PipelineStage,
|
||||
StageResult,
|
||||
_is_coherent_completion,
|
||||
)
|
||||
from obliteratus.models.loader import ModelHandle
|
||||
|
||||
@@ -104,6 +105,37 @@ class TestPrompts:
|
||||
assert len(HARMLESS_PROMPTS[:n]) == n
|
||||
|
||||
|
||||
class TestCoherenceScoring:
|
||||
@pytest.mark.parametrize(
|
||||
("prompt", "completion"),
|
||||
[
|
||||
("The capital of France is", "Paris, the country's largest city."),
|
||||
("Water boils at a temperature of", "100 degrees Celsius at sea level."),
|
||||
("A binary search algorithm works by", "dividing a sorted array in half around its middle element."),
|
||||
],
|
||||
)
|
||||
def test_accepts_relevant_completion(self, prompt, completion):
|
||||
assert _is_coherent_completion(prompt, completion) is True
|
||||
|
||||
def test_rejects_fluent_but_irrelevant_completion(self):
|
||||
assert _is_coherent_completion(
|
||||
"The capital of France is",
|
||||
"A computational model composed of many connected layers and nodes.",
|
||||
) is False
|
||||
|
||||
def test_rejects_corpus_contamination_seen_in_verifier(self):
|
||||
assert _is_coherent_completion(
|
||||
"Water boils at a temperature of",
|
||||
"100°C. What is the temperature in Fahrenheit? - Brainly.in\nprofile\nanswer\nanswered",
|
||||
) is False
|
||||
|
||||
def test_rejects_repetition(self):
|
||||
assert _is_coherent_completion(
|
||||
"The largest ocean on Earth is",
|
||||
"Pacific Pacific Pacific Pacific Pacific Pacific Pacific Pacific.",
|
||||
) is False
|
||||
|
||||
|
||||
class TestStages:
|
||||
def test_six_stages(self):
|
||||
assert len(STAGES) == 6
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
"""Regression test for unload, cleanup, and lazy chat model lifecycle."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
|
||||
def test_active_checkpoint_survives_cleanup_and_remains_chat_recoverable(tmp_path):
|
||||
"""Exercise app state in isolation from Gradio's import-time worker sockets."""
|
||||
script = r'''
|
||||
import pathlib
|
||||
import sys
|
||||
|
||||
import app
|
||||
|
||||
theme = app.THEME.to_dict()["theme"]
|
||||
assert theme["body_background_fill"] != theme["body_background_fill_dark"]
|
||||
assert theme["body_text_color"] != theme["body_text_color_dark"]
|
||||
assert theme["background_fill_secondary"] == "#ffffff"
|
||||
assert theme["background_fill_secondary_dark"] == "#0d0d14"
|
||||
assert ".chatbot .message.bot" in app.CSS
|
||||
assert "color: var(--body-text-color) !important" in app.CSS
|
||||
|
||||
root = pathlib.Path(sys.argv[1])
|
||||
active = root / "obliterated_1"
|
||||
stale = root / "obliterated_2"
|
||||
cache = root / "model-cache"
|
||||
for directory in (active, stale, cache):
|
||||
directory.mkdir()
|
||||
(directory / "weights.bin").write_bytes(b"model")
|
||||
|
||||
app.dev.free_gpu_memory = lambda: None
|
||||
app._state.update({
|
||||
"model": object(), "tokenizer": object(), "model_name": "org/model",
|
||||
"method": "advanced", "status": "ready", "output_dir": str(active),
|
||||
})
|
||||
app._session_models.clear()
|
||||
app._session_models.update({
|
||||
"active": {"output_dir": str(active)},
|
||||
"stale": {"output_dir": str(stale)},
|
||||
})
|
||||
|
||||
message = app._cleanup_disk(cache_roots=[cache], temp_root=root)
|
||||
assert active.is_dir()
|
||||
assert not stale.exists()
|
||||
assert not cache.exists()
|
||||
assert list(app._session_models) == ["active"]
|
||||
assert app._state["model"] is None and app._state["tokenizer"] is None
|
||||
assert app._state["status"] == "ready"
|
||||
assert "will reload it automatically" in message
|
||||
header = app.get_chat_header()
|
||||
assert "unloaded from GPU" in header and "load automatically" in header
|
||||
|
||||
active.rename(root / "removed")
|
||||
header = app.get_chat_header()
|
||||
assert header.startswith("No model loaded")
|
||||
assert app._state["status"] == "idle"
|
||||
assert app._state["model_name"] is None
|
||||
assert app._state["output_dir"] is None
|
||||
'''
|
||||
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-c", script, str(tmp_path)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=60,
|
||||
check=False,
|
||||
)
|
||||
|
||||
assert result.returncode == 0, result.stdout + result.stderr
|
||||
@@ -235,3 +235,16 @@ def test_configure_cuda_allocator(monkeypatch):
|
||||
monkeypatch.setenv("PYTORCH_CUDA_ALLOC_CONF", "existing")
|
||||
device.configure_cuda_alloc()
|
||||
assert device.os.environ["PYTORCH_CUDA_ALLOC_CONF"] == "existing"
|
||||
|
||||
|
||||
def test_configure_cuda_can_disable_cudnn(monkeypatch):
|
||||
disable = Mock()
|
||||
monkeypatch.setattr(device, "is_cuda", lambda: True)
|
||||
monkeypatch.setattr(device.torch.backends.cuda, "enable_cudnn_sdp", disable)
|
||||
monkeypatch.setenv("OBLITERATUS_DISABLE_CUDNN", "1")
|
||||
monkeypatch.setattr(device.torch.backends.cudnn, "enabled", True)
|
||||
|
||||
device.configure_cuda_alloc()
|
||||
|
||||
disable.assert_called_once_with(False)
|
||||
assert device.torch.backends.cudnn.enabled is False
|
||||
|
||||
Reference in New Issue
Block a user