80 lines
2.9 KiB
Python
80 lines
2.9 KiB
Python
from datetime import datetime
|
|
import os
|
|
import time
|
|
import unittest
|
|
|
|
from sia.working_memory import WorkingMemory
|
|
from sia.delete_command import DeleteCommand
|
|
from sia.reasoning_entry import ReasoningEntry
|
|
from sia.background_entry import BackgroundEntry
|
|
|
|
class DeleteCommandTest(unittest.TestCase):
|
|
def setUp(self):
|
|
"""Set up test cases with a fresh working memory"""
|
|
self.memory = WorkingMemory()
|
|
self.test_id = "test-id-1234"
|
|
self.test_timestamp = datetime(2024, 10, 31, 12, 0, 0)
|
|
|
|
def wait_for_process_start(self, entry: BackgroundEntry, timeout=1.0):
|
|
"""Wait for process to start, checking context pid"""
|
|
start_time = time.time()
|
|
while time.time() - start_time < timeout:
|
|
entry.update()
|
|
context = entry.generate_context()
|
|
if context.get("pid") is not None:
|
|
return True
|
|
time.sleep(0.1)
|
|
return False
|
|
|
|
def test_delete_running_background_entry(self):
|
|
"""Test deleting a background entry that is still running"""
|
|
# Create and start background process
|
|
entry = BackgroundEntry(self.test_id, self.test_timestamp, "sleep 10")
|
|
self.memory.add_entry(entry)
|
|
|
|
# Wait for process to start
|
|
self.assertTrue(self.wait_for_process_start(entry))
|
|
|
|
# Get PID before deletion
|
|
context = entry.generate_context()
|
|
pid = int(context.get("pid"))
|
|
|
|
# Delete entry and verify process terminated
|
|
command = DeleteCommand(entry.id)
|
|
result = command.execute(self.memory)
|
|
|
|
self.assertTrue(result.success)
|
|
self.assertFalse(result.should_stop)
|
|
self.assertEqual(result.message, "")
|
|
self.assertEqual(self.memory.get_entries_count(), 0)
|
|
|
|
# Verify process was terminated
|
|
time.sleep(0.1) # Give process time to terminate
|
|
with self.assertRaises(ProcessLookupError):
|
|
os.kill(pid, 0)
|
|
|
|
def test_delete_existing_entry(self):
|
|
"""Test deleting an existing entry"""
|
|
# Add test entry
|
|
entry = ReasoningEntry(self.test_id, self.test_timestamp, "test content")
|
|
self.memory.add_entry(entry)
|
|
|
|
# Execute delete command
|
|
command = DeleteCommand(self.test_id)
|
|
result = command.execute(self.memory)
|
|
|
|
# Verify result and memory state
|
|
self.assertTrue(result.success)
|
|
self.assertFalse(result.should_stop)
|
|
self.assertEqual(result.message, "")
|
|
self.assertEqual(self.memory.get_entries_count(), 0)
|
|
|
|
def test_delete_nonexistent_entry(self):
|
|
"""Test attempting to delete a nonexistent entry"""
|
|
command = DeleteCommand("nonexistent-id")
|
|
result = command.execute(self.memory)
|
|
|
|
# Verify error result
|
|
self.assertFalse(result.success)
|
|
self.assertFalse(result.should_stop)
|
|
self.assertIn("not found", result.message) |