diff --git a/Dockerfile b/Dockerfile index daca305..17d0b00 100644 --- a/Dockerfile +++ b/Dockerfile @@ -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 diff --git a/readme.md b/readme.md index 3fee60b..bfb3c71 100644 --- a/readme.md +++ b/readme.md @@ -318,7 +318,7 @@ stateDiagram-v2 ```mermaid classDiagram - direction LR + direction LR class BaseAgent { <> #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 { - <> - UPDATE - CONTEXT_APPROVAL - INFERENCE - RESPONSE_APPROVAL + <> + 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 <> +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 { <> - +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 diff --git a/requirements.txt b/requirements.txt index ec4f350..ed79fe8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,3 @@ bs4 -docker torch transformers \ No newline at end of file diff --git a/sia/__main__.py b/sia/__main__.py index 56a382e..7d686fa 100644 --- a/sia/__main__.py +++ b/sia/__main__.py @@ -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() \ No newline at end of file diff --git a/sia/agent_core.py b/sia/agent_core.py deleted file mode 100644 index faa5bc1..0000000 --- a/sia/agent_core.py +++ /dev/null @@ -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}") \ No newline at end of file diff --git a/sia/background_entry.py b/sia/background_entry.py new file mode 100644 index 0000000..3d87aa0 --- /dev/null +++ b/sia/background_entry.py @@ -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() \ No newline at end of file diff --git a/sia/command.py b/sia/command.py new file mode 100644 index 0000000..883854c --- /dev/null +++ b/sia/command.py @@ -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 diff --git a/sia/command_result.py b/sia/command_result.py new file mode 100644 index 0000000..331172e --- /dev/null +++ b/sia/command_result.py @@ -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) \ No newline at end of file diff --git a/sia/context_template.py b/sia/context_template.py deleted file mode 100644 index d428661..0000000 --- a/sia/context_template.py +++ /dev/null @@ -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 diff --git a/sia/delete_command.py b/sia/delete_command.py new file mode 100644 index 0000000..e17bb9f --- /dev/null +++ b/sia/delete_command.py @@ -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() \ No newline at end of file diff --git a/sia/docker_module.py b/sia/docker_module.py deleted file mode 100644 index 305deb2..0000000 --- a/sia/docker_module.py +++ /dev/null @@ -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 \ No newline at end of file diff --git a/sia/entry.py b/sia/entry.py new file mode 100644 index 0000000..acbdc67 --- /dev/null +++ b/sia/entry.py @@ -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 \ No newline at end of file diff --git a/sia/inference_result.py b/sia/inference_result.py deleted file mode 100644 index 92b3796..0000000 --- a/sia/inference_result.py +++ /dev/null @@ -1,5 +0,0 @@ -from typing import NamedTuple - -class InferenceResult(NamedTuple): - reasoning: str - actions: str \ No newline at end of file diff --git a/sia/parse_error_entry.py b/sia/parse_error_entry.py new file mode 100644 index 0000000..c90e89e --- /dev/null +++ b/sia/parse_error_entry.py @@ -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 \ No newline at end of file diff --git a/sia/read_entry.py b/sia/read_entry.py new file mode 100644 index 0000000..a5275da --- /dev/null +++ b/sia/read_entry.py @@ -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 \ No newline at end of file diff --git a/sia/reasoning_entry.py b/sia/reasoning_entry.py new file mode 100644 index 0000000..eca4d59 --- /dev/null +++ b/sia/reasoning_entry.py @@ -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 \ No newline at end of file diff --git a/sia/repeat_entry.py b/sia/repeat_entry.py new file mode 100644 index 0000000..9476862 --- /dev/null +++ b/sia/repeat_entry.py @@ -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 \ No newline at end of file diff --git a/sia/single_shot_entry.py b/sia/single_shot_entry.py new file mode 100644 index 0000000..908dfea --- /dev/null +++ b/sia/single_shot_entry.py @@ -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 \ No newline at end of file diff --git a/sia/standard_io_buffer.py b/sia/standard_io_buffer.py new file mode 100644 index 0000000..5b93944 --- /dev/null +++ b/sia/standard_io_buffer.py @@ -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) \ No newline at end of file diff --git a/sia/stop_command.py b/sia/stop_command.py new file mode 100644 index 0000000..d66adc2 --- /dev/null +++ b/sia/stop_command.py @@ -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() \ No newline at end of file diff --git a/sia/util.py b/sia/util.py index a87f571..c9ae6e2 100644 --- a/sia/util.py +++ b/sia/util.py @@ -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 \ No newline at end of file + 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('&', '&') \ + .replace('<', '<') \ + .replace('>', '>') \ + .replace('"', '"') \ + .replace("'", ''') + + # Otherwise wrap in CDATA + return f'' \ No newline at end of file diff --git a/sia/web_io_buffer.py b/sia/web_io_buffer.py new file mode 100644 index 0000000..bcb60d1 --- /dev/null +++ b/sia/web_io_buffer.py @@ -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 = "" \ No newline at end of file diff --git a/sia/working_memory.py b/sia/working_memory.py new file mode 100644 index 0000000..ef7ee29 --- /dev/null +++ b/sia/working_memory.py @@ -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] \ No newline at end of file diff --git a/sia/write_entry.py b/sia/write_entry.py new file mode 100644 index 0000000..9a1d1f3 --- /dev/null +++ b/sia/write_entry.py @@ -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 \ No newline at end of file diff --git a/sia/xml_validator.py b/sia/xml_validator.py new file mode 100644 index 0000000..f25d5c2 --- /dev/null +++ b/sia/xml_validator.py @@ -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() \ No newline at end of file diff --git a/test/background_entry_test.py b/test/background_entry_test.py new file mode 100644 index 0000000..c56d2f9 --- /dev/null +++ b/test/background_entry_test.py @@ -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("")) + + 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) \ No newline at end of file diff --git a/test/context_template_test.py b/test/context_template_test.py deleted file mode 100644 index bb9577a..0000000 --- a/test/context_template_test.py +++ /dev/null @@ -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) - self.assertIn("", context) - self.assertIn("", 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) - self.assertIn("", context) - -if __name__ == '__main__': - unittest.main() diff --git a/test/delete_command_test.py b/test/delete_command_test.py new file mode 100644 index 0000000..590acfd --- /dev/null +++ b/test/delete_command_test.py @@ -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) \ No newline at end of file diff --git a/test/docker_module_test.py b/test/docker_module_test.py deleted file mode 100644 index 5841866..0000000 --- a/test/docker_module_test.py +++ /dev/null @@ -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() diff --git a/test/llm_engine_test.py b/test/llm_engine_test.py index 93b89c6..a11b9c5 100644 --- a/test/llm_engine_test.py +++ b/test/llm_engine_test.py @@ -3,7 +3,6 @@ import unittest from itertools import tee from . import test_data -from . import test_util from sia.llm_engine import LlmEngine diff --git a/test/parse_error_entry_test.py b/test/parse_error_entry_test.py new file mode 100644 index 0000000..788371d --- /dev/null +++ b/test/parse_error_entry_test.py @@ -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"") + + # Verify content element + content_elem = element.find("content") + self.assertIsNotNone(content_elem) + self.assertEqual(content_elem.text, f"") + + 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"") + self.assertEqual(element.find("error").text, f"") + + 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, "") + self.assertEqual(element.find("error").text, "") + + 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"") + self.assertEqual(element.find("error").text, f"") \ No newline at end of file diff --git a/test/read_entry_test.py b/test/read_entry_test.py new file mode 100644 index 0000000..a24fa2b --- /dev/null +++ b/test/read_entry_test.py @@ -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"") + + 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"") \ No newline at end of file diff --git a/test/reasoning_entry_test.py b/test/reasoning_entry_test.py new file mode 100644 index 0000000..8041f1d --- /dev/null +++ b/test/reasoning_entry_test.py @@ -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"") + + 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"") + + 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, "") + + 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"") + + 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"") \ No newline at end of file diff --git a/test/repeat_entry_test.py b/test/repeat_entry_test.py new file mode 100644 index 0000000..18c97dd --- /dev/null +++ b/test/repeat_entry_test.py @@ -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"") + 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("")) + + self.assertEqual(element.find("stderr").text, "") + + 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, "") \ No newline at end of file diff --git a/test/single_shot_entry_test.py b/test/single_shot_entry_test.py new file mode 100644 index 0000000..ac7a08a --- /dev/null +++ b/test/single_shot_entry_test.py @@ -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, "") + + # 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"") + + # Get stdout and remove trailing newline for comparison + stdout_text = element.find("stdout").text + self.assertTrue(stdout_text.startswith("")) + + self.assertEqual(element.find("stderr").text, "") + 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") \ No newline at end of file diff --git a/test/standard_io_buffer_test.py b/test/standard_io_buffer_test.py new file mode 100644 index 0000000..4a8b5f5 --- /dev/null +++ b/test/standard_io_buffer_test.py @@ -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}') \ No newline at end of file diff --git a/test/stop_command_test.py b/test/stop_command_test.py new file mode 100644 index 0000000..c7a0e21 --- /dev/null +++ b/test/stop_command_test.py @@ -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) \ No newline at end of file diff --git a/test/test_util.py b/test/test_util.py deleted file mode 100644 index 695c61f..0000000 --- a/test/test_util.py +++ /dev/null @@ -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 \ No newline at end of file diff --git a/test/util_test.py b/test/util_test.py index bd4ed8d..fc2aac7 100644 --- a/test/util_test.py +++ b/test/util_test.py @@ -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\ncontent" - valid_elements = {'test_tag'} - result = util.split_response(response, valid_elements) - self.assertEqual(result.reasoning, "Some reasoning here") - self.assertEqual(result.actions, "content") \ No newline at end of file + # 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), '') + self.assertEqual(util.escape_text_for_xml(3.14), '') + self.assertEqual(util.escape_text_for_xml(-100), '') + self.assertEqual(util.escape_text_for_xml(0), '') + + def test_simple_strings(self): + """Test that simple strings are wrapped in CDATA""" + self.assertEqual(util.escape_text_for_xml('hello'), '') + self.assertEqual(util.escape_text_for_xml(''), '') + self.assertEqual(util.escape_text_for_xml(' '), '') + + 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(''), ']]>') + self.assertEqual(util.escape_text_for_xml('a & b'), '') + self.assertEqual(util.escape_text_for_xml('"quoted"'), '') + self.assertEqual(util.escape_text_for_xml("'single'"), "") + self.assertEqual(util.escape_text_for_xml('') + + 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(']]>'), ']]>') + + # CDATA end sequence with other special characters + self.assertEqual( + util.escape_text_for_xml('Text with ]]> and & ampersands'), + 'Text with ]]> and <tags> & ampersands' + ) + + # Multiple CDATA end sequences + self.assertEqual( + util.escape_text_for_xml('Multiple ]]> end ]]> sequences'), + 'Multiple ]]> end ]]> 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), '') + + multiline_with_cdata = """Line 1 + Line ]]> 2 + Line 3""" + self.assertEqual( + util.escape_text_for_xml(multiline_with_cdata), + 'Line 1\n Line ]]> 2\n Line 3' + ) + + def test_edge_cases(self): + """Test edge cases and unusual inputs""" + # Empty string + self.assertEqual(util.escape_text_for_xml(''), '') + + # String with only whitespace + self.assertEqual(util.escape_text_for_xml('\n\t '), '') + + # String with Unicode characters + self.assertEqual(util.escape_text_for_xml('Hello δΈ–η•Œ'), '') + + # String with control characters + self.assertEqual(util.escape_text_for_xml('Hello\0World'), '') + + # Very long string + long_string = 'x' * 1000 + self.assertEqual(util.escape_text_for_xml(long_string), f'') + + def test_boolean_values(self): + """Test boolean values are properly converted""" + self.assertEqual(util.escape_text_for_xml(True), '') + self.assertEqual(util.escape_text_for_xml(False), '') \ No newline at end of file diff --git a/test/web_io_buffer_test.py b/test/web_io_buffer_test.py new file mode 100644 index 0000000..8ecefb2 --- /dev/null +++ b/test/web_io_buffer_test.py @@ -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(), "") \ No newline at end of file diff --git a/test/working_memory_test.py b/test/working_memory_test.py new file mode 100644 index 0000000..e933406 --- /dev/null +++ b/test/working_memory_test.py @@ -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 = "" + 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")) \ No newline at end of file diff --git a/test/write_entry_test.py b/test/write_entry_test.py new file mode 100644 index 0000000..af0a2d6 --- /dev/null +++ b/test/write_entry_test.py @@ -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"") + + # 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"") + + 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"") \ No newline at end of file diff --git a/test/xml_validator_test.py b/test/xml_validator_test.py new file mode 100644 index 0000000..faf5198 --- /dev/null +++ b/test/xml_validator_test.py @@ -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 = """ + + + + + + + + + + + + + + + + + + + """ + + 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 content' + result = self.validator.validate(xml) + self.assertIsNone(result) + + # Test without optional attribute + xml = 'test content' + result = self.validator.validate(xml) + self.assertIsNone(result) + + # Test simple element + xml = 'test content' + result = self.validator.validate(xml) + self.assertIsNone(result) + + def test_invalid_root_element(self): + """Test validation with invalid root element""" + xml = 'test content' + 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 content' + 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 content' + 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 content' + 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 = '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 content + + """ + result = self.validator.validate(xml) + self.assertIsNone(result) + + def test_empty_content(self): + """Test validation with empty element content""" + xml = '' + result = self.validator.validate(xml) + self.assertIsNone(result) + + xml = '' + result = self.validator.validate(xml) + self.assertIsNone(result) + + def test_special_characters(self): + """Test validation with special characters""" + xml = ' " \' chars]]>' + result = self.validator.validate(xml) + self.assertIsNone(result) \ No newline at end of file