summaryrefslogtreecommitdiff
path: root/tests/integration/api_exec_test.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/integration/api_exec_test.py')
-rw-r--r--tests/integration/api_exec_test.py181
1 files changed, 144 insertions, 37 deletions
diff --git a/tests/integration/api_exec_test.py b/tests/integration/api_exec_test.py
index 1a5a4e5..554e862 100644
--- a/tests/integration/api_exec_test.py
+++ b/tests/integration/api_exec_test.py
@@ -1,15 +1,54 @@
-from docker.utils.socket import next_frame_size
+from ..helpers import assert_cat_socket_detached_with_keys
+from ..helpers import ctrl_with
+from ..helpers import requires_api_version
+from .base import BaseAPIIntegrationTest
+from .base import TEST_IMG
+from docker.utils.proxy import ProxyConfig
+from docker.utils.socket import next_frame_header
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_with_proxy_env(self):
+ # Set a custom proxy config on the client
+ self.client._proxy_configs = ProxyConfig(
+ ftp='a', https='b', http='c', no_proxy='d'
+ )
+
+ container = self.client.create_container(
+ TEST_IMG, 'cat', detach=True, stdin_open=True,
+ )
+ self.client.start(container)
+ self.tmp_containers.append(container)
+
+ cmd = 'sh -c "env | grep -i proxy"'
+
+ # First, just make sure the environment variables from the custom
+ # config are set
+
+ res = self.client.exec_create(container, cmd=cmd)
+ output = self.client.exec_start(res).decode('utf-8').split('\n')
+ expected = [
+ 'ftp_proxy=a', 'https_proxy=b', 'http_proxy=c', 'no_proxy=d',
+ 'FTP_PROXY=a', 'HTTPS_PROXY=b', 'HTTP_PROXY=c', 'NO_PROXY=d'
+ ]
+ for item in expected:
+ assert item in output
+
+ # Overwrite some variables with a custom environment
+ env = {'https_proxy': 'xxx', 'HTTPS_PROXY': 'XXX'}
+
+ res = self.client.exec_create(container, cmd=cmd, environment=env)
+ output = self.client.exec_start(res).decode('utf-8').split('\n')
+ expected = [
+ 'ftp_proxy=a', 'https_proxy=xxx', 'http_proxy=c', 'no_proxy=d',
+ 'FTP_PROXY=a', 'HTTPS_PROXY=XXX', 'HTTP_PROXY=c', 'NO_PROXY=d'
+ ]
+ for item in expected:
+ assert item in output
+
def test_execute_command(self):
- container = self.client.create_container(BUSYBOX, 'cat',
+ container = self.client.create_container(TEST_IMG, 'cat',
detach=True, stdin_open=True)
id = container['Id']
self.client.start(id)
@@ -22,7 +61,7 @@ class ExecTest(BaseAPIIntegrationTest):
assert exec_log == b'hello\n'
def test_exec_command_string(self):
- container = self.client.create_container(BUSYBOX, 'cat',
+ container = self.client.create_container(TEST_IMG, 'cat',
detach=True, stdin_open=True)
id = container['Id']
self.client.start(id)
@@ -35,20 +74,20 @@ class ExecTest(BaseAPIIntegrationTest):
assert exec_log == b'hello world\n'
def test_exec_command_as_user(self):
- container = self.client.create_container(BUSYBOX, 'cat',
+ container = self.client.create_container(TEST_IMG, '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')
+ res = self.client.exec_create(id, 'whoami', user='postgres')
assert 'Id' in res
exec_log = self.client.exec_start(res)
- assert exec_log == b'default\n'
+ assert exec_log == b'postgres\n'
def test_exec_command_as_root(self):
- container = self.client.create_container(BUSYBOX, 'cat',
+ container = self.client.create_container(TEST_IMG, 'cat',
detach=True, stdin_open=True)
id = container['Id']
self.client.start(id)
@@ -61,7 +100,7 @@ class ExecTest(BaseAPIIntegrationTest):
assert exec_log == b'root\n'
def test_exec_command_streaming(self):
- container = self.client.create_container(BUSYBOX, 'cat',
+ container = self.client.create_container(TEST_IMG, 'cat',
detach=True, stdin_open=True)
id = container['Id']
self.tmp_containers.append(id)
@@ -76,7 +115,7 @@ class ExecTest(BaseAPIIntegrationTest):
assert res == b'hello\nworld\n'
def test_exec_start_socket(self):
- container = self.client.create_container(BUSYBOX, 'cat',
+ container = self.client.create_container(TEST_IMG, 'cat',
detach=True, stdin_open=True)
container_id = container['Id']
self.client.start(container_id)
@@ -91,13 +130,14 @@ class ExecTest(BaseAPIIntegrationTest):
socket = self.client.exec_start(exec_id, socket=True)
self.addCleanup(socket.close)
- next_size = next_frame_size(socket)
+ (stream, next_size) = next_frame_header(socket)
+ assert stream == 1 # stdout (0 = stdin, 1 = stdout, 2 = stderr)
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',
+ container = self.client.create_container(TEST_IMG, 'cat',
detach=True, stdin_open=True)
container_id = container['Id']
self.client.start(container_id)
@@ -112,7 +152,7 @@ class ExecTest(BaseAPIIntegrationTest):
assert response == ""
def test_exec_inspect(self):
- container = self.client.create_container(BUSYBOX, 'cat',
+ container = self.client.create_container(TEST_IMG, 'cat',
detach=True, stdin_open=True)
id = container['Id']
self.client.start(id)
@@ -127,7 +167,7 @@ class ExecTest(BaseAPIIntegrationTest):
@requires_api_version('1.25')
def test_exec_command_with_env(self):
- container = self.client.create_container(BUSYBOX, 'cat',
+ container = self.client.create_container(TEST_IMG, 'cat',
detach=True, stdin_open=True)
id = container['Id']
self.client.start(id)
@@ -142,18 +182,18 @@ class ExecTest(BaseAPIIntegrationTest):
@requires_api_version('1.35')
def test_exec_command_with_workdir(self):
container = self.client.create_container(
- BUSYBOX, 'cat', detach=True, stdin_open=True
+ TEST_IMG, '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')
+ res = self.client.exec_create(container, 'pwd', workdir='/var/opt')
exec_log = self.client.exec_start(res)
- assert exec_log == b'/var/www\n'
+ assert exec_log == b'/var/opt\n'
def test_detach_with_default(self):
container = self.client.create_container(
- BUSYBOX, 'cat', detach=True, stdin_open=True
+ TEST_IMG, 'cat', detach=True, stdin_open=True
)
id = container['Id']
self.client.start(id)
@@ -172,7 +212,7 @@ class ExecTest(BaseAPIIntegrationTest):
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
+ TEST_IMG, 'cat', detach=True, stdin_open=True
)
id = container['Id']
self.client.start(id)
@@ -186,20 +226,87 @@ class ExecTest(BaseAPIIntegrationTest):
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
+class ExecDemuxTest(BaseAPIIntegrationTest):
+ cmd = 'sh -c "{}"'.format(' ; '.join([
+ # Write something on stdout
+ 'echo hello out',
+ # Busybox's sleep does not handle sub-second times.
+ # This loops takes ~0.3 second to execute on my machine.
+ 'sleep 0.5',
+ # Write something on stderr
+ 'echo hello err >&2'])
+ )
+
+ def setUp(self):
+ super(ExecDemuxTest, self).setUp()
+ self.container = self.client.create_container(
+ TEST_IMG, 'cat', detach=True, stdin_open=True
)
- sock = self.client.exec_start(exec_id, tty=True, socket=True)
- self.addCleanup(sock.close)
+ self.client.start(self.container)
+ self.tmp_containers.append(self.container)
- assert_cat_socket_detached_with_keys(sock, [ctrl_with('x')])
+ def test_exec_command_no_stream_no_demux(self):
+ # tty=False, stream=False, demux=False
+ res = self.client.exec_create(self.container, self.cmd)
+ exec_log = self.client.exec_start(res)
+ assert b'hello out\n' in exec_log
+ assert b'hello err\n' in exec_log
+
+ def test_exec_command_stream_no_demux(self):
+ # tty=False, stream=True, demux=False
+ res = self.client.exec_create(self.container, self.cmd)
+ exec_log = list(self.client.exec_start(res, stream=True))
+ assert len(exec_log) == 2
+ assert b'hello out\n' in exec_log
+ assert b'hello err\n' in exec_log
+
+ def test_exec_command_no_stream_demux(self):
+ # tty=False, stream=False, demux=True
+ res = self.client.exec_create(self.container, self.cmd)
+ exec_log = self.client.exec_start(res, demux=True)
+ assert exec_log == (b'hello out\n', b'hello err\n')
+
+ def test_exec_command_stream_demux(self):
+ # tty=False, stream=True, demux=True
+ res = self.client.exec_create(self.container, self.cmd)
+ exec_log = list(self.client.exec_start(res, demux=True, stream=True))
+ assert len(exec_log) == 2
+ assert (b'hello out\n', None) in exec_log
+ assert (None, b'hello err\n') in exec_log
+
+ def test_exec_command_tty_no_stream_no_demux(self):
+ # tty=True, stream=False, demux=False
+ res = self.client.exec_create(self.container, self.cmd, tty=True)
+ exec_log = self.client.exec_start(res)
+ assert exec_log == b'hello out\r\nhello err\r\n'
+
+ def test_exec_command_tty_stream_no_demux(self):
+ # tty=True, stream=True, demux=False
+ res = self.client.exec_create(self.container, self.cmd, tty=True)
+ exec_log = list(self.client.exec_start(res, stream=True))
+ assert b'hello out\r\n' in exec_log
+ if len(exec_log) == 2:
+ assert b'hello err\r\n' in exec_log
+ else:
+ assert len(exec_log) == 3
+ assert b'hello err' in exec_log
+ assert b'\r\n' in exec_log
+
+ def test_exec_command_tty_no_stream_demux(self):
+ # tty=True, stream=False, demux=True
+ res = self.client.exec_create(self.container, self.cmd, tty=True)
+ exec_log = self.client.exec_start(res, demux=True)
+ assert exec_log == (b'hello out\r\nhello err\r\n', None)
+
+ def test_exec_command_tty_stream_demux(self):
+ # tty=True, stream=True, demux=True
+ res = self.client.exec_create(self.container, self.cmd, tty=True)
+ exec_log = list(self.client.exec_start(res, demux=True, stream=True))
+ assert (b'hello out\r\n', None) in exec_log
+ if len(exec_log) == 2:
+ assert (b'hello err\r\n', None) in exec_log
+ else:
+ assert len(exec_log) == 3
+ assert (b'hello err', None) in exec_log
+ assert (b'\r\n', None) in exec_log
generated by cgit v1.2.3 (git 2.25.1) at 2024-04-19 23:01:03 +0000