docker
/
docker-py
mirror of https://github.com/docker/docker-py.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Update detach tests to work with AF_INET as well 

Signed-off-by: Joffrey F <joffrey@docker.com>
This commit is contained in:
Joffrey F 2018-01-29 17:07:15 -08:00
parent dd858648a0
commit e304f91b46
4 changed files with 71 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Makefile
View File

@ -54,7 +54,7 @@ integration-dind-py2: build
-H tcp://0.0.0.0:2375 --experimental
docker run -t --rm --env="DOCKER_HOST=tcp://docker:2375" --env="DOCKER_TEST_API_VERSION=${TEST_API_VERSION}"\
--link=dpy-dind-py2:docker docker-sdk-python py.test tests/integration
docker rm -vf dpy-dind-py3
docker rm -vf dpy-dind-py2
.PHONY: integration-dind-py3
integration-dind-py3: build-py3

15
tests/helpers.py
View File

@ -6,8 +6,8 @@ import tarfile
import tempfile
import time
import re
import socket
import six
import socket
import docker
import pytest
@ -107,16 +107,23 @@ def swarm_listen_addr():
return '0.0.0.0:{0}'.format(random.randrange(10000, 25000))
def assert_socket_closed_with_keys(sock, inputs):
def assert_cat_socket_detached_with_keys(sock, inputs):
if six.PY3:
sock = sock._sock
for i in inputs:
sock.send(i)
time.sleep(1)
time.sleep(0.5)
with pytest.raises(socket.error):
# If we're using a Unix socket, the sock.send call will fail with a
# BrokenPipeError ; INET sockets will just stop receiving / sending data
# but will not raise an error
if sock.family == getattr(socket, 'AF_UNIX', -1):
with pytest.raises(socket.error):
sock.send(b'make sure the socket is closed\n')
else:
sock.send(b"make sure the socket is closed\n")
assert sock.recv(32) == b''
def ctrl_with(char):

31
tests/integration/api_container_test.py
View File

@ -17,7 +17,7 @@ import six
from .base import BUSYBOX, BaseAPIIntegrationTest
from .. import helpers
from ..helpers import (
requires_api_version, ctrl_with, assert_socket_closed_with_keys
requires_api_version, ctrl_with, assert_cat_socket_detached_with_keys
)
@ -1227,55 +1227,54 @@ class AttachContainerTest(BaseAPIIntegrationTest):
def test_detach_with_default(self):
container = self.client.create_container(
BUSYBOX, '/bin/sh',
BUSYBOX, 'cat',
detach=True, stdin_open=True, tty=True
)
id = container['Id']
self.tmp_containers.append(id)
self.client.start(id)
self.tmp_containers.append(container)
self.client.start(container)
sock = self.client.attach_socket(
container,
{'stdin': True, 'stream': True}
)
assert_socket_closed_with_keys(sock, [ctrl_with('p'), ctrl_with('q')])
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, '/bin/sh',
BUSYBOX, 'cat',
detach=True, stdin_open=True, tty=True
)
id = container['Id']
self.tmp_containers.append(id)
self.client.start(id)
self.tmp_containers.append(container)
self.client.start(container)
sock = self.client.attach_socket(
container,
{'stdin': True, 'stream': True}
)
assert_socket_closed_with_keys(sock, [ctrl_with('p')])
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, '/bin/sh',
BUSYBOX, 'cat',
detach=True, stdin_open=True, tty=True
)
id = container['Id']
self.tmp_containers.append(id)
self.client.start(id)
self.tmp_containers.append(container)
self.client.start(container)
sock = self.client.attach_socket(
container,
{'stdin': True, 'stream': True, 'detachKeys': 'ctrl-x'}
)
assert_socket_closed_with_keys(sock, [ctrl_with('x')])
assert_cat_socket_detached_with_keys(sock, [ctrl_with('x')])
class PauseTest(BaseAPIIntegrationTest):

76
tests/integration/api_exec_test.py
View File

@ -3,7 +3,7 @@ from docker.utils.socket import read_exactly
from .base import BaseAPIIntegrationTest, BUSYBOX
from ..helpers import (
requires_api_version, ctrl_with, assert_socket_closed_with_keys
requires_api_version, ctrl_with, assert_cat_socket_detached_with_keys
)
@ -152,42 +152,54 @@ class ExecTest(BaseAPIIntegrationTest):
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, '/bin/sh', stdin=True, tty=True)
sock = self.client.exec_start(exec_id, tty=True, socket=True)
assert_socket_closed_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, '/bin/sh', stdin=True, tty=True)
sock = self.client.exec_start(exec_id, tty=True, socket=True)
assert_socket_closed_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)
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, '/bin/sh',
stdin=True, tty=True, detach_keys='ctrl-x'
id, 'cat', stdin=True, tty=True, stdout=True
)
sock = self.client.exec_start(exec_id, tty=True, socket=True)
self.addCleanup(sock.close)
assert_socket_closed_with_keys(sock, [ctrl_with('x')])
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')])