Merge remote-tracking branch 'origin/dev'

This commit is contained in:
pewdiepie-archdaemon
2026-06-30 10:26:46 +00:00
34 changed files with 607 additions and 316 deletions
+2
View File
@@ -34,6 +34,8 @@ def atomic_write_json(path: str, data: Any, *, indent: Optional[int] = None) ->
def atomic_write_text(path: str, text: str) -> None: def atomic_write_text(path: str, text: str) -> None:
if not isinstance(text, str):
raise TypeError("atomic_write_text expects a string")
os.makedirs(os.path.dirname(path) or ".", exist_ok=True) os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
tmp = f"{path}.tmp.{os.getpid()}" tmp = f"{path}.tmp.{os.getpid()}"
with open(tmp, "w", encoding="utf-8") as f: with open(tmp, "w", encoding="utf-8") as f:
+15 -11
View File
@@ -38,23 +38,27 @@ def _preview_text(value, limit: int = 200) -> str:
return text[:limit] return text[:limit]
def _text_field(value) -> str:
return value if isinstance(value, str) else ""
def _serialize_image(i: "GalleryImage") -> dict: def _serialize_image(i: "GalleryImage") -> dict:
return { return {
"id": i.id, "id": i.id,
"filename": i.filename, "filename": _text_field(i.filename),
"prompt": _preview_text(i.prompt), "prompt": _preview_text(i.prompt),
"model": i.model or "", "model": _text_field(i.model),
"size": i.size or "", "size": _text_field(i.size),
"tags": i.tags or "", "tags": _text_field(i.tags),
"favorite": bool(i.favorite), "favorite": bool(i.favorite),
"album_id": i.album_id or "", "album_id": _text_field(i.album_id),
"session_id": i.session_id or "", "session_id": _text_field(i.session_id),
"width": i.width, "width": i.width,
"height": i.height, "height": i.height,
"file_size": i.file_size, "file_size": i.file_size,
"taken_at": i.taken_at.isoformat() if i.taken_at else "", "taken_at": i.taken_at.isoformat() if i.taken_at else "",
"camera_make": i.camera_make or "", "camera_make": _text_field(i.camera_make),
"camera_model": i.camera_model or "", "camera_model": _text_field(i.camera_model),
"created_at": i.created_at.isoformat() if i.created_at else "", "created_at": i.created_at.isoformat() if i.created_at else "",
} }
@@ -93,11 +97,11 @@ def cmd_show(args):
if not i: if not i:
fail(f"no image with id {args.id!r}") fail(f"no image with id {args.id!r}")
out = _serialize_image(i) out = _serialize_image(i)
out["prompt_full"] = i.prompt or "" out["prompt_full"] = _text_field(i.prompt)
out["ai_tags"] = i.ai_tags or "" out["ai_tags"] = _text_field(i.ai_tags)
out["gps_lat"] = i.gps_lat or "" out["gps_lat"] = i.gps_lat or ""
out["gps_lng"] = i.gps_lng or "" out["gps_lng"] = i.gps_lng or ""
out["file_hash"] = i.file_hash or "" out["file_hash"] = _text_field(i.file_hash)
emit(out, args) emit(out, args)
finally: finally:
db.close() db.close()
+2
View File
@@ -108,6 +108,8 @@ def _q(name: str) -> str:
def _split_recipients(value: str) -> list[str]: def _split_recipients(value: str) -> list[str]:
if not isinstance(value, str):
return []
return [r.strip() for r in (value or "").split(",") if r.strip()] return [r.strip() for r in (value or "").split(",") if r.strip()]
+3 -1
View File
@@ -36,7 +36,9 @@ def _load_items(raw) -> list:
items = json.loads(raw) items = json.loads(raw)
except (TypeError, json.JSONDecodeError): except (TypeError, json.JSONDecodeError):
return [] return []
return items if isinstance(items, list) else [] if not isinstance(items, list):
return []
return [item for item in items if isinstance(item, dict)]
def _serialize(n: "Note") -> dict: def _serialize(n: "Note") -> dict:
+3
View File
@@ -103,6 +103,9 @@ def compute_serve_profiles(system, model, serve_weights_gb=None, serve_quant=Non
in the actual serving knobs (n_cpu_moe, KV-cache type, context). serve_quant in the actual serving knobs (n_cpu_moe, KV-cache type, context). serve_quant
is the file's quant label (e.g. "Q4_K_M") just for display. is the file's quant label (e.g. "Q4_K_M") just for display.
""" """
if not isinstance(system, dict) or not isinstance(model, dict):
return []
vram = float(system.get("gpu_vram_gb") or 0) vram = float(system.get("gpu_vram_gb") or 0)
if vram <= 0: if vram <= 0:
return [] return []
+9 -2
View File
@@ -18,6 +18,13 @@ DEFAULT_BUDGET = 6000
DEFAULT_HEADROOM = 0.85 DEFAULT_HEADROOM = 0.85
def _int_or_zero(value) -> int:
try:
return int(value or 0)
except (TypeError, ValueError):
return 0
def compute_input_token_budget( def compute_input_token_budget(
configured: int, configured: int,
context_length: int, context_length: int,
@@ -48,8 +55,8 @@ def compute_input_token_budget(
- When the window is unknown (context_length <= 0), use the conservative - When the window is unknown (context_length <= 0), use the conservative
``default`` budget and do NOT scale off the fallback. ``default`` budget and do NOT scale off the fallback.
""" """
configured = int(configured or 0) configured = _int_or_zero(configured)
context_length = int(context_length or 0) context_length = _int_or_zero(context_length)
if explicit and configured > 0: if explicit and configured > 0:
return min(configured, context_length) if context_length > 0 else configured return min(configured, context_length) if context_length > 0 else configured
+15 -3
View File
@@ -345,6 +345,18 @@ def _normalize_ollama_url(url: str) -> str:
return base.rstrip("/") + "/chat" return base.rstrip("/") + "/chat"
def _normalize_openai_chat_url(url: str) -> str:
"""Ensure an OpenAI-compatible base URL points at /chat/completions."""
base = (url or "").strip().rstrip("/")
if not base:
return base
if base.endswith("/chat/completions") or base.endswith("/completions"):
return base
if base.endswith("/models"):
base = base[: -len("/models")].rstrip("/")
return base + "/chat/completions"
def _ollama_normalize_messages(messages: List[Dict]) -> List[Dict]: def _ollama_normalize_messages(messages: List[Dict]) -> List[Dict]:
"""Adapt Odysseus' canonical OpenAI-style messages to native Ollama /api/chat. """Adapt Odysseus' canonical OpenAI-style messages to native Ollama /api/chat.
@@ -1564,7 +1576,7 @@ def llm_call(url: str, model: str, messages: List[Dict], temperature: float = LL
stream=False, num_ctx=get_context_length(url, model), stream=False, num_ctx=get_context_length(url, model),
) )
else: else:
target_url = url target_url = _normalize_openai_chat_url(url)
if provider == "copilot": if provider == "copilot":
from src.copilot import apply_request_headers from src.copilot import apply_request_headers
apply_request_headers(h, messages_copy) apply_request_headers(h, messages_copy)
@@ -1768,7 +1780,7 @@ async def llm_call_async(
stream=False, num_ctx=get_context_length(url, model), stream=False, num_ctx=get_context_length(url, model),
) )
else: else:
target_url = url target_url = _normalize_openai_chat_url(url)
h = _provider_headers(provider, headers) h = _provider_headers(provider, headers)
if provider == "copilot": if provider == "copilot":
from src.copilot import apply_request_headers from src.copilot import apply_request_headers
@@ -1891,7 +1903,7 @@ async def stream_llm(url: str, model: str, messages: List[Dict], temperature: fl
h = _provider_headers(provider, headers) h = _provider_headers(provider, headers)
payload = _build_chatgpt_responses_payload(model, messages_copy, temperature, max_tokens, stream=True) payload = _build_chatgpt_responses_payload(model, messages_copy, temperature, max_tokens, stream=True)
else: else:
target_url = url target_url = _normalize_openai_chat_url(url)
payload = { payload = {
"model": model, "model": model,
"messages": messages_copy, "messages": messages_copy,
+4 -1
View File
@@ -68,6 +68,8 @@ def read_text_file(path: str) -> str:
def split_chunks(text: str, size: int = config.CHUNK_SIZE, overlap: int = config.CHUNK_OVERLAP) -> List[str]: def split_chunks(text: str, size: int = config.CHUNK_SIZE, overlap: int = config.CHUNK_OVERLAP) -> List[str]:
"""Split text into overlapping chunks.""" """Split text into overlapping chunks."""
if not isinstance(text, str):
return []
text = text.strip() text = text.strip()
if not text: if not text:
return [] return []
@@ -87,7 +89,8 @@ def split_chunks(text: str, size: int = config.CHUNK_SIZE, overlap: int = config
def tokenize(s: str) -> Set[str]: def tokenize(s: str) -> Set[str]:
"""Tokenize string into words, excluding stop words.""" """Tokenize string into words, excluding stop words."""
tokens = re.findall(r"[A-Za-z0-9_\-]+", (s or "").lower()) text = s if isinstance(s, str) else ""
tokens = re.findall(r"[A-Za-z0-9_\-]+", text.lower())
return set(t for t in tokens if t not in config.STOP_WORDS and len(t) > 1) return set(t for t in tokens if t not in config.STOP_WORDS and len(t) > 1)
def load_personal_index( def load_personal_index(
+6
View File
@@ -79,12 +79,18 @@ def check_outbound_url(
if not raw_ips: if not raw_ips:
return False, "host does not resolve" return False, "host does not resolve"
saw_ip = False
for raw in raw_ips: for raw in raw_ips:
if not isinstance(raw, str):
continue
try: try:
ip = ipaddress.ip_address(raw.split("%")[0]) # strip IPv6 zone id ip = ipaddress.ip_address(raw.split("%")[0]) # strip IPv6 zone id
except ValueError: except ValueError:
continue continue
saw_ip = True
reason = _classify(ip, block_private=block_private) reason = _classify(ip, block_private=block_private)
if reason: if reason:
return False, reason return False, reason
if not saw_ip:
return False, "host does not resolve to an IP"
return True, "ok" return True, "ok"
+1
View File
@@ -32,6 +32,7 @@
* @returns {HTMLCanvasElement|null} * @returns {HTMLCanvasElement|null}
*/ */
export function layerUnionAlpha(w, h, layers) { export function layerUnionAlpha(w, h, layers) {
if (!Array.isArray(layers)) return null;
const visible = layers.filter(l => l.visible); const visible = layers.filter(l => l.visible);
if (visible.length < 2) return null; if (visible.length < 2) return null;
const bgId = visible[0].id; const bgId = visible[0].id;
+4 -2
View File
@@ -23,8 +23,10 @@
* @returns {{x: number, y: number, guides: Array}} * @returns {{x: number, y: number, guides: Array}}
*/ */
export function computeSnap(layer, nx, ny, ctx) { export function computeSnap(layer, nx, ny, ctx) {
const SNAP_PX = 6 / Math.max(ctx.zoom, 0.0001); if (!layer || !layer.canvas || !ctx) return { x: nx, y: ny, guides: [] };
const cw = ctx.canvasW, ch = ctx.canvasH; const zoom = Number.isFinite(Number(ctx.zoom)) ? Number(ctx.zoom) : 1;
const SNAP_PX = 6 / Math.max(zoom, 0.0001);
const cw = Number(ctx.canvasW) || 0, ch = Number(ctx.canvasH) || 0;
const w = layer.canvas.width, h = layer.canvas.height; const w = layer.canvas.width, h = layer.canvas.height;
const vTargets = [ const vTargets = [
+7 -2
View File
@@ -15,16 +15,21 @@ def test_preview_text_ignores_non_string(monkeypatch):
assert cli._preview_text(None) == "" assert cli._preview_text(None) == ""
assert cli._preview_text(123) == "" assert cli._preview_text(123) == ""
assert cli._preview_text("p" * 250) == "p" * 200 assert cli._preview_text("p" * 250) == "p" * 200
assert cli._text_field("ok") == "ok"
assert cli._text_field(123) == ""
def test_serialize_image_does_not_crash_on_non_string_prompt(monkeypatch): def test_serialize_image_does_not_crash_on_non_string_prompt(monkeypatch):
make_core_db_stub(monkeypatch, models=["GalleryImage", "GalleryAlbum"]) make_core_db_stub(monkeypatch, models=["GalleryImage", "GalleryAlbum"])
cli = load_script("odysseus-gallery") cli = load_script("odysseus-gallery")
img = SimpleNamespace( img = SimpleNamespace(
id="i1", filename="a.png", prompt=123, model=None, size=None, tags=None, id="i1", filename=123, prompt=123, model=123, size=None, tags=["bad"],
favorite=0, album_id=None, session_id=None, width=1, height=1, file_size=1, favorite=0, album_id=None, session_id=None, width=1, height=1, file_size=1,
taken_at=None, camera_make=None, camera_model=None, created_at=None, taken_at=None, camera_make=123, camera_model=None, created_at=None,
) )
out = cli._serialize_image(img) out = cli._serialize_image(img)
assert out["prompt"] == "" assert out["prompt"] == ""
assert out["filename"] == ""
assert out["model"] == ""
assert out["tags"] == ""
assert out["id"] == "i1" assert out["id"] == "i1"
+7
View File
@@ -48,3 +48,10 @@ def test_recipient_list_rejects_empty_envelope(monkeypatch):
assert exc.code == 1 assert exc.code == 1
else: else:
raise AssertionError("expected empty recipient list to exit") raise AssertionError("expected empty recipient list to exit")
def test_split_recipients_ignores_non_string_values(monkeypatch):
cli = _load_mail_cli(monkeypatch)
assert cli._split_recipients(None) == []
assert cli._split_recipients(["a@example.test"]) == []
+22
View File
@@ -46,3 +46,25 @@ def test_serialize_keeps_list_note_items(monkeypatch):
) )
assert cli._serialize(note)["items"] == [{"text": "done"}] assert cli._serialize(note)["items"] == [{"text": "done"}]
def test_serialize_skips_invalid_note_item_rows(monkeypatch):
make_core_db_stub(monkeypatch, models=["Note"])
cli = load_script("odysseus-notes")
note = SimpleNamespace(
id="n1",
title="Checklist",
content="",
items='[{"text": "done"}, "bad", null, 3]',
note_type="checklist",
color=None,
label=None,
pinned=False,
archived=False,
due_date=None,
source=None,
created_at=None,
updated_at=None,
)
assert cli._serialize(note)["items"] == [{"text": "done"}]
+8
View File
@@ -0,0 +1,8 @@
"""Shared imports for calendar route tests."""
def import_calendar_routes():
"""Import the calendar routes module after test stubs are installed."""
import routes.calendar_routes as cal
return cal
+10
View File
@@ -138,6 +138,16 @@ def test_atomic_write_text_leaves_no_tmp_file(tmp_path):
assert _tmp_siblings(tmp_path, "note.txt") == [] assert _tmp_siblings(tmp_path, "note.txt") == []
def test_atomic_write_text_rejects_non_string_before_tmp_file(tmp_path):
target = tmp_path / "note.txt"
with pytest.raises(TypeError):
atomic_write_text(str(target), 123)
assert not target.exists()
assert _tmp_siblings(tmp_path, "note.txt") == []
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# atomic_write_text — failure path: target preserved when replace fails. # atomic_write_text — failure path: target preserved when replace fails.
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
+4 -4
View File
@@ -13,7 +13,7 @@ The fallback now normalizes to UTC and strips tz, exactly like the ISO path.
""" """
import pytest import pytest
from tests.test_null_owner_gates import _import_calendar_helpers from tests.helpers.calendar_routes import import_calendar_routes
# Inputs datetime.fromisoformat() rejects (so they hit the dateutil fallback) # Inputs datetime.fromisoformat() rejects (so they hit the dateutil fallback)
# but that carry a numeric UTC offset dateutil resolves to tz-aware. # but that carry a numeric UTC offset dateutil resolves to tz-aware.
@@ -25,7 +25,7 @@ _OFFSET_NONISO = [
@pytest.mark.parametrize("s", _OFFSET_NONISO) @pytest.mark.parametrize("s", _OFFSET_NONISO)
def test_parse_dt_dateutil_fallback_returns_naive(s): def test_parse_dt_dateutil_fallback_returns_naive(s):
cal = _import_calendar_helpers() cal = import_calendar_routes()
d = cal._parse_dt(s) d = cal._parse_dt(s)
assert d.tzinfo is None, f"{s!r} leaked tz-aware: {d!r}" assert d.tzinfo is None, f"{s!r} leaked tz-aware: {d!r}"
# +0900 14:00 -> 05:00 UTC, naive. # +0900 14:00 -> 05:00 UTC, naive.
@@ -34,13 +34,13 @@ def test_parse_dt_dateutil_fallback_returns_naive(s):
@pytest.mark.parametrize("s", _OFFSET_NONISO) @pytest.mark.parametrize("s", _OFFSET_NONISO)
def test_parse_dt_pair_fallback_returns_naive(s): def test_parse_dt_pair_fallback_returns_naive(s):
cal = _import_calendar_helpers() cal = import_calendar_routes()
dt, _is_utc = cal._parse_dt_pair(s) dt, _is_utc = cal._parse_dt_pair(s)
assert dt.tzinfo is None, f"{s!r} leaked tz-aware via _parse_dt_pair: {dt!r}" assert dt.tzinfo is None, f"{s!r} leaked tz-aware via _parse_dt_pair: {dt!r}"
def test_parse_dt_naive_input_unchanged(): def test_parse_dt_naive_input_unchanged():
cal = _import_calendar_helpers() cal = import_calendar_routes()
d = cal._parse_dt("January 5, 2026 14:00") # no offset -> stays as parsed d = cal._parse_dt("January 5, 2026 14:00") # no offset -> stays as parsed
assert d.tzinfo is None assert d.tzinfo is None
assert (d.hour, d.minute) == (14, 0) assert (d.hour, d.minute) == (14, 0)
+22 -22
View File
@@ -1,8 +1,8 @@
"""Regression tests for calendar recurrence expansion. """Regression tests for calendar recurrence expansion.
Tests _expand_rrule and _resolve_base_uid — imported directly from Tests _expand_rrule and _resolve_base_uid — imported directly from
routes/calendar_routes using the same stub-friendly import pattern routes/calendar_routes using the shared stub-friendly test helper.
as test_null_owner_gates.py. No live DB or FastAPI test client needed. No live DB or FastAPI test client needed.
""" """
from datetime import datetime, timedelta from datetime import datetime, timedelta
@@ -10,34 +10,34 @@ from types import SimpleNamespace
import pytest import pytest
from tests.test_null_owner_gates import _import_calendar_helpers from tests.helpers.calendar_routes import import_calendar_routes
# ── _resolve_base_uid ────────────────────────────────────────────────── # ── _resolve_base_uid ──────────────────────────────────────────────────
def test_resolve_base_uid_plain_passthrough(): def test_resolve_base_uid_plain_passthrough():
cal = _import_calendar_helpers() cal = import_calendar_routes()
assert cal._resolve_base_uid("evt-123") == "evt-123" assert cal._resolve_base_uid("evt-123") == "evt-123"
def test_resolve_base_uid_compound_strips_suffix_date(): def test_resolve_base_uid_compound_strips_suffix_date():
cal = _import_calendar_helpers() cal = import_calendar_routes()
assert cal._resolve_base_uid("evt-123::2026-06-15") == "evt-123" assert cal._resolve_base_uid("evt-123::2026-06-15") == "evt-123"
def test_resolve_base_uid_compound_strips_suffix_datetime(): def test_resolve_base_uid_compound_strips_suffix_datetime():
cal = _import_calendar_helpers() cal = import_calendar_routes()
assert cal._resolve_base_uid("evt-123::2026-06-15T09:00") == "evt-123" assert cal._resolve_base_uid("evt-123::2026-06-15T09:00") == "evt-123"
def test_resolve_base_uid_rejects_empty(): def test_resolve_base_uid_rejects_empty():
cal = _import_calendar_helpers() cal = import_calendar_routes()
with pytest.raises(ValueError, match="empty uid"): with pytest.raises(ValueError, match="empty uid"):
cal._resolve_base_uid("") cal._resolve_base_uid("")
def test_resolve_base_uid_rejects_missing_base(): def test_resolve_base_uid_rejects_missing_base():
cal = _import_calendar_helpers() cal = import_calendar_routes()
with pytest.raises(ValueError, match="malformed compound UID"): with pytest.raises(ValueError, match="malformed compound UID"):
cal._resolve_base_uid("::2026-06-15") cal._resolve_base_uid("::2026-06-15")
@@ -74,7 +74,7 @@ def _make_event(**overrides):
def test_expand_non_recurring_returns_single(): def test_expand_non_recurring_returns_single():
"""Non-recurring events pass through unchanged with series_uid=uid.""" """Non-recurring events pass through unchanged with series_uid=uid."""
cal = _import_calendar_helpers() cal = import_calendar_routes()
ev = _make_event(rrule="") ev = _make_event(rrule="")
results = cal._expand_rrule(ev, datetime(2026, 5, 1), datetime(2026, 7, 1)) results = cal._expand_rrule(ev, datetime(2026, 5, 1), datetime(2026, 7, 1))
@@ -108,7 +108,7 @@ def test_expand_yearly_old_dtstart_later_year_single_occurrence():
This is the explicit regression case from PR review feedback. This is the explicit regression case from PR review feedback.
""" """
cal = _import_calendar_helpers() cal = import_calendar_routes()
ev = _make_event( ev = _make_event(
uid="evt-bday-001", uid="evt-bday-001",
summary="Annual Review", summary="Annual Review",
@@ -136,7 +136,7 @@ def test_expand_yearly_narrow_window_after_dtstart_returns_one():
"""DTSTART=2020, query just two months in 2029 — should return """DTSTART=2020, query just two months in 2029 — should return
exactly one occurrence (the one that falls in that window). exactly one occurrence (the one that falls in that window).
""" """
cal = _import_calendar_helpers() cal = import_calendar_routes()
ev = _make_event( ev = _make_event(
uid="evt-ann", uid="evt-ann",
dtstart=datetime(2020, 3, 1), dtstart=datetime(2020, 3, 1),
@@ -155,7 +155,7 @@ def test_expand_yearly_strict_before_window_returns_empty():
"""DTSTART=2020, query a window that ends before the yearly """DTSTART=2020, query a window that ends before the yearly
occurrence in that year. Should return zero. occurrence in that year. Should return zero.
""" """
cal = _import_calendar_helpers() cal = import_calendar_routes()
ev = _make_event( ev = _make_event(
uid="evt-late", uid="evt-late",
dtstart=datetime(2020, 12, 25), dtstart=datetime(2020, 12, 25),
@@ -172,7 +172,7 @@ def test_expand_yearly_strict_after_window_returns_empty():
"""DTSTART=2020. Query a window that starts after the occurrence in """DTSTART=2020. Query a window that starts after the occurrence in
that year. Should return zero. that year. Should return zero.
""" """
cal = _import_calendar_helpers() cal = import_calendar_routes()
ev = _make_event( ev = _make_event(
uid="evt-early", uid="evt-early",
dtstart=datetime(2020, 1, 15), dtstart=datetime(2020, 1, 15),
@@ -189,7 +189,7 @@ def test_expand_weekly_unique_no_overwrites():
"""Multiple occurrences from the same series must have unique UIDs """Multiple occurrences from the same series must have unique UIDs
so _allEvents[uid] = ev doesn't overwrite earlier ones. so _allEvents[uid] = ev doesn't overwrite earlier ones.
""" """
cal = _import_calendar_helpers() cal = import_calendar_routes()
ev = _make_event( ev = _make_event(
uid="evt-wk", uid="evt-wk",
dtstart=datetime(2026, 6, 1, 9, 0), dtstart=datetime(2026, 6, 1, 9, 0),
@@ -210,7 +210,7 @@ def test_expand_weekly_unique_no_overwrites():
def test_expand_monthly_all_day(): def test_expand_monthly_all_day():
cal = _import_calendar_helpers() cal = import_calendar_routes()
ev = _make_event( ev = _make_event(
uid="evt-rent", uid="evt-rent",
dtstart=datetime(2026, 1, 1), dtstart=datetime(2026, 1, 1),
@@ -228,7 +228,7 @@ def test_expand_monthly_all_day():
def test_expand_bad_rrule_graceful(): def test_expand_bad_rrule_graceful():
"""Malformed rrule should fall back to returning the base event, """Malformed rrule should fall back to returning the base event,
but only when the base event overlaps the requested window.""" but only when the base event overlaps the requested window."""
cal = _import_calendar_helpers() cal = import_calendar_routes()
ev = _make_event( ev = _make_event(
uid="evt-broken", uid="evt-broken",
rrule="FREQ=GARBAGE", rrule="FREQ=GARBAGE",
@@ -243,7 +243,7 @@ def test_expand_bad_rrule_graceful():
def test_expand_bad_rrule_fallback_rejects_non_overlapping(): def test_expand_bad_rrule_fallback_rejects_non_overlapping():
"""Malformed rrule with a base event outside the requested window """Malformed rrule with a base event outside the requested window
must return zero results, not leak the event into an unrelated range.""" must return zero results, not leak the event into an unrelated range."""
cal = _import_calendar_helpers() cal = import_calendar_routes()
ev = _make_event( ev = _make_event(
uid="evt-old-broken", uid="evt-old-broken",
dtstart=datetime(2020, 1, 1, 9, 0), dtstart=datetime(2020, 1, 1, 9, 0),
@@ -261,7 +261,7 @@ def test_expand_bad_rrule_fallback_rejects_non_overlapping():
def test_expand_exclusive_end_boundary(): def test_expand_exclusive_end_boundary():
"""An occurrence whose start equals the window end must be excluded. """An occurrence whose start equals the window end must be excluded.
The contract is [start, end), same as the non-recurring SQL filter.""" The contract is [start, end), same as the non-recurring SQL filter."""
cal = _import_calendar_helpers() cal = import_calendar_routes()
ev = _make_event( ev = _make_event(
uid="evt-daily", uid="evt-daily",
dtstart=datetime(2026, 6, 1, 9, 0), dtstart=datetime(2026, 6, 1, 9, 0),
@@ -278,7 +278,7 @@ def test_expand_exclusive_end_boundary():
def test_expand_multi_day_crossing_range_start(): def test_expand_multi_day_crossing_range_start():
"""A multi-day occurrence that starts before the window but ends inside """A multi-day occurrence that starts before the window but ends inside
it must be included (matching non-recurring overlap: dtend > start).""" it must be included (matching non-recurring overlap: dtend > start)."""
cal = _import_calendar_helpers() cal = import_calendar_routes()
ev = _make_event( ev = _make_event(
uid="evt-weekly-multi", uid="evt-weekly-multi",
summary="Weekend Trip", summary="Weekend Trip",
@@ -303,7 +303,7 @@ def test_expand_multi_day_crossing_range_start():
def test_expand_multi_day_fully_before_window(): def test_expand_multi_day_fully_before_window():
"""A multi-day occurrence that ends exactly at the window start """A multi-day occurrence that ends exactly at the window start
must be excluded (occ_end <= start).""" must be excluded (occ_end <= start)."""
cal = _import_calendar_helpers() cal = import_calendar_routes()
ev = _make_event( ev = _make_event(
uid="evt-multi", uid="evt-multi",
dtstart=datetime(2026, 5, 29, 18, 0), dtstart=datetime(2026, 5, 29, 18, 0),
@@ -319,7 +319,7 @@ def test_expand_multi_day_fully_before_window():
def test_expand_metadata_inheritance(): def test_expand_metadata_inheritance():
"""Occurrence dicts must carry the base event's metadata """Occurrence dicts must carry the base event's metadata
(summary, importance, event_type, color, location).""" (summary, importance, event_type, color, location)."""
cal = _import_calendar_helpers() cal = import_calendar_routes()
ev = _make_event( ev = _make_event(
uid="evt-meta", uid="evt-meta",
summary="Board Meeting", summary="Board Meeting",
@@ -341,7 +341,7 @@ def test_expand_metadata_inheritance():
def test_expand_daily_rrule_large_window_is_capped_and_marked_truncated(): def test_expand_daily_rrule_large_window_is_capped_and_marked_truncated():
"""Wide recurring windows must not materialize unbounded occurrence lists.""" """Wide recurring windows must not materialize unbounded occurrence lists."""
cal = _import_calendar_helpers() cal = import_calendar_routes()
ev = _make_event( ev = _make_event(
uid="evt-daily-cap", uid="evt-daily-cap",
dtstart=datetime(2020, 1, 1, 9, 0), dtstart=datetime(2020, 1, 1, 9, 0),
+2 -2
View File
@@ -24,7 +24,7 @@ UNTIL must expand to all of its occurrences.
from datetime import datetime from datetime import datetime
from types import SimpleNamespace from types import SimpleNamespace
from tests.test_null_owner_gates import _import_calendar_helpers from tests.helpers.calendar_routes import import_calendar_routes
_MOCK_CAL = SimpleNamespace(name="Personal", color="#5b8abf") _MOCK_CAL = SimpleNamespace(name="Personal", color="#5b8abf")
@@ -55,7 +55,7 @@ def _make_event(**overrides):
def test_expand_rrule_with_utc_until_keeps_all_occurrences(): def test_expand_rrule_with_utc_until_keeps_all_occurrences():
"""FREQ=DAILY;UNTIL=...Z must expand to every occurrence, not collapse """FREQ=DAILY;UNTIL=...Z must expand to every occurrence, not collapse
to a single non-recurring event.""" to a single non-recurring event."""
cal = _import_calendar_helpers() cal = import_calendar_routes()
ev = _make_event(rrule="FREQ=DAILY;UNTIL=20240105T090000Z") ev = _make_event(rrule="FREQ=DAILY;UNTIL=20240105T090000Z")
results = cal._expand_rrule(ev, datetime(2024, 1, 1), datetime(2024, 1, 10)) results = cal._expand_rrule(ev, datetime(2024, 1, 1), datetime(2024, 1, 10))
+5
View File
@@ -33,6 +33,11 @@ def test_unknown_window_falls_back_to_configured():
assert compute_input_token_budget(0, 0, explicit=False) == 6000 # default assert compute_input_token_budget(0, 0, explicit=False) == 6000 # default
def test_invalid_numeric_inputs_fall_back_cleanly():
assert compute_input_token_budget("bad", "also bad", explicit=False) == 6000
assert compute_input_token_budget("bad", 128000, explicit=True) == int(128000 * 0.85)
def test_is_setting_overridden_reads_raw_saved_file(tmp_path, monkeypatch): def test_is_setting_overridden_reads_raw_saved_file(tmp_path, monkeypatch):
import src.settings as settings import src.settings as settings
@@ -0,0 +1,32 @@
"""Pin harmonize mask helpers against invalid layer lists.
Driven through `node --input-type=module`; skips without node.
"""
import json
import shutil
import subprocess
from pathlib import Path
import pytest
_REPO = Path(__file__).resolve().parent.parent
_HELPER = _REPO / "static" / "js" / "editor" / "harmonize-masks.js"
_HAS_NODE = shutil.which("node") is not None
@pytest.mark.skipif(not _HAS_NODE, reason="node binary not on PATH")
def test_layer_union_alpha_returns_null_for_non_array_layers():
js = f"""
import {{ layerUnionAlpha, seamMask, layerBodyMask }} from '{_HELPER.as_posix()}';
console.log(JSON.stringify([
layerUnionAlpha(10, 10, null),
seamMask(10, 10, {{"bad": true}}),
layerBodyMask(10, 10, "bad")
]));
"""
proc = subprocess.run(
["node", "--input-type=module"],
input=js, capture_output=True, text=True, cwd=str(_REPO), timeout=30,
)
assert proc.returncode == 0, proc.stderr
assert json.loads(proc.stdout.strip()) == [None, None, None]
+4 -4
View File
@@ -1,9 +1,9 @@
"""Tests for iCalendar TEXT escaping in calendar export (RFC 5545 §3.3.11).""" """Tests for iCalendar TEXT escaping in calendar export (RFC 5545 §3.3.11)."""
from tests.test_null_owner_gates import _import_calendar_helpers from tests.helpers.calendar_routes import import_calendar_routes
def _esc(): def _esc():
return _import_calendar_helpers()._ics_escape return import_calendar_routes()._ics_escape
def test_escapes_comma_and_semicolon(): def test_escapes_comma_and_semicolon():
@@ -26,7 +26,7 @@ def test_empty_and_none_safe():
def test_safe_ics_filename_strips_header_metacharacters(): def test_safe_ics_filename_strips_header_metacharacters():
safe_filename = _import_calendar_helpers()._safe_ics_filename safe_filename = import_calendar_routes()._safe_ics_filename
assert ( assert (
safe_filename('Work\r\nX-Injected: yes";/..\\evil') safe_filename('Work\r\nX-Injected: yes";/..\\evil')
@@ -35,7 +35,7 @@ def test_safe_ics_filename_strips_header_metacharacters():
def test_safe_ics_filename_falls_back_for_empty_names(): def test_safe_ics_filename_falls_back_for_empty_names():
safe_filename = _import_calendar_helpers()._safe_ics_filename safe_filename = import_calendar_routes()._safe_ics_filename
assert safe_filename("////") == "calendar.ics" assert safe_filename("////") == "calendar.ics"
assert safe_filename(None) == "calendar.ics" assert safe_filename(None) == "calendar.ics"
+82
View File
@@ -9,6 +9,12 @@ def test_detects_ollama_cloud_native_provider():
assert llm_core._detect_provider("https://ollama.com/api/chat") == "ollama" assert llm_core._detect_provider("https://ollama.com/api/chat") == "ollama"
def test_detects_bare_local_ollama_as_native_provider():
assert llm_core._detect_provider("http://localhost:11434") == "ollama"
assert llm_core._detect_provider("http://127.0.0.1:11434/") == "ollama"
assert llm_core._detect_provider("http://localhost:11434/v1") == "openai"
def test_llm_call_posts_native_ollama_payload(monkeypatch): def test_llm_call_posts_native_ollama_payload(monkeypatch):
seen = {} seen = {}
@@ -43,6 +49,82 @@ def test_llm_call_posts_native_ollama_payload(monkeypatch):
assert seen["json"]["options"] == {"temperature": 0.2, "num_predict": 7} assert seen["json"]["options"] == {"temperature": 0.2, "num_predict": 7}
def test_llm_call_posts_bare_local_ollama_to_native_api(monkeypatch):
seen = {}
def fake_post(url, headers=None, json=None, timeout=None):
seen["url"] = url
seen["json"] = json
request = httpx.Request("POST", url)
return httpx.Response(
200,
request=request,
json={"message": {"content": "OK"}, "done": True},
)
monkeypatch.setattr(llm_core.httpx, "post", fake_post)
result = llm_core.llm_call(
"http://localhost:11434",
"llama3.2",
[{"role": "user", "content": "Say OK"}],
)
assert result == "OK"
assert seen["url"] == "http://localhost:11434/api/chat"
assert seen["json"]["stream"] is False
def test_openai_compatible_chat_url_shapes(monkeypatch):
seen = []
def fake_post(url, headers=None, json=None, timeout=None):
seen.append(url)
request = httpx.Request("POST", url)
return httpx.Response(
200,
request=request,
json={"choices": [{"message": {"content": "OK"}}]},
)
monkeypatch.setattr(llm_core.httpx, "post", fake_post)
llm_core._response_cache.clear()
cases = [
("http://localhost:11434/v1", "http://localhost:11434/v1/chat/completions"),
(
"http://localhost:11434/v1/chat/completions",
"http://localhost:11434/v1/chat/completions",
),
]
for i, (base_url, expected_url) in enumerate(cases):
result = llm_core.llm_call(
base_url,
f"openai-compatible-{i}",
[{"role": "user", "content": f"Say OK {i}"}],
)
assert result == "OK"
assert seen[-1] == expected_url
def test_list_model_ids_from_openai_compatible_v1(monkeypatch):
seen = {}
def fake_get(url, headers=None, timeout=None):
seen["url"] = url
request = httpx.Request("GET", url)
return httpx.Response(
200,
request=request,
json={"data": [{"id": "qwen2.5-coder:7b"}]},
)
monkeypatch.setattr(llm_core.httpx, "get", fake_get)
assert llm_core.list_model_ids("http://localhost:11434/v1") == ["qwen2.5-coder:7b"]
assert seen["url"] == "http://localhost:11434/v1/models"
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Tool-call argument serialization for native Ollama # Tool-call argument serialization for native Ollama
# #
+7 -17
View File
@@ -18,6 +18,8 @@ import pytest
from types import SimpleNamespace from types import SimpleNamespace
from unittest.mock import MagicMock from unittest.mock import MagicMock
from tests.helpers.calendar_routes import import_calendar_routes
# `tests/conftest.py` stubs the heavy optional deps. We additionally # `tests/conftest.py` stubs the heavy optional deps. We additionally
# stub `core.database` here because the real module instantiates # stub `core.database` here because the real module instantiates
# SQLAlchemy declarative classes at import-time — which blows up under # SQLAlchemy declarative classes at import-time — which blows up under
@@ -64,20 +66,8 @@ from fastapi import HTTPException
# calendar._get_or_404_calendar / _get_or_404_event # calendar._get_or_404_calendar / _get_or_404_event
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
def _import_calendar_helpers():
"""Import the two private gate helpers without booting the full
calendar router. We patch sys.modules so the module-load side
effects (DB import) don't blow up under the conftest stubs."""
mod_name = "routes.calendar_routes"
if mod_name in sys.modules:
return sys.modules[mod_name]
# core.database is stubbed by conftest already; the module should
# import cleanly.
return __import__(mod_name, fromlist=["_get_or_404_calendar", "_get_or_404_event"])
def test_calendar_gate_rejects_null_owner_for_authenticated_user(): def test_calendar_gate_rejects_null_owner_for_authenticated_user():
cal_mod = _import_calendar_helpers() cal_mod = import_calendar_routes()
db = MagicMock() db = MagicMock()
cal = SimpleNamespace(id="c1", owner=None) cal = SimpleNamespace(id="c1", owner=None)
db.query.return_value.filter.return_value.first.return_value = cal db.query.return_value.filter.return_value.first.return_value = cal
@@ -87,7 +77,7 @@ def test_calendar_gate_rejects_null_owner_for_authenticated_user():
def test_calendar_gate_rejects_cross_owner(): def test_calendar_gate_rejects_cross_owner():
cal_mod = _import_calendar_helpers() cal_mod = import_calendar_routes()
db = MagicMock() db = MagicMock()
cal = SimpleNamespace(id="c1", owner="bob") cal = SimpleNamespace(id="c1", owner="bob")
db.query.return_value.filter.return_value.first.return_value = cal db.query.return_value.filter.return_value.first.return_value = cal
@@ -97,7 +87,7 @@ def test_calendar_gate_rejects_cross_owner():
def test_calendar_gate_accepts_matching_owner(): def test_calendar_gate_accepts_matching_owner():
cal_mod = _import_calendar_helpers() cal_mod = import_calendar_routes()
db = MagicMock() db = MagicMock()
cal = SimpleNamespace(id="c1", owner="alice") cal = SimpleNamespace(id="c1", owner="alice")
db.query.return_value.filter.return_value.first.return_value = cal db.query.return_value.filter.return_value.first.return_value = cal
@@ -106,7 +96,7 @@ def test_calendar_gate_accepts_matching_owner():
def test_calendar_event_gate_rejects_null_owner_calendar(): def test_calendar_event_gate_rejects_null_owner_calendar():
cal_mod = _import_calendar_helpers() cal_mod = import_calendar_routes()
db = MagicMock() db = MagicMock()
cal = SimpleNamespace(owner=None) cal = SimpleNamespace(owner=None)
ev = SimpleNamespace(uid="e1", calendar=cal) ev = SimpleNamespace(uid="e1", calendar=cal)
@@ -117,7 +107,7 @@ def test_calendar_event_gate_rejects_null_owner_calendar():
def test_calendar_event_gate_rejects_cross_owner(): def test_calendar_event_gate_rejects_cross_owner():
cal_mod = _import_calendar_helpers() cal_mod = import_calendar_routes()
db = MagicMock() db = MagicMock()
cal = SimpleNamespace(owner="bob") cal = SimpleNamespace(owner="bob")
ev = SimpleNamespace(uid="e1", calendar=cal) ev = SimpleNamespace(uid="e1", calendar=cal)
+15 -1
View File
@@ -1,4 +1,4 @@
from src.personal_docs import retrieve_personal_keyword from src.personal_docs import retrieve_personal_keyword, split_chunks
def test_retrieve_personal_keyword_skips_non_dict_rows(): def test_retrieve_personal_keyword_skips_non_dict_rows():
@@ -19,3 +19,17 @@ def test_retrieve_personal_keyword_tolerates_missing_chunks_key():
index = [{"name": "empty.txt"}, {"name": "doc.txt", "chunks": ["alpha beta gamma"]}] index = [{"name": "empty.txt"}, {"name": "doc.txt", "chunks": ["alpha beta gamma"]}]
out = retrieve_personal_keyword(index, "beta", k=5) out = retrieve_personal_keyword(index, "beta", k=5)
assert out == ["[doc.txt :: chunk 1]\nalpha beta gamma"] assert out == ["[doc.txt :: chunk 1]\nalpha beta gamma"]
def test_retrieve_personal_keyword_ignores_non_string_text():
index = [{"name": "doc.txt", "chunks": [None, ["beta"], "alpha beta gamma"]}]
assert retrieve_personal_keyword(index, ["beta"], k=5) == []
assert retrieve_personal_keyword(index, "beta", k=5) == [
"[doc.txt :: chunk 3]\nalpha beta gamma"
]
def test_split_chunks_ignores_non_string_text():
assert split_chunks(None, size=1000, overlap=200) == []
assert split_chunks(["hello"], size=1000, overlap=200) == []
-244
View File
@@ -1,244 +0,0 @@
"""Provider / endpoint resolution tests against the REAL resolver.
`test_endpoint_resolver.py` deliberately *copies* the pure functions to avoid
import side effects. The downside is that those copies silently drift from the
shipped code — they already lag `src/endpoint_resolver.py` (no OpenRouter
headers, no `anthropic.com` host matching). This module instead imports the
real `src.endpoint_resolver`, so it fails the moment the shipped resolution
logic stops matching documented provider behavior. `conftest.py` stubs the
heavy deps (sqlalchemy, `src.database`), so the import is side-effect free.
Covers every provider named in ROADMAP.md "Provider setup/probing audit":
Anthropic, Gemini, Groq, xAI, OpenRouter, OpenAI, DeepSeek — plus Ollama
(local + cloud) and the Tailscale self-host fallback.
"""
import json
import socket
import types
import pytest
from src import endpoint_resolver as er
@pytest.fixture
def no_dns(monkeypatch):
"""Neutralize resolve_url so URL-building tests never touch DNS/Tailscale.
build_chat_url/build_models_url call the module-global resolve_url first;
patching it on the module makes those calls a no-op (functions resolve
globals by name at call time).
"""
monkeypatch.setattr(er, "resolve_url", lambda u: u)
# (id, base_url, expected_chat_url, expected_models_url)
PROVIDER_CASES = [
("openai", "https://api.openai.com/v1",
"https://api.openai.com/v1/chat/completions",
"https://api.openai.com/v1/models"),
("openai_pathless", "https://api.openai.com",
"https://api.openai.com/v1/chat/completions",
"https://api.openai.com/v1/models"),
("anthropic", "https://api.anthropic.com",
"https://api.anthropic.com/v1/messages",
"https://api.anthropic.com/v1/models"),
# Anthropic base that already carries /v1 must not become /v1/v1/messages.
("anthropic_v1", "https://api.anthropic.com/v1",
"https://api.anthropic.com/v1/messages",
"https://api.anthropic.com/v1/models"),
("openrouter", "https://openrouter.ai/api/v1",
"https://openrouter.ai/api/v1/chat/completions",
"https://openrouter.ai/api/v1/models"),
("groq", "https://api.groq.com/openai/v1",
"https://api.groq.com/openai/v1/chat/completions",
"https://api.groq.com/openai/v1/models"),
("nvidia", "https://integrate.api.nvidia.com/v1",
"https://integrate.api.nvidia.com/v1/chat/completions",
"https://integrate.api.nvidia.com/v1/models"),
("xai", "https://api.x.ai/v1",
"https://api.x.ai/v1/chat/completions",
"https://api.x.ai/v1/models"),
("deepseek", "https://api.deepseek.com",
"https://api.deepseek.com/chat/completions",
"https://api.deepseek.com/v1/models"),
# Gemini's OpenAI-compatible surface — treated as a generic OpenAI endpoint.
("gemini_openai", "https://generativelanguage.googleapis.com/v1beta/openai",
"https://generativelanguage.googleapis.com/v1beta/openai/chat/completions",
"https://generativelanguage.googleapis.com/v1beta/openai/models"),
("ollama_local", "http://localhost:11434/api",
"http://localhost:11434/api/chat",
"http://localhost:11434/api/tags"),
("ollama_cloud", "https://ollama.com",
"https://ollama.com/api/chat",
"https://ollama.com/api/tags"),
]
@pytest.mark.parametrize(
"base,expected", [(c[1], c[2]) for c in PROVIDER_CASES],
ids=[c[0] for c in PROVIDER_CASES],
)
def test_build_chat_url(no_dns, base, expected):
assert er.build_chat_url(base) == expected
@pytest.mark.parametrize(
"base,expected", [(c[1], c[3]) for c in PROVIDER_CASES],
ids=[c[0] for c in PROVIDER_CASES],
)
def test_build_models_url(no_dns, base, expected):
assert er.build_models_url(base) == expected
def test_chat_url_never_double_prefixes_anthropic(no_dns):
"""Regression guard: the /v1 collapse must not produce /v1/v1/messages."""
url = er.build_chat_url("https://api.anthropic.com/v1")
assert "/v1/v1/" not in url
assert url.count("/v1/messages") == 1
# ── Auth headers per provider ──
def test_headers_anthropic_uses_x_api_key():
h = er.build_headers("secret", "https://api.anthropic.com")
assert h["x-api-key"] == "secret"
assert h["anthropic-version"] == "2023-06-01"
assert "Authorization" not in h
def test_headers_anthropic_without_key_still_sends_version():
h = er.build_headers(None, "https://api.anthropic.com")
assert h["anthropic-version"] == "2023-06-01"
assert "x-api-key" not in h
@pytest.mark.parametrize("base", [
"https://api.openai.com/v1",
"https://api.x.ai/v1",
"https://api.deepseek.com",
"https://api.groq.com/openai/v1",
"https://integrate.api.nvidia.com/v1",
"https://generativelanguage.googleapis.com/v1beta/openai",
])
def test_headers_openai_style_use_bearer(base):
h = er.build_headers("secret", base)
assert h["Authorization"] == "Bearer secret"
assert "HTTP-Referer" not in h
assert "x-api-key" not in h
def test_headers_openrouter_adds_attribution():
h = er.build_headers("secret", "https://openrouter.ai/api/v1")
assert h["Authorization"] == "Bearer secret"
# OpenRouter ranks/labels apps via these headers.
assert h["HTTP-Referer"].startswith("https://github.com/")
assert h["X-OpenRouter-Title"] == "Odysseus"
def test_headers_omit_authorization_when_no_key():
assert er.build_headers(None, "https://api.openai.com/v1") == {}
# ── normalize_base: strip whatever path the user pasted ──
@pytest.mark.parametrize("raw,expected", [
("https://api.openai.com/v1/chat/completions", "https://api.openai.com/v1"),
("https://api.openai.com/v1/completions", "https://api.openai.com/v1"),
("https://api.openai.com/v1/models/", "https://api.openai.com/v1"),
("https://api.anthropic.com/v1/messages", "https://api.anthropic.com"),
("http://localhost:11434/api/chat", "http://localhost:11434/api"),
("http://localhost:11434/api/tags", "http://localhost:11434/api"),
("http://localhost:11434/api/generate", "http://localhost:11434/api"),
("https://api.openai.com/v1/", "https://api.openai.com/v1"),
(" https://api.openai.com/v1 ", "https://api.openai.com/v1"),
("", ""),
(None, ""),
])
def test_normalize_base(raw, expected):
assert er.normalize_base(raw) == expected
# ── _first_chat_model: never auto-pick an embedding/tts/etc. model ──
def test_first_chat_model_skips_non_chat():
models = ["text-embedding-ada-002", "whisper-1", "gpt-4o", "dall-e-3"]
assert er._first_chat_model(models) == "gpt-4o"
def test_first_chat_model_falls_back_to_first_when_all_non_chat():
models = ["text-embedding-3-large", "text-embedding-3-small"]
assert er._first_chat_model(models) == "text-embedding-3-large"
@pytest.mark.parametrize("models", [[], None])
def test_first_chat_model_empty(models):
assert er._first_chat_model(models) is None
# ── provider-root helpers ──
@pytest.mark.parametrize("base,expected", [
("https://api.anthropic.com/v1", "https://api.anthropic.com"),
("https://api.anthropic.com", "https://api.anthropic.com"),
# /v1 on a non-Anthropic host (OpenAI-compatible) must be preserved.
("https://api.openai.com/v1", "https://api.openai.com/v1"),
])
def test_anthropic_api_root(base, expected):
assert er._anthropic_api_root(base) == expected
@pytest.mark.parametrize("base,expected", [
("https://ollama.com", "https://ollama.com/api"),
("http://localhost:11434/api", "http://localhost:11434/api"),
# A non-Ollama host is returned untouched.
("https://api.openai.com/v1", "https://api.openai.com/v1"),
])
def test_ollama_api_root(base, expected):
assert er._ollama_api_root(base) == expected
# ── resolve_url: Tailscale self-host fallback ──
# ROADMAP flags plain-HTTP Tailscale URLs as a self-host trap; resolve_url is
# the hop that rewrites an unresolvable hostname to its Tailscale IP.
class TestResolveUrlTailscale:
def setup_method(self):
# The module memoizes hostname→IP; clear it so cases don't bleed.
er._tailscale_cache.clear()
def test_dns_success_returns_url_unchanged(self, monkeypatch):
monkeypatch.setattr(
er.socket, "getaddrinfo",
lambda *a, **k: [(2, 1, 6, "", ("1.2.3.4", 0))],
)
assert er.resolve_url("http://myhost:7000/api") == "http://myhost:7000/api"
def test_dns_failure_rewrites_to_tailscale_ip(self, monkeypatch):
def _fail(*a, **k):
raise socket.gaierror("no DNS")
monkeypatch.setattr(er.socket, "getaddrinfo", _fail)
peers = {"Peer": {"x": {
"HostName": "myhost",
"DNSName": "myhost.tail.ts.net.",
"TailscaleIPs": ["100.64.0.5"],
}}}
monkeypatch.setattr(
er.subprocess, "run",
lambda *a, **k: types.SimpleNamespace(returncode=0, stdout=json.dumps(peers)),
)
# Port is preserved, host swapped for the Tailscale IP.
assert er.resolve_url("http://myhost:7000/api") == "http://100.64.0.5:7000/api"
def test_dns_failure_no_peer_match_keeps_url(self, monkeypatch):
def _fail(*a, **k):
raise socket.gaierror("no DNS")
monkeypatch.setattr(er.socket, "getaddrinfo", _fail)
monkeypatch.setattr(
er.subprocess, "run",
lambda *a, **k: types.SimpleNamespace(returncode=0, stdout=json.dumps({"Peer": {}})),
)
assert er.resolve_url("http://myhost:7000/api") == "http://myhost:7000/api"
def test_url_without_hostname_is_returned_as_is(self):
assert er.resolve_url("") == ""
+49
View File
@@ -0,0 +1,49 @@
"""Provider endpoint auth-header tests.
Covers ``build_headers`` for every provider: Anthropic (x-api-key + version
header), OpenAI-style providers (Bearer token), OpenRouter (Bearer + attribution
headers), and the no-key case.
"""
import pytest
from src import endpoint_resolver as er
def test_headers_anthropic_uses_x_api_key():
h = er.build_headers("secret", "https://api.anthropic.com")
assert h["x-api-key"] == "secret"
assert h["anthropic-version"] == "2023-06-01"
assert "Authorization" not in h
def test_headers_anthropic_without_key_still_sends_version():
h = er.build_headers(None, "https://api.anthropic.com")
assert h["anthropic-version"] == "2023-06-01"
assert "x-api-key" not in h
@pytest.mark.parametrize("base", [
"https://api.openai.com/v1",
"https://api.x.ai/v1",
"https://api.deepseek.com",
"https://api.groq.com/openai/v1",
"https://integrate.api.nvidia.com/v1",
"https://generativelanguage.googleapis.com/v1beta/openai",
])
def test_headers_openai_style_use_bearer(base):
h = er.build_headers("secret", base)
assert h["Authorization"] == "Bearer secret"
assert "HTTP-Referer" not in h
assert "x-api-key" not in h
def test_headers_openrouter_adds_attribution():
h = er.build_headers("secret", "https://openrouter.ai/api/v1")
assert h["Authorization"] == "Bearer secret"
# OpenRouter ranks/labels apps via these headers.
assert h["HTTP-Referer"].startswith("https://github.com/")
assert h["X-OpenRouter-Title"] == "Odysseus"
def test_headers_omit_authorization_when_no_key():
assert er.build_headers(None, "https://api.openai.com/v1") == {}
+25
View File
@@ -0,0 +1,25 @@
"""Provider endpoint model-selection tests.
Covers ``_first_chat_model``: auto-picking the first usable chat model from a
provider's model list, skipping embedding/tts/image models when possible.
"""
import pytest
from src import endpoint_resolver as er
# ── _first_chat_model: never auto-pick an embedding/tts/etc. model ──
def test_first_chat_model_skips_non_chat():
models = ["text-embedding-ada-002", "whisper-1", "gpt-4o", "dall-e-3"]
assert er._first_chat_model(models) == "gpt-4o"
def test_first_chat_model_falls_back_to_first_when_all_non_chat():
models = ["text-embedding-3-large", "text-embedding-3-small"]
assert er._first_chat_model(models) == "text-embedding-3-large"
@pytest.mark.parametrize("models", [[], None])
def test_first_chat_model_empty(models):
assert er._first_chat_model(models) is None
@@ -0,0 +1,49 @@
"""Provider endpoint normalization tests.
Covers ``normalize_base`` (strip whatever path the user pasted), and the
provider-root helpers ``_anthropic_api_root`` and ``_ollama_api_root``.
"""
import pytest
from src import endpoint_resolver as er
# ── normalize_base: strip whatever path the user pasted ──
@pytest.mark.parametrize("raw,expected", [
("https://api.openai.com/v1/chat/completions", "https://api.openai.com/v1"),
("https://api.openai.com/v1/completions", "https://api.openai.com/v1"),
("https://api.openai.com/v1/models/", "https://api.openai.com/v1"),
("https://api.anthropic.com/v1/messages", "https://api.anthropic.com"),
("http://localhost:11434/api/chat", "http://localhost:11434/api"),
("http://localhost:11434/api/tags", "http://localhost:11434/api"),
("http://localhost:11434/api/generate", "http://localhost:11434/api"),
("https://api.openai.com/v1/", "https://api.openai.com/v1"),
(" https://api.openai.com/v1 ", "https://api.openai.com/v1"),
("", ""),
(None, ""),
])
def test_normalize_base(raw, expected):
assert er.normalize_base(raw) == expected
# ── provider-root helpers ──
@pytest.mark.parametrize("base,expected", [
("https://api.anthropic.com/v1", "https://api.anthropic.com"),
("https://api.anthropic.com", "https://api.anthropic.com"),
# /v1 on a non-Anthropic host (OpenAI-compatible) must be preserved.
("https://api.openai.com/v1", "https://api.openai.com/v1"),
])
def test_anthropic_api_root(base, expected):
assert er._anthropic_api_root(base) == expected
@pytest.mark.parametrize("base,expected", [
("https://ollama.com", "https://ollama.com/api"),
("http://localhost:11434/api", "http://localhost:11434/api"),
# A non-Ollama host is returned untouched.
("https://api.openai.com/v1", "https://api.openai.com/v1"),
])
def test_ollama_api_root(base, expected):
assert er._ollama_api_root(base) == expected
@@ -0,0 +1,59 @@
"""Provider endpoint Tailscale URL-resolution tests.
Covers ``resolve_url``: the hop that rewrites an unresolvable hostname to its
Tailscale IP. ROADMAP flags plain-HTTP Tailscale URLs as a self-host trap;
resolve_url is the gate that handles that fallback.
"""
import json
import socket
import types
import pytest
from src import endpoint_resolver as er
# ── resolve_url: Tailscale self-host fallback ──
# ROADMAP flags plain-HTTP Tailscale URLs as a self-host trap; resolve_url is
# the hop that rewrites an unresolvable hostname to its Tailscale IP.
class TestResolveUrlTailscale:
def setup_method(self):
# The module memoizes hostname→IP; clear it so cases don't bleed.
er._tailscale_cache.clear()
def test_dns_success_returns_url_unchanged(self, monkeypatch):
monkeypatch.setattr(
er.socket, "getaddrinfo",
lambda *a, **k: [(2, 1, 6, "", ("1.2.3.4", 0))],
)
assert er.resolve_url("http://myhost:7000/api") == "http://myhost:7000/api"
def test_dns_failure_rewrites_to_tailscale_ip(self, monkeypatch):
def _fail(*a, **k):
raise socket.gaierror("no DNS")
monkeypatch.setattr(er.socket, "getaddrinfo", _fail)
peers = {"Peer": {"x": {
"HostName": "myhost",
"DNSName": "myhost.tail.ts.net.",
"TailscaleIPs": ["100.64.0.5"],
}}}
monkeypatch.setattr(
er.subprocess, "run",
lambda *a, **k: types.SimpleNamespace(returncode=0, stdout=json.dumps(peers)),
)
# Port is preserved, host swapped for the Tailscale IP.
assert er.resolve_url("http://myhost:7000/api") == "http://100.64.0.5:7000/api"
def test_dns_failure_no_peer_match_keeps_url(self, monkeypatch):
def _fail(*a, **k):
raise socket.gaierror("no DNS")
monkeypatch.setattr(er.socket, "getaddrinfo", _fail)
monkeypatch.setattr(
er.subprocess, "run",
lambda *a, **k: types.SimpleNamespace(returncode=0, stdout=json.dumps({"Peer": {}})),
)
assert er.resolve_url("http://myhost:7000/api") == "http://myhost:7000/api"
def test_url_without_hostname_is_returned_as_is(self):
assert er.resolve_url("") == ""
@@ -0,0 +1,86 @@
"""Provider endpoint URL-building tests.
Covers ``build_chat_url`` and ``build_models_url`` for every provider named in
ROADMAP.md: Anthropic, Gemini, Groq, xAI, OpenRouter, OpenAI, DeepSeek, Ollama
(local + cloud).
"""
import pytest
from src import endpoint_resolver as er
@pytest.fixture
def no_dns(monkeypatch):
"""Neutralize resolve_url so URL-building tests never touch DNS/Tailscale.
build_chat_url/build_models_url call the module-global resolve_url first;
patching it on the module makes those calls a no-op (functions resolve
globals by name at call time).
"""
monkeypatch.setattr(er, "resolve_url", lambda u: u)
# (id, base_url, expected_chat_url, expected_models_url)
PROVIDER_CASES = [
("openai", "https://api.openai.com/v1",
"https://api.openai.com/v1/chat/completions",
"https://api.openai.com/v1/models"),
("openai_pathless", "https://api.openai.com",
"https://api.openai.com/v1/chat/completions",
"https://api.openai.com/v1/models"),
("anthropic", "https://api.anthropic.com",
"https://api.anthropic.com/v1/messages",
"https://api.anthropic.com/v1/models"),
# Anthropic base that already carries /v1 must not become /v1/v1/messages.
("anthropic_v1", "https://api.anthropic.com/v1",
"https://api.anthropic.com/v1/messages",
"https://api.anthropic.com/v1/models"),
("openrouter", "https://openrouter.ai/api/v1",
"https://openrouter.ai/api/v1/chat/completions",
"https://openrouter.ai/api/v1/models"),
("groq", "https://api.groq.com/openai/v1",
"https://api.groq.com/openai/v1/chat/completions",
"https://api.groq.com/openai/v1/models"),
("nvidia", "https://integrate.api.nvidia.com/v1",
"https://integrate.api.nvidia.com/v1/chat/completions",
"https://integrate.api.nvidia.com/v1/models"),
("xai", "https://api.x.ai/v1",
"https://api.x.ai/v1/chat/completions",
"https://api.x.ai/v1/models"),
("deepseek", "https://api.deepseek.com",
"https://api.deepseek.com/chat/completions",
"https://api.deepseek.com/v1/models"),
# Gemini's OpenAI-compatible surface — treated as a generic OpenAI endpoint.
("gemini_openai", "https://generativelanguage.googleapis.com/v1beta/openai",
"https://generativelanguage.googleapis.com/v1beta/openai/chat/completions",
"https://generativelanguage.googleapis.com/v1beta/openai/models"),
("ollama_local", "http://localhost:11434/api",
"http://localhost:11434/api/chat",
"http://localhost:11434/api/tags"),
("ollama_cloud", "https://ollama.com",
"https://ollama.com/api/chat",
"https://ollama.com/api/tags"),
]
@pytest.mark.parametrize(
"base,expected", [(c[1], c[2]) for c in PROVIDER_CASES],
ids=[c[0] for c in PROVIDER_CASES],
)
def test_build_chat_url(no_dns, base, expected):
assert er.build_chat_url(base) == expected
@pytest.mark.parametrize(
"base,expected", [(c[1], c[3]) for c in PROVIDER_CASES],
ids=[c[0] for c in PROVIDER_CASES],
)
def test_build_models_url(no_dns, base, expected):
assert er.build_models_url(base) == expected
def test_chat_url_never_double_prefixes_anthropic(no_dns):
"""Regression guard: the /v1 collapse must not produce /v1/v1/messages."""
url = er.build_chat_url("https://api.anthropic.com/v1")
assert "/v1/v1/" not in url
assert url.count("/v1/messages") == 1
+6
View File
@@ -28,6 +28,12 @@ def _sys(vram, family="rdna"):
return {"backend": "rocm", "gpu_vram_gb": vram, "gpu_family": family} return {"backend": "rocm", "gpu_vram_gb": vram, "gpu_family": family}
def test_compute_serve_profiles_ignores_invalid_inputs():
assert compute_serve_profiles(None, _DENSE_8B) == []
assert compute_serve_profiles(_sys(8), None) == []
assert compute_serve_profiles(["bad"], _DENSE_8B) == []
def test_big_moe_on_small_card_offloads_not_fails(): def test_big_moe_on_small_card_offloads_not_fails():
"""A 35B MoE can't hold its weights on 16 GB, so the Quality profile must """A 35B MoE can't hold its weights on 16 GB, so the Quality profile must
offload experts to CPU (n_cpu_moe > 0) rather than be dropped.""" offload experts to CPU (n_cpu_moe > 0) rather than be dropped."""
@@ -36,6 +36,28 @@ def test_compute_snap_tolerates_non_array_other_layers():
assert r["x"] == 10 and r["y"] == 10 and r["guides"] == [] assert r["x"] == 10 and r["y"] == 10 and r["guides"] == []
@pytest.mark.skipif(not _HAS_NODE, reason="node binary not on PATH")
def test_compute_snap_tolerates_missing_layer_or_context():
js = f"""
import {{ computeSnap }} from '{_HELPER.as_posix()}';
console.log(JSON.stringify([
computeSnap(null, 10, 20, {{ zoom: 1, canvasW: 800, canvasH: 600 }}),
computeSnap({{ id: 'L1' }}, 11, 21, {{ zoom: 1, canvasW: 800, canvasH: 600 }}),
computeSnap({{ id: 'L1', canvas: {{ width: 100, height: 50 }} }}, 12, 22, null)
]));
"""
proc = subprocess.run(
["node", "--input-type=module"],
input=js, capture_output=True, text=True, cwd=str(_REPO), timeout=30,
)
assert proc.returncode == 0, proc.stderr
assert json.loads(proc.stdout.strip()) == [
{"x": 10, "y": 20, "guides": []},
{"x": 11, "y": 21, "guides": []},
{"x": 12, "y": 22, "guides": []},
]
@pytest.mark.skipif(not _HAS_NODE, reason="node binary not on PATH") @pytest.mark.skipif(not _HAS_NODE, reason="node binary not on PATH")
def test_compute_snap_still_snaps_to_a_layer_edge(): def test_compute_snap_still_snaps_to_a_layer_edge():
other = [{"id": "L2", "visible": True, "offset": {"x": 12, "y": 300}, other = [{"id": "L2", "visible": True, "offset": {"x": 12, "y": 300},
+20
View File
@@ -68,3 +68,23 @@ def test_unresolvable_host_blocked():
ok, reason = check_outbound_url("http://does-not-resolve.invalid", resolver=PUBLIC) ok, reason = check_outbound_url("http://does-not-resolve.invalid", resolver=PUBLIC)
assert ok is False assert ok is False
assert "resolve" in reason assert "resolve" in reason
def test_resolver_values_must_include_a_parseable_ip():
ok, reason = check_outbound_url(
"https://example.test",
resolver=lambda _host: [None, 123, "not-an-ip"],
)
assert ok is False
assert "does not resolve to an IP" in reason
def test_resolver_skips_invalid_values_but_accepts_public_ip():
ok, reason = check_outbound_url(
"https://example.test",
resolver=lambda _host: [None, "not-an-ip", "93.184.216.34"],
)
assert ok is True
assert reason == "ok"