from docker.utils.socket import next_frame_size from docker.utils.socket import read_exactly from .base import BaseAPIIntegrationTest, BUSYBOX from ..helpers import ( requires_api_version, ctrl_with, assert_cat_socket_detached_with_keys ) class ExecTest(BaseAPIIntegrationTest): def test_execute_command(self): container = self.client.create_container(BUSYBOX, 'cat', detach=True, stdin_open=True) id = container['Id'] self.client.start(id) self.tmp_containers.append(id) res = self.client.exec_create(id, ['echo', 'hello']) assert 'Id' in res exec_log = self.client.exec_start(res) assert exec_log == b'hello\n' def test_exec_command_string(self): container = self.client.create_container(BUSYBOX, 'cat', detach=True, stdin_open=True) id = container['Id'] self.client.start(id) self.tmp_containers.append(id) res = self.client.exec_create(id, 'echo hello world') assert 'Id' in res exec_log = self.client.exec_start(res) assert exec_log == b'hello world\n' def test_exec_command_as_user(self): container = self.client.create_container(BUSYBOX, 'cat', detach=True, stdin_open=True) id = container['Id'] self.client.start(id) self.tmp_containers.append(id) res = self.client.exec_create(id, 'whoami', user='default') assert 'Id' in res exec_log = self.client.exec_start(res) assert exec_log == b'default\n' def test_exec_command_as_root(self): container = self.client.create_container(BUSYBOX, 'cat', detach=True, stdin_open=True) id = container['Id'] self.client.start(id) self.tmp_containers.append(id) res = self.client.exec_create(id, 'whoami') assert 'Id' in res exec_log = self.client.exec_start(res) assert exec_log == b'root\n' def test_exec_command_streaming(self): container = self.client.create_container(BUSYBOX, 'cat', detach=True, stdin_open=True) id = container['Id'] self.tmp_containers.append(id) self.client.start(id) exec_id = self.client.exec_create(id, ['echo', 'hello\nworld']) assert 'Id' in exec_id res = b'' for chunk in self.client.exec_start(exec_id, stream=True): res += chunk assert res == b'hello\nworld\n' def test_exec_start_socket(self): container = self.client.create_container(BUSYBOX, 'cat', detach=True, stdin_open=True) container_id = container['Id'] self.client.start(container_id) self.tmp_containers.append(container_id) line = 'yay, interactive exec!' # `echo` appends CRLF, `printf` doesn't exec_id = self.client.exec_create( container_id, ['printf', line], tty=True) assert 'Id' in exec_id socket = self.client.exec_start(exec_id, socket=True) self.addCleanup(socket.close) next_size = next_frame_size(socket) assert next_size == len(line) data = read_exactly(socket, next_size) assert data.decode('utf-8') == line def test_exec_start_detached(self): container = self.client.create_container(BUSYBOX, 'cat', detach=True, stdin_open=True) container_id = container['Id'] self.client.start(container_id) self.tmp_containers.append(container_id) exec_id = self.client.exec_create( container_id, ['printf', "asdqwe"]) assert 'Id' in exec_id response = self.client.exec_start(exec_id, detach=True) assert response == "" def test_exec_inspect(self): container = self.client.create_container(BUSYBOX, 'cat', detach=True, stdin_open=True) id = container['Id'] self.client.start(id) self.tmp_containers.append(id) exec_id = self.client.exec_create(id, ['mkdir', '/does/not/exist']) assert 'Id' in exec_id self.client.exec_start(exec_id) exec_info = self.client.exec_inspect(exec_id) assert 'ExitCode' in exec_info assert exec_info['ExitCode'] != 0 @requires_api_version('1.25') def test_exec_command_with_env(self): container = self.client.create_container(BUSYBOX, 'cat', detach=True, stdin_open=True) id = container['Id'] self.client.start(id) self.tmp_containers.append(id) res = self.client.exec_create(id, 'env', environment=["X=Y"]) assert 'Id' in res exec_log = self.client.exec_start(res) assert b'X=Y\n' in exec_log @requires_api_version('1.35') def test_exec_command_with_workdir(self): container = self.client.create_container( BUSYBOX, 'cat', detach=True, stdin_open=True ) self.tmp_containers.append(container) self.client.start(container) res = self.client.exec_create(container, 'pwd', workdir='/var/www') exec_log = self.client.exec_start(res) assert exec_log == b'/var/www\n' def test_detach_with_default(self): container = self.client.create_container( BUSYBOX, 'cat', detach=True, stdin_open=True ) id = container['Id'] self.client.start(id) self.tmp_containers.append(id) exec_id = self.client.exec_create( id, 'cat', stdin=True, tty=True, stdout=True ) sock = self.client.exec_start(exec_id, tty=True, socket=True) self.addCleanup(sock.close) assert_cat_socket_detached_with_keys( sock, [ctrl_with('p'), ctrl_with('q')] ) def test_detach_with_config_file(self): self.client._general_configs['detachKeys'] = 'ctrl-p' container = self.client.create_container( BUSYBOX, 'cat', detach=True, stdin_open=True ) id = container['Id'] self.client.start(id) self.tmp_containers.append(id) exec_id = self.client.exec_create( id, 'cat', stdin=True, tty=True, stdout=True ) sock = self.client.exec_start(exec_id, tty=True, socket=True) self.addCleanup(sock.close) assert_cat_socket_detached_with_keys(sock, [ctrl_with('p')]) def test_detach_with_arg(self): self.client._general_configs['detachKeys'] = 'ctrl-p' container = self.client.create_container( BUSYBOX, 'cat', detach=True, stdin_open=True ) id = container['Id'] self.client.start(id) self.tmp_containers.append(id) exec_id = self.client.exec_create( id, 'cat', stdin=True, tty=True, detach_keys='ctrl-x', stdout=True ) sock = self.client.exec_start(exec_id, tty=True, socket=True) self.addCleanup(sock.close) assert_cat_socket_detached_with_keys(sock, [ctrl_with('x')])