45 lines
1.6 KiB
Python
45 lines
1.6 KiB
Python
import unittest
|
|
from datetime import datetime
|
|
import time
|
|
|
|
from sia.working_memory import WorkingMemory
|
|
from sia.stop_command import StopCommand
|
|
from sia.reasoning_entry import ReasoningEntry
|
|
from sia.background_entry import BackgroundEntry
|
|
|
|
class StopCommandTest(unittest.TestCase):
|
|
def setUp(self):
|
|
"""Set up test cases with a fresh working memory"""
|
|
self.memory = WorkingMemory()
|
|
self.test_timestamp = datetime(2024, 10, 31, 12, 0, 0)
|
|
|
|
def test_stop_command_cleanup(self):
|
|
"""Test stop command cleans up and removes all entries"""
|
|
# Add a mix of entry types including a background process
|
|
entries = [
|
|
ReasoningEntry("test content", "id-1", self.test_timestamp),
|
|
BackgroundEntry("sleep 10", "id-2", self.test_timestamp)
|
|
]
|
|
|
|
for entry in entries:
|
|
self.memory.add_entry(entry)
|
|
|
|
# Start the background process
|
|
background_entry = self.memory.get_entry("id-2")
|
|
background_entry.update()
|
|
self.assertIsNotNone(background_entry.process)
|
|
process_pid = background_entry.process.pid
|
|
|
|
# Execute stop command
|
|
command = StopCommand()
|
|
result = command.execute(self.memory)
|
|
|
|
# Verify all entries removed
|
|
self.assertTrue(result.should_stop)
|
|
self.assertEqual(self.memory.get_entries_count(), 0)
|
|
|
|
# Verify background process was terminated
|
|
time.sleep(0.1) # Give process time to terminate
|
|
with self.assertRaises(ProcessLookupError):
|
|
import os
|
|
os.kill(process_pid, 0) |