175 lines
6.6 KiB
Python
175 lines
6.6 KiB
Python
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}'
|
|
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}') |