92a8f5d894
Tests cover: - Smoke: /health, /auth/status, /distill/status (test_health.py) - Persona validation: path traversal, bad names, list_personas (test_persona.py) - Chat API: persona routing, session persistence, error handling (test_api_chat.py) - Files API: ALLOWED set enforcement, read/write, missing files (test_api_files.py) - Webhooks: NC Talk HMAC accept/reject, Google Chat JWT (test_webhooks.py) - Tools: scratch read/write/append/clear, tasks CRUD, cron parser + tools (test_tools.py) - Security: path traversal, replay attack, known gaps documented (test_security.py) All LLM calls mocked — suite runs in ~1.4s. Run: cd cortex && .venv/bin/pytest Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
272 lines
9.9 KiB
Python
272 lines
9.9 KiB
Python
"""
|
|
Unit tests for Inara's internal tools — scratch, tasks, cron.
|
|
|
|
These test the sync implementation functions directly, using a temp directory.
|
|
No HTTP, no LLM calls.
|
|
"""
|
|
import json
|
|
import pytest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
|
|
@pytest.fixture
|
|
def persona_dir(tmp_path) -> Path:
|
|
"""A temp persona directory pre-populated with empty tool files."""
|
|
d = tmp_path / "inara"
|
|
d.mkdir()
|
|
(d / "SCRATCH.md").write_text("")
|
|
(d / "TASKS.json").write_text("[]")
|
|
(d / "CRONS.json").write_text("[]")
|
|
(d / "REMINDERS.md").write_text("")
|
|
return d
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def patch_persona_path(persona_dir):
|
|
"""
|
|
Route all persona_path() calls to the temp dir.
|
|
Each tool does `from persona import persona_path`, so we must patch
|
|
the name in each module's namespace, not just in persona itself.
|
|
"""
|
|
with (
|
|
patch("tools.scratch.persona_path", return_value=persona_dir),
|
|
patch("tools.tasks.persona_path", return_value=persona_dir),
|
|
patch("tools.cron.persona_path", return_value=persona_dir),
|
|
patch("cron_runner._persona_path", return_value=persona_dir),
|
|
):
|
|
yield
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scratch
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestScratch:
|
|
def test_read_empty(self):
|
|
from tools.scratch import _scratch_read
|
|
result = _scratch_read()
|
|
assert "empty" in result.lower()
|
|
|
|
def test_write_and_read(self, persona_dir):
|
|
from tools.scratch import _scratch_write, _scratch_read
|
|
_scratch_write("# My notes\n\nSome content here.")
|
|
content = _scratch_read()
|
|
assert "My notes" in content
|
|
assert "Some content here" in content
|
|
|
|
def test_append_adds_section(self, persona_dir):
|
|
from tools.scratch import _scratch_write, _scratch_append, _scratch_read
|
|
_scratch_write("# Existing")
|
|
_scratch_append("New section content", heading="Section A")
|
|
content = _scratch_read()
|
|
assert "Existing" in content
|
|
assert "Section A" in content
|
|
assert "New section content" in content
|
|
|
|
def test_append_auto_heading(self, persona_dir):
|
|
from tools.scratch import _scratch_append, _scratch_read
|
|
_scratch_append("Content without explicit heading")
|
|
content = _scratch_read()
|
|
assert "UTC" in content # auto heading includes timestamp
|
|
|
|
def test_clear(self, persona_dir):
|
|
from tools.scratch import _scratch_write, _scratch_clear, _scratch_read
|
|
_scratch_write("Some content")
|
|
_scratch_clear()
|
|
assert "empty" in _scratch_read().lower()
|
|
|
|
def test_write_strips_trailing_whitespace(self, persona_dir):
|
|
from tools.scratch import _scratch_write, _scratch_read
|
|
_scratch_write("Content \n\n\n")
|
|
content = (persona_dir / "SCRATCH.md").read_text()
|
|
assert content.endswith("\n")
|
|
assert not content.endswith("\n\n")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tasks
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestTasks:
|
|
def _mk(self, title, description=None, priority="normal"):
|
|
from tools.tasks import _task_create
|
|
return _task_create(title, description, priority)
|
|
|
|
def _id(self, result: str) -> str:
|
|
import re
|
|
m = re.search(r't_\w+', result)
|
|
assert m, f"No task ID in: {result}"
|
|
return m.group()
|
|
|
|
def test_list_empty(self):
|
|
from tools.tasks import _task_list
|
|
assert "No tasks" in _task_list(status=None)
|
|
|
|
def test_create_and_list(self):
|
|
from tools.tasks import _task_list
|
|
self._mk("Buy coffee", description="Dark roast", priority="high")
|
|
result = _task_list(status=None)
|
|
assert "Buy coffee" in result
|
|
assert "[high]" in result
|
|
|
|
def test_create_bad_priority_defaults_to_normal(self):
|
|
from tools.tasks import _task_list
|
|
self._mk("Test task", priority="urgent") # invalid — becomes "normal"
|
|
result = _task_list(status=None)
|
|
assert "Test task" in result
|
|
assert "[normal]" not in result # normal priority not shown in brackets
|
|
|
|
def test_update_status(self):
|
|
from tools.tasks import _task_update, _task_list
|
|
tid = self._id(self._mk("Work item"))
|
|
_task_update(tid, status="in_progress", title=None, description=None, priority=None)
|
|
assert "Work item" in _task_list(status="in_progress")
|
|
|
|
def test_complete(self):
|
|
from tools.tasks import _task_complete, _task_list
|
|
tid = self._id(self._mk("Finish this"))
|
|
_task_complete(tid)
|
|
assert "Finish this" in _task_list(status="done")
|
|
assert "Finish this" not in _task_list(status="todo")
|
|
|
|
def test_filter_by_status(self):
|
|
from tools.tasks import _task_list
|
|
self._mk("A task")
|
|
assert "A task" in _task_list(status="todo")
|
|
assert "A task" not in _task_list(status="done")
|
|
|
|
def test_update_unknown_id(self):
|
|
from tools.tasks import _task_update
|
|
result = _task_update("t_doesnotexist", status="done",
|
|
title=None, description=None, priority=None)
|
|
assert "not found" in result.lower()
|
|
|
|
def test_persistence(self, persona_dir):
|
|
"""Tasks survive between _load() calls (written to TASKS.json)."""
|
|
self._mk("Persistent task")
|
|
data = json.loads((persona_dir / "TASKS.json").read_text())
|
|
assert any(t["title"] == "Persistent task" for t in data)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cron
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCronRunner:
|
|
def test_parse_hourly(self):
|
|
from cron_runner import parse_schedule
|
|
assert parse_schedule("hourly") == {"minute": 0}
|
|
|
|
def test_parse_daily_default(self):
|
|
from cron_runner import parse_schedule
|
|
r = parse_schedule("daily")
|
|
assert r["hour"] == 9
|
|
assert r["minute"] == 0
|
|
|
|
def test_parse_daily_custom_time(self):
|
|
from cron_runner import parse_schedule
|
|
r = parse_schedule("daily:14:30")
|
|
assert r["hour"] == 14
|
|
assert r["minute"] == 30
|
|
|
|
def test_parse_weekly(self):
|
|
from cron_runner import parse_schedule
|
|
r = parse_schedule("weekly:mon")
|
|
assert r["day_of_week"] == "mon"
|
|
assert r["hour"] == 9
|
|
|
|
def test_parse_weekly_with_time(self):
|
|
from cron_runner import parse_schedule
|
|
r = parse_schedule("weekly:fri:17:00")
|
|
assert r["day_of_week"] == "fri"
|
|
assert r["hour"] == 17
|
|
assert r["minute"] == 0
|
|
|
|
def test_parse_full_day_names(self):
|
|
from cron_runner import parse_schedule
|
|
assert parse_schedule("weekly:monday")["day_of_week"] == "mon"
|
|
assert parse_schedule("weekly:friday")["day_of_week"] == "fri"
|
|
|
|
def test_parse_unknown_schedule(self):
|
|
from cron_runner import parse_schedule
|
|
with pytest.raises(ValueError, match="Unrecognised"):
|
|
parse_schedule("every-tuesday")
|
|
|
|
def test_parse_bad_dow(self):
|
|
from cron_runner import parse_schedule
|
|
with pytest.raises(ValueError, match="day of week"):
|
|
parse_schedule("weekly:funday")
|
|
|
|
def test_parse_bad_time_non_integer(self):
|
|
from cron_runner import parse_schedule
|
|
with pytest.raises((ValueError, Exception)):
|
|
parse_schedule("daily:noon") # non-integer — parse error
|
|
|
|
|
|
class TestCronTools:
|
|
def test_list_empty(self):
|
|
from tools.cron import _cron_list
|
|
result = _cron_list()
|
|
assert "No crons" in result
|
|
|
|
def test_add_and_list(self):
|
|
from tools.cron import _cron_add, _cron_list
|
|
with patch("tools.cron._scheduler_add"):
|
|
r = _cron_add("Morning reminder", "daily:09:00", "remind", "Check in.")
|
|
assert "c_" in r
|
|
listing = _cron_list()
|
|
assert "Morning reminder" in listing
|
|
assert "daily:09:00" in listing
|
|
|
|
def test_add_bad_schedule(self):
|
|
from tools.cron import _cron_add
|
|
r = _cron_add("Bad job", "every-day", "remind", "Hello")
|
|
assert "Bad schedule" in r
|
|
|
|
def test_add_bad_type(self):
|
|
from tools.cron import _cron_add
|
|
r = _cron_add("Bad job", "daily", "email", "Hello")
|
|
assert "Bad type" in r
|
|
|
|
def _extract_id(self, result: str) -> str:
|
|
import re
|
|
m = re.search(r'c_\w+', result)
|
|
assert m, f"No cron ID in: {result}"
|
|
return m.group()
|
|
|
|
def test_remove(self):
|
|
from tools.cron import _cron_add, _cron_remove, _cron_list
|
|
with patch("tools.cron._scheduler_add"), patch("tools.cron._scheduler_remove"):
|
|
r = _cron_add("To remove", "hourly", "note", "Tick")
|
|
cron_id = self._extract_id(r)
|
|
result = _cron_remove(cron_id)
|
|
assert "Removed" in result
|
|
assert "To remove" not in _cron_list()
|
|
|
|
def test_remove_unknown(self):
|
|
from tools.cron import _cron_remove
|
|
result = _cron_remove("c_doesnotexist")
|
|
assert "Not found" in result
|
|
|
|
def test_toggle_pause_resume(self):
|
|
from tools.cron import _cron_add, _cron_toggle, _cron_list
|
|
with patch("tools.cron._scheduler_add"), patch("tools.cron._scheduler_pause"), patch("tools.cron._scheduler_resume"):
|
|
r = _cron_add("Toggleable", "daily", "note", "Content")
|
|
cron_id = self._extract_id(r)
|
|
|
|
result = _cron_toggle(cron_id)
|
|
assert "Paused" in result
|
|
assert "PAUSED" in _cron_list()
|
|
|
|
result = _cron_toggle(cron_id)
|
|
assert "Resumed" in result
|
|
assert "enabled" in _cron_list()
|
|
|
|
def test_reminders_clear(self, persona_dir):
|
|
from tools.cron import _reminders_clear
|
|
(persona_dir / "REMINDERS.md").write_text("## Some reminder\n\nContent")
|
|
result = _reminders_clear()
|
|
assert "cleared" in result.lower()
|
|
assert (persona_dir / "REMINDERS.md").read_text() == ""
|