213 lines
8.7 KiB
Python
213 lines
8.7 KiB
Python
import unittest
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
import xml.etree.ElementTree as ET
|
|
|
|
from sia.command import Command
|
|
from sia.delete_command import DeleteCommand
|
|
from sia.stop_command import StopCommand
|
|
from sia.entry import Entry
|
|
from sia.entry.background_entry import BackgroundEntry
|
|
from sia.entry.parse_error_entry import ParseErrorEntry
|
|
from sia.entry.read_entry import ReadEntry
|
|
from sia.entry.reasoning_entry import ReasoningEntry
|
|
from sia.entry.repeat_entry import RepeatEntry
|
|
from sia.entry.single_entry import SingleEntry
|
|
from sia.entry.write_entry import WriteEntry
|
|
from sia.web_io_buffer import WebIOBuffer
|
|
from sia.response_parser import ResponseParser
|
|
|
|
class ResponseParserTest(unittest.TestCase):
|
|
def setUp(self):
|
|
"""Set up test cases with parser and IO buffer"""
|
|
self.io_buffer = WebIOBuffer()
|
|
self.work_dir = Path("/tmp") # Use a temporary directory for tests
|
|
self.parser = ResponseParser(self.work_dir, self.io_buffer)
|
|
|
|
def test_delete_command(self):
|
|
"""Test parsing delete command"""
|
|
xml = '<delete id="test-id-123"/>'
|
|
result = self.parser.parse(datetime.now(), xml)
|
|
|
|
self.assertIsInstance(result, DeleteCommand)
|
|
self.assertEqual(result.id, "test-id-123")
|
|
|
|
def test_delete_command_missing_id(self):
|
|
"""Test parsing delete command without ID returns error entry"""
|
|
xml = '<delete/>'
|
|
result = self.parser.parse(datetime.now(), xml)
|
|
|
|
self.assertIsInstance(result, ParseErrorEntry)
|
|
self.assertEqual(result.content, xml)
|
|
self.assertIn("missing required 'id'", result.error)
|
|
|
|
def test_stop_command(self):
|
|
"""Test parsing stop command"""
|
|
xml = '<stop/>'
|
|
result = self.parser.parse(datetime.now(), xml)
|
|
|
|
self.assertIsInstance(result, StopCommand)
|
|
|
|
def test_background_entry(self):
|
|
"""Test parsing background entry"""
|
|
xml = '<background>echo test</background>'
|
|
result = self.parser.parse(datetime.now(), xml)
|
|
|
|
self.assertIsInstance(result, BackgroundEntry)
|
|
self.assertEqual(result.script, "echo test")
|
|
|
|
def test_background_entry_empty_script(self):
|
|
"""Test parsing background entry with empty script returns error entry"""
|
|
xml = '<background/>'
|
|
result = self.parser.parse(datetime.now(), xml)
|
|
|
|
self.assertIsInstance(result, ParseErrorEntry)
|
|
self.assertEqual(result.content, xml)
|
|
self.assertIn("Background entry requires (only) script content", result.error)
|
|
|
|
def test_repeat_entry(self):
|
|
"""Test parsing repeat entry"""
|
|
xml = '<repeat>ls -l</repeat>'
|
|
result = self.parser.parse(datetime.now(), xml)
|
|
|
|
self.assertIsInstance(result, RepeatEntry)
|
|
self.assertEqual(result.script, "ls -l")
|
|
|
|
def test_repeat_entry_empty_script(self):
|
|
"""Test parsing repeat entry with empty script returns error entry"""
|
|
xml = '<repeat/>'
|
|
result = self.parser.parse(datetime.now(), xml)
|
|
|
|
self.assertIsInstance(result, ParseErrorEntry)
|
|
self.assertEqual(result.content, xml)
|
|
self.assertIn("Repeat entry requires (only) script content", result.error)
|
|
|
|
def test_single_entry(self):
|
|
"""Test parsing single shot entry"""
|
|
xml = '<single>echo "one time"</single>'
|
|
result = self.parser.parse(datetime.now(), xml)
|
|
|
|
self.assertIsInstance(result, SingleEntry)
|
|
self.assertEqual(result.script, 'echo "one time"')
|
|
|
|
def test_single_entry_empty_script(self):
|
|
"""Test parsing single shot entry with empty script returns error entry"""
|
|
xml = '<single/>'
|
|
result = self.parser.parse(datetime.now(), xml)
|
|
|
|
self.assertIsInstance(result, ParseErrorEntry)
|
|
self.assertEqual(result.content, xml)
|
|
self.assertIn("Single entry requires (only) script content", result.error)
|
|
|
|
def test_reasoning_entry(self):
|
|
"""Test parsing reasoning entry"""
|
|
xml = '<reasoning>test reasoning</reasoning>'
|
|
result = self.parser.parse(datetime.now(), xml)
|
|
|
|
self.assertIsInstance(result, ReasoningEntry)
|
|
self.assertEqual(result.content, "test reasoning")
|
|
|
|
def test_reasoning_entry_empty_content(self):
|
|
"""Test parsing reasoning entry with empty content returns error entry"""
|
|
xml = '<reasoning/>'
|
|
result = self.parser.parse(datetime.now(), xml)
|
|
|
|
self.assertIsInstance(result, ParseErrorEntry)
|
|
self.assertEqual(result.content, xml)
|
|
self.assertIn("Reasoning entry requires (only) text content", result.error)
|
|
|
|
def test_read_stdin_entry(self):
|
|
"""Test parsing read stdin entry"""
|
|
xml = '<read_stdin/>'
|
|
result = self.parser.parse(datetime.now(), xml)
|
|
|
|
self.assertIsInstance(result, ReadEntry)
|
|
|
|
def test_write_stdout_entry(self):
|
|
"""Test parsing write stdout entry"""
|
|
xml = '<write_stdout>test output</write_stdout>'
|
|
result = self.parser.parse(datetime.now(), xml)
|
|
|
|
self.assertIsInstance(result, WriteEntry)
|
|
self.assertEqual(result.content, "test output")
|
|
self.assertEqual(result._io_buffer, self.io_buffer)
|
|
|
|
def test_write_stdout_entry_empty_content(self):
|
|
"""Test parsing write stdout entry with empty content returns error entry"""
|
|
xml = '<write_stdout/>'
|
|
result = self.parser.parse(datetime.now(), xml)
|
|
|
|
self.assertIsInstance(result, ParseErrorEntry)
|
|
self.assertEqual(result.content, xml)
|
|
self.assertIn("Write stdout entry requires (only) text content", result.error)
|
|
|
|
def test_unknown_element(self):
|
|
"""Test parsing unknown element returns error entry"""
|
|
xml = '<unknown>test</unknown>'
|
|
result = self.parser.parse(datetime.now(), xml)
|
|
|
|
self.assertIsInstance(result, ParseErrorEntry)
|
|
self.assertEqual(result.content, xml)
|
|
self.assertIn("Unknown root element", result.error)
|
|
|
|
def test_malformed_xml(self):
|
|
"""Test parsing malformed XML returns error entry"""
|
|
xml = '<unclosed>'
|
|
result = self.parser.parse(datetime.now(), xml)
|
|
|
|
self.assertIsInstance(result, ParseErrorEntry)
|
|
self.assertEqual(result.content, xml)
|
|
# Update expected error message to match implementation
|
|
self.assertIn("Unknown root element", result.error)
|
|
|
|
def test_entry_id_generation(self):
|
|
"""Test that unique IDs are generated for entries"""
|
|
xml1 = '<reasoning>test 1</reasoning>'
|
|
xml2 = '<reasoning>test 2</reasoning>'
|
|
|
|
# Use two different timestamps to ensure different IDs
|
|
timestamp1 = datetime.now()
|
|
timestamp2 = timestamp1 + timedelta(seconds=1)
|
|
|
|
result1 = self.parser.parse(timestamp1, xml1)
|
|
result2 = self.parser.parse(timestamp2, xml2)
|
|
|
|
self.assertNotEqual(result1.id, result2.id)
|
|
|
|
def test_whitespace_handling(self):
|
|
"""Test handling of whitespace in XML"""
|
|
xml = """
|
|
<reasoning>
|
|
test content
|
|
with multiple lines
|
|
</reasoning>
|
|
"""
|
|
result = self.parser.parse(datetime.now(), xml)
|
|
|
|
self.assertIsInstance(result, ReasoningEntry)
|
|
self.assertTrue(result.content.strip())
|
|
|
|
def test_cdata_content(self):
|
|
"""Test parsing content with CDATA sections"""
|
|
xml = """
|
|
<reasoning><![CDATA[Test content with <special> & chars]]></reasoning>
|
|
"""
|
|
result = self.parser.parse(datetime.now(), xml)
|
|
self.assertIsInstance(result, ReasoningEntry)
|
|
self.assertEqual(result.content, "Test content with <special> & chars")
|
|
|
|
def test_empty_element_with_attributes(self):
|
|
"""Test parsing empty element with attributes"""
|
|
xml = '<read_stdin id="123"/>'
|
|
result = self.parser.parse(datetime.now(), xml)
|
|
self.assertIsInstance(result, ReadEntry)
|
|
|
|
def test_error_in_content_extraction(self):
|
|
"""Test handling errors during content extraction"""
|
|
xml = '<single><![CDATA[test]]>' # Malformed CDATA
|
|
result = self.parser.parse(datetime.now(), xml)
|
|
|
|
self.assertIsInstance(result, ParseErrorEntry)
|
|
self.assertEqual(result.content, xml)
|
|
# Update expected error message to match implementation
|
|
self.assertIn("Single entry requires", result.error) |