Fix superfluous CDATA tokens

This commit is contained in:
2024-11-11 14:23:08 +01:00
parent 3a1d8a449c
commit 59f482bc0e
17 changed files with 102 additions and 324 deletions

View File

@@ -4,7 +4,6 @@ import xml.etree.ElementTree as ET
from typing import Optional from typing import Optional
from .entry import Entry from .entry import Entry
from .util import escape_text_for_xml
class BackgroundEntry(Entry): class BackgroundEntry(Entry):
""" """
@@ -66,13 +65,10 @@ class BackgroundEntry(Entry):
""" """
if self._process is not None: if self._process is not None:
try: try:
# Close file handles if they're open
if self._process.stdout: if self._process.stdout:
self._process.stdout.close() self._process.stdout.close()
if self._process.stderr: if self._process.stderr:
self._process.stderr.close() self._process.stderr.close()
# Terminate process if it's still running
if self._process.poll() is None: if self._process.poll() is None:
self._process.terminate() self._process.terminate()
try: try:
@@ -90,65 +86,48 @@ class BackgroundEntry(Entry):
Start the process if not running and collect any new output. Start the process if not running and collect any new output.
Updates stdout and stderr with any new output. Updates stdout and stderr with any new output.
""" """
try: if self._process is None and self._exit_code is None:
# Start process if not running self._process = subprocess.Popen(
if self._process is None and self._exit_code is None: self._script,
self._process = subprocess.Popen( shell=True,
self._script, stdout=subprocess.PIPE,
shell=True, stderr=subprocess.PIPE,
stdout=subprocess.PIPE, text=True,
stderr=subprocess.PIPE, bufsize=1 # Line buffered
text=True, )
bufsize=1 # Line buffered return
) if self._process is None:
return # Return after starting to allow process to generate output return
exit_code = self._process.poll()
if exit_code is not None:
try:
remaining_out, remaining_err = self._process.communicate(timeout=0.1)
self._stdout += remaining_out
self._stderr += remaining_err
except subprocess.TimeoutExpired:
pass # Process didn't finish communicating, try again next update
if self._process is None: self._exit_code = exit_code
return # Process already completed
# Check if process has finished
exit_code = self._process.poll()
if exit_code is not None:
# Process has terminated, collect remaining output
try:
remaining_out, remaining_err = self._process.communicate(timeout=0.1)
self._stdout += remaining_out
self._stderr += remaining_err
except subprocess.TimeoutExpired:
pass # Process didn't finish communicating, try again next update
self._exit_code = exit_code
self.cleanup()
return
# Read stdout if available
if self._process.stdout:
while True:
try:
line = self._process.stdout.readline()
if not line:
break
self._stdout += line
except:
break
# Read stderr if available
if self._process.stderr:
while True:
try:
line = self._process.stderr.readline()
if not line:
break
self._stderr += line
except:
break
except Exception as e:
# Handle any errors
error_msg = f"Error handling background process: {str(e)}"
self._stderr += f"\n{error_msg}"
self._exit_code = -1
self.cleanup() self.cleanup()
return
if self._process.stdout:
while True:
try:
line = self._process.stdout.readline()
if not line:
break
self._stdout += line
except:
break
if self._process.stderr:
while True:
try:
line = self._process.stderr.readline()
if not line:
break
self._stderr += line
except:
break
def generate_context(self) -> ET.Element: def generate_context(self) -> ET.Element:
""" """
@@ -157,23 +136,15 @@ class BackgroundEntry(Entry):
Returns: Returns:
ET.Element: XML element containing the entry's data ET.Element: XML element containing the entry's data
""" """
# Create root element with appropriate status
element = ET.Element("background", {"id": self._id}) element = ET.Element("background", {"id": self._id})
if self._process is not None: if self._process is not None:
element.set("pid", str(self._process.pid)) element.set("pid", str(self._process.pid))
elif self._exit_code is not None: elif self._exit_code is not None:
element.set("exit_code", str(self._exit_code)) element.set("exit_code", str(self._exit_code))
element.text = self._script
# Add script as CDATA or escaped text
element.text = escape_text_for_xml(self._script)
# Add stdout element with accumulated output
stdout_elem = ET.SubElement(element, "stdout") stdout_elem = ET.SubElement(element, "stdout")
stdout_elem.text = escape_text_for_xml(self._stdout) stdout_elem.text = self._stdout
# Add stderr element with accumulated output
stderr_elem = ET.SubElement(element, "stderr") stderr_elem = ET.SubElement(element, "stderr")
stderr_elem.text = escape_text_for_xml(self._stderr) stderr_elem.text = self._stderr
return element return element

View File

@@ -2,7 +2,6 @@ from datetime import datetime
import xml.etree.ElementTree as ET import xml.etree.ElementTree as ET
from .entry import Entry from .entry import Entry
from .util import escape_text_for_xml
class ParseErrorEntry(Entry): class ParseErrorEntry(Entry):
""" """
@@ -43,15 +42,10 @@ class ParseErrorEntry(Entry):
Returns: Returns:
ET.Element: XML element containing the entry's data ET.Element: XML element containing the entry's data
""" """
# Create root element
element = ET.Element("parse_error", {"id": self._id}) element = ET.Element("parse_error", {"id": self._id})
# Add error subelement
error_elem = ET.SubElement(element, "error") error_elem = ET.SubElement(element, "error")
error_elem.text = escape_text_for_xml(self._error) error_elem.text = self._error
# Add content subelement
content_elem = ET.SubElement(element, "content") content_elem = ET.SubElement(element, "content")
content_elem.text = escape_text_for_xml(self._content) content_elem.text = self._content
return element return element

View File

@@ -2,7 +2,6 @@ from datetime import datetime
import xml.etree.ElementTree as ET import xml.etree.ElementTree as ET
from .entry import Entry from .entry import Entry
from .util import escape_text_for_xml
class ReadEntry(Entry): class ReadEntry(Entry):
""" """
@@ -51,11 +50,8 @@ class ReadEntry(Entry):
Returns: Returns:
ET.Element: XML element containing the entry's data ET.Element: XML element containing the entry's data
""" """
# Create root element
element = ET.Element("read_stdin", {"id": self._id}) element = ET.Element("read_stdin", {"id": self._id})
# Add content as CDATA or escaped text if content has been read
if self._read: if self._read:
element.text = escape_text_for_xml(self._content) element.text = self._content
return element return element

View File

@@ -2,7 +2,6 @@ from datetime import datetime
import xml.etree.ElementTree as ET import xml.etree.ElementTree as ET
from .entry import Entry from .entry import Entry
from .util import escape_text_for_xml
class ReasoningEntry(Entry): class ReasoningEntry(Entry):
""" """
@@ -42,10 +41,7 @@ class ReasoningEntry(Entry):
Returns: Returns:
ET.Element: XML element containing the entry's data ET.Element: XML element containing the entry's data
""" """
# Create root element
element = ET.Element("reasoning", {"id": self._id}) element = ET.Element("reasoning", {"id": self._id})
element.text = self._content
# Add content as CDATA or escaped text
element.text = escape_text_for_xml(self._content)
return element return element

View File

@@ -4,7 +4,6 @@ import xml.etree.ElementTree as ET
from typing import Optional from typing import Optional
from .entry import Entry from .entry import Entry
from .util import escape_text_for_xml
class RepeatEntry(Entry): class RepeatEntry(Entry):
""" """
@@ -51,31 +50,15 @@ class RepeatEntry(Entry):
Execute the script and update the output. Execute the script and update the output.
Captures stdout, stderr and exit code from each execution. Captures stdout, stderr and exit code from each execution.
""" """
try: process = subprocess.run(
# Run script and capture output self._script,
process = subprocess.run( shell=True,
self._script, capture_output=True,
shell=True, text=True
capture_output=True, )
text=True self._stdout = process.stdout
) self._stderr = process.stderr
self._exit_code = process.returncode
# Store results
self._stdout = process.stdout
self._stderr = process.stderr
self._exit_code = process.returncode
except subprocess.SubprocessError as e:
# Handle subprocess errors by capturing the error message
self._stdout = ""
self._stderr = str(e)
self._exit_code = -1
except Exception as e:
# Handle any other errors
self._stdout = ""
self._stderr = f"Error executing script: {str(e)}"
self._exit_code = -1
def generate_context(self) -> ET.Element: def generate_context(self) -> ET.Element:
""" """
@@ -84,21 +67,14 @@ class RepeatEntry(Entry):
Returns: Returns:
ET.Element: XML element containing the entry's data ET.Element: XML element containing the entry's data
""" """
# Create root element
element = ET.Element("repeat", { element = ET.Element("repeat", {
"id": self.id, "id": self.id,
"exit_code": str(self._exit_code) if self._exit_code is not None else "-1" "exit_code": str(self._exit_code) if self._exit_code is not None else "-1"
}) })
element.text = self.script
# Add script as CDATA or escaped text
element.text = escape_text_for_xml(self.script)
# Add stdout element
stdout_elem = ET.SubElement(element, "stdout") stdout_elem = ET.SubElement(element, "stdout")
stdout_elem.text = escape_text_for_xml(self._stdout) stdout_elem.text = self._stdout
# Add stderr element
stderr_elem = ET.SubElement(element, "stderr") stderr_elem = ET.SubElement(element, "stderr")
stderr_elem.text = escape_text_for_xml(self._stderr) stderr_elem.text = self._stderr
return element return element

View File

@@ -4,7 +4,6 @@ import xml.etree.ElementTree as ET
from typing import Optional from typing import Optional
from .entry import Entry from .entry import Entry
from .util import escape_text_for_xml
class SingleEntry(Entry): class SingleEntry(Entry):
""" """
@@ -61,33 +60,15 @@ class SingleEntry(Entry):
""" """
if self._executed: if self._executed:
return return
process = subprocess.run(
try: self.script,
# Run script and capture output shell=True,
process = subprocess.run( capture_output=True,
self.script, text=True
shell=True, )
capture_output=True, self._stdout = process.stdout
text=True self._stderr = process.stderr
) self._exit_code = process.returncode
# Store results
self._stdout = process.stdout
self._stderr = process.stderr
self._exit_code = process.returncode
except subprocess.SubprocessError as e:
# Handle subprocess errors by capturing the error message
self._stdout = ""
self._stderr = str(e)
self._exit_code = -1
except Exception as e:
# Handle any other errors
self._stdout = ""
self._stderr = f"Error executing script: {str(e)}"
self._exit_code = -1
self._executed = True self._executed = True
def generate_context(self) -> ET.Element: def generate_context(self) -> ET.Element:
@@ -97,25 +78,14 @@ class SingleEntry(Entry):
Returns: Returns:
ET.Element: XML element containing the entry's data ET.Element: XML element containing the entry's data
""" """
# Create root element
element = ET.Element("single", { element = ET.Element("single", {
"id": self.id, "id": self.id,
}) })
element.text = self.script
# Add script as CDATA or escaped text
element.text = escape_text_for_xml(self.script)
# Only add output elements if script has executed
if self._executed: if self._executed:
# Add stdout element if there is output
stdout_elem = ET.SubElement(element, "stdout") stdout_elem = ET.SubElement(element, "stdout")
stdout_elem.text = escape_text_for_xml(self._stdout) stdout_elem.text = self._stdout
# Add stderr element if there are errors
stderr_elem = ET.SubElement(element, "stderr") stderr_elem = ET.SubElement(element, "stderr")
stderr_elem.text = escape_text_for_xml(self._stderr) stderr_elem.text = self._stderr
# Add exit code attribute after execution
element.set("exit_code", str(self._exit_code)) element.set("exit_code", str(self._exit_code))
return element return element

View File

@@ -23,12 +23,6 @@ def stop_before_value(iterator: Iterator[str], stop_value: str) -> Iterator[str]
break break
yield item yield item
def escape_text_for_xml(text: str) -> str:
"""Convert any value to string representation for XML storage."""
if text is None:
return ''
return str(text)
def pretty_print_element(elem: ET.Element, level: int = 0, max_line: int = 80) -> str: def pretty_print_element(elem: ET.Element, level: int = 0, max_line: int = 80) -> str:
"""Convert ElementTree element to pretty-printed string with custom formatting.""" """Convert ElementTree element to pretty-printed string with custom formatting."""
indent = ' ' * level indent = ' ' * level
@@ -73,30 +67,3 @@ def pretty_print_element(elem: ET.Element, level: int = 0, max_line: int = 80) -
parts.append(f'{indent}</{tag}>') parts.append(f'{indent}</{tag}>')
return '\n'.join(parts) return '\n'.join(parts)
def escape_text_for_xml(text: str) -> str:
"""
Convert any text value to a properly escaped XML string.
Args:
text: Text to escape, can be string, number, or None
Returns:
str: XML-escaped string representation
"""
if text is None:
return ''
# Convert numbers to strings
text_str = str(text)
# Standard XML escaping if string contains CDATA end sequence
if ']]>' in text_str:
return text_str.replace('&', '&amp;') \
.replace('<', '&lt;') \
.replace('>', '&gt;') \
.replace('"', '&quot;') \
.replace("'", '&apos;')
# Otherwise wrap in CDATA
return f'<![CDATA[{text_str}]]>'

View File

@@ -2,7 +2,6 @@ from datetime import datetime
import xml.etree.ElementTree as ET import xml.etree.ElementTree as ET
from .entry import Entry from .entry import Entry
from .util import escape_text_for_xml
class WriteEntry(Entry): class WriteEntry(Entry):
""" """
@@ -45,10 +44,6 @@ class WriteEntry(Entry):
Returns: Returns:
ET.Element: XML element containing the entry's data ET.Element: XML element containing the entry's data
""" """
# Create root element
element = ET.Element("write_stdout", {"id": self.id}) element = ET.Element("write_stdout", {"id": self.id})
element.text = self.content
# Add content as CDATA or escaped text
element.text = escape_text_for_xml(self.content)
return element return element

View File

@@ -44,7 +44,7 @@ class BackgroundEntryTest(unittest.TestCase):
context = entry.generate_context() context = entry.generate_context()
# Initial context should have script but no pid or exit_code # Initial context should have script but no pid or exit_code
self.assertEqual(context.text, "<![CDATA[echo test]]>") self.assertEqual(context.text, "echo test")
self.assertIsNone(context.get("pid")) self.assertIsNone(context.get("pid"))
self.assertIsNone(context.get("exit_code")) self.assertIsNone(context.get("exit_code"))
@@ -60,8 +60,8 @@ class BackgroundEntryTest(unittest.TestCase):
# Verify output # Verify output
context = entry.generate_context() context = entry.generate_context()
self.assertTrue("<![CDATA[test" in context.find("stdout").text) self.assertEqual(context.find("stdout").text, "test\n")
self.assertEqual(context.find("stderr").text, "<![CDATA[]]>") self.assertEqual(context.find("stderr").text, "")
def test_continuous_output(self): def test_continuous_output(self):
"""Test process that generates continuous output""" """Test process that generates continuous output"""
@@ -96,7 +96,7 @@ class BackgroundEntryTest(unittest.TestCase):
# Verify error captured # Verify error captured
context = entry.generate_context() context = entry.generate_context()
self.assertTrue(len(context.find("stderr").text) > 0) self.assertTrue(len(context.find("stderr").text) > 0)
self.assertEqual(context.find("stdout").text, "<![CDATA[]]>") self.assertEqual(context.find("stdout").text, "")
def test_cleanup_running_process(self): def test_cleanup_running_process(self):
"""Test cleanup of running process""" """Test cleanup of running process"""

View File

@@ -162,7 +162,7 @@ class BaseAgentTest(unittest.TestCase):
self.assertIsNotNone(reasoning_elem) self.assertIsNotNone(reasoning_elem)
self.assertEqual(reasoning_elem.get("id"), "test-id") self.assertEqual(reasoning_elem.get("id"), "test-id")
# Strip whitespace before comparing # Strip whitespace before comparing
self.assertEqual(reasoning_elem.text.strip(), "<![CDATA[test reasoning]]>") self.assertEqual(reasoning_elem.text.strip(), "test reasoning")
def test_multiple_entries(self): def test_multiple_entries(self):
"""Test handling multiple entries in working memory.""" """Test handling multiple entries in working memory."""
@@ -186,4 +186,4 @@ class BaseAgentTest(unittest.TestCase):
self.assertEqual(len(reasoning_elems), 3) self.assertEqual(len(reasoning_elems), 3)
for i, elem in enumerate(reasoning_elems): for i, elem in enumerate(reasoning_elems):
self.assertEqual(elem.get("id"), f"id-{i}") self.assertEqual(elem.get("id"), f"id-{i}")
self.assertEqual(elem.text.strip(), f"<![CDATA[test {i}]]>") self.assertEqual(elem.text.strip(), f"test {i}")

View File

@@ -48,12 +48,12 @@ class ParseErrorEntryTest(unittest.TestCase):
# Verify error element # Verify error element
error_elem = element.find("error") error_elem = element.find("error")
self.assertIsNotNone(error_elem) self.assertIsNotNone(error_elem)
self.assertEqual(error_elem.text, f"<![CDATA[{error}]]>") self.assertEqual(error_elem.text, f"{error}")
# Verify content element # Verify content element
content_elem = element.find("content") content_elem = element.find("content")
self.assertIsNotNone(content_elem) self.assertIsNotNone(content_elem)
self.assertEqual(content_elem.text, f"<![CDATA[{content}]]>") self.assertEqual(content_elem.text, f"{content}")
def test_special_characters(self): def test_special_characters(self):
"""Test handling content and errors with special characters""" """Test handling content and errors with special characters"""
@@ -62,13 +62,13 @@ class ParseErrorEntryTest(unittest.TestCase):
entry = ParseErrorEntry(content, error, self.test_id, self.test_timestamp) entry = ParseErrorEntry(content, error, self.test_id, self.test_timestamp)
element = entry.generate_context() element = entry.generate_context()
self.assertEqual(element.find("content").text, f"<![CDATA[{content}]]>") self.assertEqual(element.find("content").text, f"{content}")
self.assertEqual(element.find("error").text, f"<![CDATA[{error}]]>") self.assertEqual(element.find("error").text, f"{error}")
def test_empty_content_and_error(self): def test_empty_content_and_error(self):
"""Test handling empty content and error messages""" """Test handling empty content and error messages"""
entry = ParseErrorEntry("", "", self.test_id, self.test_timestamp) entry = ParseErrorEntry("", "", self.test_id, self.test_timestamp)
element = entry.generate_context() element = entry.generate_context()
self.assertEqual(element.find("content").text, "<![CDATA[]]>") self.assertEqual(element.find("content").text, "")
self.assertEqual(element.find("error").text, "<![CDATA[]]>") self.assertEqual(element.find("error").text, "")

View File

@@ -94,7 +94,7 @@ class ReadEntryTest(unittest.TestCase):
# Verify XML structure # Verify XML structure
self.assertEqual(element.tag, "read_stdin") self.assertEqual(element.tag, "read_stdin")
self.assertEqual(element.get("id"), self.test_id) self.assertEqual(element.get("id"), self.test_id)
self.assertEqual(element.text, f"<![CDATA[{test_input}]]>") self.assertEqual(element.text, f"{test_input}")
def test_unicode_content(self): def test_unicode_content(self):
"""Test reading Unicode content""" """Test reading Unicode content"""
@@ -120,4 +120,4 @@ class ReadEntryTest(unittest.TestCase):
# Verify XML escaping for multiline content # Verify XML escaping for multiline content
element = entry.generate_context() element = entry.generate_context()
self.assertEqual(element.text, f"<![CDATA[{test_input}]]>") self.assertEqual(element.text, f"{test_input}")

View File

@@ -42,7 +42,7 @@ class ReasoningEntryTest(unittest.TestCase):
# Verify XML structure # Verify XML structure
self.assertEqual(element.tag, "reasoning") self.assertEqual(element.tag, "reasoning")
self.assertEqual(element.get("id"), self.test_id) self.assertEqual(element.get("id"), self.test_id)
self.assertEqual(element.text, f"<![CDATA[{content}]]>") self.assertEqual(element.text, f"{content}")
def test_special_characters(self): def test_special_characters(self):
"""Test handling reasoning text with special characters""" """Test handling reasoning text with special characters"""
@@ -50,14 +50,14 @@ class ReasoningEntryTest(unittest.TestCase):
entry = ReasoningEntry(content, self.test_id, self.test_timestamp) entry = ReasoningEntry(content, self.test_id, self.test_timestamp)
element = entry.generate_context() element = entry.generate_context()
self.assertEqual(element.text, f"<![CDATA[{content}]]>") self.assertEqual(element.text, f"{content}")
def test_empty_content(self): def test_empty_content(self):
"""Test handling empty reasoning text""" """Test handling empty reasoning text"""
entry = ReasoningEntry("", self.test_id, self.test_timestamp) entry = ReasoningEntry("", self.test_id, self.test_timestamp)
element = entry.generate_context() element = entry.generate_context()
self.assertEqual(element.text, "<![CDATA[]]>") self.assertEqual(element.text, "")
def test_large_content(self): def test_large_content(self):
"""Test handling large reasoning text""" """Test handling large reasoning text"""
@@ -65,7 +65,7 @@ class ReasoningEntryTest(unittest.TestCase):
entry = ReasoningEntry(content, self.test_id, self.test_timestamp) entry = ReasoningEntry(content, self.test_id, self.test_timestamp)
element = entry.generate_context() element = entry.generate_context()
self.assertEqual(element.text, f"<![CDATA[{content}]]>") self.assertEqual(element.text, f"{content}")
def test_multiline_content(self): def test_multiline_content(self):
"""Test handling multiline reasoning text""" """Test handling multiline reasoning text"""
@@ -75,4 +75,4 @@ class ReasoningEntryTest(unittest.TestCase):
entry = ReasoningEntry(content, self.test_id, self.test_timestamp) entry = ReasoningEntry(content, self.test_id, self.test_timestamp)
element = entry.generate_context() element = entry.generate_context()
self.assertEqual(element.text, f"<![CDATA[{content}]]>") self.assertEqual(element.text, f"{content}")

View File

@@ -95,19 +95,14 @@ class RepeatEntryTest(unittest.TestCase):
entry = RepeatEntry(script, self.test_id, self.test_timestamp) entry = RepeatEntry(script, self.test_id, self.test_timestamp)
entry.update() entry.update()
element = entry.generate_context() element = entry.generate_context()
# Verify XML structure # Verify XML structure
self.assertEqual(element.tag, "repeat") self.assertEqual(element.tag, "repeat")
self.assertEqual(element.get("id"), self.test_id) self.assertEqual(element.get("id"), self.test_id)
self.assertEqual(element.text, f"<![CDATA[{script}]]>") self.assertEqual(element.text, f"{script}")
self.assertEqual(element.get("exit_code"), "0") self.assertEqual(element.get("exit_code"), "0")
# Get stdout and remove trailing newline for comparison
stdout_text = element.find("stdout").text stdout_text = element.find("stdout").text
self.assertTrue(stdout_text.startswith("<![CDATA[test")) self.assertEqual(stdout_text, "test\n")
self.assertTrue(stdout_text.endswith("]]>")) self.assertEqual(element.find("stderr").text, "")
self.assertEqual(element.find("stderr").text, "<![CDATA[]]>")
def test_changing_output(self): def test_changing_output(self):
"""Test handling of changing command output""" """Test handling of changing command output"""

View File

@@ -98,7 +98,7 @@ class SingleEntryTest(unittest.TestCase):
# Verify XML structure # Verify XML structure
self.assertEqual(element.tag, "single") self.assertEqual(element.tag, "single")
self.assertEqual(element.get("id"), self.test_id) self.assertEqual(element.get("id"), self.test_id)
self.assertEqual(element.text, "<![CDATA[echo test]]>") self.assertEqual(element.text, "echo test")
# Verify no output elements exist # Verify no output elements exist
self.assertIsNone(element.find("stdout")) self.assertIsNone(element.find("stdout"))
@@ -116,14 +116,12 @@ class SingleEntryTest(unittest.TestCase):
# Verify XML structure # Verify XML structure
self.assertEqual(element.tag, "single") self.assertEqual(element.tag, "single")
self.assertEqual(element.get("id"), self.test_id) self.assertEqual(element.get("id"), self.test_id)
self.assertEqual(element.text, f"<![CDATA[{script}]]>") self.assertEqual(element.text, f"{script}")
# Get stdout and remove trailing newline for comparison # Get stdout and remove trailing newline for comparison
stdout_text = element.find("stdout").text stdout_text = element.find("stdout").text
self.assertTrue(stdout_text.startswith("<![CDATA[test")) self.assertEqual(stdout_text, "test\n")
self.assertTrue(stdout_text.endswith("]]>")) self.assertEqual(element.find("stderr").text, "")
self.assertEqual(element.find("stderr").text, "<![CDATA[]]>")
self.assertEqual(element.get("exit_code"), "0") self.assertEqual(element.get("exit_code"), "0")
def test_multiple_commands(self): def test_multiple_commands(self):

View File

@@ -36,86 +36,6 @@ class UtilTest(unittest.TestCase):
result = list(util.stop_before_value(create_iterator(input_sequence), 'STOP')) result = list(util.stop_before_value(create_iterator(input_sequence), 'STOP'))
self.assertEqual(result, []) self.assertEqual(result, [])
def test_none_value(self):
"""Test that None values are converted to empty strings"""
self.assertEqual(util.escape_text_for_xml(None), '')
def test_number_values(self):
"""Test that numbers are properly converted to strings and wrapped in CDATA"""
self.assertEqual(util.escape_text_for_xml(42), '<![CDATA[42]]>')
self.assertEqual(util.escape_text_for_xml(3.14), '<![CDATA[3.14]]>')
self.assertEqual(util.escape_text_for_xml(-100), '<![CDATA[-100]]>')
self.assertEqual(util.escape_text_for_xml(0), '<![CDATA[0]]>')
def test_simple_strings(self):
"""Test that simple strings are wrapped in CDATA"""
self.assertEqual(util.escape_text_for_xml('hello'), '<![CDATA[hello]]>')
self.assertEqual(util.escape_text_for_xml(''), '<![CDATA[]]>')
self.assertEqual(util.escape_text_for_xml(' '), '<![CDATA[ ]]>')
def test_strings_with_special_characters(self):
"""Test strings containing XML special characters but no CDATA end sequence"""
self.assertEqual(util.escape_text_for_xml('<hello>'), '<![CDATA[<hello>]]>')
self.assertEqual(util.escape_text_for_xml('a & b'), '<![CDATA[a & b]]>')
self.assertEqual(util.escape_text_for_xml('"quoted"'), '<![CDATA["quoted"]]>')
self.assertEqual(util.escape_text_for_xml("'single'"), "<![CDATA['single']]>")
self.assertEqual(util.escape_text_for_xml('<![CDATA['), '<![CDATA[<![CDATA[]]>')
def test_strings_with_cdata_end_sequence(self):
"""Test strings containing CDATA end sequence are properly escaped"""
# Simple case with just the CDATA end sequence
self.assertEqual(util.escape_text_for_xml(']]>'), ']]&gt;')
# CDATA end sequence with other special characters
self.assertEqual(
util.escape_text_for_xml('Text with ]]> and <tags> & ampersands'),
'Text with ]]&gt; and &lt;tags&gt; &amp; ampersands'
)
# Multiple CDATA end sequences
self.assertEqual(
util.escape_text_for_xml('Multiple ]]> end ]]> sequences'),
'Multiple ]]&gt; end ]]&gt; sequences'
)
def test_multiline_strings(self):
"""Test multiline strings are properly handled"""
multiline = """Line 1
Line 2
Line 3"""
self.assertEqual(util.escape_text_for_xml(multiline), '<![CDATA[Line 1\n Line 2\n Line 3]]>')
multiline_with_cdata = """Line 1
Line ]]> 2
Line 3"""
self.assertEqual(
util.escape_text_for_xml(multiline_with_cdata),
'Line 1\n Line ]]&gt; 2\n Line 3'
)
def test_edge_cases(self):
"""Test edge cases and unusual inputs"""
# Empty string
self.assertEqual(util.escape_text_for_xml(''), '<![CDATA[]]>')
# String with only whitespace
self.assertEqual(util.escape_text_for_xml('\n\t '), '<![CDATA[\n\t ]]>')
# String with Unicode characters
self.assertEqual(util.escape_text_for_xml('Hello 世界'), '<![CDATA[Hello 世界]]>')
# String with control characters
self.assertEqual(util.escape_text_for_xml('Hello\0World'), '<![CDATA[Hello\0World]]>')
# Very long string
long_string = 'x' * 1000
self.assertEqual(util.escape_text_for_xml(long_string), f'<![CDATA[{long_string}]]>')
def test_boolean_values(self):
"""Test boolean values are properly converted"""
self.assertEqual(util.escape_text_for_xml(True), '<![CDATA[True]]>')
self.assertEqual(util.escape_text_for_xml(False), '<![CDATA[False]]>')
def test_pretty_print_element_basic(self): def test_pretty_print_element_basic(self):
"""Test basic element printing""" """Test basic element printing"""
elem = ET.Element('root') elem = ET.Element('root')

View File

@@ -85,14 +85,14 @@ class WriteEntryTest(unittest.TestCase):
# Verify XML structure # Verify XML structure
self.assertEqual(element.tag, "write_stdout") self.assertEqual(element.tag, "write_stdout")
self.assertEqual(element.get("id"), self.test_id) self.assertEqual(element.get("id"), self.test_id)
self.assertEqual(element.text, f"<![CDATA[{content}]]>") self.assertEqual(element.text, f"{content}")
# Write content and verify context remains the same # Write content and verify context remains the same
entry.update() entry.update()
element_after = entry.generate_context() element_after = entry.generate_context()
self.assertEqual(element_after.tag, "write_stdout") self.assertEqual(element_after.tag, "write_stdout")
self.assertEqual(element_after.get("id"), self.test_id) self.assertEqual(element_after.get("id"), self.test_id)
self.assertEqual(element_after.text, f"<![CDATA[{content}]]>") self.assertEqual(element_after.text, f"{content}")
def test_unicode_content(self): def test_unicode_content(self):
"""Test writing Unicode content""" """Test writing Unicode content"""
@@ -114,4 +114,4 @@ class WriteEntryTest(unittest.TestCase):
# Verify XML escaping for multiline content # Verify XML escaping for multiline content
element = entry.generate_context() element = entry.generate_context()
self.assertEqual(element.text, f"<![CDATA[{content}]]>") self.assertEqual(element.text, f"{content}")