Harden framework: tests, VRAM Python, dashboard spec, security, CI
- Rewrite vram_detect in Python with fixed config parsing and 10KB read limit
- Add pytest suite (72 tests) covering dashboard core, app security, and VRAM
- Standardize all prompts to .automaton/tasks/{task-name}/ path
- Reconcile dashboard spec with web implementation; remove themes.py
- Remove half-implemented refresh.py file watcher
- Harden dashboard static-file serving and task-name validation
- Add uncommitted-change guard to update.sh and real Gitea URLs
- Add AGENTS.md, Gitea CI workflow, and template documentation
This commit is contained in:
@@ -0,0 +1,88 @@
|
||||
"""Tests for automaton.dashboard.ui.app."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import io
|
||||
|
||||
import pytest
|
||||
|
||||
from automaton.dashboard.ui.app import DashboardHandler
|
||||
|
||||
|
||||
def test_validate_task_name() -> None:
|
||||
assert DashboardHandler._validate_task_name("good-task") is True
|
||||
assert DashboardHandler._validate_task_name("bad/../task") is False
|
||||
assert DashboardHandler._validate_task_name("bad\\task") is False
|
||||
assert DashboardHandler._validate_task_name("") is False
|
||||
|
||||
|
||||
def test_find_tasks_dir(tmp_path: Path) -> None:
|
||||
(tmp_path / ".automaton" / "tasks").mkdir(parents=True)
|
||||
tasks_dir = DashboardHandler._find_tasks_dir(tmp_path)
|
||||
assert tasks_dir == tmp_path / ".automaton" / "tasks"
|
||||
|
||||
|
||||
def test_find_tasks_dir_missing(tmp_path: Path) -> None:
|
||||
tasks_dir = DashboardHandler._find_tasks_dir(tmp_path)
|
||||
assert tasks_dir == tmp_path / ".automaton" / "tasks"
|
||||
|
||||
|
||||
def test_path_traversal_attempt() -> None:
|
||||
"""Task names with path traversal should be rejected."""
|
||||
assert DashboardHandler._validate_task_name("../etc/passwd") is False
|
||||
assert DashboardHandler._validate_task_name("task%2f..%2fetc") is False
|
||||
|
||||
|
||||
def test_static_path_traversal_symlink(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Static file serving must reject symlinks that resolve outside the html directory."""
|
||||
html_dir = tmp_path / "html"
|
||||
html_dir.mkdir()
|
||||
outside = tmp_path / "secret.txt"
|
||||
outside.write_text("secret")
|
||||
symlink = html_dir / "link.txt"
|
||||
symlink.symlink_to(outside)
|
||||
|
||||
monkeypatch.setattr(DashboardHandler, "dashboard_path", html_dir)
|
||||
|
||||
handler = DashboardHandler.__new__(DashboardHandler)
|
||||
handler.path = "/link.txt"
|
||||
errors: list[tuple[int, str]] = []
|
||||
|
||||
def capture_error(code: int, message: str) -> None:
|
||||
errors.append((code, message))
|
||||
|
||||
handler._send_error = capture_error
|
||||
handler._serve_static()
|
||||
assert errors == [(403, "Forbidden")]
|
||||
|
||||
|
||||
def test_static_valid_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Static file serving returns a valid html file."""
|
||||
html_dir = tmp_path / "html"
|
||||
html_dir.mkdir()
|
||||
(html_dir / "index.html").write_text("<html></html>")
|
||||
|
||||
monkeypatch.setattr(DashboardHandler, "dashboard_path", html_dir)
|
||||
|
||||
handler = DashboardHandler.__new__(DashboardHandler)
|
||||
handler.path = "/"
|
||||
|
||||
response_status: list[int] = []
|
||||
response_headers: list[tuple[str, str]] = []
|
||||
|
||||
def fake_send_response(code: int) -> None:
|
||||
response_status.append(code)
|
||||
|
||||
def fake_send_header(key: str, value: str) -> None:
|
||||
response_headers.append((key, value))
|
||||
|
||||
handler.send_response = fake_send_response
|
||||
handler.send_header = fake_send_header
|
||||
handler.end_headers = lambda: None
|
||||
handler.wfile = io.BytesIO()
|
||||
handler._send_error = lambda code, msg: None
|
||||
|
||||
handler._serve_static()
|
||||
assert response_status == [200]
|
||||
assert any(h[0] == "Content-Type" and h[1] == "text/html" for h in response_headers)
|
||||
assert handler.wfile.getvalue() == b"<html></html>"
|
||||
@@ -0,0 +1,54 @@
|
||||
"""Tests for automaton.dashboard.core.board."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from automaton.dashboard.core.board import KanbanBoard
|
||||
from automaton.dashboard.core.task import determine_task_state
|
||||
|
||||
|
||||
def _task(name: str, tmp_path: Path, artifacts: dict[str, str]) -> object:
|
||||
from automaton.dashboard.core.task import Task
|
||||
|
||||
task_dir = tmp_path / name
|
||||
task_dir.mkdir(parents=True)
|
||||
for filename, content in artifacts.items():
|
||||
(task_dir / filename).write_text(content)
|
||||
state, artifact_map = determine_task_state(task_dir)
|
||||
return Task(name=name, folder_path=task_dir, state=state, artifacts=artifact_map)
|
||||
|
||||
|
||||
def test_board_columns(tmp_path: Path) -> None:
|
||||
tasks = [
|
||||
_task("research", tmp_path, {"SPEC.md": "# Spec"}),
|
||||
_task("done", tmp_path, {"VERDICT.md": "## Status: PASS"}),
|
||||
]
|
||||
board = KanbanBoard(tasks)
|
||||
assert board.columns[tasks[0].state] == [tasks[0]]
|
||||
assert board.columns[tasks[1].state] == [tasks[1]]
|
||||
|
||||
|
||||
def test_board_filter_by_phase(tmp_path: Path) -> None:
|
||||
research = _task("research", tmp_path, {"SPEC.md": "# Spec"})
|
||||
done = _task("done", tmp_path, {"VERDICT.md": "## Status: PASS"})
|
||||
board = KanbanBoard([research, done])
|
||||
filtered = board.filter_columns(phase_filter=research.state)
|
||||
assert research in filtered[research.state]
|
||||
assert done not in filtered[research.state]
|
||||
|
||||
|
||||
def test_board_search(tmp_path: Path) -> None:
|
||||
apple = _task("apple-task", tmp_path, {"SPEC.md": "# Spec"})
|
||||
banana = _task("banana-task", tmp_path, {"SPEC.md": "# Spec"})
|
||||
board = KanbanBoard([apple, banana])
|
||||
filtered = board.filter_columns(search_query="apple")
|
||||
assert apple in filtered[apple.state]
|
||||
assert banana not in filtered[banana.state]
|
||||
|
||||
|
||||
def test_board_wip_count(tmp_path: Path) -> None:
|
||||
research = _task("research", tmp_path, {"SPEC.md": "# Spec"})
|
||||
done = _task("done", tmp_path, {"VERDICT.md": "## Status: PASS"})
|
||||
board = KanbanBoard([research, done])
|
||||
assert board.get_wip_count() == 1
|
||||
@@ -0,0 +1,55 @@
|
||||
"""Tests for automaton.dashboard.config."""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from automaton.dashboard.config import DashboardConfig, get_config_path
|
||||
|
||||
|
||||
def test_config_defaults() -> None:
|
||||
config = DashboardConfig()
|
||||
assert config.auto_refresh_interval == 2
|
||||
assert config.default_view == "board"
|
||||
assert config.column_width == 30
|
||||
assert config.show_timelines is True
|
||||
assert config.theme == "default"
|
||||
|
||||
|
||||
def test_config_validation() -> None:
|
||||
config = DashboardConfig(auto_refresh_interval=0, default_view="invalid", column_width=5, theme="neon")
|
||||
errors = config.validate()
|
||||
assert any("auto_refresh_interval" in e for e in errors)
|
||||
assert any("default_view" in e for e in errors)
|
||||
assert any("column_width" in e for e in errors)
|
||||
assert any("theme" in e for e in errors)
|
||||
|
||||
|
||||
def test_config_from_dict() -> None:
|
||||
config = DashboardConfig.from_dict({"theme": "dark", "column_width": 50})
|
||||
assert config.theme == "dark"
|
||||
assert config.column_width == 50
|
||||
assert config.default_view == "board" # default preserved
|
||||
|
||||
|
||||
def test_config_from_file(tmp_path: Path) -> None:
|
||||
config_path = tmp_path / "dashboard-config.json"
|
||||
config_path.write_text(json.dumps({"theme": "light", "auto_refresh_interval": 5}))
|
||||
config = DashboardConfig.from_file(config_path)
|
||||
assert config.theme == "light"
|
||||
assert config.auto_refresh_interval == 5
|
||||
|
||||
|
||||
def test_config_save(tmp_path: Path) -> None:
|
||||
config_path = tmp_path / "subdir" / "dashboard-config.json"
|
||||
config = DashboardConfig(theme="dark")
|
||||
config.save(config_path)
|
||||
assert config_path.exists()
|
||||
data = json.loads(config_path.read_text())
|
||||
assert data["theme"] == "dark"
|
||||
|
||||
|
||||
def test_get_config_path(tmp_path: Path) -> None:
|
||||
path = get_config_path(tmp_path)
|
||||
assert path == tmp_path / ".automaton" / "dashboard-config.json"
|
||||
@@ -0,0 +1,44 @@
|
||||
"""Regression test for task path conventions in prompts and templates."""
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
PROMPTS_DIR = ROOT / "prompts"
|
||||
TEMPLATES_DIR = ROOT / "templates"
|
||||
|
||||
# Legacy path pattern that should no longer appear.
|
||||
LEGACY_PATH = re.compile(r"\{project\}/tasks/\{task-name\}/")
|
||||
# Canonical path pattern that should be used instead.
|
||||
CANONICAL_PATH = re.compile(r"\{project\}/\.automaton/tasks/\{task-name\}/")
|
||||
|
||||
|
||||
def _markdown_files(*directories: Path) -> list[Path]:
|
||||
files = []
|
||||
for directory in directories:
|
||||
if directory.exists():
|
||||
files.extend(directory.rglob("*.md"))
|
||||
return files
|
||||
|
||||
|
||||
@pytest.mark.parametrize("path", _markdown_files(PROMPTS_DIR, TEMPLATES_DIR))
|
||||
def test_no_legacy_task_paths(path: Path) -> None:
|
||||
"""Every prompt/template must use the canonical .automaton/tasks path."""
|
||||
text = path.read_text(encoding="utf-8")
|
||||
legacy_matches = LEGACY_PATH.findall(text)
|
||||
assert not legacy_matches, (
|
||||
f"Found legacy task path in {path.relative_to(ROOT)}: {legacy_matches}\n"
|
||||
"Use {project}/.automaton/tasks/{task-name}/ instead."
|
||||
)
|
||||
|
||||
|
||||
def test_canonical_path_present_in_prompts() -> None:
|
||||
"""At least one prompt uses the canonical path (sanity check)."""
|
||||
found = False
|
||||
for path in _markdown_files(PROMPTS_DIR):
|
||||
if CANONICAL_PATH.search(path.read_text(encoding="utf-8")):
|
||||
found = True
|
||||
break
|
||||
assert found, "No prompt uses the canonical {project}/.automaton/tasks/{task-name}/ path"
|
||||
@@ -0,0 +1,62 @@
|
||||
"""Tests for automaton.dashboard.core.scope."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from automaton.dashboard.core.scope import detect_scope, find_automaton_root
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_project(tmp_path: Path) -> Path:
|
||||
"""Create a temporary automaton project."""
|
||||
(tmp_path / ".automaton").mkdir()
|
||||
return tmp_path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def nested_project(tmp_path: Path) -> Path:
|
||||
"""Create a nested automaton project."""
|
||||
project = tmp_path / "my-project"
|
||||
(project / ".automaton").mkdir(parents=True)
|
||||
return project
|
||||
|
||||
|
||||
def test_find_automaton_root_in_project(temp_project: Path) -> None:
|
||||
root = find_automaton_root(temp_project)
|
||||
assert root == temp_project
|
||||
|
||||
|
||||
def test_find_automaton_root_from_subdir(temp_project: Path) -> None:
|
||||
subdir = temp_project / "src" / "deep"
|
||||
subdir.mkdir(parents=True)
|
||||
root = find_automaton_root(subdir)
|
||||
assert root == temp_project
|
||||
|
||||
|
||||
def test_find_automaton_root_not_found(tmp_path: Path) -> None:
|
||||
root = find_automaton_root(tmp_path)
|
||||
assert root is None
|
||||
|
||||
|
||||
def test_detect_scope_project(temp_project: Path) -> None:
|
||||
root, scope = detect_scope(temp_project)
|
||||
assert root == temp_project
|
||||
assert scope == "project"
|
||||
|
||||
|
||||
def test_detect_scope_framework(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||
"""Framework mode is detected when .automaton is directly in home."""
|
||||
fake_home = tmp_path / "home"
|
||||
fake_home.mkdir(parents=True)
|
||||
(fake_home / ".automaton").mkdir()
|
||||
monkeypatch.setattr(Path, "home", lambda: fake_home)
|
||||
root, scope = detect_scope(fake_home)
|
||||
assert root == fake_home
|
||||
assert scope == "framework"
|
||||
|
||||
|
||||
def test_detect_scope_none(tmp_path: Path) -> None:
|
||||
root, scope = detect_scope(tmp_path)
|
||||
assert root is None
|
||||
assert scope == "none"
|
||||
@@ -0,0 +1,56 @@
|
||||
"""Tests for automaton.dashboard.core.stats."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from automaton.dashboard.core.stats import TaskStats
|
||||
from automaton.dashboard.core.task import determine_task_state
|
||||
|
||||
|
||||
def _task(name: str, tmp_path: Path, artifacts: dict[str, str]) -> object:
|
||||
from automaton.dashboard.core.task import Task
|
||||
|
||||
task_dir = tmp_path / name
|
||||
task_dir.mkdir(parents=True)
|
||||
for filename, content in artifacts.items():
|
||||
(task_dir / filename).write_text(content)
|
||||
state, artifact_map = determine_task_state(task_dir)
|
||||
return Task(name=name, folder_path=task_dir, state=state, artifacts=artifact_map)
|
||||
|
||||
|
||||
def test_empty_stats() -> None:
|
||||
stats = TaskStats([])
|
||||
assert stats.total_tasks == 0
|
||||
assert stats.pass_rate == 0.0
|
||||
assert stats.fail_rate == 0.0
|
||||
|
||||
|
||||
def test_counts(tmp_path: Path) -> None:
|
||||
tasks = [
|
||||
_task("done-1", tmp_path, {"VERDICT.md": "## Status: PASS"}),
|
||||
_task("done-2", tmp_path, {"VERDICT.md": "## Status: PASS"}),
|
||||
_task("blocked", tmp_path, {"VERDICT.md": "## Status: FAIL"}),
|
||||
_task("research", tmp_path, {"SPEC.md": "# Spec"}),
|
||||
]
|
||||
stats = TaskStats(tasks)
|
||||
assert stats.total_tasks == 4
|
||||
assert stats.pass_count == 2
|
||||
assert stats.fail_count == 1
|
||||
assert stats.in_progress_count == 1
|
||||
assert stats.backlog_count == 0
|
||||
assert stats.pass_rate == 50.0
|
||||
|
||||
|
||||
def test_sub_task_stats(tmp_path: Path) -> None:
|
||||
from automaton.dashboard.core.task import SubTask, TaskState
|
||||
|
||||
parent = _task("parent", tmp_path, {"SPEC.md": "# Spec"})
|
||||
parent.sub_tasks = [
|
||||
SubTask(name="a", state=TaskState.DONE, has_spec=True, has_verdict=True, verdict_status="PASS"),
|
||||
SubTask(name="b", state=TaskState.BLOCKED, has_spec=True, has_verdict=True, verdict_status="FAIL"),
|
||||
]
|
||||
stats = TaskStats([parent])
|
||||
assert stats.sub_task_stats == {
|
||||
"parent": {"total": 2, "passed": 1, "failed": 1, "needs_review": 0, "incomplete": 0}
|
||||
}
|
||||
@@ -0,0 +1,134 @@
|
||||
"""Tests for automaton.dashboard.core.task."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from automaton.dashboard.core.task import (
|
||||
determine_task_state,
|
||||
discover_tasks,
|
||||
parse_sub_tasks,
|
||||
TaskState,
|
||||
)
|
||||
|
||||
|
||||
def _make_task(tmp_path: Path, name: str, artifacts: dict[str, str]) -> Path:
|
||||
"""Create a task folder with the given artifact contents."""
|
||||
task_dir = tmp_path / name
|
||||
task_dir.mkdir(parents=True)
|
||||
for filename, content in artifacts.items():
|
||||
(task_dir / filename).write_text(content)
|
||||
return task_dir
|
||||
|
||||
|
||||
def test_backlog_state(tmp_path: Path) -> None:
|
||||
task_dir = _make_task(tmp_path, "backlog-task", {})
|
||||
state, artifacts = determine_task_state(task_dir)
|
||||
assert state == TaskState.BACKLOG
|
||||
assert not artifacts
|
||||
|
||||
|
||||
def test_research_state(tmp_path: Path) -> None:
|
||||
task_dir = _make_task(tmp_path, "research-task", {"SPEC.md": "# Spec"})
|
||||
state, artifacts = determine_task_state(task_dir)
|
||||
assert state == TaskState.RESEARCH
|
||||
assert "SPEC.md" in artifacts
|
||||
|
||||
|
||||
def test_implementation_state(tmp_path: Path) -> None:
|
||||
task_dir = _make_task(
|
||||
tmp_path,
|
||||
"impl-task",
|
||||
{"SPEC.md": "# Spec", "IMPLEMENTATION.md": "# Impl"},
|
||||
)
|
||||
state, _ = determine_task_state(task_dir)
|
||||
assert state == TaskState.IMPLEMENT
|
||||
|
||||
|
||||
def test_bug_find_state(tmp_path: Path) -> None:
|
||||
task_dir = _make_task(
|
||||
tmp_path,
|
||||
"bug-task",
|
||||
{"SPEC.md": "# Spec", "IMPLEMENTATION.md": "# Impl", "BUG_REPORT.md": "# Bugs"},
|
||||
)
|
||||
state, _ = determine_task_state(task_dir)
|
||||
assert state == TaskState.BUG_FIND
|
||||
|
||||
|
||||
def test_adv_bug_find_state(tmp_path: Path) -> None:
|
||||
task_dir = _make_task(
|
||||
tmp_path,
|
||||
"adv-task",
|
||||
{
|
||||
"SPEC.md": "# Spec",
|
||||
"IMPLEMENTATION.md": "# Impl",
|
||||
"BUG_REPORT.md": "# Bugs",
|
||||
"ADVERSARIAL_BUG_REPORT.md": "# Adv",
|
||||
},
|
||||
)
|
||||
state, _ = determine_task_state(task_dir)
|
||||
assert state == TaskState.ADV_BUG_FIND
|
||||
|
||||
|
||||
def test_done_state(tmp_path: Path) -> None:
|
||||
task_dir = _make_task(
|
||||
tmp_path,
|
||||
"done-task",
|
||||
{"SPEC.md": "# Spec", "VERDICT.md": "## Status: PASS"},
|
||||
)
|
||||
state, _ = determine_task_state(task_dir)
|
||||
assert state == TaskState.DONE
|
||||
|
||||
|
||||
def test_blocked_state(tmp_path: Path) -> None:
|
||||
task_dir = _make_task(
|
||||
tmp_path,
|
||||
"blocked-task",
|
||||
{"SPEC.md": "# Spec", "VERDICT.md": "## Status: FAIL"},
|
||||
)
|
||||
state, _ = determine_task_state(task_dir)
|
||||
assert state == TaskState.BLOCKED
|
||||
|
||||
|
||||
def test_empty_verdict_is_blocked(tmp_path: Path) -> None:
|
||||
task_dir = _make_task(tmp_path, "empty-verdict", {"VERDICT.md": ""})
|
||||
state, _ = determine_task_state(task_dir)
|
||||
assert state == TaskState.BLOCKED
|
||||
|
||||
|
||||
def test_parse_sub_tasks(tmp_path: Path) -> None:
|
||||
parent = tmp_path / "parent"
|
||||
parent.mkdir()
|
||||
sub = parent / "subtasks" / "sub-a"
|
||||
sub.mkdir(parents=True)
|
||||
(sub / "SPEC.md").write_text("# Spec")
|
||||
(sub / "VERDICT.md").write_text("## Status: PASS")
|
||||
|
||||
sub_tasks = parse_sub_tasks(parent)
|
||||
assert len(sub_tasks) == 1
|
||||
assert sub_tasks[0].name == "sub-a"
|
||||
assert sub_tasks[0].has_verdict is True
|
||||
assert sub_tasks[0].verdict_status == "PASS"
|
||||
|
||||
|
||||
def test_discover_tasks(tmp_path: Path) -> None:
|
||||
_make_task(tmp_path, "aaa", {"SPEC.md": "# Spec"})
|
||||
_make_task(tmp_path, "zzz", {"VERDICT.md": "## Status: PASS"})
|
||||
tasks = discover_tasks(tmp_path)
|
||||
assert len(tasks) == 2
|
||||
# Done task should be first (sorted by advancement).
|
||||
assert tasks[0].state == TaskState.DONE
|
||||
assert tasks[1].state == TaskState.RESEARCH
|
||||
|
||||
|
||||
def test_discover_tasks_skips_subtasks_root(tmp_path: Path) -> None:
|
||||
parent = tmp_path / "parent"
|
||||
parent.mkdir()
|
||||
(parent / "SPEC.md").write_text("# Spec")
|
||||
(parent / "subtasks" / "sub-a").mkdir(parents=True)
|
||||
(parent / "subtasks" / "sub-a" / "SPEC.md").write_text("# Sub")
|
||||
|
||||
tasks = discover_tasks(tmp_path)
|
||||
assert len(tasks) == 1
|
||||
assert tasks[0].name == "parent"
|
||||
assert len(tasks[0].sub_tasks) == 1
|
||||
@@ -0,0 +1,110 @@
|
||||
"""Tests for scripts/vram_detect.py."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
import scripts.vram_detect as vram
|
||||
|
||||
|
||||
def test_lookup_model_context() -> None:
|
||||
assert vram._lookup_model_context("gpt-4o") == 128_000
|
||||
assert vram._lookup_model_context("claude-3-5-sonnet") == 200_000
|
||||
assert vram._lookup_model_context("unknown-model") == 0
|
||||
|
||||
|
||||
def test_parse_token_value() -> None:
|
||||
assert vram._parse_token_value("128k") == 128_000
|
||||
assert vram._parse_token_value("128K") == 128_000
|
||||
assert vram._parse_token_value("128000") == 128_000
|
||||
assert vram._parse_token_value("nonsense") is None
|
||||
|
||||
|
||||
def test_extract_value() -> None:
|
||||
assert vram._extract_value("- **Model**: gpt-4o") == "gpt-4o"
|
||||
assert vram._extract_value("model = gpt-4o # comment") == "gpt-4o"
|
||||
assert vram._extract_value("- **Model**: auto") == "auto"
|
||||
|
||||
|
||||
def test_parse_config_model(tmp_path: Path) -> None:
|
||||
config = tmp_path / "config.md"
|
||||
config.write_text(
|
||||
"## VRAM Configuration\n"
|
||||
"- **Auto-detect**: Yes\n"
|
||||
"- **Target context**: 16k tokens\n"
|
||||
"- **Headroom**: 25%\n"
|
||||
"\n"
|
||||
"## Model Configuration\n"
|
||||
"- **Model**: gpt-4o\n"
|
||||
"- **Override context window**: 128k\n"
|
||||
)
|
||||
model, override = vram._parse_config_model(config)
|
||||
assert model == "gpt-4o"
|
||||
assert override == 128_000
|
||||
|
||||
|
||||
def test_parse_config_model_skips_code_blocks(tmp_path: Path) -> None:
|
||||
"""Example code blocks should not be parsed as live config."""
|
||||
config = tmp_path / "config.md"
|
||||
config.write_text(
|
||||
"## VRAM Configuration\n"
|
||||
"- **Headroom**: 25%\n"
|
||||
"```\n"
|
||||
"- **Headroom**: 99%\n"
|
||||
"```\n"
|
||||
)
|
||||
parsed = vram.parse_vram_config(config)
|
||||
assert parsed["headroom_pct"] == 25
|
||||
|
||||
|
||||
def test_parse_vram_config_manual(tmp_path: Path) -> None:
|
||||
config = tmp_path / "config.md"
|
||||
config.write_text(
|
||||
"## VRAM Configuration\n"
|
||||
"- **Auto-detect**: No\n"
|
||||
"- **Target context**: 8k\n"
|
||||
"- **Headroom**: 30%\n"
|
||||
"- **Max peak context per sub-task**: 5.6k\n"
|
||||
)
|
||||
parsed = vram.parse_vram_config(config)
|
||||
assert parsed["auto_detect"] is False
|
||||
assert parsed["target_context_kb"] == 8000
|
||||
assert parsed["headroom_pct"] == 30
|
||||
assert parsed["max_peak_kb"] == 5600
|
||||
|
||||
|
||||
def test_recommend_context_api_model() -> None:
|
||||
config = {"auto_detect": True, "headroom_pct": 25, "target_context_kb": 0, "max_peak_kb": 0}
|
||||
_, recommended_kb, max_peak_kb = vram.recommend_context(
|
||||
gpu_vram_gb=0,
|
||||
ram_gb=16,
|
||||
model_context_kb=128_000,
|
||||
overhead_tokens=4000,
|
||||
config=config,
|
||||
)
|
||||
assert recommended_kb > 0
|
||||
assert max_peak_kb > 0
|
||||
assert recommended_kb <= 128_000 * 0.75 # after headroom
|
||||
|
||||
|
||||
def test_recommend_context_manual_mode() -> None:
|
||||
config = {"auto_detect": False, "headroom_pct": 30, "target_context_kb": 8000, "max_peak_kb": 0}
|
||||
headroom, recommended_kb, max_peak_kb = vram.recommend_context(
|
||||
gpu_vram_gb=0,
|
||||
ram_gb=16,
|
||||
model_context_kb=0,
|
||||
overhead_tokens=0,
|
||||
config=config,
|
||||
)
|
||||
assert headroom == 30
|
||||
assert recommended_kb == 8000
|
||||
assert max_peak_kb == 5600
|
||||
|
||||
|
||||
def test_extract_model_from_file_respects_10kb_limit(tmp_path: Path) -> None:
|
||||
"""Only the first 10KB of an API config file is scanned."""
|
||||
env_file = tmp_path / ".env"
|
||||
# Put the model name far beyond 10KB.
|
||||
env_file.write_text("x" * 11_000 + "\nMODEL=far-away-model\n")
|
||||
model = vram._extract_model_from_file(env_file)
|
||||
assert model is None
|
||||
Reference in New Issue
Block a user