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

start implementation on new architecture

Browse Source
This commit is contained in:
Geens
2024-11-01 09:56:30 +01:00
parent ee089e5be7
commit 2c1e134c6e
43 changed files with 2978 additions and 619 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
Dockerfile
View File

@@ -1,12 +1,4 @@
FROM huggingface/transformers-pytorch-gpu AS requirements
RUN apt update
RUN apt install -y \
apt-transport-https ca-certificates \
curl software-properties-common
RUN curl -fsSL https://download.docker.com/linux/ubuntu/gpg | apt-key add -
RUN add-apt-repository "deb [arch=amd64] https://download.docker.com/linux/ubuntu bionic stable"
RUN apt update
RUN apt install -y docker-ce
COPY requirements.txt /requirements.txt
RUN pip3 install -r /requirements.txt
RUN rm -rf /requirements.txt

53
readme.md
View File

@@ -318,7 +318,7 @@ stateDiagram-v2
```mermaid
classDiagram
direction LR
direction LR
class BaseAgent {
<<abstract>>
#working_memory WorkingMemory
@@ -331,8 +331,8 @@ classDiagram
}
class LLMEngine {
+LLMEngine(path str)
+set_model_path(path str)
+LLMEngine(model_path str)
+set_model_path(model_path str)
+inference(context str) Iterator~str~
}
@@ -348,14 +348,15 @@ classDiagram
}
class WebAgentState {
<<enumeration>>
UPDATE
CONTEXT_APPROVAL
INFERENCE
RESPONSE_APPROVAL
<<enumeration>>
UPDATE
CONTEXT_APPROVAL
INFERENCE
RESPONSE_APPROVAL
}
class WorkingMemory {
+WorkingMemory()
-entries List~Entry~
+add_entry(entry Entry)
+remove_entry(id str)
@@ -369,12 +370,13 @@ classDiagram
class XMLValidator {
-schema ElementTree
+XMLValidator(schema_path str)
+XMLValidator(schema str)
+validate(xml str) Optional~str~
}
class ResponseParser {
-io_buffer IOBuffer
+ResponseParser(io_buffer IOBuffer)
+parse(xml str) Command | Entry
}
@@ -382,6 +384,7 @@ classDiagram
<<abstract>>
+id str
+timestamp datetime
+Entry(id str, timestamp datetime)
+update()*
+generate_context() ElementTree*
}
@@ -391,6 +394,7 @@ classDiagram
+stdout str
+stderr str
+exit_code int
+SingleShotEntry(script str, id str, timestamp datetime)
+update()
+generate_context() ElementTree
}
@@ -400,6 +404,7 @@ classDiagram
+stdout str
+stderr str
+exit_code int
+RepeatEntry(script str, id str, timestamp datetime)
+update()
+generate_context() ElementTree
}
@@ -409,12 +414,14 @@ classDiagram
+stdout str
+stderr str
+process Process
+BackgroundEntry(script str, id str, timestamp datetime)
+update()
+generate_context() ElementTree
}
class ReasoningEntry {
+content str
+ReasoningEntry(content str, id str, timestamp datetime)
+update()
+generate_context() ElementTree
}
@@ -422,18 +429,21 @@ classDiagram
class ParseErrorEntry {
+content str
+error str
+ParseErrorEntry(content str, error str, id str, timestamp datetime)
+update()
+generate_context() ElementTree
}
class ReadEntry {
+content str
+ReadEntry(io_buffer IOBuffer, id str, timestamp datetime)
+update()
+generate_context() ElementTree
}
class WriteEntry {
+content str
+WriteEntry(content str, io_buffer IOBuffer, id str, timestamp datetime)
+update()
+generate_context() ElementTree
}
@@ -446,6 +456,7 @@ classDiagram
}
class StandardIOBuffer {
+StandardIOBuffer()
+read() str
+write(content str)
+buffer_length() int
@@ -454,6 +465,7 @@ classDiagram
class WebIOBuffer {
-stdin_buffer str
-stdout_buffer str
+WebIOBuffer()
+read() str
+write(content str)
+buffer_length() int
@@ -462,23 +474,35 @@ classDiagram
+clear_stdout()
}
class CommandResult {
+should_stop bool
+success bool
+message str
+CommandResult(should_stop bool, success bool, message str)
+static success() CommandResult
+static failure(message str) CommandResult
+static stop() CommandResult
}
class Command {
<<abstract>>
+execute(memory &WorkingMemory)*
+execute(memory &WorkingMemory) CommandResult*
}
class DeleteCommand {
+id: str
+execute(memory &WorkingMemory)
+id str
+DeleteCommand(id str)
+execute(memory &WorkingMemory) CommandResult
}
class StopCommand {
+execute(memory &WorkingMemory)
+execute(memory &WorkingMemory) CommandResult
}
class WebServer {
-agent WebAgent
-clients List~WebSocket~
+WebServer(agent WebAgent)
-broadcast_state_change()
}
@@ -491,7 +515,6 @@ classDiagram
BaseAgent "1" *-- "1" LLMEngine
BaseAgent "1" *-- "1" ResponseParser
WebServer "1" *-- "1" WebIOBuffer
WebServer "1" *-- "1" WebAgent
@@ -509,7 +532,7 @@ classDiagram
Command <|-- DeleteCommand
Command <|-- StopCommand
Command -- CommandResult
IOBuffer <|.. WebIOBuffer
IOBuffer <|.. StandardIOBuffer

1
requirements.txt
View File

@@ -1,4 +1,3 @@
bs4
docker
torch
transformers

10
sia/__main__.py
View File

@@ -1,6 +1,4 @@
from .agent_core import AgentCore
from .llm_engine import LlmEngine
from .docker_module import DockerModule
def main():
"""Main entry point for the SIA application."""
@@ -13,13 +11,7 @@ def main():
with open(action_schema_path, 'r') as f:
action_schema = f.read()
agent_core = AgentCore(
system_prompt=f"{system_prompt}{action_schema}",
action_schema=action_schema,
docker_module=DockerModule(),
llm_engine=LlmEngine(model_path=model_path)
)
agent_core.run_iteration()
print("todo")
if __name__ == "__main__":
main()

51
sia/agent_core.py
View File

@@ -1,51 +0,0 @@
from itertools import tee
from typing import Optional, Dict, List
from .docker_module import DockerModule
from .llm_engine import LlmEngine
from .context_template import generate_context
from .util import get_valid_root_elements, split_response
class AgentCore:
"""
Core orchestration class for SIA that manages the interaction between
different modules and runs the main agent loop.
"""
def __init__(
self,
system_prompt: str,
action_schema: str,
docker_module: DockerModule,
llm_engine: LlmEngine
):
"""
Initialize the AgentCore with required components.
Args:
system_prompt: System prompt to use for the LLM
action_schema_path: Path to the XML schema defining valid actions
docker_module: DockerModule instance
llm_engine: LLmEngine instance
"""
self.system_prompt = system_prompt
self.action_schema = action_schema
self.docker_module = docker_module
self.llm_engine = llm_engine
self.valid_elements = get_valid_root_elements(self.action_schema)
def run_iteration(self) -> None:
"""Run a single iteration of the main agent loop."""
containers = self.docker_module.get_all_container_statuses()
context = generate_context(containers)
tokens = self.llm_engine.infer(
self.system_prompt,
context
)
print_tokens, response_tokens = tee(tokens)
for token in print_tokens:
print(token, end="", flush=True)
response_tokens = ''.join(response_tokens)
result = split_response(response_tokens, self.valid_elements)
print(f"result: {result}")

175
sia/background_entry.py Normal file
View File

@@ -0,0 +1,175 @@
from datetime import datetime
import subprocess
import xml.etree.ElementTree as ET
from typing import Optional
from .entry import Entry
from .util import escape_text_for_xml
class BackgroundEntry(Entry):
"""
Entry type for long-running background processes.
Attributes:
script: The script/command to execute
stdout: Captured standard output since last update
stderr: Captured standard error since last update
process: The running subprocess.Popen instance
_accumulated_stdout: Total stdout collected so far
_accumulated_stderr: Total stderr collected so far
_exit_code: Exit code when process completes
"""
def __init__(self, script: str, id: str, timestamp: datetime):
"""
Initialize a new background entry.
Args:
script: The script/command to execute
id: Unique identifier for this entry
timestamp: Creation timestamp for this entry
"""
super().__init__(id, timestamp)
self.script = script
self.stdout = ""
self.stderr = ""
self.process: Optional[subprocess.Popen] = None
self._accumulated_stdout = ""
self._accumulated_stderr = ""
self._exit_code: Optional[int] = None
def cleanup(self) -> None:
"""
Clean up the background process if it's still running.
Ensures process is terminated and file handles are closed.
"""
self._cleanup_process()
def _cleanup_process(self):
"""Clean up process and file handles"""
if self.process is not None:
try:
# Close file handles if they're open
if self.process.stdout:
self.process.stdout.close()
if self.process.stderr:
self.process.stderr.close()
# Terminate process if it's still running
if self.process.poll() is None:
self.process.terminate()
try:
self.process.wait(timeout=1.0)
except subprocess.TimeoutExpired:
self.process.kill()
self.process.wait()
except:
pass # Ignore cleanup errors
finally:
self.process = None
def update(self) -> None:
"""
Start the process if not running and collect any new output.
Updates stdout and stderr with any new output since last update.
"""
try:
# Reset current output buffers
self.stdout = ""
self.stderr = ""
# Start process if not running
if self.process is None and self._exit_code is None:
self.process = subprocess.Popen(
self.script,
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1 # Line buffered
)
return # Return after starting to allow process to generate output
if self.process is None:
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
self._accumulated_stdout += remaining_out
self._accumulated_stderr += remaining_err
except subprocess.TimeoutExpired:
pass # Process didn't finish communicating, try again next update
self._exit_code = exit_code
self._cleanup_process()
return
# Read stdout if available
if self.process.stdout:
while True:
try:
line = self.process.stdout.readline()
if not line:
break
self.stdout += line
self._accumulated_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
self._accumulated_stderr += line
except:
break
except Exception as e:
# Handle any errors
error_msg = f"Error handling background process: {str(e)}"
self.stderr = error_msg
self._accumulated_stderr += f"\n{error_msg}"
self._exit_code = -1
self._cleanup_process()
def generate_context(self) -> ET.Element:
"""
Generate an XML Element representing this background entry.
Returns:
ET.Element: XML element containing the entry's data
"""
# Create root element with appropriate status
element = ET.Element("background", {"id": self.id})
if self.process is not None:
element.set("pid", str(self.process.pid))
elif self._exit_code is not None:
element.set("exit_code", str(self._exit_code))
# 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.text = escape_text_for_xml(self._accumulated_stdout)
# Add stderr element with accumulated output
stderr_elem = ET.SubElement(element, "stderr")
stderr_elem.text = escape_text_for_xml(self._accumulated_stderr)
return element
def __del__(self):
"""Ensure process is terminated and file handles are closed when entry is destroyed"""
self._cleanup_process()

25
sia/command.py Normal file
View File

@@ -0,0 +1,25 @@
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .working_memory import WorkingMemory
from .command_result import CommandResult
class Command(ABC):
"""
Abstract base class for all commands that can be executed on working memory.
Commands represent immediate actions that modify the system state.
"""
@abstractmethod
def execute(self, memory: 'WorkingMemory') -> 'CommandResult':
"""
Execute the command on the given working memory.
Args:
memory: WorkingMemory instance to execute command on
Returns:
CommandResult: Result of command execution
"""
pass

35
sia/command_result.py Normal file
View File

@@ -0,0 +1,35 @@
from dataclasses import dataclass
@dataclass
class CommandResult:
"""
Result of a command execution.
Attributes:
message: Optional message describing the result
success: Whether the command executed successfully
should_stop: Whether the agent should stop after this command
"""
message: str
success: bool = True
should_stop: bool = False
@staticmethod
def success() -> 'CommandResult':
"""Create a successful command result."""
return CommandResult(message="", success=True, should_stop=False)
@staticmethod
def failure(message: str) -> 'CommandResult':
"""
Create a failed command result.
Args:
message: Description of the failure
"""
return CommandResult(message=message, success=False, should_stop=False)
@staticmethod
def stop() -> 'CommandResult':
"""Create a command result indicating the agent should stop."""
return CommandResult(message="", success=True, should_stop=True)

40
sia/context_template.py
View File

@@ -1,40 +0,0 @@
from bs4 import BeautifulSoup
from typing import List, Dict
import xml.etree.ElementTree as ET
def generate_context(containers: List[Dict[str, any]]) -> str:
"""Generate full context XML string.
Args:
containers: List of container status dictionaries
Returns:
Formatted context XML string
"""
context_elem = ET.Element("context")
context_elem.append(format_containers_xml(containers))
return BeautifulSoup(ET.tostring(context_elem, encoding='unicode'), "xml").prettify()
def format_containers_xml(containers: List[Dict[str, any]]) -> ET.Element:
"""Format container statuses as XML string.
Args:
containers: List of dictionaries containing container information
Returns:
ElementTree.Element representing the containers
"""
containers_elem = ET.Element("containers")
for container in containers:
container_elem = ET.SubElement(containers_elem, "container")
container_elem.set("name", container['name'])
container_elem.set("status", container['status'])
container_elem.set("started_at", container['started_at'])
container_elem.set("stdout", str(container['stdout_size']))
container_elem.set("stderr", str(container['stderr_size']))
if 'exit_code' in container.keys():
container_elem.set("exit_code", str(container['exit_code']))
if 'error' in container.keys():
container_elem.set("error", container['error'])
return containers_elem

42
sia/delete_command.py Normal file
View File

@@ -0,0 +1,42 @@
from .command import Command
from .command_result import CommandResult
from .working_memory import WorkingMemory
class DeleteCommand(Command):
"""
Command to delete an entry from working memory.
Ensures proper cleanup of entry resources before removal.
Attributes:
id: Unique identifier of entry to delete
"""
def __init__(self, id: str):
"""
Initialize delete command.
Args:
id: Unique identifier of entry to delete
"""
self.id = id
def execute(self, memory: WorkingMemory) -> CommandResult:
"""
Delete the specified entry from working memory.
Performs cleanup on the entry before removal.
Args:
memory: WorkingMemory instance to delete entry from
Returns:
CommandResult: Success if entry was found and deleted,
failure with message if entry not found
"""
entry = memory.get_entry(self.id)
if entry is None:
return CommandResult.failure(f"Entry with id '{self.id}' not found")
# Perform cleanup before removing
entry.cleanup()
memory.remove_entry(self.id)
return CommandResult.success()

271
sia/docker_module.py
View File

@@ -1,271 +0,0 @@
from docker.models.containers import Container
from typing import Dict, List, Optional, Tuple
import docker
import logging
import requests
import subprocess
import time
class DockerModule:
"""
Module for handling Docker container operations in SIA.
Manages container lifecycle, I/O operations, and status monitoring.
"""
def _start_docker_engine():
"""
Start the Docker engine
"""
subprocess.run(['/etc/init.d/docker', 'start'])
time.sleep(1)
def __init__(self):
DockerModule._start_docker_engine()
self.client = docker.DockerClient(base_url='unix://var/run/docker.sock')
self.containers: Dict[str, (Container, docker.APIClient.AttachSocket)] = {}
self.logger = logging.getLogger(__name__)
def __enter__(self):
return self
def __exit__(self, exec_type, exec_value, traceback):
del self
def __del__(self):
"""Cleanup all resources before shutdown."""
try:
# Stop and remove all containers
for (container, socket) in self.containers.values():
self._wait_and_stop(container, socket, 0)
# Close the Docker client
if hasattr(self, 'client'):
self.client.close()
except Exception as e:
self.logger.error(f"Error during cleanup: {str(e)}")
def _wait_and_stop(self, container: Container, socket, timeout: int) -> Tuple[int, str]:
"""Helper method for waiting on long- and short-lived containers."""
status_code = None
if timeout > 0:
try:
result = container.wait(timeout=timeout/1000)
status_code = result.get('StatusCode')
except requests.exceptions.ReadTimeout as e:
container.stop()
status_code = -1
except requests.exceptions.ConnectionError as e:
container.stop()
status_code = -1
else:
container.stop()
status_code = -1
try:
socket.close()
except Exception as e:
self.logger.error(f"Error closing socket: {str(e)}")
output = container.logs(stdout=True, stderr=True).decode('utf-8')
try:
container.remove(force=True)
except Exception as e:
self.logger.error(f"Error removing container: {str(e)}")
return status_code, output
def start_container(
self,
image: str,
name: Optional[str] = None,
timeout: int = -1,
command: Optional[str] = None,
arguments: Optional[List[str]] = None,
volumes: Optional[Dict[str, str]] = None,
ports: Optional[Dict[str, str]] = None,
environment: Optional[Dict[str, str]] = None,
) -> Optional[str]:
"""Start a new Docker container with the specified configuration.
Args:
image: Docker image to use
name: Unique container name for long running containers
timeout: Timeout in milliseconds for short running containers
command: Main command to run in container
arguments: List of command line arguments
volumes: Dictionary mapping host paths to container paths
ports: Dictionary mapping host ports to container ports
environment: Dictionary of environment variables and values
Returns:
For short-lived containers (with timeout): Container output
For long-running containers: None
"""
# Validate inputs
if name is None and timeout < 0:
raise ValueError("Either name or timeout must be provided")
cmd = []
if command:
cmd.append(command)
if arguments:
cmd.extend(arguments)
volume_binds = {}
if volumes:
for host_path, container_path in volumes.items():
volume_binds[host_path] = {'bind': container_path, 'mode': 'rw'}
port_bindings = {}
if ports:
for host_port, container_port in ports.items():
port_bindings[container_port] = host_port
self.client.images.pull(image)
container = self.client.containers.create(
image,
command=cmd if cmd else None,
name=name,
detach=True,
volumes=volume_binds,
ports=port_bindings,
environment=environment,
tty=False,
stdin_open=True
)
socket = container.attach_socket(params={
'stdin': 1,
'stdout': 1,
'stderr': 1,
'stream': 1,
'demux': True
})
container.start()
if timeout >= 0:
# For short-lived containers
return self._wait_and_stop(container, socket, timeout)[1]
else:
# For long-running containers
self.containers[name] = (container, socket)
def write_container_stdin(self, name: str, data: str) -> None:
"""Write data to a container's standard input.
Args:
name: Name of the target container
data: Data to write to stdin
Raises:
KeyError: If container name not found
docker.errors.APIError: If write operation fails
"""
if name not in self.containers:
raise KeyError(f"Container {name} not found")
try:
_, socket = self.containers[name]
socket._sock.send(data.encode('utf-8'))
except Exception as e:
self.logger.error(f"Error writing to container {name} stdin: {str(e)}")
raise
def read_container_stdout(self, name: str, n: int = -1) -> str:
"""Read from a container's standard output buffer.
Args:
name: Name of the container
n: Number of bytes to read; -1 means read all available
Returns:
Data read from stdout
Raises:
KeyError: If container name not found
docker.errors.APIError: If read operation fails
"""
if name not in self.containers:
raise KeyError(f"Container {name} not found")
try:
container, socket = self.containers[name]
logs = container.logs(stdout=True, stderr=False, tail=n if n > 0 else 'all')
return logs.decode('utf-8')
except Exception as e:
self.logger.error(f"Error reading container {name} stdout: {str(e)}")
raise
def read_container_stderr(self, name: str, n: int = -1) -> str:
"""Read from a container's standard error buffer.
Args:
name: Name of the container
n: Number of bytes to read; -1 means read all available
Returns:
Data read from stderr
Raises:
KeyError: If container name not found
docker.errors.APIError: If read operation fails
"""
if name not in self.containers:
raise KeyError(f"Container {name} not found")
try:
container, socket = self.containers[name]
logs = container.logs(stdout=False, stderr=True, tail=n if n > 0 else 'all')
return logs.decode('utf-8')
except Exception as e:
self.logger.error(f"Error reading container {name} stderr: {str(e)}")
raise
def wait_container(self, name: str, timeout: int) -> Tuple[int, str]:
"""Wait for a container to finish execution.
Args:
name: Name of the container to wait for
timeout: Time to wait in milliseconds
Returns:
Tuple of (exit_code, output)
"""
if name not in self.containers:
raise KeyError(f"Container {name} not found")
container, socket = self.containers.pop(name)
return self._wait_and_stop(container, socket, timeout)
def get_container_status(self, name: str) -> Dict[str, any]:
"""Get current status information for a container.
Args:
name: Name of the container
Returns:
Dictionary with container status information
Raises:
KeyError: If container name not found
docker.errors.APIError: If status check fails
"""
if name not in self.containers:
raise KeyError(f"Container {name} not found")
try:
container, socket = self.containers[name]
container.reload() # Refresh container info
return {
'name': name,
'status': container.status,
'started_at': container.attrs['State']['StartedAt'],
'exit_code': container.attrs['State']['ExitCode'],
'error': container.attrs['State']['Error'],
'stdout_size': len(container.logs(stdout=True, stderr=False)),
'stderr_size': len(container.logs(stdout=False, stderr=True))
}
except Exception as e:
self.logger.error(f"Error getting status for container {name}: {str(e)}")
raise
def get_all_container_statuses(self) -> List[Dict[str, any]]:
"""Get current status information for all containers.
Returns:
List of dictionaries with container status information
"""
statuses = []
for name, (container, socket) in self.containers.items():
statuses.append(self.get_container_status(name))
return statuses

50
sia/entry.py Normal file
View File

@@ -0,0 +1,50 @@
from abc import ABC, abstractmethod
from datetime import datetime
import xml.etree.ElementTree as ET
class Entry(ABC):
"""
Abstract base class for all entry types in the working memory.
Attributes:
id: Unique identifier for the entry
timestamp: When the entry was created
"""
def __init__(self, id: str, timestamp: datetime):
"""
Initialize a new entry with provided id and timestamp.
Args:
id: Unique identifier for this entry
timestamp: Creation timestamp for this entry
"""
self.id = id
self.timestamp = timestamp
@abstractmethod
def update(self) -> None:
"""
Update the entry's state.
Must be implemented by concrete classes.
"""
pass
@abstractmethod
def generate_context(self) -> ET.Element:
"""
Generate an XML Element representing this entry's context.
Must be implemented by concrete classes.
Returns:
ET.Element: XML element containing the entry's data
"""
pass
def cleanup(self) -> None:
"""
Clean up any resources used by this entry.
Should be overridden by classes that need cleanup.
Default implementation does nothing.
"""
pass

5
sia/inference_result.py
View File

@@ -1,5 +0,0 @@
from typing import NamedTuple
class InferenceResult(NamedTuple):
reasoning: str
actions: str

52
sia/parse_error_entry.py Normal file
View File

@@ -0,0 +1,52 @@
from datetime import datetime
import xml.etree.ElementTree as ET
from .entry import Entry
from .util import escape_text_for_xml
class ParseErrorEntry(Entry):
"""
Entry type for parse and validation errors.
Attributes:
content: The original content that failed to parse
error: The error message describing what went wrong
"""
def __init__(self, content: str, error: str, id: str, timestamp: datetime):
"""
Initialize a new parse error entry.
Args:
content: Original content that failed to parse
error: Error message describing the failure
id: Unique identifier for this entry
timestamp: Creation timestamp for this entry
"""
super().__init__(id, timestamp)
self.content = content
self.error = error
def update(self) -> None:
"""No update needed for parse error entries."""
pass
def generate_context(self) -> ET.Element:
"""
Generate an XML Element representing this parse error entry.
Returns:
ET.Element: XML element containing the entry's data
"""
# Create root element
element = ET.Element("parse_error", {"id": self.id})
# Add error subelement
error_elem = ET.SubElement(element, "error")
error_elem.text = escape_text_for_xml(self.error)
# Add content subelement
content_elem = ET.SubElement(element, "content")
content_elem.text = escape_text_for_xml(self.content)
return element

54
sia/read_entry.py Normal file
View File

@@ -0,0 +1,54 @@
from datetime import datetime
import xml.etree.ElementTree as ET
from .entry import Entry
from .util import escape_text_for_xml
class ReadEntry(Entry):
"""
Entry type for reading content from standard input.
Attributes:
content: Content read from stdin, empty until first update
io_buffer: Buffer to use for IO operations
_read: Whether content has been read
"""
def __init__(self, io_buffer, id: str, timestamp: datetime):
"""
Initialize a new read entry.
Args:
io_buffer: Buffer to use for IO operations
id: Unique identifier for this entry
timestamp: Creation timestamp for this entry
"""
super().__init__(id, timestamp)
self.content = ""
self.io_buffer = io_buffer
self._read = False
def update(self) -> None:
"""
Read from stdin if not already read.
Uses the provided IO buffer for the actual read operation.
"""
if not self._read:
self.content = self.io_buffer.read()
self._read = True
def generate_context(self) -> ET.Element:
"""
Generate an XML Element representing this read entry.
Returns:
ET.Element: XML element containing the entry's data
"""
# Create root element
element = ET.Element("read_stdin", {"id": self.id})
# Add content as CDATA or escaped text if content has been read
if self._read:
element.text = escape_text_for_xml(self.content)
return element

44
sia/reasoning_entry.py Normal file
View File

@@ -0,0 +1,44 @@
from datetime import datetime
import xml.etree.ElementTree as ET
from .entry import Entry
from .util import escape_text_for_xml
class ReasoningEntry(Entry):
"""
Entry type for agent reasoning steps.
Attributes:
content: The reasoning text
"""
def __init__(self, content: str, id: str, timestamp: datetime):
"""
Initialize a new reasoning entry.
Args:
content: The reasoning text
id: Unique identifier for this entry
timestamp: Creation timestamp for this entry
"""
super().__init__(id, timestamp)
self.content = content
def update(self) -> None:
"""No update needed for reasoning entries."""
pass
def generate_context(self) -> ET.Element:
"""
Generate an XML Element representing this reasoning entry.
Returns:
ET.Element: XML element containing the entry's data
"""
# Create root element
element = ET.Element("reasoning", {"id": self.id})
# Add content as CDATA or escaped text
element.text = escape_text_for_xml(self.content)
return element

90
sia/repeat_entry.py Normal file
View File

@@ -0,0 +1,90 @@
from datetime import datetime
import subprocess
import xml.etree.ElementTree as ET
from typing import Optional
from .entry import Entry
from .util import escape_text_for_xml
class RepeatEntry(Entry):
"""
Entry type for scripts that are executed on every update.
Attributes:
script: The script/command to execute
stdout: Captured standard output from most recent execution
stderr: Captured standard error from most recent execution
exit_code: Process exit code from most recent execution
"""
def __init__(self, script: str, id: str, timestamp: datetime):
"""
Initialize a new repeat entry.
Args:
script: The script/command to execute
id: Unique identifier for this entry
timestamp: Creation timestamp for this entry
"""
super().__init__(id, timestamp)
self.script = script
self.stdout = ""
self.stderr = ""
self.exit_code: Optional[int] = None
def update(self) -> None:
"""
Execute the script and update the output.
Captures stdout, stderr and exit code from each execution.
"""
try:
# Run script and capture output
process = subprocess.run(
self.script,
shell=True,
capture_output=True,
text=True
)
# 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:
"""
Generate an XML Element representing this repeat entry.
Returns:
ET.Element: XML element containing the entry's data
"""
# Create root element
element = ET.Element("repeat", {
"id": self.id,
"exit_code": str(self.exit_code) if self.exit_code is not None else "-1"
})
# 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.text = escape_text_for_xml(self.stdout)
# Add stderr element
stderr_elem = ET.SubElement(element, "stderr")
stderr_elem.text = escape_text_for_xml(self.stderr)
return element

101
sia/single_shot_entry.py Normal file
View File

@@ -0,0 +1,101 @@
from datetime import datetime
import subprocess
import xml.etree.ElementTree as ET
from typing import Optional
from .entry import Entry
from .util import escape_text_for_xml
class SingleShotEntry(Entry):
"""
Entry type for one-time script executions.
Attributes:
script: The script/command to execute
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
"""
def __init__(self, script: str, id: str, timestamp: datetime):
"""
Initialize a new single shot entry.
Args:
script: The script/command to execute
id: Unique identifier for this entry
timestamp: Creation timestamp for this entry
"""
super().__init__(id, timestamp)
self.script = script
self.stdout = ""
self.stderr = ""
self.exit_code: Optional[int] = None
self._executed = False
def update(self) -> None:
"""
Execute the script if not already executed.
Captures stdout, stderr and exit code.
"""
if self._executed:
return
try:
# Run script and capture output
process = subprocess.run(
self.script,
shell=True,
capture_output=True,
text=True
)
# 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
def generate_context(self) -> ET.Element:
"""
Generate an XML Element representing this single shot entry.
Returns:
ET.Element: XML element containing the entry's data
"""
# Create root element
element = ET.Element("single_shot", {
"id": self.id,
})
# 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:
# Add stdout element if there is output
stdout_elem = ET.SubElement(element, "stdout")
stdout_elem.text = escape_text_for_xml(self.stdout)
# Add stderr element if there are errors
stderr_elem = ET.SubElement(element, "stderr")
stderr_elem.text = escape_text_for_xml(self.stderr)
# Add exit code attribute after execution
element.set("exit_code", str(self.exit_code))
return element

66
sia/standard_io_buffer.py Normal file
View File

@@ -0,0 +1,66 @@
import sys
class StandardIOBuffer:
"""
IOBuffer implementation that uses system standard input/output.
This class provides direct access to stdin/stdout for IO operations.
It implements a basic line-based input buffer to handle cases where
multiple reads are needed to process all available input.
"""
def __init__(self):
"""Initialize the standard IO buffer."""
self._input_buffer: str = ""
def read(self) -> str:
"""
Read available input from stdin.
If there is buffered input from a previous read, return that first.
Otherwise, try to read new input from stdin if available.
Returns:
str: Content read from stdin, or empty string if no input available
"""
# Return and clear any existing buffered input
if self._input_buffer:
content = self._input_buffer
self._input_buffer = ""
return content
# Check if there's input available
if not sys.stdin.isatty() and sys.stdin.readable():
try:
content = sys.stdin.read()
if content:
return content
except Exception:
pass # Ignore any read errors
return ""
def write(self, content: str) -> None:
"""
Write content to stdout.
Args:
content: String content to write
"""
if not content:
return
try:
sys.stdout.write(content)
sys.stdout.flush()
except Exception:
pass # Ignore write errors
def buffer_length(self) -> int:
"""
Get the current length of buffered input.
Returns:
int: Number of characters in the input buffer
"""
return len(self._input_buffer)

30
sia/stop_command.py Normal file
View File

@@ -0,0 +1,30 @@
from .command import Command
from .command_result import CommandResult
from .working_memory import WorkingMemory
class StopCommand(Command):
"""
Command to stop the agent.
Performs cleanup on all entries and clears working memory.
"""
def execute(self, memory: WorkingMemory) -> CommandResult:
"""
Signal that the agent should stop.
Cleans up all entries and clears the working memory.
Args:
memory: WorkingMemory instance to clear
Returns:
CommandResult: Stop result
"""
# Get a copy of entries to iterate over
entries = memory.get_entries()
# Clean up each entry individually
for entry in entries:
entry.cleanup()
memory.remove_entry(entry.id)
return CommandResult.stop()

75
sia/util.py
View File

@@ -1,52 +1,6 @@
from typing import Iterator
import re
import xml.etree.ElementTree as ET
from .inference_result import InferenceResult
def get_valid_root_elements(schema: str) -> set:
"""
Extract valid root element names from the XML schema.
Args:
schema: XML schema string
Returns:
set: Set of valid root element names
"""
try:
schema = schema.strip()
ET.register_namespace('xs', 'http://www.w3.org/2001/XMLSchema')
root = ET.fromstring(schema)
ns = {'xs': 'http://www.w3.org/2001/XMLSchema'}
elements = root.findall(".//xs:element", ns)
return {elem.get('name') for elem in elements if elem.get('name')}
except ET.ParseError as e:
print(f"Error parsing schema: {e}")
return set()
def split_response(response: str, valid_elements: set) -> InferenceResult:
"""
Split the response into reasoning and actions based on valid XML elements.
Args:
response: Raw response string from the model
valid_elements: Set of valid root element names from the schema
Returns:
InferenceResult: Tuple containing reasoning and actions
"""
elements_pattern = '|'.join(map(re.escape, valid_elements))
pattern = f"<({elements_pattern})[^>]*>"
matches = list(re.finditer(pattern, response))
if not matches:
return InferenceResult(response.strip(), "")
last_match = matches[-1]
split_point = last_match.start()
reasoning = response[:split_point].strip()
actions = response[split_point:].strip()
return InferenceResult(reasoning, actions)
def stop_before_value(iterator: Iterator[str], stop_value: str) -> Iterator[str]:
"""
Creates an iterator that yields values from the input iterator
@@ -66,4 +20,31 @@ def stop_before_value(iterator: Iterator[str], stop_value: str) -> Iterator[str]
if split_point > 0:
yield item[:split_point]
break
yield item
yield item
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}]]>'

66
sia/web_io_buffer.py Normal file
View File

@@ -0,0 +1,66 @@
class WebIOBuffer:
"""
IOBuffer implementation for web interface communication.
This class manages input/output through memory buffers instead of system IO,
allowing for asynchronous communication through a web interface. It maintains
separate buffers for stdin and stdout to simulate bidirectional communication.
"""
def __init__(self):
"""Initialize empty input and output buffers."""
self._stdin_buffer = ""
self._stdout_buffer = ""
def read(self) -> str:
"""
Read and clear all available input from the stdin buffer.
Returns:
str: Content from the stdin buffer, or empty string if buffer is empty
"""
content = self._stdin_buffer
self._stdin_buffer = ""
return content
def write(self, content: str) -> None:
"""
Append content to the stdout buffer.
Args:
content: String content to append to the buffer
"""
if content:
self._stdout_buffer += content
def buffer_length(self) -> int:
"""
Get the current length of the stdin buffer.
Returns:
int: Number of characters in the stdin buffer
"""
return len(self._stdin_buffer)
def append_stdin(self, content: str) -> None:
"""
Append content to the stdin buffer.
Args:
content: String content to append to the stdin buffer
"""
if content:
self._stdin_buffer += content
def get_stdout(self) -> str:
"""
Get the current content of the stdout buffer.
Returns:
str: Current content of the stdout buffer
"""
return self._stdout_buffer
def clear_stdout(self) -> None:
"""Clear the stdout buffer."""
self._stdout_buffer = ""

116
sia/working_memory.py Normal file
View File

@@ -0,0 +1,116 @@
from typing import List, Optional
import xml.etree.ElementTree as ET
from .entry import Entry
class WorkingMemory:
"""
Manages a collection of entries that represent the current state of the system.
The working memory stores different types of entries (scripts, reasoning, errors, etc.)
and provides methods to add, remove, and update them. It also generates XML context
representing the current state.
"""
def __init__(self):
"""Initialize an empty working memory."""
self._entries: List[Entry] = []
def add_entry(self, entry: Entry) -> None:
"""
Add a new entry to working memory.
Args:
entry: Entry object to add
"""
if not isinstance(entry, Entry):
raise TypeError("Entry must be an instance of Entry class")
self._entries.append(entry)
def remove_entry(self, id: str) -> None:
"""
Remove an entry from working memory by its ID.
Ensures cleanup is performed before removal.
Args:
id: Unique identifier of entry to remove
"""
entry = self.get_entry(id)
if entry is not None:
entry.cleanup()
self._entries = [e for e in self._entries if e.id != id]
def clear(self) -> None:
"""
Remove all entries from working memory.
Ensures cleanup is performed on all entries.
"""
for entry in self._entries:
entry.cleanup()
self._entries.clear()
def __del__(self):
"""
Clean up all entries when memory is deleted.
"""
if hasattr(self, '_entries'): # Check in case of partial initialization
self.clear()
def get_entry(self, id: str) -> Optional[Entry]:
"""
Get an entry by its ID.
Args:
id: Unique identifier of entry to retrieve
Returns:
Entry if found, None otherwise
"""
for entry in self._entries:
if entry.id == id:
return entry
return None
def get_entries(self) -> List[Entry]:
"""
Get all entries in working memory.
Returns:
List[Entry]: List of all entries
"""
return self._entries.copy() # Return a copy to prevent external modification
def get_entries_count(self) -> int:
"""
Get the total number of entries.
Returns:
int: Number of entries in working memory
"""
return len(self._entries)
def get_entries_by_type(self, entry_type: type) -> List[Entry]:
"""
Get all entries of a specific type.
Args:
entry_type: Type of entries to retrieve
Returns:
List[Entry]: List of entries matching the specified type
"""
return [e for e in self._entries if isinstance(e, entry_type)]
def update(self) -> None:
"""Update all entries in working memory."""
for entry in self._entries:
entry.update()
def generate_context(self) -> List[ET.Element]:
"""
Generate XML Elements representing all entries.
Returns:
List[ET.Element]: List of XML elements for each entry
"""
return [entry.generate_context() for entry in self._entries]

54
sia/write_entry.py Normal file
View File

@@ -0,0 +1,54 @@
from datetime import datetime
import xml.etree.ElementTree as ET
from .entry import Entry
from .util import escape_text_for_xml
class WriteEntry(Entry):
"""
Entry type for writing content to standard output.
Attributes:
content: Text content to write to stdout
io_buffer: Buffer to use for IO operations
_written: Whether the content has been written
"""
def __init__(self, content: str, io_buffer, id: str, timestamp: datetime):
"""
Initialize a new write entry.
Args:
content: Text to write to stdout
io_buffer: Buffer to use for IO operations
id: Unique identifier for this entry
timestamp: Creation timestamp for this entry
"""
super().__init__(id, timestamp)
self.content = content
self.io_buffer = io_buffer
self._written = False
def update(self) -> None:
"""
Write the content to stdout if not already written.
Uses the provided IO buffer for the actual write operation.
"""
if not self._written:
self.io_buffer.write(self.content)
self._written = True
def generate_context(self) -> ET.Element:
"""
Generate an XML Element representing this write entry.
Returns:
ET.Element: XML element containing the entry's data
"""
# Create root element
element = ET.Element("write_stdout", {"id": self.id})
# Add content as CDATA or escaped text
element.text = escape_text_for_xml(self.content)
return element

97
sia/xml_validator.py Normal file
View File

@@ -0,0 +1,97 @@
import xml.etree.ElementTree as ET
from typing import Optional, Set
class XMLValidator:
"""
Validates XML content against a schema.
Attributes:
_schema: The parsed XML schema to validate against
_valid_root_elements: Set of valid root element names from schema
"""
def __init__(self, schema: str):
"""
Initialize validator with XML schema.
Args:
schema: XML schema string
"""
# Register namespace used in schema
ET.register_namespace('xs', 'http://www.w3.org/2001/XMLSchema')
try:
# Parse schema
self._schema = ET.fromstring(schema.strip())
# Extract valid root elements
ns = {'xs': 'http://www.w3.org/2001/XMLSchema'}
elements = self._schema.findall(".//xs:element", ns)
self._valid_root_elements = {elem.get('name') for elem in elements if elem.get('name')}
except ET.ParseError as e:
raise ValueError(f"Invalid schema: {e}")
def validate(self, xml: str) -> Optional[str]:
"""
Validate XML content against the schema.
Args:
xml: XML string to validate
Returns:
str: Error message if validation fails, None if validation succeeds
"""
try:
# Parse XML
root = ET.fromstring(xml.strip())
# Check root element is valid
if root.tag not in self._valid_root_elements:
return f"Invalid root element: {root.tag}. Expected one of: {sorted(self._valid_root_elements)}"
# Get schema definition for this element
ns = {'xs': 'http://www.w3.org/2001/XMLSchema'}
element_schema = self._schema.find(f".//xs:element[@name='{root.tag}']", ns)
if element_schema is None:
return f"Schema definition not found for element: {root.tag}"
# Validate attributes if complex type defined
complex_type = element_schema.find('xs:complexType', ns)
if complex_type is not None:
# Check required attributes
for attr in complex_type.findall('.//xs:attribute[@use="required"]', ns):
attr_name = attr.get('name')
if attr_name not in root.attrib:
return f"Missing required attribute '{attr_name}' on element '{root.tag}'"
# Check attribute types
for attr_name, attr_value in root.attrib.items():
attr_schema = complex_type.find(f'.//xs:attribute[@name="{attr_name}"]', ns)
if attr_schema is None:
return f"Unexpected attribute '{attr_name}' on element '{root.tag}'"
attr_type = attr_schema.get('type')
if attr_type == 'xs:string':
continue # All string values are valid
elif attr_type == 'xs:integer':
try:
int(attr_value)
except ValueError:
return f"Invalid integer value '{attr_value}' for attribute '{attr_name}'"
return None # Validation successful
except ET.ParseError as e:
return f"Invalid XML: {e}"
except Exception as e:
return f"Validation error: {e}"
def get_valid_root_elements(self) -> Set[str]:
"""
Get set of valid root element names from schema.
Returns:
Set[str]: Set of valid root element names
"""
return self._valid_root_elements.copy()

227
test/background_entry_test.py Normal file
View File

@@ -0,0 +1,227 @@
import unittest
from datetime import datetime
import os
import time
import signal
import platform
from sia.background_entry import BackgroundEntry
class BackgroundEntryTest(unittest.TestCase):
def setUp(self):
"""Set up test cases with fixed id and timestamp"""
self.test_id = "test-id-1234"
self.test_timestamp = datetime(2024, 10, 31, 12, 0, 0)
# Create temporary files for testing
self.test_filename = "test_output.txt"
self.counter_file = "counter.txt"
self.cleanup_files()
def tearDown(self):
"""Clean up test files and ensure no processes are left running"""
self.cleanup_files()
def cleanup_files(self):
"""Helper to remove test files"""
for filename in [self.test_filename, self.counter_file]:
if os.path.exists(filename):
os.remove(filename)
def wait_for_process_start(self, entry, timeout=1.0):
"""Wait for process to start, with timeout"""
start_time = time.time()
while entry.process is None and time.time() - start_time < timeout:
entry.update()
time.sleep(0.1)
self.assertIsNotNone(entry.process, "Process failed to start")
def wait_for_process_exit(self, entry, timeout=1.0):
"""Wait for process to exit, with timeout"""
start_time = time.time()
while entry.process is not None and time.time() - start_time < timeout:
entry.update()
time.sleep(0.1)
self.assertIsNone(entry.process, "Process failed to exit")
def test_initialization(self):
"""Test entry initialization"""
script = "echo test"
entry = BackgroundEntry(script, self.test_id, self.test_timestamp)
self.assertEqual(entry.script, script)
self.assertEqual(entry.stdout, "")
self.assertEqual(entry.stderr, "")
self.assertIsNone(entry.process)
self.assertEqual(entry.id, self.test_id)
self.assertEqual(entry.timestamp, self.test_timestamp)
def test_short_running_process(self):
"""Test execution of a quick process"""
script = "echo 'test'"
entry = BackgroundEntry(script, self.test_id, self.test_timestamp)
# Start and wait for completion
entry.update()
self.wait_for_process_start(entry)
self.wait_for_process_exit(entry)
# Verify output
self.assertEqual(entry._accumulated_stdout.strip(), "test")
self.assertEqual(entry._accumulated_stderr, "")
self.assertEqual(entry._exit_code, 0)
def test_continuous_output(self):
"""Test process that generates continuous output"""
# Create a Python script that counts to 3 with minimal delay
script = (
'python3 -c "'
'import sys\n'
'for i in range(3):\n'
' sys.stdout.write(f\\\"count {i+1}\\n\\\")\n'
' sys.stdout.flush()\n'
'"'
)
entry = BackgroundEntry(script, self.test_id, self.test_timestamp)
# Start process and wait for completion
entry.update()
self.wait_for_process_start(entry)
self.wait_for_process_exit(entry, timeout=2.0)
# One more update to ensure we get all output
entry.update()
# Verify all output was captured
output = entry._accumulated_stdout
self.assertIn('count 1', output, "Missing count 1")
self.assertIn('count 2', output, "Missing count 2")
self.assertIn('count 3', output, "Missing count 3")
def test_process_failure(self):
"""Test handling of process failures"""
entry = BackgroundEntry("nonexistentcommand", self.test_id, self.test_timestamp)
# Start and wait for failure
entry.update()
self.wait_for_process_exit(entry)
# Verify error was captured
self.assertTrue(len(entry._accumulated_stderr) > 0)
self.assertEqual(entry._accumulated_stdout, "")
self.assertNotEqual(entry._exit_code, 0)
def test_generate_context_running(self):
"""Test XML context generation for running process"""
script = "sleep 1"
entry = BackgroundEntry(script, self.test_id, self.test_timestamp)
entry.update() # Start process
self.wait_for_process_start(entry)
element = entry.generate_context()
# Verify XML structure
self.assertEqual(element.tag, "background")
self.assertEqual(element.get("id"), self.test_id)
self.assertIsNotNone(element.get("pid"))
self.assertIsNone(element.get("exit_code"))
# Clean up
entry._cleanup_process()
def test_generate_context_completed(self):
"""Test XML context generation for completed process"""
script = "echo 'test'"
entry = BackgroundEntry(script, self.test_id, self.test_timestamp)
# Start and wait for completion
entry.update()
self.wait_for_process_start(entry)
self.wait_for_process_exit(entry)
element = entry.generate_context()
# Verify XML structure
self.assertEqual(element.tag, "background")
self.assertEqual(element.get("id"), self.test_id)
self.assertIsNone(element.get("pid"))
self.assertEqual(element.get("exit_code"), "0")
stdout_text = element.find("stdout").text
self.assertTrue(stdout_text.startswith("<![CDATA["))
self.assertIn("test", stdout_text)
self.assertTrue(stdout_text.endswith("]]>"))
def test_cleanup_on_deletion(self):
"""Test process cleanup when entry is deleted"""
script = "sleep 10"
entry = BackgroundEntry(script, self.test_id, self.test_timestamp)
entry.update()
self.wait_for_process_start(entry)
# Verify process is running
self.assertIsNotNone(entry.process)
pid = entry.process.pid
# Delete the entry
entry._cleanup_process()
del entry
# Give process time to be cleaned up
time.sleep(0.1)
# Verify process was terminated
with self.assertRaises(ProcessLookupError):
os.kill(pid, 0) # Check if process exists
def test_cleanup_running_process(self):
"""Test cleanup of running process"""
script = "sleep 10"
entry = BackgroundEntry(script, self.test_id, self.test_timestamp)
# Start the process
entry.update()
self.assertIsNotNone(entry.process)
process_pid = entry.process.pid
# Cleanup should terminate process
entry.cleanup()
# Verify process was terminated
self.assertIsNone(entry.process)
time.sleep(0.1) # Give process time to terminate
with self.assertRaises(ProcessLookupError):
import os
os.kill(process_pid, 0)
def test_cleanup_completed_process(self):
"""Test cleanup of already completed process"""
script = "echo 'test'"
entry = BackgroundEntry(script, self.test_id, self.test_timestamp)
# Run and wait for completion
entry.update()
self.wait_for_process_exit(entry)
# Cleanup should be safe for completed process
entry.cleanup()
self.assertIsNone(entry.process)
def test_multiple_cleanup_calls(self):
"""Test that multiple cleanup calls are safe"""
script = "sleep 10"
entry = BackgroundEntry(script, self.test_id, self.test_timestamp)
# Start process
entry.update()
self.assertIsNotNone(entry.process)
# Multiple cleanups should be safe
entry.cleanup()
entry.cleanup()
entry.cleanup()
self.assertIsNone(entry.process)

33
test/context_template_test.py
View File

@@ -1,33 +0,0 @@
import unittest
from datetime import datetime
from sia.context_template import generate_context
class TestContextTemplate(unittest.TestCase):
def test_empty_containers(self):
"""Test context generation with no containers."""
context = generate_context([])
self.assertIn("<context>", context)
self.assertIn("<containers/>", context)
self.assertIn("</context>", context)
def test_single_container(self):
"""Test context generation with a single container."""
container_status = {
'name': 'test-container',
'status': 'running',
'started_at': '2024-10-25T10:00:00Z',
'stdout_size': 100,
'stderr_size': 50
}
context = generate_context([container_status])
self.assertIn("<context>", context)
self.assertIn("<containers", context)
self.assertIn('name="test-container"', context)
self.assertIn('status="running"', context)
self.assertIn('started_at="2024-10-25T10:00:00Z"', context)
self.assertIn('stdout="100"', context)
self.assertIn('stderr="50"', context)
self.assertIn("</context>", context)
if __name__ == '__main__':
unittest.main()

66
test/delete_command_test.py Normal file
View File

@@ -0,0 +1,66 @@
import unittest
from datetime import datetime
import time
from sia.working_memory import WorkingMemory
from sia.delete_command import DeleteCommand
from sia.reasoning_entry import ReasoningEntry
from sia.background_entry import BackgroundEntry
class DeleteCommandTest(unittest.TestCase):
def setUp(self):
"""Set up test cases with a fresh working memory"""
self.memory = WorkingMemory()
self.test_id = "test-id-1234"
self.test_timestamp = datetime(2024, 10, 31, 12, 0, 0)
def test_delete_existing_entry(self):
"""Test deleting an existing entry"""
# Add test entry
entry = ReasoningEntry("test content", self.test_id, self.test_timestamp)
self.memory.add_entry(entry)
# Execute delete command
command = DeleteCommand(self.test_id)
result = command.execute(self.memory)
# Verify result and memory state
self.assertTrue(result.success)
self.assertFalse(result.should_stop)
self.assertEqual(result.message, "")
self.assertEqual(self.memory.get_entries_count(), 0)
def test_delete_running_background_entry(self):
"""Test deleting a background entry that is still running"""
# Create a background entry with a long-running command
entry = BackgroundEntry("sleep 10", self.test_id, self.test_timestamp)
self.memory.add_entry(entry)
# Start the process
entry.update()
self.assertIsNotNone(entry.process)
process_pid = entry.process.pid
# Execute delete command
command = DeleteCommand(self.test_id)
result = command.execute(self.memory)
# Verify process was terminated
time.sleep(0.1) # Give process time to terminate
self.assertTrue(result.success)
self.assertIsNone(entry.process)
# Verify process no longer exists
with self.assertRaises(ProcessLookupError):
import os
os.kill(process_pid, 0)
def test_delete_nonexistent_entry(self):
"""Test attempting to delete a nonexistent entry"""
command = DeleteCommand("nonexistent-id")
result = command.execute(self.memory)
# Verify error result
self.assertFalse(result.success)
self.assertFalse(result.should_stop)
self.assertIn("not found", result.message)

121
test/docker_module_test.py
View File

@@ -1,121 +0,0 @@
import unittest
import time
from sia.docker_module import DockerModule
class DockerModuleTest(unittest.TestCase):
def test_initialization(self):
"""Test DockerModule initialization."""
with DockerModule() as docker_module:
self.assertIsInstance(docker_module, DockerModule)
def test_start_container_short_lived(self):
"""Test starting a short-lived container."""
with DockerModule() as docker_module:
image = "busybox:latest"
timeout = 1000
command = "echo"
arguments = ["Hello World"]
output = docker_module.start_container(
image=image,
timeout=timeout,
command=command,
arguments=arguments
)
self.assertEqual(output, "Hello World\n")
def test_start_container_long_running(self):
"""Test starting a long-running container."""
with DockerModule() as docker_module:
name = "test_start_container_long_running"
image = "busybox:latest"
container_name = docker_module.start_container(
image=image,
name=name
)
self.assertEqual(container_name, name)
self.assertIn(name, docker_module.containers)
docker_module.wait_container(container_name, 0)
self.assertNotIn(name, docker_module.containers)
def test_start_container_with_volumes(self):
"""Test starting a container with volume mappings."""
with DockerModule() as docker_module:
arguments = ["sh", "-c", "echo 'Hello World' > /write_here/test_start_container_with_volumes"]
volumes = {"/tmp": "/write_here"}
docker_module.start_container(
image="busybox:latest",
timeout=1000,
arguments=arguments,
volumes=volumes
)
with open("/tmp/test_start_container_with_volumes", "r") as f:
self.assertEqual(f.read(), "Hello World\n")
def test_invalid_container_start(self):
"""Test error handling for invalid container start."""
with DockerModule() as docker_module:
with self.assertRaises(ValueError):
docker_module.start_container(image="busybox:latest")
def test_container_io_operations(self):
"""Test container I/O operations."""
with DockerModule() as docker_module:
arguments = [
"sh",
"-c",
"read input; echo \"$input\" | tr '[:lower:]' '[:upper:]'; echo \"$input\" | tr '[:upper:]' '[:lower:]' >&2"
]
name = "test_container_io_operations"
container_name = docker_module.start_container(
image="busybox:latest",
name=name,
arguments=arguments
)
test_input = "Test Input\n"
docker_module.write_container_stdin(name, test_input)
time.sleep(1) # for processing
stdout = docker_module.read_container_stdout(name)
self.assertEqual(stdout, "TEST INPUT\n")
stderr = docker_module.read_container_stderr(name)
self.assertEqual(stderr, "test input\n")
docker_module.wait_container(container_name, 0)
def test_wait_container(self):
"""Test waiting for container completion."""
with DockerModule() as docker_module:
name = "test_wait_container"
docker_module.start_container(
image="busybox:latest",
name=name
)
exit_code, output = docker_module.wait_container(name, timeout=1000)
self.assertEqual(exit_code, -1)
self.assertEqual(output, "")
def test_get_container_status(self):
"""Test getting container status information."""
with DockerModule() as docker_module:
arguments = [
"sh",
"-c",
"echo 'standard output'; echo 'standard error' >&2; sleep 3"
]
name = "test_get_container_status"
docker_module.start_container(
image="busybox:latest",
name=name,
arguments=arguments
)
time.sleep(1)
status = docker_module.get_container_status(name)
self.assertEqual(status['name'], name)
self.assertEqual(status['status'], "running")
self.assertGreater(len(status['started_at']), 0)
self.assertEqual(status['exit_code'], 0)
self.assertEqual(status['error'], "")
self.assertEqual(status['stdout_size'], 16)
self.assertEqual(status['stderr_size'], 15)
docker_module.wait_container(name, 0)
if __name__ == '__main__':
unittest.main()

1
test/llm_engine_test.py
View File

@@ -3,7 +3,6 @@ import unittest
from itertools import tee
from . import test_data
from . import test_util
from sia.llm_engine import LlmEngine

88
test/parse_error_entry_test.py Normal file
View File

@@ -0,0 +1,88 @@
import unittest
from datetime import datetime
from sia.parse_error_entry import ParseErrorEntry
class ParseErrorEntryTest(unittest.TestCase):
def setUp(self):
"""Set up test cases with fixed id and timestamp"""
self.test_id = "test-id-1234"
self.test_timestamp = datetime(2024, 10, 31, 12, 0, 0)
def test_initialization(self):
"""Test entry initialization"""
content = "invalid content"
error = "parsing failed"
entry = ParseErrorEntry(content, error, self.test_id, self.test_timestamp)
self.assertEqual(entry.content, content)
self.assertEqual(entry.error, error)
self.assertEqual(entry.id, self.test_id)
self.assertEqual(entry.timestamp, self.test_timestamp)
def test_update_does_nothing(self):
"""Test that update operation has no effect"""
content = "invalid content"
error = "parsing failed"
entry = ParseErrorEntry(content, error, self.test_id, self.test_timestamp)
# Store initial state
initial_content = entry.content
initial_error = entry.error
# Perform update
entry.update()
# Verify state hasn't changed
self.assertEqual(entry.content, initial_content)
self.assertEqual(entry.error, initial_error)
def test_generate_context(self):
"""Test XML context generation"""
content = "invalid content"
error = "parsing failed"
entry = ParseErrorEntry(content, error, self.test_id, self.test_timestamp)
element = entry.generate_context()
# Verify XML structure
self.assertEqual(element.tag, "parse_error")
self.assertEqual(element.get("id"), self.test_id)
# Verify error element
error_elem = element.find("error")
self.assertIsNotNone(error_elem)
self.assertEqual(error_elem.text, f"<![CDATA[{error}]]>")
# Verify content element
content_elem = element.find("content")
self.assertIsNotNone(content_elem)
self.assertEqual(content_elem.text, f"<![CDATA[{content}]]>")
def test_special_characters(self):
"""Test handling content and errors with special characters"""
content = "Special content: \n\t\r\'\"\\"
error = "Special error: \n\t\r\'\"\\"
entry = ParseErrorEntry(content, error, self.test_id, self.test_timestamp)
element = entry.generate_context()
self.assertEqual(element.find("content").text, f"<![CDATA[{content}]]>")
self.assertEqual(element.find("error").text, f"<![CDATA[{error}]]>")
def test_empty_content_and_error(self):
"""Test handling empty content and error messages"""
entry = ParseErrorEntry("", "", self.test_id, self.test_timestamp)
element = entry.generate_context()
self.assertEqual(element.find("content").text, "<![CDATA[]]>")
self.assertEqual(element.find("error").text, "<![CDATA[]]>")
def test_large_content(self):
"""Test handling large content and error messages"""
content = "x" * 10000
error = "y" * 10000
entry = ParseErrorEntry(content, error, self.test_id, self.test_timestamp)
element = entry.generate_context()
self.assertEqual(element.find("content").text, f"<![CDATA[{content}]]>")
self.assertEqual(element.find("error").text, f"<![CDATA[{error}]]>")

130
test/read_entry_test.py Normal file
View File

@@ -0,0 +1,130 @@
import unittest
from datetime import datetime
from sia.web_io_buffer import WebIOBuffer
from sia.read_entry import ReadEntry
class ReadEntryTest(unittest.TestCase):
def setUp(self):
"""Set up test cases with fixed id, timestamp, and IO buffer"""
self.test_id = "test-id-1234"
self.test_timestamp = datetime(2024, 10, 31, 12, 0, 0)
self.io_buffer = WebIOBuffer()
def test_initialization(self):
"""Test entry initialization"""
entry = ReadEntry(self.io_buffer, self.test_id, self.test_timestamp)
self.assertEqual(entry.content, "")
self.assertEqual(entry.id, self.test_id)
self.assertEqual(entry.timestamp, self.test_timestamp)
self.assertFalse(entry._read)
def test_single_read(self):
"""Test reading content once"""
test_input = "test message"
self.io_buffer.append_stdin(test_input)
entry = ReadEntry(self.io_buffer, self.test_id, self.test_timestamp)
entry.update()
# Verify content was read
self.assertEqual(entry.content, test_input)
self.assertTrue(entry._read)
self.assertEqual(self.io_buffer.buffer_length(), 0) # Buffer should be cleared
def test_multiple_updates(self):
"""Test that content is only read once even with multiple updates"""
test_input = "initial input"
self.io_buffer.append_stdin(test_input)
entry = ReadEntry(self.io_buffer, self.test_id, self.test_timestamp)
entry.update()
initial_content = entry.content
# Add more input and update again
self.io_buffer.append_stdin("additional input")
entry.update()
entry.update()
# Verify content hasn't changed after first read
self.assertEqual(entry.content, initial_content)
self.assertTrue(entry._read)
def test_empty_input(self):
"""Test reading when no input is available"""
entry = ReadEntry(self.io_buffer, self.test_id, self.test_timestamp)
entry.update()
self.assertEqual(entry.content, "")
self.assertTrue(entry._read)
def test_special_characters(self):
"""Test reading content with special characters"""
test_input = "Special chars: \n\t\r\'\"\\"
self.io_buffer.append_stdin(test_input)
entry = ReadEntry(self.io_buffer, self.test_id, self.test_timestamp)
entry.update()
self.assertEqual(entry.content, test_input)
def test_large_content(self):
"""Test reading large content"""
test_input = "x" * 10000
self.io_buffer.append_stdin(test_input)
entry = ReadEntry(self.io_buffer, self.test_id, self.test_timestamp)
entry.update()
self.assertEqual(entry.content, test_input)
def test_generate_context_before_read(self):
"""Test XML context generation before reading"""
entry = ReadEntry(self.io_buffer, self.test_id, self.test_timestamp)
element = entry.generate_context()
# Verify XML structure
self.assertEqual(element.tag, "read_stdin")
self.assertEqual(element.get("id"), self.test_id)
self.assertIsNone(element.text) # No content before read
def test_generate_context_after_read(self):
"""Test XML context generation after reading"""
test_input = "test message"
self.io_buffer.append_stdin(test_input)
entry = ReadEntry(self.io_buffer, self.test_id, self.test_timestamp)
entry.update()
element = entry.generate_context()
# Verify XML structure
self.assertEqual(element.tag, "read_stdin")
self.assertEqual(element.get("id"), self.test_id)
self.assertEqual(element.text, f"<![CDATA[{test_input}]]>")
def test_unicode_content(self):
"""Test reading Unicode content"""
test_input = "Hello 世界 😊"
self.io_buffer.append_stdin(test_input)
entry = ReadEntry(self.io_buffer, self.test_id, self.test_timestamp)
entry.update()
self.assertEqual(entry.content, test_input)
def test_multiline_content(self):
"""Test reading multiline content"""
test_input = """Line 1
Line 2
Line 3"""
self.io_buffer.append_stdin(test_input)
entry = ReadEntry(self.io_buffer, self.test_id, self.test_timestamp)
entry.update()
self.assertEqual(entry.content, test_input)
# Verify XML escaping for multiline content
element = entry.generate_context()
self.assertEqual(element.text, f"<![CDATA[{test_input}]]>")

78
test/reasoning_entry_test.py Normal file
View File

@@ -0,0 +1,78 @@
import unittest
from datetime import datetime
from sia.reasoning_entry import ReasoningEntry
class ReasoningEntryTest(unittest.TestCase):
def setUp(self):
"""Set up test cases with fixed id and timestamp"""
self.test_id = "test-id-1234"
self.test_timestamp = datetime(2024, 10, 31, 12, 0, 0)
def test_initialization(self):
"""Test entry initialization"""
content = "test reasoning"
entry = ReasoningEntry(content, self.test_id, self.test_timestamp)
self.assertEqual(entry.content, content)
self.assertEqual(entry.id, self.test_id)
self.assertEqual(entry.timestamp, self.test_timestamp)
def test_update_does_nothing(self):
"""Test that update operation has no effect"""
content = "test reasoning"
entry = ReasoningEntry(content, self.test_id, self.test_timestamp)
# Store initial state
initial_content = entry.content
# Perform update
entry.update()
# Verify state hasn't changed
self.assertEqual(entry.content, initial_content)
def test_generate_context(self):
"""Test XML context generation"""
content = "test reasoning"
entry = ReasoningEntry(content, self.test_id, self.test_timestamp)
element = entry.generate_context()
# Verify XML structure
self.assertEqual(element.tag, "reasoning")
self.assertEqual(element.get("id"), self.test_id)
self.assertEqual(element.text, f"<![CDATA[{content}]]>")
def test_special_characters(self):
"""Test handling reasoning text with special characters"""
content = "Special reasoning: \n\t\r\'\"\\"
entry = ReasoningEntry(content, self.test_id, self.test_timestamp)
element = entry.generate_context()
self.assertEqual(element.text, f"<![CDATA[{content}]]>")
def test_empty_content(self):
"""Test handling empty reasoning text"""
entry = ReasoningEntry("", self.test_id, self.test_timestamp)
element = entry.generate_context()
self.assertEqual(element.text, "<![CDATA[]]>")
def test_large_content(self):
"""Test handling large reasoning text"""
content = "x" * 10000
entry = ReasoningEntry(content, self.test_id, self.test_timestamp)
element = entry.generate_context()
self.assertEqual(element.text, f"<![CDATA[{content}]]>")
def test_multiline_content(self):
"""Test handling multiline reasoning text"""
content = """Line 1
Line 2
Line 3"""
entry = ReasoningEntry(content, self.test_id, self.test_timestamp)
element = entry.generate_context()
self.assertEqual(element.text, f"<![CDATA[{content}]]>")

164
test/repeat_entry_test.py Normal file
View File

@@ -0,0 +1,164 @@
import unittest
from datetime import datetime
import os
from sia.repeat_entry import RepeatEntry
class RepeatEntryTest(unittest.TestCase):
def setUp(self):
"""Set up test cases with fixed id and timestamp"""
self.test_id = "test-id-1234"
self.test_timestamp = datetime(2024, 10, 31, 12, 0, 0)
# Create a temporary file for testing
self.test_filename = "test_output.txt"
self.counter_file = "counter.txt"
self.cleanup_files()
def tearDown(self):
"""Clean up any test files"""
self.cleanup_files()
def cleanup_files(self):
"""Helper to remove test files"""
for filename in [self.test_filename, self.counter_file]:
if os.path.exists(filename):
os.remove(filename)
def test_initialization(self):
"""Test entry initialization"""
script = "echo test"
entry = RepeatEntry(script, self.test_id, self.test_timestamp)
self.assertEqual(entry.script, script)
self.assertEqual(entry.stdout, "")
self.assertEqual(entry.stderr, "")
self.assertIsNone(entry.exit_code)
self.assertEqual(entry.id, self.test_id)
self.assertEqual(entry.timestamp, self.test_timestamp)
def test_successful_execution(self):
"""Test successful script execution"""
script = "echo 'test'"
entry = RepeatEntry(script, self.test_id, self.test_timestamp)
entry.update()
# Verify entry state
self.assertEqual(entry.stdout.strip(), "test")
self.assertEqual(entry.stderr, "")
self.assertEqual(entry.exit_code, 0)
def test_failed_execution(self):
"""Test failed script execution with invalid command"""
entry = RepeatEntry("nonexistentcommand", self.test_id, self.test_timestamp)
entry.update()
# Verify entry state
self.assertEqual(entry.stdout, "")
self.assertTrue(len(entry.stderr) > 0) # Error message will vary by OS
self.assertNotEqual(entry.exit_code, 0) # Exit code will be non-zero
def test_repeated_execution(self):
"""Test that script executes on every update"""
# Create a counter file
with open(self.counter_file, 'w') as f:
f.write('0')
# Script that increments and reads the counter
script = (
f'python3 -c "'
f'import os; '
f'f=open(\'{self.counter_file}\', \'r+\'); '
f'n=int(f.read()); '
f'f.seek(0); '
f'f.write(str(n+1)); '
f'f.truncate(); '
f'f.close(); '
f'print(n+1)'
f'"'
)
entry = RepeatEntry(script, self.test_id, self.test_timestamp)
# Run update multiple times
outputs = []
for _ in range(3):
entry.update()
outputs.append(int(entry.stdout.strip()))
# Verify outputs are sequential numbers
self.assertEqual(outputs, [1, 2, 3])
def test_generate_context(self):
"""Test XML context generation"""
script = "echo 'test'"
entry = RepeatEntry(script, self.test_id, self.test_timestamp)
entry.update()
element = entry.generate_context()
# Verify XML structure
self.assertEqual(element.tag, "repeat")
self.assertEqual(element.get("id"), self.test_id)
self.assertEqual(element.text, f"<![CDATA[{script}]]>")
self.assertEqual(element.get("exit_code"), "0")
# Get stdout and remove trailing newline for comparison
stdout_text = element.find("stdout").text
self.assertTrue(stdout_text.startswith("<![CDATA[test"))
self.assertTrue(stdout_text.endswith("]]>"))
self.assertEqual(element.find("stderr").text, "<![CDATA[]]>")
def test_changing_output(self):
"""Test handling of changing command output"""
# Create a counter file
with open(self.counter_file, 'w') as f:
f.write('0')
# Script that outputs an incrementing number
script = (
f'python3 -c "'
f'import os; '
f'f=open(\'{self.counter_file}\', \'r+\'); '
f'n=int(f.read()); '
f'f.seek(0); '
f'f.write(str(n+1)); '
f'f.truncate(); '
f'f.close(); '
f'print(\'Output\', n+1)'
f'"'
)
entry = RepeatEntry(script, self.test_id, self.test_timestamp)
# Execute multiple times and verify output changes
entry.update()
first_output = entry.stdout.strip()
entry.update()
second_output = entry.stdout.strip()
entry.update()
third_output = entry.stdout.strip()
# Outputs should be different for each run
outputs = [first_output, second_output, third_output]
unique_outputs = set(outputs)
self.assertEqual(len(unique_outputs), 3)
self.assertEqual(outputs, ['Output 1', 'Output 2', 'Output 3'])
def test_error_recovery(self):
"""Test that entry recovers from errors in subsequent updates"""
# Start with a failing command
entry = RepeatEntry("nonexistentcommand", self.test_id, self.test_timestamp)
entry.update()
self.assertNotEqual(entry.exit_code, 0)
# Change to a working command
entry.script = "echo 'test'"
entry.update()
# Verify recovery
self.assertEqual(entry.exit_code, 0)
self.assertEqual(entry.stdout.strip(), "test")
self.assertEqual(entry.stderr, "")

141
test/single_shot_entry_test.py Normal file
View File

@@ -0,0 +1,141 @@
import unittest
from datetime import datetime
import os
from sia.single_shot_entry import SingleShotEntry
class SingleShotEntryTest(unittest.TestCase):
def setUp(self):
"""Set up test cases with fixed id and timestamp"""
self.test_id = "test-id-1234"
self.test_timestamp = datetime(2024, 10, 31, 12, 0, 0)
# Create a temporary file for testing
self.test_filename = "test_output.txt"
self.cleanup_files()
def tearDown(self):
"""Clean up any test files"""
self.cleanup_files()
def cleanup_files(self):
"""Helper to remove test files"""
if os.path.exists(self.test_filename):
os.remove(self.test_filename)
def test_initialization(self):
"""Test entry initialization"""
script = "echo test"
entry = SingleShotEntry(script, self.test_id, self.test_timestamp)
self.assertEqual(entry.script, script)
self.assertEqual(entry.stdout, "")
self.assertEqual(entry.stderr, "")
self.assertIsNone(entry.exit_code)
self.assertEqual(entry.id, self.test_id)
self.assertEqual(entry.timestamp, self.test_timestamp)
self.assertFalse(entry._executed)
def test_successful_execution(self):
"""Test successful script execution"""
script = "echo 'test'"
entry = SingleShotEntry(script, self.test_id, self.test_timestamp)
entry.update()
# Verify entry state
self.assertEqual(entry.stdout.strip(), "test")
self.assertEqual(entry.stderr, "")
self.assertEqual(entry.exit_code, 0)
self.assertTrue(entry._executed)
def test_failed_execution(self):
"""Test failed script execution with invalid command"""
entry = SingleShotEntry("nonexistentcommand", self.test_id, self.test_timestamp)
entry.update()
# Verify entry state
self.assertEqual(entry.stdout, "")
self.assertTrue(len(entry.stderr) > 0) # Error message will vary by OS
self.assertNotEqual(entry.exit_code, 0) # Exit code will be non-zero
self.assertTrue(entry._executed)
def test_file_creation(self):
"""Test script that creates a file"""
script = f"echo 'test' > {self.test_filename}"
entry = SingleShotEntry(script, self.test_id, self.test_timestamp)
entry.update()
# Verify file was created and contains expected content
self.assertTrue(os.path.exists(self.test_filename))
with open(self.test_filename, 'r') as f:
content = f.read().strip()
self.assertEqual(content, "test")
def test_single_execution(self):
"""Test that script only executes once"""
script = f"echo 'test' >> {self.test_filename}"
entry = SingleShotEntry(script, self.test_id, self.test_timestamp)
# Run update multiple times
entry.update()
entry.update()
entry.update()
# Verify file only contains one line
with open(self.test_filename, 'r') as f:
lines = f.readlines()
self.assertEqual(len(lines), 1)
self.assertEqual(lines[0].strip(), "test")
def test_generate_context_before_execution(self):
"""Test XML context generation before execution"""
entry = SingleShotEntry("echo test", self.test_id, self.test_timestamp)
element = entry.generate_context()
# Verify XML structure
self.assertEqual(element.tag, "single_shot")
self.assertEqual(element.get("id"), self.test_id)
self.assertEqual(element.text, "<![CDATA[echo test]]>")
# Verify no output elements exist
self.assertIsNone(element.find("stdout"))
self.assertIsNone(element.find("stderr"))
self.assertIsNone(element.get("exit_code"))
def test_generate_context_after_execution(self):
"""Test XML context generation after execution"""
script = "echo 'test'"
entry = SingleShotEntry(script, self.test_id, self.test_timestamp)
entry.update()
element = entry.generate_context()
# Verify XML structure
self.assertEqual(element.tag, "single_shot")
self.assertEqual(element.get("id"), self.test_id)
self.assertEqual(element.text, f"<![CDATA[{script}]]>")
# Get stdout and remove trailing newline for comparison
stdout_text = element.find("stdout").text
self.assertTrue(stdout_text.startswith("<![CDATA[test"))
self.assertTrue(stdout_text.endswith("]]>"))
self.assertEqual(element.find("stderr").text, "<![CDATA[]]>")
self.assertEqual(element.get("exit_code"), "0")
def test_multiple_commands(self):
"""Test executing multiple commands in one script"""
script = f"echo 'first' > {self.test_filename} && echo 'second' >> {self.test_filename}"
entry = SingleShotEntry(script, self.test_id, self.test_timestamp)
entry.update()
# Verify file contains both lines
with open(self.test_filename, 'r') as f:
lines = f.readlines()
self.assertEqual(len(lines), 2)
self.assertEqual(lines[0].strip(), "first")
self.assertEqual(lines[1].strip(), "second")

180
test/standard_io_buffer_test.py Normal file
View File

@@ -0,0 +1,180 @@
import unittest
import os
import subprocess
import sys
import tempfile
from pathlib import Path
class StandardIOBufferTest(unittest.TestCase):
def setUp(self):
"""Create temporary directory for test files"""
self.test_dir = tempfile.mkdtemp()
self.sia_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
self.write_test_script()
def tearDown(self):
"""Clean up temporary test files"""
for file in Path(self.test_dir).glob('*'):
file.unlink()
os.rmdir(self.test_dir)
def write_test_script(self):
"""Create a Python script that uses StandardIOBuffer"""
script_path = os.path.join(self.test_dir, 'test_script.py')
with open(script_path, 'w', newline='') as f:
f.write(f'''
import sys
import os
import time
# Add the sia package directory to Python path
sys.path.insert(0, {repr(self.sia_path)})
from sia.standard_io_buffer import StandardIOBuffer
def run_buffer_test(test_type):
buffer = StandardIOBuffer()
if test_type == 'read':
content = buffer.read()
sys.stdout.buffer.write(f"Read: {{content}}".encode('utf-8'))
sys.stdout.buffer.flush()
elif test_type == 'write':
buffer.write("Test output")
sys.stdout.buffer.flush()
elif test_type == 'multiple_write':
buffer.write("Part 1")
buffer.write("Part 2")
sys.stdout.buffer.flush()
elif test_type == 'buffer_length':
buffer._input_buffer = "test"
sys.stdout.buffer.write(str(buffer.buffer_length()).encode('utf-8'))
sys.stdout.buffer.flush()
elif test_type == 'read_lines':
content1 = buffer.read()
sys.stdout.buffer.write(f"First read: {{content1}}".encode('utf-8'))
sys.stdout.buffer.flush()
time.sleep(0.1)
content2 = buffer.read()
sys.stdout.buffer.write(f"Second read: {{content2}}".encode('utf-8'))
sys.stdout.buffer.flush()
elif test_type == 'debug':
print("Python path:", sys.path)
print("Current directory:", os.getcwd())
print("Buffer import successful")
print("Python version:", sys.version)
print("Platform:", sys.platform)
if __name__ == '__main__':
if len(sys.argv) < 2:
print("Test type required")
sys.exit(1)
test_type = sys.argv[1]
run_buffer_test(test_type)
''')
self.script_path = script_path
def run_script(self, test_type, input_data=None):
"""Run test script with given input and return output"""
process = subprocess.Popen(
[sys.executable, self.script_path, test_type],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env={**os.environ, 'PYTHONPATH': self.sia_path},
# Use binary mode for consistent line endings
universal_newlines=False
)
try:
# Convert input data to bytes if provided
input_bytes = input_data.encode('utf-8') if input_data is not None else None
stdout_bytes, stderr_bytes = process.communicate(input=input_bytes, timeout=5)
# Decode output using utf-8, preserving special characters
stdout = stdout_bytes.decode('utf-8')
stderr = stderr_bytes.decode('utf-8')
return stdout, stderr, process.returncode
except subprocess.TimeoutExpired:
process.kill()
stdout_bytes, stderr_bytes = process.communicate()
stdout = stdout_bytes.decode('utf-8')
stderr = stderr_bytes.decode('utf-8')
return stdout, stderr, -1
def test_debug_environment(self):
"""Test that the environment is properly set up"""
stdout, stderr, code = self.run_script('debug')
self.assertEqual(code, 0, f"Debug script failed with stderr: {stderr}")
self.assertIn("Buffer import successful", stdout)
def test_read_available_input(self):
"""Test reading available input without mocks"""
stdout, stderr, code = self.run_script('read', 'test input\n')
self.assertEqual(code, 0, f"Script failed with stderr: {stderr}")
self.assertEqual(stdout, 'Read: test input\n')
def test_read_no_input(self):
"""Test reading when no input is provided"""
stdout, stderr, code = self.run_script('read')
self.assertEqual(code, 0, f"Script failed with stderr: {stderr}")
self.assertEqual(stdout, 'Read: ')
def test_write_output(self):
"""Test writing output without mocks"""
stdout, stderr, code = self.run_script('write')
self.assertEqual(code, 0, f"Script failed with stderr: {stderr}")
self.assertEqual(stdout, 'Test output')
def test_multiple_writes(self):
"""Test multiple write operations"""
stdout, stderr, code = self.run_script('multiple_write')
self.assertEqual(code, 0, f"Script failed with stderr: {stderr}")
self.assertEqual(stdout, 'Part 1Part 2')
def test_buffer_length(self):
"""Test buffer length reporting"""
stdout, stderr, code = self.run_script('buffer_length')
self.assertEqual(code, 0, f"Script failed with stderr: {stderr}")
self.assertEqual(stdout.strip(), '4')
def test_multiple_reads(self):
"""Test multiple read operations with actual input"""
input_data = 'line1\nline2\n'
stdout, stderr, code = self.run_script('read_lines', input_data)
self.assertEqual(code, 0, f"Script failed with stderr: {stderr}")
self.assertIn('First read: line1\nline2\n', stdout)
self.assertIn('Second read: ', stdout)
def test_special_characters(self):
"""Test handling special characters in input/output"""
# Use raw string to ensure exact character representation
special_chars = 'Special chars: \n\t\r\'"'
stdout, stderr, code = self.run_script('read', special_chars)
self.assertEqual(code, 0, f"Script failed with stderr: {stderr}")
expected = f'Read: {special_chars}'
# Print actual bytes for debugging
if stdout != expected:
print("\nDebug output:")
print("Expected bytes:", expected.encode('utf-8'))
print("Actual bytes:", stdout.encode('utf-8'))
self.assertEqual(stdout, expected)
def test_empty_input(self):
"""Test handling empty input"""
stdout, stderr, code = self.run_script('read', '')
self.assertEqual(code, 0, f"Script failed with stderr: {stderr}")
self.assertEqual(stdout, 'Read: ')
def test_large_input(self):
"""Test handling large input"""
large_input = 'x' * 10000
stdout, stderr, code = self.run_script('read', large_input)
self.assertEqual(code, 0, f"Script failed with stderr: {stderr}")
self.assertEqual(stdout, f'Read: {large_input}')

45
test/stop_command_test.py Normal file
View File

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

8
test/test_util.py
View File

@@ -1,8 +0,0 @@
import unittest
class SequentialTestLoader(unittest.TestLoader):
def getTestCaseNames(self, testCaseClass):
test_names = super().getTestCaseNames(testCaseClass)
testcase_methods = list(testCaseClass.__dict__.keys())
test_names.sort(key=testcase_methods.index)
return test_names

119
test/util_test.py
View File

@@ -1,17 +1,118 @@
import unittest
from typing import Iterator
from . import test_data
from sia import util
class UtilTest(unittest.TestCase):
def test_get_valid_root_elements_single(self):
valid_elements = util.get_valid_root_elements(test_data.echo_action_schema)
self.assertEqual(valid_elements, {'test_tag'})
def test_stop_before_value(self):
# Helper function to create iterator from list
def create_iterator(items: list) -> Iterator[str]:
for item in items:
yield item
def test_split_response_single_element(self):
response = "Some reasoning here\n<test_tag>content</test_tag>"
valid_elements = {'test_tag'}
result = util.split_response(response, valid_elements)
self.assertEqual(result.reasoning, "Some reasoning here")
self.assertEqual(result.actions, "<test_tag>content</test_tag>")
# Test case 1: Stop value in middle of sequence
input_sequence = ['hello', 'world', 'STOP', 'ignored']
result = list(util.stop_before_value(create_iterator(input_sequence), 'STOP'))
self.assertEqual(result, ['hello', 'world'])
# Test case 2: Stop value not in sequence
input_sequence = ['hello', 'world']
result = list(util.stop_before_value(create_iterator(input_sequence), 'STOP'))
self.assertEqual(result, ['hello', 'world'])
# Test case 3: Stop value at start of sequence
input_sequence = ['STOP', 'ignored']
result = list(util.stop_before_value(create_iterator(input_sequence), 'STOP'))
self.assertEqual(result, [])
# Test case 4: Stop value as part of an item
input_sequence = ['hello', 'woSTOPrld', 'ignored']
result = list(util.stop_before_value(create_iterator(input_sequence), 'STOP'))
self.assertEqual(result, ['hello', 'wo'])
# Test case 5: Empty sequence
input_sequence = []
result = list(util.stop_before_value(create_iterator(input_sequence), 'STOP'))
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]]>')

106
test/web_io_buffer_test.py Normal file
View File

@@ -0,0 +1,106 @@
import unittest
from sia.web_io_buffer import WebIOBuffer
class WebIOBufferTest(unittest.TestCase):
def setUp(self):
"""Set up a fresh WebIOBuffer for each test."""
self.buffer = WebIOBuffer()
def test_initialization(self):
"""Test initial state of buffer"""
self.assertEqual(self.buffer.buffer_length(), 0)
self.assertEqual(self.buffer.get_stdout(), "")
self.assertEqual(self.buffer.read(), "")
def test_append_stdin(self):
"""Test appending to stdin buffer"""
test_input = "test input"
self.buffer.append_stdin(test_input)
self.assertEqual(self.buffer.buffer_length(), len(test_input))
def test_read_clears_buffer(self):
"""Test that reading clears the stdin buffer"""
test_input = "test input"
self.buffer.append_stdin(test_input)
content = self.buffer.read()
self.assertEqual(content, test_input)
self.assertEqual(self.buffer.buffer_length(), 0)
self.assertEqual(self.buffer.read(), "")
def test_write_output(self):
"""Test writing to stdout buffer"""
test_output = "test output"
self.buffer.write(test_output)
self.assertEqual(self.buffer.get_stdout(), test_output)
def test_multiple_writes(self):
"""Test multiple write operations"""
self.buffer.write("Part 1")
self.buffer.write("Part 2")
self.assertEqual(self.buffer.get_stdout(), "Part 1Part 2")
def test_clear_stdout(self):
"""Test clearing stdout buffer"""
self.buffer.write("test output")
self.buffer.clear_stdout()
self.assertEqual(self.buffer.get_stdout(), "")
def test_multiple_stdin_appends(self):
"""Test multiple stdin append operations"""
self.buffer.append_stdin("Line 1\n")
self.buffer.append_stdin("Line 2\n")
content = self.buffer.read()
self.assertEqual(content, "Line 1\nLine 2\n")
def test_empty_write(self):
"""Test writing empty content"""
self.buffer.write("")
self.assertEqual(self.buffer.get_stdout(), "")
def test_empty_append_stdin(self):
"""Test appending empty content to stdin"""
self.buffer.append_stdin("")
self.assertEqual(self.buffer.buffer_length(), 0)
def test_special_characters(self):
"""Test handling special characters"""
special_chars = "Special chars: \n\t\r\'\"\\"
self.buffer.append_stdin(special_chars)
content = self.buffer.read()
self.assertEqual(content, special_chars)
self.buffer.write(special_chars)
self.assertEqual(self.buffer.get_stdout(), special_chars)
def test_large_content(self):
"""Test handling large content"""
large_input = "x" * 10000
self.buffer.append_stdin(large_input)
content = self.buffer.read()
self.assertEqual(content, large_input)
self.buffer.write(large_input)
self.assertEqual(self.buffer.get_stdout(), large_input)
def test_multiple_operations(self):
"""Test mixed read/write operations"""
# Write some output
self.buffer.write("Output 1\n")
self.buffer.write("Output 2\n")
# Add some input
self.buffer.append_stdin("Input 1\n")
self.buffer.append_stdin("Input 2\n")
# Verify stdout accumulated correctly
self.assertEqual(self.buffer.get_stdout(), "Output 1\nOutput 2\n")
# Read input and verify it's cleared
content = self.buffer.read()
self.assertEqual(content, "Input 1\nInput 2\n")
self.assertEqual(self.buffer.buffer_length(), 0)
# Clear stdout and verify
self.buffer.clear_stdout()
self.assertEqual(self.buffer.get_stdout(), "")

246
test/working_memory_test.py Normal file
View File

@@ -0,0 +1,246 @@
from datetime import datetime
import time
import unittest
import xml.etree.ElementTree as ET
from sia.background_entry import BackgroundEntry
from sia.entry import Entry
from sia.parse_error_entry import ParseErrorEntry
from sia.read_entry import ReadEntry
from sia.reasoning_entry import ReasoningEntry
from sia.web_io_buffer import WebIOBuffer
from sia.working_memory import WorkingMemory
from sia.write_entry import WriteEntry
class MockEntry(Entry):
"""Mock entry class for testing"""
def __init__(self, id: str, timestamp: datetime):
super().__init__(id, timestamp)
self.updated = False
def update(self) -> None:
self.updated = True
def generate_context(self) -> ET.Element:
elem = ET.Element("mock", {"id": self.id})
elem.text = "<![CDATA[mock content]]>"
return elem
class WorkingMemoryTest(unittest.TestCase):
def setUp(self):
"""Set up test cases with a fresh working memory"""
self.memory = WorkingMemory()
self.test_timestamp = datetime(2024, 10, 31, 12, 0, 0)
def test_initialization(self):
"""Test initial state of working memory"""
self.assertEqual(self.memory.get_entries_count(), 0)
self.assertEqual(self.memory.generate_context(), [])
def test_add_entry(self):
"""Test adding entries"""
entry = MockEntry("test-id-1", self.test_timestamp)
self.memory.add_entry(entry)
self.assertEqual(self.memory.get_entries_count(), 1)
self.assertEqual(self.memory.get_entry("test-id-1"), entry)
def test_add_invalid_entry(self):
"""Test adding invalid entry type"""
with self.assertRaises(TypeError):
self.memory.add_entry("not an entry")
def test_remove_entry(self):
"""Test removing entries"""
entry1 = MockEntry("test-id-1", self.test_timestamp)
entry2 = MockEntry("test-id-2", self.test_timestamp)
self.memory.add_entry(entry1)
self.memory.add_entry(entry2)
self.memory.remove_entry("test-id-1")
self.assertEqual(self.memory.get_entries_count(), 1)
self.assertIsNone(self.memory.get_entry("test-id-1"))
self.assertEqual(self.memory.get_entry("test-id-2"), entry2)
def test_remove_nonexistent_entry(self):
"""Test removing entry that doesn't exist"""
entry = MockEntry("test-id-1", self.test_timestamp)
self.memory.add_entry(entry)
self.memory.remove_entry("nonexistent-id")
self.assertEqual(self.memory.get_entries_count(), 1)
self.assertEqual(self.memory.get_entry("test-id-1"), entry)
def test_update_entries(self):
"""Test updating all entries"""
entries = [
MockEntry(f"test-id-{i}", self.test_timestamp)
for i in range(3)
]
for entry in entries:
self.memory.add_entry(entry)
self.memory.update()
for entry in entries:
self.assertTrue(entry.updated)
def test_generate_context(self):
"""Test generating XML context"""
entry1 = MockEntry("test-id-1", self.test_timestamp)
entry2 = MockEntry("test-id-2", self.test_timestamp)
self.memory.add_entry(entry1)
self.memory.add_entry(entry2)
context = self.memory.generate_context()
self.assertEqual(len(context), 2)
self.assertEqual(context[0].tag, "mock")
self.assertEqual(context[0].get("id"), "test-id-1")
self.assertEqual(context[1].tag, "mock")
self.assertEqual(context[1].get("id"), "test-id-2")
def test_get_entries_by_type(self):
"""Test retrieving entries by type"""
# Create IO buffer for IO entries
io_buffer = WebIOBuffer()
# Add different types of entries
entries = [
ReasoningEntry("test reasoning", "reasoning-1", self.test_timestamp),
ParseErrorEntry("bad content", "error msg", "error-1", self.test_timestamp),
ReadEntry(io_buffer, "read-1", self.test_timestamp),
WriteEntry("test output", io_buffer, "write-1", self.test_timestamp)
]
for entry in entries:
self.memory.add_entry(entry)
# Test filtering by each type
reasoning_entries = self.memory.get_entries_by_type(ReasoningEntry)
self.assertEqual(len(reasoning_entries), 1)
self.assertIsInstance(reasoning_entries[0], ReasoningEntry)
error_entries = self.memory.get_entries_by_type(ParseErrorEntry)
self.assertEqual(len(error_entries), 1)
self.assertIsInstance(error_entries[0], ParseErrorEntry)
read_entries = self.memory.get_entries_by_type(ReadEntry)
self.assertEqual(len(read_entries), 1)
self.assertIsInstance(read_entries[0], ReadEntry)
write_entries = self.memory.get_entries_by_type(WriteEntry)
self.assertEqual(len(write_entries), 1)
self.assertIsInstance(write_entries[0], WriteEntry)
def test_empty_context_generation(self):
"""Test context generation with no entries"""
context = self.memory.generate_context()
self.assertEqual(context, [])
def test_get_nonexistent_entry(self):
"""Test retrieving entry that doesn't exist"""
self.assertIsNone(self.memory.get_entry("nonexistent-id"))
def test_remove_entry_cleanup(self):
"""Test that removing an entry triggers cleanup"""
# Create a background process entry
entry = BackgroundEntry("sleep 10", "test-id", datetime(2024, 10, 31, 12, 0, 0))
self.memory.add_entry(entry)
# Start the process
entry.update()
self.assertIsNotNone(entry.process)
process_pid = entry.process.pid
# Remove entry should trigger cleanup
self.memory.remove_entry("test-id")
# Verify process was terminated
time.sleep(0.1) # Give process time to terminate
with self.assertRaises(ProcessLookupError):
import os
os.kill(process_pid, 0)
def test_cleanup_on_memory_clear(self):
"""Test that clearing memory properly cleans up all entries"""
# Add multiple background processes
processes = []
for i in range(3):
entry = BackgroundEntry(
"sleep 10",
f"id-{i}",
datetime(2024, 10, 31, 12, 0, 0)
)
self.memory.add_entry(entry)
entry.update()
processes.append(entry.process.pid)
# Clear memory
self.memory.clear()
# Verify all processes were terminated
time.sleep(0.1) # Give processes time to terminate
for pid in processes:
with self.assertRaises(ProcessLookupError):
import os
os.kill(pid, 0)
self.assertEqual(self.memory.get_entries_count(), 0)
def test_cleanup_on_memory_deletion(self):
"""Test that deleting memory properly cleans up all entries"""
# Add a background process
entry = BackgroundEntry(
"sleep 10",
"test-id",
datetime(2024, 10, 31, 12, 0, 0)
)
self.memory.add_entry(entry)
entry.update()
process_pid = entry.process.pid
# Delete memory
del self.memory
# Verify process was terminated
time.sleep(0.1) # Give process time to terminate
with self.assertRaises(ProcessLookupError):
import os
os.kill(process_pid, 0)
def test_get_entries(self):
"""Test getting all entries"""
# Add multiple entries
entries = [
ReasoningEntry(f"content {i}", f"id-{i}", self.test_timestamp)
for i in range(3)
]
for entry in entries:
self.memory.add_entry(entry)
# Get all entries
retrieved_entries = self.memory.get_entries()
# Verify count and contents
self.assertEqual(len(retrieved_entries), len(entries))
for entry in entries:
self.assertIn(entry, retrieved_entries)
def test_get_entries_returns_copy(self):
"""Test that get_entries returns a copy of the list"""
# Add an entry
entry = ReasoningEntry("test content", "test-id", self.test_timestamp)
self.memory.add_entry(entry)
# Get entries and modify the returned list
entries = self.memory.get_entries()
entries.clear()
# Verify original memory is unchanged
self.assertEqual(self.memory.get_entries_count(), 1)
self.assertIsNotNone(self.memory.get_entry("test-id"))

117
test/write_entry_test.py Normal file
View File

@@ -0,0 +1,117 @@
import unittest
from datetime import datetime
from sia.web_io_buffer import WebIOBuffer
from sia.write_entry import WriteEntry
class WriteEntryTest(unittest.TestCase):
def setUp(self):
"""Set up test cases with fixed id, timestamp, and IO buffer"""
self.test_id = "test-id-1234"
self.test_timestamp = datetime(2024, 10, 31, 12, 0, 0)
self.io_buffer = WebIOBuffer()
def test_initialization(self):
"""Test entry initialization"""
content = "test message"
entry = WriteEntry(content, self.io_buffer, self.test_id, self.test_timestamp)
self.assertEqual(entry.content, content)
self.assertEqual(entry.id, self.test_id)
self.assertEqual(entry.timestamp, self.test_timestamp)
self.assertFalse(entry._written)
self.assertEqual(self.io_buffer.get_stdout(), "")
def test_single_write(self):
"""Test writing content once"""
content = "test message"
entry = WriteEntry(content, self.io_buffer, self.test_id, self.test_timestamp)
# Perform update and verify content was written
entry.update()
self.assertEqual(self.io_buffer.get_stdout(), content)
self.assertTrue(entry._written)
def test_multiple_updates(self):
"""Test that content is only written once even with multiple updates"""
content = "test message"
entry = WriteEntry(content, self.io_buffer, self.test_id, self.test_timestamp)
# Perform multiple updates
entry.update()
initial_output = self.io_buffer.get_stdout()
self.io_buffer.clear_stdout()
entry.update()
entry.update()
# Verify no additional content was written
self.assertEqual(self.io_buffer.get_stdout(), "")
self.assertTrue(entry._written)
self.assertEqual(initial_output, content)
def test_empty_content(self):
"""Test writing empty content"""
entry = WriteEntry("", self.io_buffer, self.test_id, self.test_timestamp)
entry.update()
self.assertEqual(self.io_buffer.get_stdout(), "")
self.assertTrue(entry._written)
def test_special_characters(self):
"""Test writing content with special characters"""
content = "Special chars: \n\t\r\'\"\\"
entry = WriteEntry(content, self.io_buffer, self.test_id, self.test_timestamp)
entry.update()
self.assertEqual(self.io_buffer.get_stdout(), content)
def test_large_content(self):
"""Test writing large content"""
content = "x" * 10000
entry = WriteEntry(content, self.io_buffer, self.test_id, self.test_timestamp)
entry.update()
self.assertEqual(self.io_buffer.get_stdout(), content)
def test_generate_context(self):
"""Test XML context generation"""
content = "test message"
entry = WriteEntry(content, self.io_buffer, self.test_id, self.test_timestamp)
# Generate context before writing
element = entry.generate_context()
# Verify XML structure
self.assertEqual(element.tag, "write_stdout")
self.assertEqual(element.get("id"), self.test_id)
self.assertEqual(element.text, f"<![CDATA[{content}]]>")
# Write content and verify context remains the same
entry.update()
element_after = entry.generate_context()
self.assertEqual(element_after.tag, "write_stdout")
self.assertEqual(element_after.get("id"), self.test_id)
self.assertEqual(element_after.text, f"<![CDATA[{content}]]>")
def test_unicode_content(self):
"""Test writing Unicode content"""
content = "Hello 世界 😊"
entry = WriteEntry(content, self.io_buffer, self.test_id, self.test_timestamp)
entry.update()
self.assertEqual(self.io_buffer.get_stdout(), content)
def test_multiline_content(self):
"""Test writing multiline content"""
content = """Line 1
Line 2
Line 3"""
entry = WriteEntry(content, self.io_buffer, self.test_id, self.test_timestamp)
entry.update()
self.assertEqual(self.io_buffer.get_stdout(), content)
# Verify XML escaping for multiline content
element = entry.generate_context()
self.assertEqual(element.text, f"<![CDATA[{content}]]>")

116
test/xml_validator_test.py Normal file
View File

@@ -0,0 +1,116 @@
import unittest
from sia.xml_validator import XMLValidator
class XMLValidatorTest(unittest.TestCase):
def setUp(self):
"""Set up test schema"""
self.test_schema = """<?xml version="1.0" encoding="UTF-8"?>
<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema">
<xs:element name="test_element">
<xs:complexType>
<xs:simpleContent>
<xs:extension base="xs:string">
<xs:attribute name="id" type="xs:string" use="required"/>
<xs:attribute name="count" type="xs:integer"/>
</xs:extension>
</xs:simpleContent>
</xs:complexType>
</xs:element>
<xs:element name="simple_element">
<xs:complexType>
<xs:simpleContent>
<xs:extension base="xs:string"/>
</xs:simpleContent>
</xs:complexType>
</xs:element>
</xs:schema>"""
self.validator = XMLValidator(self.test_schema)
def test_initialization(self):
"""Test validator initialization and root element extraction"""
valid_elements = self.validator.get_valid_root_elements()
self.assertEqual(valid_elements, {'test_element', 'simple_element'})
def test_invalid_schema(self):
"""Test handling invalid schema"""
with self.assertRaises(ValueError):
XMLValidator("invalid schema")
def test_valid_xml(self):
"""Test validation of valid XML"""
xml = '<test_element id="123" count="42">test content</test_element>'
result = self.validator.validate(xml)
self.assertIsNone(result)
# Test without optional attribute
xml = '<test_element id="123">test content</test_element>'
result = self.validator.validate(xml)
self.assertIsNone(result)
# Test simple element
xml = '<simple_element>test content</simple_element>'
result = self.validator.validate(xml)
self.assertIsNone(result)
def test_invalid_root_element(self):
"""Test validation with invalid root element"""
xml = '<invalid_element>test content</invalid_element>'
result = self.validator.validate(xml)
self.assertIsNotNone(result)
self.assertIn("Invalid root element", result)
def test_missing_required_attribute(self):
"""Test validation with missing required attribute"""
xml = '<test_element>test content</test_element>'
result = self.validator.validate(xml)
self.assertIsNotNone(result)
self.assertIn("Missing required attribute", result)
def test_invalid_integer_attribute(self):
"""Test validation with invalid integer attribute"""
xml = '<test_element id="123" count="not_a_number">test content</test_element>'
result = self.validator.validate(xml)
self.assertIsNotNone(result)
self.assertIn("Invalid integer value", result)
def test_unexpected_attribute(self):
"""Test validation with unexpected attribute"""
xml = '<test_element id="123" invalid_attr="value">test content</test_element>'
result = self.validator.validate(xml)
self.assertIsNotNone(result)
self.assertIn("Unexpected attribute", result)
def test_invalid_xml_syntax(self):
"""Test validation with invalid XML syntax"""
xml = '<test_element>unclosed tag'
result = self.validator.validate(xml)
self.assertIsNotNone(result)
self.assertIn("Invalid XML", result)
def test_whitespace_handling(self):
"""Test validation with whitespace in XML"""
xml = """
<test_element id="123">
test content
</test_element>
"""
result = self.validator.validate(xml)
self.assertIsNone(result)
def test_empty_content(self):
"""Test validation with empty element content"""
xml = '<test_element id="123"></test_element>'
result = self.validator.validate(xml)
self.assertIsNone(result)
xml = '<test_element id="123"/>'
result = self.validator.validate(xml)
self.assertIsNone(result)
def test_special_characters(self):
"""Test validation with special characters"""
xml = '<test_element id="123"><![CDATA[Special & < > " \' chars]]></test_element>'
result = self.validator.validate(xml)
self.assertIsNone(result)