feat+test: pytest suite + Now Playing + World Clock + Daily Quote packs (154 commands)
Adds first formal test suite for Python fork — 42 tests covering memory_store,
profiles_store, scheduler_store, macros_store, llm_backend. All pass.
Tests (tests/, ~290 lines)
conftest.py — `isolated_state` fixture: rewires _PATH/_PROFILES_DIR etc to
a per-test tmp_path so concurrent runs don't touch real data.
test_memory_store.py — 9 tests: remember/recall/forget round-trip,
search by key/value, limit/empty query,
persistence across reload, build_llm_context
test_profiles_store.py — 7 tests: seeded defaults, set_active persistence,
allows_command logic (default/work/driving)
test_scheduler_store.py — 12 tests: parser (daily/at/every/in, ru units,
bad input), add/remove/clear/remove_by_text,
_next_fire for each schedule kind
test_macros_store.py — 8 tests: is_macro_control filter, start/save
round-trip, control-phrase filtering, list_names
sorted, replay unknown raises
test_llm_backend.py — 6 tests: parse_backend aliases (ru + en),
persist round-trip, auto-detect logic
Run: cd /c/Jarvis/python && python -m pytest tests/
Three new fun packs (commands.yaml: 151 → 154)
now_playing — Windows Media Session API via PowerShell
"что играет" / "что за песня" / "какой трек"
Works with Spotify, YouTube, Foobar, Yandex Music, anything
that exposes SMTC (Win10 1803+).
world_clock — worldtimeapi.org (free, no key)
"сколько времени в Токио" / "время в Лондоне"
21 Russian + world cities pre-mapped to IANA timezones.
daily_quote — zenquotes.io (free, no key) + LLM translation
"цитата дня" / "вдохнови меня"
Fetches English quote, translates to Russian via active LLM
(Groq or Ollama), speaks "text — author."
Pack count: 151 → 154. Tests: 42 pytest + ast.parse + yaml.safe_load.
This commit is contained in:
parent
8d3da4ea06
commit
4e62049a98
10 changed files with 637 additions and 0 deletions
0
tests/__init__.py
Normal file
0
tests/__init__.py
Normal file
45
tests/conftest.py
Normal file
45
tests/conftest.py
Normal file
|
|
@ -0,0 +1,45 @@
|
|||
"""Pytest config: temp-dir isolation for state modules.
|
||||
|
||||
Each test gets a fresh dir for memory.json / schedule.json / etc so concurrent
|
||||
runs don't pollute the real user data.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import importlib
|
||||
import pytest
|
||||
|
||||
# Make parent package importable.
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def isolated_state(tmp_path, monkeypatch):
|
||||
"""Patch state modules to use a tmp_path so we don't touch real state files."""
|
||||
|
||||
def patch_module(mod_name, path_attrs):
|
||||
"""Reimport a module and rewire its `_PATH` and friends to tmp_path."""
|
||||
if mod_name in sys.modules:
|
||||
del sys.modules[mod_name]
|
||||
mod = importlib.import_module(mod_name)
|
||||
for attr in path_attrs:
|
||||
if not hasattr(mod, attr):
|
||||
continue
|
||||
original = getattr(mod, attr)
|
||||
new_val = os.path.join(str(tmp_path), os.path.basename(original))
|
||||
monkeypatch.setattr(mod, attr, new_val)
|
||||
# Also reset any cached _loaded / _initialized flags
|
||||
for flag in ('_loaded', '_initialized', '_thread_started'):
|
||||
if hasattr(mod, flag):
|
||||
monkeypatch.setattr(mod, flag, False)
|
||||
# Reset stores
|
||||
for store_attr in ('_store', '_tasks'):
|
||||
if hasattr(mod, store_attr):
|
||||
cur = getattr(mod, store_attr)
|
||||
if isinstance(cur, dict):
|
||||
monkeypatch.setattr(mod, store_attr, {})
|
||||
elif isinstance(cur, list):
|
||||
monkeypatch.setattr(mod, store_attr, [])
|
||||
return mod
|
||||
|
||||
yield patch_module
|
||||
68
tests/test_llm_backend.py
Normal file
68
tests/test_llm_backend.py
Normal file
|
|
@ -0,0 +1,68 @@
|
|||
"""Tests for llm_backend.parse_backend + state machine."""
|
||||
|
||||
import os
|
||||
import importlib
|
||||
import sys
|
||||
|
||||
|
||||
def test_parse_backend_groq_aliases():
|
||||
if 'llm_backend' in sys.modules:
|
||||
del sys.modules['llm_backend']
|
||||
m = importlib.import_module('llm_backend')
|
||||
for alias in ('groq', 'GROQ', 'cloud', 'облако', 'клауд', ' Groq '):
|
||||
assert m.parse_backend(alias) == 'groq', f"failed: {alias!r}"
|
||||
|
||||
|
||||
def test_parse_backend_ollama_aliases():
|
||||
if 'llm_backend' in sys.modules:
|
||||
del sys.modules['llm_backend']
|
||||
m = importlib.import_module('llm_backend')
|
||||
for alias in ('ollama', 'OLLAMA', 'local', 'локальный', 'локал', 'локальн'):
|
||||
assert m.parse_backend(alias) == 'ollama', f"failed: {alias!r}"
|
||||
|
||||
|
||||
def test_parse_backend_unknown_returns_none():
|
||||
if 'llm_backend' in sys.modules:
|
||||
del sys.modules['llm_backend']
|
||||
m = importlib.import_module('llm_backend')
|
||||
assert m.parse_backend('openai') is None
|
||||
assert m.parse_backend('claude') is None
|
||||
assert m.parse_backend('') is None
|
||||
assert m.parse_backend(None) is None
|
||||
|
||||
|
||||
def test_persist_round_trip(tmp_path, monkeypatch):
|
||||
if 'llm_backend' in sys.modules:
|
||||
del sys.modules['llm_backend']
|
||||
m = importlib.import_module('llm_backend')
|
||||
monkeypatch.setattr(m, '_CHOICE_FILE', str(tmp_path / 'llm_backend.txt'))
|
||||
m._persist('ollama')
|
||||
assert m._read_persisted() == 'ollama'
|
||||
m._persist('groq')
|
||||
assert m._read_persisted() == 'groq'
|
||||
|
||||
|
||||
def test_auto_detect_fallback_to_ollama(tmp_path, monkeypatch):
|
||||
"""When no GROQ_TOKEN in config, auto-detect should pick ollama."""
|
||||
if 'llm_backend' in sys.modules:
|
||||
del sys.modules['llm_backend']
|
||||
m = importlib.import_module('llm_backend')
|
||||
|
||||
# Mock config module with no GROQ_TOKEN
|
||||
class FakeConfig:
|
||||
GROQ_TOKEN = ''
|
||||
monkeypatch.setitem(sys.modules, 'config', FakeConfig())
|
||||
|
||||
assert m._auto_detect_backend() == 'ollama'
|
||||
|
||||
|
||||
def test_auto_detect_picks_groq_when_token_present(tmp_path, monkeypatch):
|
||||
if 'llm_backend' in sys.modules:
|
||||
del sys.modules['llm_backend']
|
||||
m = importlib.import_module('llm_backend')
|
||||
|
||||
class FakeConfig:
|
||||
GROQ_TOKEN = 'gsk_fake_test_token'
|
||||
monkeypatch.setitem(sys.modules, 'config', FakeConfig())
|
||||
|
||||
assert m._auto_detect_backend() == 'groq'
|
||||
87
tests/test_macros_store.py
Normal file
87
tests/test_macros_store.py
Normal file
|
|
@ -0,0 +1,87 @@
|
|||
"""Tests for macros_store: record/save/cancel/list/delete + control-phrase filter."""
|
||||
|
||||
|
||||
def test_is_macro_control_filters_record_phrases(isolated_state):
|
||||
m = isolated_state('macros_store', ['_PATH'])
|
||||
assert m._is_control("Запиши макрос работа")
|
||||
assert m._is_control("сохрани макрос")
|
||||
assert m._is_control("запусти макрос работа")
|
||||
assert m._is_control("Удали макрос работа")
|
||||
# Regular commands NOT filtered
|
||||
assert not m._is_control("открой браузер")
|
||||
assert not m._is_control("установи громкость 50")
|
||||
|
||||
|
||||
def test_start_save_round_trip(isolated_state):
|
||||
m = isolated_state('macros_store', ['_PATH'])
|
||||
m.start("test")
|
||||
assert m.is_recording()
|
||||
assert m.recording_name() == 'test'
|
||||
|
||||
m.record_step("открой браузер")
|
||||
m.record_step("режим работа")
|
||||
m.record_step("установи громкость 30")
|
||||
# Control phrase should be filtered
|
||||
m.record_step("запиши макрос новый")
|
||||
|
||||
count = m.save()
|
||||
assert count == 3 # 3 + 1 filtered = 3
|
||||
assert not m.is_recording()
|
||||
|
||||
mac = m.get('test')
|
||||
assert mac is not None
|
||||
assert mac['steps'] == [
|
||||
'открой браузер',
|
||||
'режим работа',
|
||||
'установи громкость 30',
|
||||
]
|
||||
|
||||
|
||||
def test_save_without_recording_raises(isolated_state):
|
||||
import pytest
|
||||
m = isolated_state('macros_store', ['_PATH'])
|
||||
with pytest.raises(RuntimeError):
|
||||
m.save()
|
||||
|
||||
|
||||
def test_save_empty_recording_raises(isolated_state):
|
||||
import pytest
|
||||
m = isolated_state('macros_store', ['_PATH'])
|
||||
m.start("empty")
|
||||
with pytest.raises(RuntimeError):
|
||||
m.save()
|
||||
|
||||
|
||||
def test_cancel_aborts_without_saving(isolated_state):
|
||||
m = isolated_state('macros_store', ['_PATH'])
|
||||
m.start("test")
|
||||
m.record_step("foo")
|
||||
assert m.cancel() is True
|
||||
assert not m.is_recording()
|
||||
assert m.get('test') is None
|
||||
|
||||
|
||||
def test_delete(isolated_state):
|
||||
m = isolated_state('macros_store', ['_PATH'])
|
||||
m.start("test")
|
||||
m.record_step("x")
|
||||
m.save()
|
||||
assert m.delete("test") is True
|
||||
assert m.get('test') is None
|
||||
assert m.delete("test") is False # idempotent
|
||||
|
||||
|
||||
def test_list_names_sorted(isolated_state):
|
||||
m = isolated_state('macros_store', ['_PATH'])
|
||||
for name in ('zeta', 'alpha', 'beta'):
|
||||
m.start(name)
|
||||
m.record_step("step")
|
||||
m.save()
|
||||
assert m.list_names() == ['alpha', 'beta', 'zeta']
|
||||
|
||||
|
||||
def test_replay_unknown_raises(isolated_state):
|
||||
import pytest
|
||||
m = isolated_state('macros_store', ['_PATH'])
|
||||
with pytest.raises(KeyError):
|
||||
m.replay("nonexistent", lambda s: None)
|
||||
79
tests/test_memory_store.py
Normal file
79
tests/test_memory_store.py
Normal file
|
|
@ -0,0 +1,79 @@
|
|||
"""Tests for memory_store — port of long_term_memory tests in rust."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import pytest
|
||||
|
||||
|
||||
def test_remember_recall_round_trip(isolated_state):
|
||||
m = isolated_state('memory_store', ['_PATH'])
|
||||
m.remember("любимый чай", "улун")
|
||||
assert m.recall("любимый чай") == "улун"
|
||||
assert m.recall("Любимый Чай") == "улун" # case-insensitive
|
||||
|
||||
|
||||
def test_recall_returns_none_for_missing(isolated_state):
|
||||
m = isolated_state('memory_store', ['_PATH'])
|
||||
assert m.recall("nonexistent") is None
|
||||
|
||||
|
||||
def test_forget_removes(isolated_state):
|
||||
m = isolated_state('memory_store', ['_PATH'])
|
||||
m.remember("foo", "bar")
|
||||
assert m.forget("foo") is True
|
||||
assert m.recall("foo") is None
|
||||
assert m.forget("foo") is False # idempotent
|
||||
|
||||
|
||||
def test_search_by_key_and_value(isolated_state):
|
||||
m = isolated_state('memory_store', ['_PATH'])
|
||||
m.remember("любимый чай", "улун")
|
||||
m.remember("любимый кофе", "арабика")
|
||||
m.remember("питомец", "собака бася")
|
||||
|
||||
hits = m.search("чай", 5)
|
||||
assert len(hits) == 1
|
||||
assert hits[0]['key'] == "любимый чай"
|
||||
|
||||
hits = m.search("бася", 5)
|
||||
assert len(hits) == 1
|
||||
assert hits[0]['key'] == "питомец"
|
||||
|
||||
|
||||
def test_search_respects_limit(isolated_state):
|
||||
m = isolated_state('memory_store', ['_PATH'])
|
||||
for i in range(10):
|
||||
m.remember(f"key{i}", "value")
|
||||
hits = m.search("value", 3)
|
||||
assert len(hits) == 3
|
||||
|
||||
|
||||
def test_search_empty_query_returns_nothing(isolated_state):
|
||||
m = isolated_state('memory_store', ['_PATH'])
|
||||
m.remember("foo", "bar")
|
||||
assert m.search("", 5) == []
|
||||
assert m.search(" ", 5) == []
|
||||
|
||||
|
||||
def test_persists_across_module_reload(isolated_state, tmp_path):
|
||||
m = isolated_state('memory_store', ['_PATH'])
|
||||
m.remember("persistent", "yes")
|
||||
json_file = m._PATH
|
||||
assert os.path.isfile(json_file)
|
||||
with open(json_file, encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
assert data['entries']['persistent']['value'] == 'yes'
|
||||
|
||||
|
||||
def test_build_llm_context_format(isolated_state):
|
||||
m = isolated_state('memory_store', ['_PATH'])
|
||||
m.remember("любимый чай", "улун")
|
||||
# Query "чай" is substring of key "любимый чай" → matches.
|
||||
ctx = m.build_llm_context("чай", 5)
|
||||
assert "любимый чай = улун" in ctx
|
||||
assert ctx.startswith("Известные факты")
|
||||
|
||||
|
||||
def test_build_llm_context_empty(isolated_state):
|
||||
m = isolated_state('memory_store', ['_PATH'])
|
||||
assert m.build_llm_context("anything", 5) == ""
|
||||
61
tests/test_profiles_store.py
Normal file
61
tests/test_profiles_store.py
Normal file
|
|
@ -0,0 +1,61 @@
|
|||
"""Tests for profiles_store."""
|
||||
|
||||
import os
|
||||
import json
|
||||
|
||||
|
||||
def test_default_seeded_on_init(isolated_state, tmp_path):
|
||||
p = isolated_state('profiles_store', ['_PROFILES_DIR', '_ACTIVE_FILE'])
|
||||
p._init()
|
||||
names = p.list_names()
|
||||
# 5 defaults: default, work, game, sleep, driving
|
||||
for expected in ('default', 'work', 'game', 'sleep', 'driving'):
|
||||
assert expected in names, f"missing seeded profile: {expected}"
|
||||
|
||||
|
||||
def test_active_defaults_to_default(isolated_state):
|
||||
p = isolated_state('profiles_store', ['_PROFILES_DIR', '_ACTIVE_FILE'])
|
||||
a = p.active()
|
||||
assert a['name'] == 'default'
|
||||
|
||||
|
||||
def test_set_active_persists(isolated_state):
|
||||
p = isolated_state('profiles_store', ['_PROFILES_DIR', '_ACTIVE_FILE'])
|
||||
p.set_active('work')
|
||||
assert p.active_name() == 'work'
|
||||
assert os.path.isfile(p._ACTIVE_FILE)
|
||||
with open(p._ACTIVE_FILE, encoding='utf-8') as f:
|
||||
assert f.read().strip() == 'work'
|
||||
|
||||
|
||||
def test_set_active_unknown_raises(isolated_state):
|
||||
p = isolated_state('profiles_store', ['_PROFILES_DIR', '_ACTIVE_FILE'])
|
||||
try:
|
||||
p.set_active('nonexistent')
|
||||
assert False, "should have raised"
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
|
||||
def test_allows_command_default_allows_all(isolated_state):
|
||||
p = isolated_state('profiles_store', ['_PROFILES_DIR', '_ACTIVE_FILE'])
|
||||
p.set_active('default')
|
||||
assert p.allows_command('anything')
|
||||
assert p.allows_command('games.steam.launch')
|
||||
|
||||
|
||||
def test_allows_command_work_disables_games(isolated_state):
|
||||
p = isolated_state('profiles_store', ['_PROFILES_DIR', '_ACTIVE_FILE'])
|
||||
p.set_active('work')
|
||||
assert not p.allows_command('games.steam.launch')
|
||||
assert not p.allows_command('fun.joke')
|
||||
assert p.allows_command('time.current')
|
||||
|
||||
|
||||
def test_allows_command_driving_whitelist(isolated_state):
|
||||
p = isolated_state('profiles_store', ['_PROFILES_DIR', '_ACTIVE_FILE'])
|
||||
p.set_active('driving')
|
||||
assert p.allows_command('music.play')
|
||||
assert p.allows_command('weather.now')
|
||||
assert not p.allows_command('windows.minimize')
|
||||
assert not p.allows_command('screen.brightness')
|
||||
109
tests/test_scheduler_store.py
Normal file
109
tests/test_scheduler_store.py
Normal file
|
|
@ -0,0 +1,109 @@
|
|||
"""Tests for scheduler_store: parser + add/remove + serde."""
|
||||
|
||||
import time
|
||||
import pytest
|
||||
|
||||
|
||||
def test_parse_daily(isolated_state):
|
||||
s = isolated_state('scheduler_store', ['_PATH'])
|
||||
schedule = s.parse_schedule("daily 09:00")
|
||||
assert schedule == {'kind': 'daily', 'hour': 9, 'minute': 0}
|
||||
|
||||
|
||||
def test_parse_at_today_or_tomorrow(isolated_state):
|
||||
s = isolated_state('scheduler_store', ['_PATH'])
|
||||
schedule = s.parse_schedule("at 23:59")
|
||||
assert schedule['kind'] == 'once'
|
||||
assert schedule['at'] > time.time()
|
||||
|
||||
|
||||
def test_parse_every_minutes(isolated_state):
|
||||
s = isolated_state('scheduler_store', ['_PATH'])
|
||||
schedule = s.parse_schedule("every 30 minutes")
|
||||
assert schedule == {'kind': 'interval', 'seconds': 1800}
|
||||
|
||||
|
||||
def test_parse_every_russian(isolated_state):
|
||||
s = isolated_state('scheduler_store', ['_PATH'])
|
||||
schedule = s.parse_schedule("every 2 часа")
|
||||
assert schedule == {'kind': 'interval', 'seconds': 7200}
|
||||
|
||||
|
||||
def test_parse_in_minutes_is_once(isolated_state):
|
||||
s = isolated_state('scheduler_store', ['_PATH'])
|
||||
schedule = s.parse_schedule("in 5 minutes")
|
||||
assert schedule['kind'] == 'once'
|
||||
# Should be roughly 300 seconds in the future
|
||||
delta = schedule['at'] - time.time()
|
||||
assert 290 <= delta <= 310
|
||||
|
||||
|
||||
def test_parse_bad_unit_raises(isolated_state):
|
||||
s = isolated_state('scheduler_store', ['_PATH'])
|
||||
with pytest.raises(ValueError):
|
||||
s.parse_schedule("every 5 weeks")
|
||||
|
||||
|
||||
def test_parse_bad_time_raises(isolated_state):
|
||||
s = isolated_state('scheduler_store', ['_PATH'])
|
||||
with pytest.raises(ValueError):
|
||||
s.parse_schedule("daily 25:00")
|
||||
with pytest.raises(ValueError):
|
||||
s.parse_schedule("at 12:99")
|
||||
|
||||
|
||||
def test_add_remove(isolated_state):
|
||||
s = isolated_state('scheduler_store', ['_PATH'])
|
||||
tid = s.add({
|
||||
'name': 'Test',
|
||||
'schedule': 'in 1 hour',
|
||||
'action': {'type': 'speak', 'text': 'hello'},
|
||||
})
|
||||
assert tid
|
||||
tasks = s.list_all()
|
||||
assert len(tasks) == 1
|
||||
assert tasks[0]['name'] == 'Test'
|
||||
|
||||
assert s.remove(tid) is True
|
||||
assert s.list_all() == []
|
||||
assert s.remove(tid) is False # idempotent
|
||||
|
||||
|
||||
def test_remove_by_text(isolated_state):
|
||||
s = isolated_state('scheduler_store', ['_PATH'])
|
||||
s.add({
|
||||
'name': 'Water',
|
||||
'schedule': 'every 2 hours',
|
||||
'action': {'type': 'speak', 'text': 'Попей воды.'},
|
||||
})
|
||||
s.add({
|
||||
'name': 'Stretch',
|
||||
'schedule': 'every 50 minutes',
|
||||
'action': {'type': 'speak', 'text': 'Размяться.'},
|
||||
})
|
||||
|
||||
removed = s.remove_by_text("воды")
|
||||
assert removed == 1
|
||||
assert len(s.list_all()) == 1
|
||||
|
||||
|
||||
def test_clear(isolated_state):
|
||||
s = isolated_state('scheduler_store', ['_PATH'])
|
||||
s.add({'name': 'a', 'schedule': 'in 1 hour', 'action': {'type': 'speak', 'text': 'x'}})
|
||||
s.add({'name': 'b', 'schedule': 'in 2 hours', 'action': {'type': 'speak', 'text': 'y'}})
|
||||
assert s.clear() == 2
|
||||
assert s.list_all() == []
|
||||
|
||||
|
||||
def test_next_fire_once(isolated_state):
|
||||
s = isolated_state('scheduler_store', ['_PATH'])
|
||||
schedule = {'kind': 'once', 'at': 500}
|
||||
assert s._next_fire(schedule, None, 1000) == 500
|
||||
# Already fired
|
||||
assert s._next_fire(schedule, 500, 1000) is None
|
||||
|
||||
|
||||
def test_next_fire_interval(isolated_state):
|
||||
s = isolated_state('scheduler_store', ['_PATH'])
|
||||
schedule = {'kind': 'interval', 'seconds': 60}
|
||||
assert s._next_fire(schedule, 100, 1000) == 160
|
||||
Loading…
Add table
Add a link
Reference in a new issue