llm/SIA
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Added timeout and limit to script entries

Browse Source
This commit is contained in:
Niels Geens
2024-11-13 17:48:39 +01:00
parent 65478e0b17
commit 834ce12e8f
21 changed files with 262 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
action_schema.xsd
View File

@@ -33,6 +33,8 @@
Single script that runs once and completes. Single script that runs once and completes.
Output is stored in context until explicitly deleted. Output is stored in context until explicitly deleted.
Used for one-time operations like file manipulation. Used for one-time operations like file manipulation.
Single scripts are limited to 1024 characters and 1 second timeout by default.
These limits can be changed with attributes.
Example: Example:
<single> <single>
ls / ls /
@@ -43,6 +45,8 @@
<xs:sequence> <xs:sequence>
<xs:any minOccurs="0" maxOccurs="unbounded" processContents="skip"/> <xs:any minOccurs="0" maxOccurs="unbounded" processContents="skip"/>
</xs:sequence> </xs:sequence>
<xs:attribute name="timeout" type="xs:float" use="optional"/>
<xs:attribute name="limit" type="xs:integer" use="optional"/>
</xs:complexType> </xs:complexType>
</xs:element> </xs:element>
@@ -51,6 +55,8 @@
After a command is issued, all repeat scripts in context are run again. After a command is issued, all repeat scripts in context are run again.
Useful for monitoring changing files or viewing results immediately after changing a file. Useful for monitoring changing files or viewing results immediately after changing a file.
Repeat scripts should execute quickly to avoid blocking the agent. Repeat scripts should execute quickly to avoid blocking the agent.
Repeat scripts are limited to 1024 characters and 1 second timeout by default.
These limits can be changed with attributes.
Example: Example:
<repeat> <repeat>
ls / ls /
@@ -61,6 +67,8 @@
<xs:sequence> <xs:sequence>
<xs:any minOccurs="0" maxOccurs="unbounded" processContents="skip"/> <xs:any minOccurs="0" maxOccurs="unbounded" processContents="skip"/>
</xs:sequence> </xs:sequence>
<xs:attribute name="timeout" type="xs:float" use="optional"/>
<xs:attribute name="limit" type="xs:integer" use="optional"/>
</xs:complexType> </xs:complexType>
</xs:element> </xs:element>
@@ -83,6 +91,7 @@
<!-- <!--
Read all available text on stdin and store it in context. Read all available text on stdin and store it in context.
Do this only if the context indicates there is data in the stdin buffer.
Example: Example:
<read_stdin/> <read_stdin/>
--> -->
@@ -93,6 +102,7 @@
<!-- <!--
Write to stdout. Write to stdout.
This is your main way of contacting the user. This is your main way of contacting the user.
Make sure you have properly reasoned about what to say and if it is necessary before issuing a write_stdout command.
Example: Example:
<write_stdout> <write_stdout>
Hello world! Hello world!

9
sia/background_entry.py
View File

@@ -17,14 +17,19 @@ class BackgroundEntry(Entry):
_exit_code: Exit code when process completes _exit_code: Exit code when process completes
""" """
def __init__(self, script: str, id: str, timestamp: datetime): def __init__(
self,
id: str,
timestamp: datetime,
script: str,
):
""" """
Initialize a new background entry. Initialize a new background entry.
Args: Args:
script: The script/command to execute
id: Unique identifier for this entry id: Unique identifier for this entry
timestamp: Creation timestamp for this entry timestamp: Creation timestamp for this entry
script: The script/command to execute
""" """
super().__init__(id, timestamp) super().__init__(id, timestamp)
self._script = script self._script = script

12
sia/parse_error_entry.py
View File

@@ -8,15 +8,21 @@ class ParseErrorEntry(Entry):
Entry type for parse and validation errors. Entry type for parse and validation errors.
""" """
def __init__(self, content: str, error: str, id: str, timestamp: datetime): def __init__(
self,
id: str,
timestamp: datetime,
content: str,
error: str,
):
""" """
Initialize a new parse error entry. Initialize a new parse error entry.
Args: Args:
content: Original content that failed to parse
error: Error message describing the failure
id: Unique identifier for this entry id: Unique identifier for this entry
timestamp: Creation timestamp for this entry timestamp: Creation timestamp for this entry
content: Original content that failed to parse
error: Error message describing the failure
""" """
super().__init__(id, timestamp) super().__init__(id, timestamp)
self._content = content self._content = content

9
sia/read_entry.py
View File

@@ -13,14 +13,19 @@ class ReadEntry(Entry):
_read: Whether content has been read _read: Whether content has been read
""" """
def __init__(self, io_buffer, id: str, timestamp: datetime): def __init__(
self,
id: str,
timestamp: datetime,
io_buffer,
):
""" """
Initialize a new read entry. Initialize a new read entry.
Args: Args:
io_buffer: Buffer to use for IO operations
id: Unique identifier for this entry id: Unique identifier for this entry
timestamp: Creation timestamp for this entry timestamp: Creation timestamp for this entry
io_buffer: Buffer to use for IO operations
""" """
super().__init__(id, timestamp) super().__init__(id, timestamp)
self._content = "" self._content = ""

9
sia/reasoning_entry.py
View File

@@ -11,14 +11,19 @@ class ReasoningEntry(Entry):
_content: The reasoning text _content: The reasoning text
""" """
def __init__(self, content: str, id: str, timestamp: datetime): def __init__(
self,
id: str,
timestamp: datetime,
content: str,
):
""" """
Initialize a new reasoning entry. Initialize a new reasoning entry.
Args: Args:
content: The reasoning text
id: Unique identifier for this entry id: Unique identifier for this entry
timestamp: Creation timestamp for this entry timestamp: Creation timestamp for this entry
content: The reasoning text
""" """
super().__init__(id, timestamp) super().__init__(id, timestamp)
self._content = content self._content = content

69
sia/repeat_entry.py
View File

@@ -8,22 +8,46 @@ from .entry import Entry
class RepeatEntry(Entry): class RepeatEntry(Entry):
""" """
Entry type for scripts that are executed on every update. Entry type for scripts that are executed on every update.
Attributes:
script: The script/command to execute
timeout: Maximum time to wait for script execution
limit: Maximum number of characters to capture from stdout/stderr
stdout: Captured standard output from script execution
stderr: Captured standard error from script execution
exit_code: Process exit code after completion
_executed: Whether the script has been executed
""" """
default_timeout = 1
default_limit = 1024
def __init__(self, script: str, id: str, timestamp: datetime): def __init__(
self,
id: str,
timestamp: datetime,
script: str,
timeout: Optional[float],
limit: Optional[int],
):
""" """
Initialize a new repeat entry. Initialize a new repeat entry.
Args: Args:
script: The script/command to execute
id: Unique identifier for this entry id: Unique identifier for this entry
timestamp: Creation timestamp for this entry timestamp: Creation timestamp for this entry
script: The script/command to execute
timeout: Maximum time to wait for script execution
limit: Maximum number of characters to capture from stdout/stderr
""" """
super().__init__(id, timestamp) super().__init__(id, timestamp)
self._script = script self._script = script
self._timeout = timeout or self.default_timeout
self._limit = limit or self.default_limit
self._stdout = "" self._stdout = ""
self._stderr = "" self._stderr = ""
self._exit_code: Optional[int] = None self._exit_code: Optional[int] = None
self._timed_out = False
@property @property
def script(self) -> str: def script(self) -> str:
@@ -50,15 +74,20 @@ 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.
""" """
process = subprocess.run( try:
self._script, process = subprocess.run(
shell=True, self._script,
capture_output=True, timeout=self._timeout,
text=True shell=True,
) capture_output=True,
self._stdout = process.stdout text=True
self._stderr = process.stderr )
self._exit_code = process.returncode self._stdout = process.stdout
self._stderr = process.stderr
self._exit_code = process.returncode
except subprocess.TimeoutExpired as e:
self._timed_out = True
def generate_context(self) -> ET.Element: def generate_context(self) -> ET.Element:
""" """
@@ -69,12 +98,20 @@ class RepeatEntry(Entry):
""" """
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"
}) })
element.text = self.script element.text = self.script
stdout_elem = ET.SubElement(element, "stdout") if self._exit_code is not None:
stdout_elem.text = self._stdout element.set("exit_code", str(self._exit_code))
stderr_elem = ET.SubElement(element, "stderr") if self._timed_out:
stderr_elem.text = self._stderr element.set("timed_out", "true")
else:
if len(self._stdout) > self._limit:
element.set("stdout_truncated", "true")
stdout_elem = ET.SubElement(element, "stdout")
stdout_elem.text = self._stdout[:self._limit]
if len(self._stderr) > self._limit:
element.set("stderr_truncated", "true")
stderr_elem = ET.SubElement(element, "stderr")
stderr_elem.text = self._stderr[:self._limit]
return element return element

50
sia/response_parser.py
View File

@@ -54,15 +54,15 @@ class ResponseParser:
try: try:
root = ET.fromstring(xml.strip()) root = ET.fromstring(xml.strip())
except ET.ParseError as e: except ET.ParseError as e:
return ParseErrorEntry(xml, f"Invalid XML: {str(e)}", entry_id, timestamp) return ParseErrorEntry(entry_id, timestamp, xml, f"Invalid XML: {str(e)}")
try: try:
if root.tag == 'delete': if root.tag == 'delete':
target_id = root.get('id') target_id = root.get('id')
if not target_id: if not target_id:
return ParseErrorEntry(xml, "Delete command missing required 'id' attribute", entry_id, timestamp) return ParseErrorEntry(entry_id, timestamp, xml, "Delete command missing required 'id' attribute")
if len(root.attrib) > 1: if len(root.attrib) > 1:
return ParseErrorEntry(xml, "Delete command should only have 'id' attribute", entry_id, timestamp) return ParseErrorEntry(entry_id, timestamp, xml, "Delete command should only have 'id' attribute")
return DeleteCommand(target_id) return DeleteCommand(target_id)
elif root.tag == 'stop': elif root.tag == 'stop':
@@ -70,34 +70,46 @@ class ResponseParser:
elif root.tag == 'background': elif root.tag == 'background':
if len(root) != 0 or root.attrib or root.text is None or root.text.strip() == '': if len(root) != 0 or root.attrib or root.text is None or root.text.strip() == '':
return ParseErrorEntry(xml, "Background entry requires (only) script content", entry_id, timestamp) return ParseErrorEntry(entry_id, timestamp, xml, "Background entry requires (only) script content")
return BackgroundEntry(root.text, entry_id, timestamp) return BackgroundEntry(entry_id, timestamp, root.text)
elif root.tag == 'repeat': elif root.tag == 'repeat':
if len(root) != 0 or root.attrib or root.text is None or root.text.strip() == '': if len(root) != 0 or root.text is None or root.text.strip() == '':
return ParseErrorEntry(xml, "Repeat entry requires (only) script content", entry_id, timestamp) return ParseErrorEntry(entry_id, timestamp, xml, "Repeat entry requires (only) script content")
return RepeatEntry(root.text, entry_id, timestamp) if len(root.attrib) > 2 or any(k not in ('timeout', 'limit') for k in root.attrib):
return ParseErrorEntry(entry_id, timestamp, xml, "Repeat entry only accepts 'timeout' and 'limit' attributes")
timeout = root.get('timeout')
limit = root.get('limit')
timeout = float(timeout) if timeout is not None else None
limit = int(limit) if limit is not None else None
return RepeatEntry(entry_id, timestamp, root.text, timeout, limit)
elif root.tag == 'single': elif root.tag == 'single':
if len(root) != 0 or root.attrib or root.text is None or root.text.strip() == '': if len(root) != 0 or root.text is None or root.text.strip() == '':
return ParseErrorEntry(xml, "Single entry requires (only) script content", entry_id, timestamp) return ParseErrorEntry(entry_id, timestamp, xml, "Single entry requires (only) script content")
return SingleEntry(root.text, entry_id, timestamp) if len(root.attrib) > 2 or any(k not in ('timeout', 'limit') for k in root.attrib):
return ParseErrorEntry(entry_id, timestamp, xml, "Single entry only accepts 'timeout' and 'limit' attributes")
timeout = root.get('timeout')
limit = root.get('limit')
timeout = float(timeout) if timeout is not None else None
limit = int(limit) if limit is not None else None
return SingleEntry(entry_id, timestamp, root.text, timeout, limit)
elif root.tag == 'reasoning': elif root.tag == 'reasoning':
if len(root) != 0 or root.attrib or root.text is None or root.text.strip() == '': if len(root) != 0 or root.attrib or root.text is None or root.text.strip() == '':
return ParseErrorEntry(xml, "Reasoning entry requires (only) text content", entry_id, timestamp) return ParseErrorEntry(entry_id, timestamp, xml, "Reasoning entry requires (only) text content")
return ReasoningEntry(root.text, entry_id, timestamp) return ReasoningEntry(entry_id, timestamp, root.text)
elif root.tag == 'read_stdin': elif root.tag == 'read_stdin':
return ReadEntry(self.io_buffer, entry_id, timestamp) return ReadEntry(entry_id, timestamp, self.io_buffer)
elif root.tag == 'write_stdout': elif root.tag == 'write_stdout':
if len(root) != 0 or root.attrib or root.text is None or root.text.strip() == '': if len(root) != 0 or root.attrib or root.text is None or root.text.strip() == '':
return ParseErrorEntry(xml, "Write stdout entry requires (only) text content", entry_id, timestamp) return ParseErrorEntry(entry_id, timestamp, xml, "Write stdout entry requires (only) text content")
return WriteEntry(root.text, self.io_buffer, entry_id, timestamp) return WriteEntry(entry_id, timestamp, root.text, self.io_buffer)
else: else:
return ParseErrorEntry(xml, f"Unknown root element: {root.tag}", entry_id, timestamp) return ParseErrorEntry(entry_id, timestamp, xml, f"Unknown root element: {root.tag}")
except Exception as e: except Exception as e:
return ParseErrorEntry(xml, f"Error parsing response: {str(e)}", entry_id, timestamp) return ParseErrorEntry(entry_id, timestamp, xml, f"Error parsing response: {str(e)}")

59
sia/single_entry.py
View File

@@ -11,27 +11,44 @@ class SingleEntry(Entry):
Attributes: Attributes:
script: The script/command to execute script: The script/command to execute
timeout: Maximum time to wait for script execution
limit: Maximum number of characters to capture from stdout/stderr
stdout: Captured standard output from script execution stdout: Captured standard output from script execution
stderr: Captured standard error from script execution stderr: Captured standard error from script execution
exit_code: Process exit code after completion exit_code: Process exit code after completion
_executed: Whether the script has been executed _executed: Whether the script has been executed
""" """
default_timeout = 1
default_limit = 1024
def __init__(self, script: str, id: str, timestamp: datetime): def __init__(
self,
id: str,
timestamp: datetime,
script: str,
timeout: Optional[float],
limit: Optional[int],
):
""" """
Initialize a new single shot entry. Initialize a new single shot entry.
Args: Args:
script: The script/command to execute
id: Unique identifier for this entry id: Unique identifier for this entry
timestamp: Creation timestamp for this entry timestamp: Creation timestamp for this entry
script: The script/command to execute
timeout: Maximum time to wait for script execution
limit: Maximum number of characters to capture from stdout/stderr
""" """
super().__init__(id, timestamp) super().__init__(id, timestamp)
self._script = script self._script = script
self._timeout = timeout or self.default_timeout
self._limit = limit or self.default_limit
self._stdout = "" self._stdout = ""
self._stderr = "" self._stderr = ""
self._exit_code: Optional[int] = None self._exit_code: Optional[int] = None
self._executed = False self._executed = False
self._timed_out = False
@property @property
def script(self) -> str: def script(self) -> str:
@@ -60,16 +77,20 @@ class SingleEntry(Entry):
""" """
if self._executed: if self._executed:
return return
process = subprocess.run(
self.script,
shell=True,
capture_output=True,
text=True
)
self._stdout = process.stdout
self._stderr = process.stderr
self._exit_code = process.returncode
self._executed = True self._executed = True
try:
process = subprocess.run(
self._script,
timeout=self._timeout,
shell=True,
capture_output=True,
text=True
)
self._stdout = process.stdout
self._stderr = process.stderr
self._exit_code = process.returncode
except subprocess.TimeoutExpired as e:
self._timed_out = True
def generate_context(self) -> ET.Element: def generate_context(self) -> ET.Element:
""" """
@@ -82,10 +103,16 @@ class SingleEntry(Entry):
"id": self.id, "id": self.id,
}) })
element.text = self.script element.text = self.script
if self._executed: if self._timed_out:
stdout_elem = ET.SubElement(element, "stdout") element.set("timed_out", "true")
stdout_elem.text = self._stdout elif self._executed:
stderr_elem = ET.SubElement(element, "stderr")
stderr_elem.text = self._stderr
element.set("exit_code", str(self._exit_code)) element.set("exit_code", str(self._exit_code))
if len(self._stdout) > self._limit:
element.set("stdout_truncated", "true")
stdout_elem = ET.SubElement(element, "stdout")
stdout_elem.text = self._stdout[:self._limit]
if len(self._stderr) > self._limit:
element.set("stderr_truncated", "true")
stderr_elem = ET.SubElement(element, "stderr")
stderr_elem.text = self._stderr[:self._limit]
return element return element

13
sia/write_entry.py
View File

@@ -2,6 +2,7 @@ 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 .io_buffer import IOBuffer
class WriteEntry(Entry): class WriteEntry(Entry):
""" """
@@ -13,15 +14,21 @@ class WriteEntry(Entry):
_written: Whether the content has been written _written: Whether the content has been written
""" """
def __init__(self, content: str, io_buffer, id: str, timestamp: datetime): def __init__(
self,
id: str,
timestamp: datetime,
content: str,
io_buffer: IOBuffer,
):
""" """
Initialize a new write entry. Initialize a new write entry.
Args: Args:
content: Text to write to stdout
io_buffer: Buffer to use for IO operations
id: Unique identifier for this entry id: Unique identifier for this entry
timestamp: Creation timestamp for this entry timestamp: Creation timestamp for this entry
content: Text to write to stdout
io_buffer: Buffer to use for IO operations
""" """
super().__init__(id, timestamp) super().__init__(id, timestamp)
self.content = content self.content = content

4
system_prompt.md
View File

@@ -30,6 +30,7 @@ Scripts and their output appear in the context.
You can use a script for starting a detached process that runs in the background. You can use a script for starting a detached process that runs in the background.
All processes can be managed by the usual Linux tools. All processes can be managed by the usual Linux tools.
The scripts defined in the script actions all run in a `bash` shell. The scripts defined in the script actions all run in a `bash` shell.
You are logged in as root.
# File system # File system
@@ -71,6 +72,9 @@ Inspect your previous actions in detail.
If something didn't work, try to understand why and don't repeat the same mistake. If something didn't work, try to understand why and don't repeat the same mistake.
Don't delete mistakes until you understand them. Don't delete mistakes until you understand them.
If you notice parse_error entries in the context you have made a mistake.
Reason about the error before trying again.
# User interaction # User interaction
You are always working for a user. You are always working for a user.

14
test/background_entry_test.py
View File

@@ -40,7 +40,7 @@ class BackgroundEntryTest(unittest.TestCase):
def test_initialization(self): def test_initialization(self):
"""Test entry initialization""" """Test entry initialization"""
entry = BackgroundEntry("echo test", self.test_id, self.test_timestamp) entry = BackgroundEntry(self.test_id, self.test_timestamp, "echo test")
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
@@ -50,7 +50,7 @@ class BackgroundEntryTest(unittest.TestCase):
def test_short_running_process(self): def test_short_running_process(self):
"""Test execution of a quick process""" """Test execution of a quick process"""
entry = BackgroundEntry("echo 'test'", self.test_id, self.test_timestamp) entry = BackgroundEntry(self.test_id, self.test_timestamp, "echo 'test'")
# Wait for process to complete # Wait for process to complete
def is_complete(ctx): def is_complete(ctx):
@@ -74,7 +74,7 @@ class BackgroundEntryTest(unittest.TestCase):
'"' '"'
) )
entry = BackgroundEntry(script, self.test_id, self.test_timestamp) entry = BackgroundEntry(self.test_id, self.test_timestamp, script)
# Wait for process to complete # Wait for process to complete
def has_all_output(ctx): def has_all_output(ctx):
@@ -85,7 +85,7 @@ class BackgroundEntryTest(unittest.TestCase):
def test_process_failure(self): def test_process_failure(self):
"""Test handling of process failures""" """Test handling of process failures"""
entry = BackgroundEntry("nonexistentcommand", self.test_id, self.test_timestamp) entry = BackgroundEntry(self.test_id, self.test_timestamp, "nonexistentcommand")
# Wait for process to fail # Wait for process to fail
def has_failed(ctx): def has_failed(ctx):
@@ -100,7 +100,7 @@ class BackgroundEntryTest(unittest.TestCase):
def test_cleanup_running_process(self): def test_cleanup_running_process(self):
"""Test cleanup of running process""" """Test cleanup of running process"""
entry = BackgroundEntry("sleep 10", self.test_id, self.test_timestamp) entry = BackgroundEntry(self.test_id, self.test_timestamp, "sleep 10")
# Wait for process to start # Wait for process to start
def has_started(ctx): def has_started(ctx):
@@ -122,7 +122,7 @@ class BackgroundEntryTest(unittest.TestCase):
def test_cleanup_completed_process(self): def test_cleanup_completed_process(self):
"""Test cleanup of already completed process""" """Test cleanup of already completed process"""
entry = BackgroundEntry("echo test", self.test_id, self.test_timestamp) entry = BackgroundEntry(self.test_id, self.test_timestamp, "echo test")
# Wait for completion # Wait for completion
def is_complete(ctx): def is_complete(ctx):
@@ -135,7 +135,7 @@ class BackgroundEntryTest(unittest.TestCase):
def test_multiple_cleanup_calls(self): def test_multiple_cleanup_calls(self):
"""Test that multiple cleanup calls are safe""" """Test that multiple cleanup calls are safe"""
entry = BackgroundEntry("sleep 10", self.test_id, self.test_timestamp) entry = BackgroundEntry(self.test_id, self.test_timestamp, "sleep 10")
# Wait for process to start # Wait for process to start
def has_started(ctx): def has_started(ctx):

11
test/base_agent_test.py
View File

@@ -145,8 +145,9 @@ class BaseAgentTest(unittest.TestCase):
self.assertEqual(len(list(root)), 0) self.assertEqual(len(list(root)), 0)
def test_compile_context_with_entries(self): def test_compile_context_with_entries(self):
# Add test entry to working memory """Test context compilation with a single entry."""
entry = ReasoningEntry("test reasoning", "test-id", self.test_timestamp) # Add test entry to working memory - note id and timestamp are now first
entry = ReasoningEntry("test-id", self.test_timestamp, "test reasoning")
self.agent._working_memory.add_entry(entry) self.agent._working_memory.add_entry(entry)
context = self.agent._compile_context() context = self.agent._compile_context()
@@ -166,9 +167,9 @@ class BaseAgentTest(unittest.TestCase):
def test_multiple_entries(self): def test_multiple_entries(self):
"""Test handling multiple entries in working memory.""" """Test handling multiple entries in working memory."""
# Add multiple entries # Add multiple entries - note id and timestamp are now first
entries = [ entries = [
ReasoningEntry(f"test {i}", f"id-{i}", self.test_timestamp) ReasoningEntry(f"id-{i}", self.test_timestamp, f"test {i}")
for i in range(3) for i in range(3)
] ]
@@ -186,4 +187,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"test {i}") self.assertEqual(elem.text.strip(), f"test {i}")

6
test/delete_command_test.py
View File

@@ -8,8 +8,6 @@ from sia.delete_command import DeleteCommand
from sia.reasoning_entry import ReasoningEntry from sia.reasoning_entry import ReasoningEntry
from sia.background_entry import BackgroundEntry from sia.background_entry import BackgroundEntry
class DeleteCommandTest(unittest.TestCase): class DeleteCommandTest(unittest.TestCase):
def setUp(self): def setUp(self):
"""Set up test cases with a fresh working memory""" """Set up test cases with a fresh working memory"""
@@ -31,7 +29,7 @@ class DeleteCommandTest(unittest.TestCase):
def test_delete_running_background_entry(self): def test_delete_running_background_entry(self):
"""Test deleting a background entry that is still running""" """Test deleting a background entry that is still running"""
# Create and start background process # Create and start background process
entry = BackgroundEntry("sleep 10", "test-id", self.test_timestamp) entry = BackgroundEntry(self.test_id, self.test_timestamp, "sleep 10")
self.memory.add_entry(entry) self.memory.add_entry(entry)
# Wait for process to start # Wait for process to start
@@ -58,7 +56,7 @@ class DeleteCommandTest(unittest.TestCase):
def test_delete_existing_entry(self): def test_delete_existing_entry(self):
"""Test deleting an existing entry""" """Test deleting an existing entry"""
# Add test entry # Add test entry
entry = ReasoningEntry("test content", self.test_id, self.test_timestamp) entry = ReasoningEntry(self.test_id, self.test_timestamp, "test content")
self.memory.add_entry(entry) self.memory.add_entry(entry)
# Execute delete command # Execute delete command

26
test/parse_error_entry_test.py
View File

@@ -13,31 +13,35 @@ class ParseErrorEntryTest(unittest.TestCase):
"""Test entry initialization""" """Test entry initialization"""
content = "invalid content" content = "invalid content"
error = "parsing failed" error = "parsing failed"
entry = ParseErrorEntry(content, error, self.test_id, self.test_timestamp) entry = ParseErrorEntry(self.test_id, self.test_timestamp, content, error)
self.assertEqual(entry.id, self.test_id) self.assertEqual(entry.id, self.test_id)
self.assertEqual(entry.timestamp, self.test_timestamp) self.assertEqual(entry.timestamp, self.test_timestamp)
self.assertEqual(entry.content, content)
self.assertEqual(entry.error, error)
def test_update_does_nothing(self): def test_update_does_nothing(self):
"""Test that update operation has no effect""" """Test that update operation has no effect"""
content = "invalid content" content = "invalid content"
error = "parsing failed" error = "parsing failed"
entry = ParseErrorEntry(content, error, self.test_id, self.test_timestamp) entry = ParseErrorEntry(self.test_id, self.test_timestamp, content, error)
# Store initial state # Store initial state
initial_context = entry.content initial_content = entry.content
initial_error = entry.error
# Perform update # Perform update
entry.update() entry.update()
# Verify state hasn't changed # Verify state hasn't changed
self.assertEqual(entry.content, initial_context) self.assertEqual(entry.content, initial_content)
self.assertEqual(entry.error, initial_error)
def test_generate_context(self): def test_generate_context(self):
"""Test XML context generation""" """Test XML context generation"""
content = "invalid content" content = "invalid content"
error = "parsing failed" error = "parsing failed"
entry = ParseErrorEntry(content, error, self.test_id, self.test_timestamp) entry = ParseErrorEntry(self.test_id, self.test_timestamp, content, error)
element = entry.generate_context() element = entry.generate_context()
@@ -48,26 +52,26 @@ 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"{error}") self.assertEqual(error_elem.text, 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"{content}") self.assertEqual(content_elem.text, 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"""
content = "Special content: \n\t\r\'\"\\" content = "Special content: \n\t\r\'\"\\"
error = "Special error: \n\t\r\'\"\\" error = "Special error: \n\t\r\'\"\\"
entry = ParseErrorEntry(content, error, self.test_id, self.test_timestamp) entry = ParseErrorEntry(self.test_id, self.test_timestamp, content, error)
element = entry.generate_context() element = entry.generate_context()
self.assertEqual(element.find("content").text, f"{content}") self.assertEqual(element.find("content").text, content)
self.assertEqual(element.find("error").text, f"{error}") self.assertEqual(element.find("error").text, 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, "") self.assertEqual(element.find("content").text, "")

20
test/read_entry_test.py
View File

@@ -13,7 +13,7 @@ class ReadEntryTest(unittest.TestCase):
def test_initialization(self): def test_initialization(self):
"""Test entry initialization""" """Test entry initialization"""
entry = ReadEntry(self.io_buffer, self.test_id, self.test_timestamp) entry = ReadEntry(self.test_id, self.test_timestamp, self.io_buffer)
self.assertEqual(entry.content, "") self.assertEqual(entry.content, "")
self.assertEqual(entry.id, self.test_id) self.assertEqual(entry.id, self.test_id)
@@ -24,7 +24,7 @@ class ReadEntryTest(unittest.TestCase):
test_input = "test message" test_input = "test message"
self.io_buffer.append_stdin(test_input) self.io_buffer.append_stdin(test_input)
entry = ReadEntry(self.io_buffer, self.test_id, self.test_timestamp) entry = ReadEntry(self.test_id, self.test_timestamp, self.io_buffer)
entry.update() entry.update()
self.assertEqual(entry.content, test_input) self.assertEqual(entry.content, test_input)
@@ -35,7 +35,7 @@ class ReadEntryTest(unittest.TestCase):
test_input = "initial input" test_input = "initial input"
self.io_buffer.append_stdin(test_input) self.io_buffer.append_stdin(test_input)
entry = ReadEntry(self.io_buffer, self.test_id, self.test_timestamp) entry = ReadEntry(self.test_id, self.test_timestamp, self.io_buffer)
entry.update() entry.update()
initial_content = entry.content initial_content = entry.content
@@ -47,7 +47,7 @@ class ReadEntryTest(unittest.TestCase):
def test_empty_input(self): def test_empty_input(self):
"""Test reading when no input is available""" """Test reading when no input is available"""
entry = ReadEntry(self.io_buffer, self.test_id, self.test_timestamp) entry = ReadEntry(self.test_id, self.test_timestamp, self.io_buffer)
entry.update() entry.update()
self.assertEqual(entry.content, "") self.assertEqual(entry.content, "")
@@ -57,7 +57,7 @@ class ReadEntryTest(unittest.TestCase):
test_input = "Special chars: \n\t\r\'\"\\" test_input = "Special chars: \n\t\r\'\"\\"
self.io_buffer.append_stdin(test_input) self.io_buffer.append_stdin(test_input)
entry = ReadEntry(self.io_buffer, self.test_id, self.test_timestamp) entry = ReadEntry(self.test_id, self.test_timestamp, self.io_buffer)
entry.update() entry.update()
self.assertEqual(entry.content, test_input) self.assertEqual(entry.content, test_input)
@@ -67,14 +67,14 @@ class ReadEntryTest(unittest.TestCase):
test_input = "x" * 10000 test_input = "x" * 10000
self.io_buffer.append_stdin(test_input) self.io_buffer.append_stdin(test_input)
entry = ReadEntry(self.io_buffer, self.test_id, self.test_timestamp) entry = ReadEntry(self.test_id, self.test_timestamp, self.io_buffer)
entry.update() entry.update()
self.assertEqual(entry.content, test_input) self.assertEqual(entry.content, test_input)
def test_generate_context_before_read(self): def test_generate_context_before_read(self):
"""Test XML context generation before reading""" """Test XML context generation before reading"""
entry = ReadEntry(self.io_buffer, self.test_id, self.test_timestamp) entry = ReadEntry(self.test_id, self.test_timestamp, self.io_buffer)
element = entry.generate_context() element = entry.generate_context()
# Verify XML structure # Verify XML structure
@@ -87,7 +87,7 @@ class ReadEntryTest(unittest.TestCase):
test_input = "test message" test_input = "test message"
self.io_buffer.append_stdin(test_input) self.io_buffer.append_stdin(test_input)
entry = ReadEntry(self.io_buffer, self.test_id, self.test_timestamp) entry = ReadEntry(self.test_id, self.test_timestamp, self.io_buffer)
entry.update() entry.update()
element = entry.generate_context() element = entry.generate_context()
@@ -101,7 +101,7 @@ class ReadEntryTest(unittest.TestCase):
test_input = "Hello 世界 😊" test_input = "Hello 世界 😊"
self.io_buffer.append_stdin(test_input) self.io_buffer.append_stdin(test_input)
entry = ReadEntry(self.io_buffer, self.test_id, self.test_timestamp) entry = ReadEntry(self.test_id, self.test_timestamp, self.io_buffer)
entry.update() entry.update()
self.assertEqual(entry.content, test_input) self.assertEqual(entry.content, test_input)
@@ -113,7 +113,7 @@ class ReadEntryTest(unittest.TestCase):
Line 3""" Line 3"""
self.io_buffer.append_stdin(test_input) self.io_buffer.append_stdin(test_input)
entry = ReadEntry(self.io_buffer, self.test_id, self.test_timestamp) entry = ReadEntry(self.test_id, self.test_timestamp, self.io_buffer)
entry.update() entry.update()
self.assertEqual(entry.content, test_input) self.assertEqual(entry.content, test_input)

14
test/reasoning_entry_test.py
View File

@@ -12,7 +12,7 @@ class ReasoningEntryTest(unittest.TestCase):
def test_initialization(self): def test_initialization(self):
"""Test entry initialization""" """Test entry initialization"""
content = "test reasoning" content = "test reasoning"
entry = ReasoningEntry(content, self.test_id, self.test_timestamp) entry = ReasoningEntry(self.test_id, self.test_timestamp, content)
self.assertEqual(entry.content, content) self.assertEqual(entry.content, content)
self.assertEqual(entry.id, self.test_id) self.assertEqual(entry.id, self.test_id)
@@ -21,7 +21,7 @@ class ReasoningEntryTest(unittest.TestCase):
def test_update_does_nothing(self): def test_update_does_nothing(self):
"""Test that update operation has no effect""" """Test that update operation has no effect"""
content = "test reasoning" content = "test reasoning"
entry = ReasoningEntry(content, self.test_id, self.test_timestamp) entry = ReasoningEntry(self.test_id, self.test_timestamp, content)
# Store initial state # Store initial state
initial_content = entry.generate_context().text initial_content = entry.generate_context().text
@@ -35,7 +35,7 @@ class ReasoningEntryTest(unittest.TestCase):
def test_generate_context(self): def test_generate_context(self):
"""Test XML context generation""" """Test XML context generation"""
content = "test reasoning" content = "test reasoning"
entry = ReasoningEntry(content, self.test_id, self.test_timestamp) entry = ReasoningEntry(self.test_id, self.test_timestamp, content)
element = entry.generate_context() element = entry.generate_context()
@@ -47,14 +47,14 @@ class ReasoningEntryTest(unittest.TestCase):
def test_special_characters(self): def test_special_characters(self):
"""Test handling reasoning text with special characters""" """Test handling reasoning text with special characters"""
content = "Special reasoning: \n\t\r\'\"\\" content = "Special reasoning: \n\t\r\'\"\\"
entry = ReasoningEntry(content, self.test_id, self.test_timestamp) entry = ReasoningEntry(self.test_id, self.test_timestamp, content)
element = entry.generate_context() element = entry.generate_context()
self.assertEqual(element.text, f"{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, "") self.assertEqual(element.text, "")
@@ -62,7 +62,7 @@ class ReasoningEntryTest(unittest.TestCase):
def test_large_content(self): def test_large_content(self):
"""Test handling large reasoning text""" """Test handling large reasoning text"""
content = "x" * 10000 content = "x" * 10000
entry = ReasoningEntry(content, self.test_id, self.test_timestamp) entry = ReasoningEntry(self.test_id, self.test_timestamp, content)
element = entry.generate_context() element = entry.generate_context()
self.assertEqual(element.text, f"{content}") self.assertEqual(element.text, f"{content}")
@@ -72,7 +72,7 @@ class ReasoningEntryTest(unittest.TestCase):
content = """Line 1 content = """Line 1
Line 2 Line 2
Line 3""" Line 3"""
entry = ReasoningEntry(content, self.test_id, self.test_timestamp) entry = ReasoningEntry(self.test_id, self.test_timestamp, content)
element = entry.generate_context() element = entry.generate_context()
self.assertEqual(element.text, f"{content}") self.assertEqual(element.text, f"{content}")

14
test/repeat_entry_test.py
View File

@@ -28,7 +28,7 @@ class RepeatEntryTest(unittest.TestCase):
def test_initialization(self): def test_initialization(self):
"""Test entry initialization""" """Test entry initialization"""
script = "echo test" script = "echo test"
entry = RepeatEntry(script, self.test_id, self.test_timestamp) entry = RepeatEntry(self.test_id, self.test_timestamp, script, None, None)
self.assertEqual(entry.script, script) self.assertEqual(entry.script, script)
self.assertEqual(entry.stdout, "") self.assertEqual(entry.stdout, "")
@@ -40,7 +40,7 @@ class RepeatEntryTest(unittest.TestCase):
def test_successful_execution(self): def test_successful_execution(self):
"""Test successful script execution""" """Test successful script execution"""
script = "echo 'test'" script = "echo 'test'"
entry = RepeatEntry(script, self.test_id, self.test_timestamp) entry = RepeatEntry(self.test_id, self.test_timestamp, script, None, None)
entry.update() entry.update()
# Verify entry state # Verify entry state
@@ -50,7 +50,7 @@ class RepeatEntryTest(unittest.TestCase):
def test_failed_execution(self): def test_failed_execution(self):
"""Test failed script execution with invalid command""" """Test failed script execution with invalid command"""
entry = RepeatEntry("nonexistentcommand", self.test_id, self.test_timestamp) entry = RepeatEntry(self.test_id, self.test_timestamp, "nonexistentcommand", None, None)
entry.update() entry.update()
# Verify entry state # Verify entry state
@@ -78,7 +78,7 @@ class RepeatEntryTest(unittest.TestCase):
f'"' f'"'
) )
entry = RepeatEntry(script, self.test_id, self.test_timestamp) entry = RepeatEntry(self.test_id, self.test_timestamp, script, None, None)
# Run update multiple times # Run update multiple times
outputs = [] outputs = []
@@ -92,13 +92,13 @@ class RepeatEntryTest(unittest.TestCase):
def test_generate_context(self): def test_generate_context(self):
"""Test XML context generation""" """Test XML context generation"""
script = "echo 'test'" script = "echo 'test'"
entry = RepeatEntry(script, self.test_id, self.test_timestamp) entry = RepeatEntry(self.test_id, self.test_timestamp, script, None, None)
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"{script}") self.assertEqual(element.text, script)
self.assertEqual(element.get("exit_code"), "0") self.assertEqual(element.get("exit_code"), "0")
stdout_text = element.find("stdout").text stdout_text = element.find("stdout").text
self.assertEqual(stdout_text, "test\n") self.assertEqual(stdout_text, "test\n")
@@ -124,7 +124,7 @@ class RepeatEntryTest(unittest.TestCase):
f'"' f'"'
) )
entry = RepeatEntry(script, self.test_id, self.test_timestamp) entry = RepeatEntry(self.test_id, self.test_timestamp, script, None, None)
# Execute multiple times and verify output changes # Execute multiple times and verify output changes
entry.update() entry.update()

16
test/single_entry_test.py
View File

@@ -26,7 +26,7 @@ class SingleEntryTest(unittest.TestCase):
def test_initialization(self): def test_initialization(self):
"""Test entry initialization""" """Test entry initialization"""
script = "echo test" script = "echo test"
entry = SingleEntry(script, self.test_id, self.test_timestamp) entry = SingleEntry(self.test_id, self.test_timestamp, script, None, None)
self.assertEqual(entry.script, script) self.assertEqual(entry.script, script)
self.assertEqual(entry.stdout, "") self.assertEqual(entry.stdout, "")
@@ -40,7 +40,7 @@ class SingleEntryTest(unittest.TestCase):
"""Test successful script execution""" """Test successful script execution"""
script = "echo 'test'" script = "echo 'test'"
entry = SingleEntry(script, self.test_id, self.test_timestamp) entry = SingleEntry(self.test_id, self.test_timestamp, script, None, None)
entry.update() entry.update()
# Verify entry state # Verify entry state
@@ -51,7 +51,7 @@ class SingleEntryTest(unittest.TestCase):
def test_failed_execution(self): def test_failed_execution(self):
"""Test failed script execution with invalid command""" """Test failed script execution with invalid command"""
entry = SingleEntry("nonexistentcommand", self.test_id, self.test_timestamp) entry = SingleEntry(self.test_id, self.test_timestamp, "nonexistentcommand", None, None)
entry.update() entry.update()
# Verify entry state # Verify entry state
@@ -64,7 +64,7 @@ class SingleEntryTest(unittest.TestCase):
"""Test script that creates a file""" """Test script that creates a file"""
script = f"echo 'test' > {self.test_filename}" script = f"echo 'test' > {self.test_filename}"
entry = SingleEntry(script, self.test_id, self.test_timestamp) entry = SingleEntry(self.test_id, self.test_timestamp, script, None, None)
entry.update() entry.update()
# Verify file was created and contains expected content # Verify file was created and contains expected content
@@ -77,7 +77,7 @@ class SingleEntryTest(unittest.TestCase):
"""Test that script only executes once""" """Test that script only executes once"""
script = f"echo 'test' >> {self.test_filename}" script = f"echo 'test' >> {self.test_filename}"
entry = SingleEntry(script, self.test_id, self.test_timestamp) entry = SingleEntry(self.test_id, self.test_timestamp, script, None, None)
# Run update multiple times # Run update multiple times
entry.update() entry.update()
@@ -92,7 +92,7 @@ class SingleEntryTest(unittest.TestCase):
def test_generate_context_before_execution(self): def test_generate_context_before_execution(self):
"""Test XML context generation before execution""" """Test XML context generation before execution"""
entry = SingleEntry("echo test", self.test_id, self.test_timestamp) entry = SingleEntry(self.test_id, self.test_timestamp, "echo test", None, None)
element = entry.generate_context() element = entry.generate_context()
# Verify XML structure # Verify XML structure
@@ -109,7 +109,7 @@ class SingleEntryTest(unittest.TestCase):
"""Test XML context generation after execution""" """Test XML context generation after execution"""
script = "echo 'test'" script = "echo 'test'"
entry = SingleEntry(script, self.test_id, self.test_timestamp) entry = SingleEntry(self.test_id, self.test_timestamp, script, None, None)
entry.update() entry.update()
element = entry.generate_context() element = entry.generate_context()
@@ -128,7 +128,7 @@ class SingleEntryTest(unittest.TestCase):
"""Test executing multiple commands in one script""" """Test executing multiple commands in one script"""
script = f"echo 'first' > {self.test_filename} && echo 'second' >> {self.test_filename}" script = f"echo 'first' > {self.test_filename} && echo 'second' >> {self.test_filename}"
entry = SingleEntry(script, self.test_id, self.test_timestamp) entry = SingleEntry(self.test_id, self.test_timestamp, script, None, None)
entry.update() entry.update()
# Verify file contains both lines # Verify file contains both lines

2
test/stop_command_test.py
View File

@@ -27,7 +27,7 @@ class StopCommandTest(unittest.TestCase):
def test_stop_command_cleanup(self): def test_stop_command_cleanup(self):
"""Test stop command cleans up and removes all entries""" """Test stop command cleans up and removes all entries"""
# Add background process # Add background process
entry = BackgroundEntry("sleep 10", "test-id", self.test_timestamp) entry = BackgroundEntry("test-id", self.test_timestamp, "sleep 10")
self.memory.add_entry(entry) self.memory.add_entry(entry)
# Wait for process to start and get PID # Wait for process to start and get PID

18
test/working_memory_test.py
View File

@@ -123,10 +123,10 @@ class WorkingMemoryTest(unittest.TestCase):
# Add different types of entries # Add different types of entries
entries = [ entries = [
ReasoningEntry("test reasoning", "reasoning-1", self.test_timestamp), ReasoningEntry("reasoning-1", self.test_timestamp, "test reasoning"),
ParseErrorEntry("bad content", "error msg", "error-1", self.test_timestamp), ParseErrorEntry("error-1", self.test_timestamp, "bad content", "error msg"),
ReadEntry(io_buffer, "read-1", self.test_timestamp), ReadEntry("read-1", self.test_timestamp, io_buffer),
WriteEntry("test output", io_buffer, "write-1", self.test_timestamp) WriteEntry("write-1", self.test_timestamp, "test output", io_buffer)
] ]
for entry in entries: for entry in entries:
@@ -161,7 +161,7 @@ class WorkingMemoryTest(unittest.TestCase):
def test_remove_entry_cleanup(self): def test_remove_entry_cleanup(self):
"""Test that removing an entry triggers cleanup""" """Test that removing an entry triggers cleanup"""
# Create and start background process # Create and start background process
entry = BackgroundEntry("sleep 10", "test-id", self.test_timestamp) entry = BackgroundEntry("test-id", self.test_timestamp, "sleep 10")
self.memory.add_entry(entry) self.memory.add_entry(entry)
# Wait for process to start and get PID # Wait for process to start and get PID
@@ -181,7 +181,7 @@ class WorkingMemoryTest(unittest.TestCase):
# Add multiple background processes # Add multiple background processes
pids = [] pids = []
for i in range(3): for i in range(3):
entry = BackgroundEntry("sleep 10", f"id-{i}", self.test_timestamp) entry = BackgroundEntry(f"id-{i}", self.test_timestamp, "sleep 10")
self.memory.add_entry(entry) self.memory.add_entry(entry)
self.assertTrue(self.wait_for_process_start(entry)) self.assertTrue(self.wait_for_process_start(entry))
context = entry.generate_context() context = entry.generate_context()
@@ -201,7 +201,7 @@ class WorkingMemoryTest(unittest.TestCase):
def test_cleanup_on_memory_deletion(self): def test_cleanup_on_memory_deletion(self):
"""Test that deleting memory properly cleans up all entries""" """Test that deleting memory properly cleans up all entries"""
# Add a background process # Add a background process
entry = BackgroundEntry("sleep 10", "test-id", self.test_timestamp) entry = BackgroundEntry("test-id", self.test_timestamp, "sleep 10")
self.memory.add_entry(entry) self.memory.add_entry(entry)
# Wait for process to start and get PID # Wait for process to start and get PID
@@ -221,7 +221,7 @@ class WorkingMemoryTest(unittest.TestCase):
"""Test getting all entries""" """Test getting all entries"""
# Add multiple entries # Add multiple entries
entries = [ entries = [
ReasoningEntry(f"content {i}", f"id-{i}", self.test_timestamp) ReasoningEntry(f"id-{i}", self.test_timestamp, f"content {i}")
for i in range(3) for i in range(3)
] ]
@@ -239,7 +239,7 @@ class WorkingMemoryTest(unittest.TestCase):
def test_get_entries_returns_copy(self): def test_get_entries_returns_copy(self):
"""Test that get_entries returns a copy of the list""" """Test that get_entries returns a copy of the list"""
# Add an entry # Add an entry
entry = ReasoningEntry("test content", "test-id", self.test_timestamp) entry = ReasoningEntry("test-id", self.test_timestamp, "test content")
self.memory.add_entry(entry) self.memory.add_entry(entry)
# Get entries and modify the returned list # Get entries and modify the returned list

18
test/write_entry_test.py
View File

@@ -14,7 +14,7 @@ class WriteEntryTest(unittest.TestCase):
def test_initialization(self): def test_initialization(self):
"""Test entry initialization""" """Test entry initialization"""
content = "test message" content = "test message"
entry = WriteEntry(content, self.io_buffer, self.test_id, self.test_timestamp) entry = WriteEntry(self.test_id, self.test_timestamp, content, self.io_buffer)
self.assertEqual(entry.content, content) self.assertEqual(entry.content, content)
self.assertEqual(entry.id, self.test_id) self.assertEqual(entry.id, self.test_id)
@@ -25,7 +25,7 @@ class WriteEntryTest(unittest.TestCase):
def test_single_write(self): def test_single_write(self):
"""Test writing content once""" """Test writing content once"""
content = "test message" content = "test message"
entry = WriteEntry(content, self.io_buffer, self.test_id, self.test_timestamp) entry = WriteEntry(self.test_id, self.test_timestamp, content, self.io_buffer)
# Perform update and verify content was written # Perform update and verify content was written
entry.update() entry.update()
@@ -35,7 +35,7 @@ class WriteEntryTest(unittest.TestCase):
def test_multiple_updates(self): def test_multiple_updates(self):
"""Test that content is only written once even with multiple updates""" """Test that content is only written once even with multiple updates"""
content = "test message" content = "test message"
entry = WriteEntry(content, self.io_buffer, self.test_id, self.test_timestamp) entry = WriteEntry(self.test_id, self.test_timestamp, content, self.io_buffer)
# Perform multiple updates # Perform multiple updates
entry.update() entry.update()
@@ -52,7 +52,7 @@ class WriteEntryTest(unittest.TestCase):
def test_empty_content(self): def test_empty_content(self):
"""Test writing empty content""" """Test writing empty content"""
entry = WriteEntry("", self.io_buffer, self.test_id, self.test_timestamp) entry = WriteEntry(self.test_id, self.test_timestamp, "", self.io_buffer)
entry.update() entry.update()
self.assertEqual(self.io_buffer.get_stdout(), "") self.assertEqual(self.io_buffer.get_stdout(), "")
@@ -61,7 +61,7 @@ class WriteEntryTest(unittest.TestCase):
def test_special_characters(self): def test_special_characters(self):
"""Test writing content with special characters""" """Test writing content with special characters"""
content = "Special chars: \n\t\r\'\"\\" content = "Special chars: \n\t\r\'\"\\"
entry = WriteEntry(content, self.io_buffer, self.test_id, self.test_timestamp) entry = WriteEntry(self.test_id, self.test_timestamp, content, self.io_buffer)
entry.update() entry.update()
self.assertEqual(self.io_buffer.get_stdout(), content) self.assertEqual(self.io_buffer.get_stdout(), content)
@@ -69,7 +69,7 @@ class WriteEntryTest(unittest.TestCase):
def test_large_content(self): def test_large_content(self):
"""Test writing large content""" """Test writing large content"""
content = "x" * 10000 content = "x" * 10000
entry = WriteEntry(content, self.io_buffer, self.test_id, self.test_timestamp) entry = WriteEntry(self.test_id, self.test_timestamp, content, self.io_buffer)
entry.update() entry.update()
self.assertEqual(self.io_buffer.get_stdout(), content) self.assertEqual(self.io_buffer.get_stdout(), content)
@@ -77,7 +77,7 @@ class WriteEntryTest(unittest.TestCase):
def test_generate_context(self): def test_generate_context(self):
"""Test XML context generation""" """Test XML context generation"""
content = "test message" content = "test message"
entry = WriteEntry(content, self.io_buffer, self.test_id, self.test_timestamp) entry = WriteEntry(self.test_id, self.test_timestamp, content, self.io_buffer)
# Generate context before writing # Generate context before writing
element = entry.generate_context() element = entry.generate_context()
@@ -97,7 +97,7 @@ class WriteEntryTest(unittest.TestCase):
def test_unicode_content(self): def test_unicode_content(self):
"""Test writing Unicode content""" """Test writing Unicode content"""
content = "Hello 世界 😊" content = "Hello 世界 😊"
entry = WriteEntry(content, self.io_buffer, self.test_id, self.test_timestamp) entry = WriteEntry(self.test_id, self.test_timestamp, content, self.io_buffer)
entry.update() entry.update()
self.assertEqual(self.io_buffer.get_stdout(), content) self.assertEqual(self.io_buffer.get_stdout(), content)
@@ -107,7 +107,7 @@ class WriteEntryTest(unittest.TestCase):
content = """Line 1 content = """Line 1
Line 2 Line 2
Line 3""" Line 3"""
entry = WriteEntry(content, self.io_buffer, self.test_id, self.test_timestamp) entry = WriteEntry(self.test_id, self.test_timestamp, content, self.io_buffer)
entry.update() entry.update()
self.assertEqual(self.io_buffer.get_stdout(), content) self.assertEqual(self.io_buffer.get_stdout(), content)