122 lines
4.8 KiB
Python
122 lines
4.8 KiB
Python
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()
|