157 lines
No EOL
4.8 KiB
Python
157 lines
No EOL
4.8 KiB
Python
"""
|
|
Unit tests for mascot_state module.
|
|
Run with: python -m pytest plugins/mascot/test_mascot_state.py -v
|
|
"""
|
|
|
|
import json
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
# Import the module under test
|
|
import sys
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
|
|
|
|
from plugins.mascot.mascot_state import (
|
|
MascotState,
|
|
MascotStateManager,
|
|
get_manager,
|
|
STATE_IDLE,
|
|
STATE_THINKING,
|
|
STATE_WORKING,
|
|
STATE_WAITING_INPUT,
|
|
STATE_ERROR,
|
|
VALID_STATES,
|
|
)
|
|
|
|
|
|
class TestMascotState(unittest.TestCase):
|
|
"""Tests for MascotState dataclass."""
|
|
|
|
def test_default_state(self):
|
|
"""Default state should be idle."""
|
|
state = MascotState()
|
|
self.assertEqual(state.status, STATE_IDLE)
|
|
self.assertIsNone(state.task)
|
|
self.assertIsNone(state.mood)
|
|
self.assertGreater(state.last_update, 0)
|
|
|
|
def test_to_dict(self):
|
|
"""to_dict should return all fields."""
|
|
state = MascotState(
|
|
status=STATE_THINKING,
|
|
task="Test task",
|
|
mood="happy",
|
|
last_update=12345.0,
|
|
)
|
|
d = state.to_dict()
|
|
self.assertEqual(d["status"], STATE_THINKING)
|
|
self.assertEqual(d["task"], "Test task")
|
|
self.assertEqual(d["mood"], "happy")
|
|
self.assertEqual(d["last_update"], 12345.0)
|
|
|
|
|
|
class TestMascotStateManager(unittest.TestCase):
|
|
"""Tests for MascotStateManager singleton."""
|
|
|
|
def setUp(self):
|
|
"""Use temp directory for state persistence."""
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.state_path = Path(self.temp_dir) / "plugins" / "mascot" / "state.json"
|
|
|
|
# Patch the state path
|
|
self.patcher = patch(
|
|
"plugins.mascot.mascot_state.MascotStateManager._state_path",
|
|
new_callable=lambda: self.state_path,
|
|
)
|
|
self.patcher.start()
|
|
|
|
# Reset singleton
|
|
MascotStateManager._instance = None
|
|
|
|
def tearDown(self):
|
|
self.patcher.stop()
|
|
|
|
def test_singleton(self):
|
|
"""get_manager should return the same instance."""
|
|
m1 = get_manager()
|
|
m2 = get_manager()
|
|
self.assertIs(m1, m2)
|
|
|
|
def test_get_state(self):
|
|
"""get_state should return a copy."""
|
|
manager = get_manager()
|
|
state1 = manager.get_state()
|
|
state2 = manager.get_state()
|
|
self.assertIsNot(state1, state2)
|
|
self.assertEqual(state1.status, state2.status)
|
|
|
|
def test_set_state(self):
|
|
"""set_state should update and persist."""
|
|
manager = get_manager()
|
|
manager.reset() # Start from known state
|
|
|
|
new_state = manager.set_state(status=STATE_WORKING, task="Testing")
|
|
self.assertEqual(new_state.status, STATE_WORKING)
|
|
self.assertEqual(new_state.task, "Testing")
|
|
|
|
# Check persistence
|
|
self.assertTrue(self.state_path.exists())
|
|
data = json.loads(self.state_path.read_text())
|
|
self.assertEqual(data["status"], STATE_WORKING)
|
|
|
|
def test_invalid_status(self):
|
|
"""Invalid status should raise ValueError."""
|
|
manager = get_manager()
|
|
with self.assertRaises(ValueError):
|
|
manager.set_state(status="invalid_status")
|
|
|
|
def test_reset(self):
|
|
"""reset should return to idle."""
|
|
manager = get_manager()
|
|
manager.set_state(status=STATE_WORKING, task="Something")
|
|
|
|
new_state = manager.reset()
|
|
self.assertEqual(new_state.status, STATE_IDLE)
|
|
self.assertIsNone(new_state.task)
|
|
self.assertIsNone(new_state.mood)
|
|
|
|
def test_subscribers(self):
|
|
"""Subscribers should be called on state change."""
|
|
manager = get_manager()
|
|
manager.reset()
|
|
|
|
received = []
|
|
def callback(state):
|
|
received.append(state.status)
|
|
|
|
manager.subscribe(callback)
|
|
manager.set_state(status=STATE_THINKING)
|
|
manager.set_state(status=STATE_WORKING)
|
|
manager.unsubscribe(callback)
|
|
manager.set_state(status=STATE_IDLE)
|
|
|
|
self.assertEqual(received, [STATE_THINKING, STATE_WORKING])
|
|
|
|
def test_transient_state_reset(self):
|
|
"""Transient states should reset on load."""
|
|
manager = get_manager()
|
|
manager.set_state(status=STATE_THINKING, task="Was thinking")
|
|
|
|
# Persist
|
|
data = json.loads(self.state_path.read_text())
|
|
self.assertEqual(data["status"], STATE_THINKING)
|
|
|
|
# Reset singleton and reload
|
|
MascotStateManager._instance = None
|
|
new_manager = get_manager()
|
|
|
|
# Thinking should be reset to idle (it's transient)
|
|
state = new_manager.get_state()
|
|
self.assertEqual(state.status, STATE_IDLE)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |