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)