circuitforge-core/tests/test_preferences.py
pyr0ball 80b0d5fd34
Some checks failed
CI / test (push) Waiting to run
Mirror / mirror (push) Has been cancelled
Release — PyPI / release (push) Has been cancelled
feat: v0.9.0 — cf-text, pipeline crystallization engine, multimodal pipeline, a11y preferences
Closes #33, #37, #38, #41, #42.

## cf-text (closes #41)
- New module: `circuitforge_core.text` — direct local inference bypassing ollama/vllm
- Backends: llama.cpp (GGUF), transformers (HF), mock
- Auto-detects backend from file extension; CF_TEXT_BACKEND env override
- Optional 4-bit/8-bit quantisation via bitsandbytes (CF_TEXT_4BIT / CF_TEXT_8BIT)
- process-level singleton + per-request `make_backend()` path

## Pipeline crystallization engine (closes #33, #37)
- FPGA→ASIC model: LLM-discovered paths → deterministic workflows after N approvals
- `models.py`: PipelineRun (incl. review_duration_ms + output_modified per #37),
  CrystallizedWorkflow, Step, hash_input()
- `recorder.py`: append-only JSON run log under ~/.config/circuitforge/pipeline/
- `crystallizer.py`: threshold check, majority/most-recent step strategy,
  rubber-stamp warning (review_duration_ms < 5s triggers warnings.warn)
- `registry.py`: exact + fuzzy match, deactivate-without-delete, colon-safe filenames
- `executor.py`: deterministic steps with transparent LLM fallback

## Multimodal chunked pipeline (closes #42)
- `pipeline/multimodal.py`: cf-docuvision pages → cf-text streaming
- `run()` yields PageResult per page (progressive, no full-doc buffer)
- `stream()` yields (page_idx, token) tuples for token-level UI rendering
- `vram_serialise` flag + `swap_fn` hook for 8GB GPU VRAM management
- `prompt_fn` callback for product-specific prompt construction

## Accessibility preferences (closes #38)
- `preferences/accessibility.py`: PREF_REDUCED_MOTION, PREF_HIGH_CONTRAST,
  PREF_FONT_SIZE, PREF_SCREEN_READER with get/set helpers
- Exported from preferences package __init__

## LLM router fix
- cf-orch backends: skip reachability pre-check; allocation starts the service
- Static backends: reachability check remains in place
2026-04-08 23:17:18 -07:00

235 lines
9.5 KiB
Python

"""Tests for circuitforge_core.preferences path utilities."""
import pytest
from circuitforge_core.preferences import get_path, set_path
class TestGetPath:
def test_top_level_key(self):
assert get_path({"a": 1}, "a") == 1
def test_nested_key(self):
data = {"affiliate": {"opt_out": False}}
assert get_path(data, "affiliate.opt_out") is False
def test_deeply_nested(self):
data = {"affiliate": {"byok_ids": {"ebay": "my-tag"}}}
assert get_path(data, "affiliate.byok_ids.ebay") == "my-tag"
def test_missing_key_returns_default(self):
assert get_path({}, "missing", default="x") == "x"
def test_missing_nested_returns_default(self):
assert get_path({"a": {}}, "a.b.c", default=42) == 42
def test_default_is_none_when_omitted(self):
assert get_path({}, "nope") is None
def test_non_dict_intermediate_returns_default(self):
assert get_path({"a": "string"}, "a.b", default="d") == "d"
class TestSetPath:
def test_top_level_key(self):
result = set_path({}, "opt_out", True)
assert result == {"opt_out": True}
def test_nested_key_created(self):
result = set_path({}, "affiliate.opt_out", True)
assert result == {"affiliate": {"opt_out": True}}
def test_deeply_nested(self):
result = set_path({}, "affiliate.byok_ids.ebay", "my-tag")
assert result == {"affiliate": {"byok_ids": {"ebay": "my-tag"}}}
def test_preserves_sibling_keys(self):
data = {"affiliate": {"opt_out": False, "byok_ids": {}}}
result = set_path(data, "affiliate.opt_out", True)
assert result["affiliate"]["opt_out"] is True
assert result["affiliate"]["byok_ids"] == {}
def test_preserves_unrelated_top_level_keys(self):
data = {"other": "value", "affiliate": {"opt_out": False}}
result = set_path(data, "affiliate.opt_out", True)
assert result["other"] == "value"
def test_does_not_mutate_original(self):
data = {"affiliate": {"opt_out": False}}
set_path(data, "affiliate.opt_out", True)
assert data["affiliate"]["opt_out"] is False
def test_overwrites_existing_value(self):
data = {"affiliate": {"byok_ids": {"ebay": "old-tag"}}}
result = set_path(data, "affiliate.byok_ids.ebay", "new-tag")
assert result["affiliate"]["byok_ids"]["ebay"] == "new-tag"
def test_non_dict_intermediate_replaced(self):
data = {"affiliate": "not-a-dict"}
result = set_path(data, "affiliate.opt_out", True)
assert result == {"affiliate": {"opt_out": True}}
def test_roundtrip_get_after_set(self):
prefs = {}
prefs = set_path(prefs, "affiliate.opt_out", True)
prefs = set_path(prefs, "affiliate.byok_ids.ebay", "tag-123")
assert get_path(prefs, "affiliate.opt_out") is True
assert get_path(prefs, "affiliate.byok_ids.ebay") == "tag-123"
import os
import tempfile
from pathlib import Path
from circuitforge_core.preferences.store import LocalFileStore
class TestLocalFileStore:
def _store(self, tmp_path) -> LocalFileStore:
return LocalFileStore(prefs_path=tmp_path / "preferences.yaml")
def test_get_returns_default_when_file_missing(self, tmp_path):
store = self._store(tmp_path)
assert store.get(user_id=None, path="affiliate.opt_out", default=False) is False
def test_set_then_get_roundtrip(self, tmp_path):
store = self._store(tmp_path)
store.set(user_id=None, path="affiliate.opt_out", value=True)
assert store.get(user_id=None, path="affiliate.opt_out", default=False) is True
def test_set_nested_path(self, tmp_path):
store = self._store(tmp_path)
store.set(user_id=None, path="affiliate.byok_ids.ebay", value="my-tag")
assert store.get(user_id=None, path="affiliate.byok_ids.ebay") == "my-tag"
def test_set_preserves_sibling_keys(self, tmp_path):
store = self._store(tmp_path)
store.set(user_id=None, path="affiliate.opt_out", value=False)
store.set(user_id=None, path="affiliate.byok_ids.ebay", value="tag")
assert store.get(user_id=None, path="affiliate.opt_out") is False
assert store.get(user_id=None, path="affiliate.byok_ids.ebay") == "tag"
def test_creates_parent_dirs(self, tmp_path):
deep_path = tmp_path / "deep" / "nested" / "preferences.yaml"
store = LocalFileStore(prefs_path=deep_path)
store.set(user_id=None, path="x", value=1)
assert deep_path.exists()
def test_user_id_ignored_for_local_store(self, tmp_path):
"""LocalFileStore is single-user; user_id is accepted but ignored."""
store = self._store(tmp_path)
store.set(user_id="u123", path="affiliate.opt_out", value=True)
assert store.get(user_id="u456", path="affiliate.opt_out", default=False) is True
from circuitforge_core.preferences import get_user_preference, set_user_preference
from circuitforge_core.preferences.accessibility import (
is_reduced_motion_preferred,
is_high_contrast,
get_font_size,
is_screen_reader_mode,
set_reduced_motion,
PREF_REDUCED_MOTION,
PREF_HIGH_CONTRAST,
PREF_FONT_SIZE,
PREF_SCREEN_READER,
)
class TestAccessibilityPreferences:
def _store(self, tmp_path) -> LocalFileStore:
return LocalFileStore(prefs_path=tmp_path / "preferences.yaml")
def test_reduced_motion_default_false(self, tmp_path):
store = self._store(tmp_path)
assert is_reduced_motion_preferred(store=store) is False
def test_set_reduced_motion_persists(self, tmp_path):
store = self._store(tmp_path)
set_reduced_motion(True, store=store)
assert is_reduced_motion_preferred(store=store) is True
def test_reduced_motion_false_roundtrip(self, tmp_path):
store = self._store(tmp_path)
set_reduced_motion(True, store=store)
set_reduced_motion(False, store=store)
assert is_reduced_motion_preferred(store=store) is False
def test_high_contrast_default_false(self, tmp_path):
store = self._store(tmp_path)
assert is_high_contrast(store=store) is False
def test_high_contrast_set_and_read(self, tmp_path):
store = self._store(tmp_path)
store.set(user_id=None, path=PREF_HIGH_CONTRAST, value=True)
assert is_high_contrast(store=store) is True
def test_font_size_default(self, tmp_path):
store = self._store(tmp_path)
assert get_font_size(store=store) == "default"
def test_font_size_large(self, tmp_path):
store = self._store(tmp_path)
store.set(user_id=None, path=PREF_FONT_SIZE, value="large")
assert get_font_size(store=store) == "large"
def test_font_size_xlarge(self, tmp_path):
store = self._store(tmp_path)
store.set(user_id=None, path=PREF_FONT_SIZE, value="xlarge")
assert get_font_size(store=store) == "xlarge"
def test_font_size_invalid_falls_back_to_default(self, tmp_path):
store = self._store(tmp_path)
store.set(user_id=None, path=PREF_FONT_SIZE, value="gigantic")
assert get_font_size(store=store) == "default"
def test_screen_reader_mode_default_false(self, tmp_path):
store = self._store(tmp_path)
assert is_screen_reader_mode(store=store) is False
def test_screen_reader_mode_set(self, tmp_path):
store = self._store(tmp_path)
store.set(user_id=None, path=PREF_SCREEN_READER, value=True)
assert is_screen_reader_mode(store=store) is True
def test_preferences_are_independent(self, tmp_path):
"""Setting one a11y pref doesn't affect others."""
store = self._store(tmp_path)
set_reduced_motion(True, store=store)
assert is_high_contrast(store=store) is False
assert get_font_size(store=store) == "default"
assert is_screen_reader_mode(store=store) is False
def test_user_id_threaded_through(self, tmp_path):
"""user_id param is accepted (LocalFileStore ignores it, but must not error)."""
store = self._store(tmp_path)
set_reduced_motion(True, user_id="u999", store=store)
assert is_reduced_motion_preferred(user_id="u999", store=store) is True
def test_accessibility_exported_from_package(self):
from circuitforge_core.preferences import accessibility
assert hasattr(accessibility, "is_reduced_motion_preferred")
assert hasattr(accessibility, "PREF_REDUCED_MOTION")
class TestPreferenceHelpers:
def _store(self, tmp_path) -> LocalFileStore:
return LocalFileStore(prefs_path=tmp_path / "preferences.yaml")
def test_get_returns_default_when_unset(self, tmp_path):
store = self._store(tmp_path)
result = get_user_preference(user_id=None, path="affiliate.opt_out",
default=False, store=store)
assert result is False
def test_set_then_get(self, tmp_path):
store = self._store(tmp_path)
set_user_preference(user_id=None, path="affiliate.opt_out", value=True, store=store)
result = get_user_preference(user_id=None, path="affiliate.opt_out",
default=False, store=store)
assert result is True
def test_default_store_is_local(self, tmp_path, monkeypatch):
"""When no store is passed, helpers use LocalFileStore at default path."""
from circuitforge_core.preferences import store as store_module
local = self._store(tmp_path)
monkeypatch.setattr(store_module, "_DEFAULT_STORE", local)
set_user_preference(user_id=None, path="x.y", value=42)
assert get_user_preference(user_id=None, path="x.y") == 42