Unverified Kaydet (Commit) 39900c55 authored tarafından Ben Firshman's avatar Ben Firshman

Move APIClient tests into single file

For some reason this was spread across two files.
Signed-off-by: 's avatarBen Firshman <ben@firshman.co.uk>
üst e055a1c8
import os
import unittest
from docker.api import APIClient
TEST_CERT_DIR = os.path.join(
os.path.dirname(__file__),
'testdata/certs',
)
class ClientTest(unittest.TestCase):
def setUp(self):
self.os_environ = os.environ.copy()
def tearDown(self):
os.environ = self.os_environ
def test_from_env(self):
"""Test that environment variables are passed through to
utils.kwargs_from_env(). KwargsFromEnvTest tests that environment
variables are parsed correctly."""
os.environ.update(DOCKER_HOST='tcp://192.168.59.103:2376',
DOCKER_CERT_PATH=TEST_CERT_DIR,
DOCKER_TLS_VERIFY='1')
client = APIClient.from_env()
self.assertEqual(client.base_url, "https://192.168.59.103:2376")
def test_from_env_with_version(self):
os.environ.update(DOCKER_HOST='tcp://192.168.59.103:2376',
DOCKER_CERT_PATH=TEST_CERT_DIR,
DOCKER_TLS_VERIFY='1')
client = APIClient.from_env(version='2.32')
self.assertEqual(client.base_url, "https://192.168.59.103:2376")
self.assertEqual(client._version, '2.32')
class DisableSocketTest(unittest.TestCase):
class DummySocket(object):
def __init__(self, timeout=60):
self.timeout = timeout
def settimeout(self, timeout):
self.timeout = timeout
def gettimeout(self):
return self.timeout
def setUp(self):
self.client = APIClient()
def test_disable_socket_timeout(self):
"""Test that the timeout is disabled on a generic socket object."""
socket = self.DummySocket()
self.client._disable_socket_timeout(socket)
self.assertEqual(socket.timeout, None)
def test_disable_socket_timeout2(self):
"""Test that the timeouts are disabled on a generic socket object
and it's _sock object if present."""
socket = self.DummySocket()
socket._sock = self.DummySocket()
self.client._disable_socket_timeout(socket)
self.assertEqual(socket.timeout, None)
self.assertEqual(socket._sock.timeout, None)
def test_disable_socket_timout_non_blocking(self):
"""Test that a non-blocking socket does not get set to blocking."""
socket = self.DummySocket()
socket._sock = self.DummySocket(0.0)
self.client._disable_socket_timeout(socket)
self.assertEqual(socket.timeout, None)
self.assertEqual(socket._sock.timeout, 0.0)
...@@ -27,6 +27,7 @@ except ImportError: ...@@ -27,6 +27,7 @@ except ImportError:
DEFAULT_TIMEOUT_SECONDS = docker.constants.DEFAULT_TIMEOUT_SECONDS DEFAULT_TIMEOUT_SECONDS = docker.constants.DEFAULT_TIMEOUT_SECONDS
TEST_CERT_DIR = os.path.join(os.path.dirname(__file__), 'testdata/certs')
def response(status_code=200, content='', headers=None, reason=None, elapsed=0, def response(status_code=200, content='', headers=None, reason=None, elapsed=0,
...@@ -484,3 +485,73 @@ class UserAgentTest(unittest.TestCase): ...@@ -484,3 +485,73 @@ class UserAgentTest(unittest.TestCase):
self.assertEqual(self.mock_send.call_count, 1) self.assertEqual(self.mock_send.call_count, 1)
headers = self.mock_send.call_args[0][0].headers headers = self.mock_send.call_args[0][0].headers
self.assertEqual(headers['User-Agent'], 'foo/bar') self.assertEqual(headers['User-Agent'], 'foo/bar')
class FromEnvTest(unittest.TestCase):
def setUp(self):
self.os_environ = os.environ.copy()
def tearDown(self):
os.environ = self.os_environ
def test_from_env(self):
"""Test that environment variables are passed through to
utils.kwargs_from_env(). KwargsFromEnvTest tests that environment
variables are parsed correctly."""
os.environ.update(DOCKER_HOST='tcp://192.168.59.103:2376',
DOCKER_CERT_PATH=TEST_CERT_DIR,
DOCKER_TLS_VERIFY='1')
client = APIClient.from_env()
self.assertEqual(client.base_url, "https://192.168.59.103:2376")
def test_from_env_with_version(self):
os.environ.update(DOCKER_HOST='tcp://192.168.59.103:2376',
DOCKER_CERT_PATH=TEST_CERT_DIR,
DOCKER_TLS_VERIFY='1')
client = APIClient.from_env(version='2.32')
self.assertEqual(client.base_url, "https://192.168.59.103:2376")
self.assertEqual(client._version, '2.32')
class DisableSocketTest(unittest.TestCase):
class DummySocket(object):
def __init__(self, timeout=60):
self.timeout = timeout
def settimeout(self, timeout):
self.timeout = timeout
def gettimeout(self):
return self.timeout
def setUp(self):
self.client = APIClient()
def test_disable_socket_timeout(self):
"""Test that the timeout is disabled on a generic socket object."""
socket = self.DummySocket()
self.client._disable_socket_timeout(socket)
self.assertEqual(socket.timeout, None)
def test_disable_socket_timeout2(self):
"""Test that the timeouts are disabled on a generic socket object
and it's _sock object if present."""
socket = self.DummySocket()
socket._sock = self.DummySocket()
self.client._disable_socket_timeout(socket)
self.assertEqual(socket.timeout, None)
self.assertEqual(socket._sock.timeout, None)
def test_disable_socket_timout_non_blocking(self):
"""Test that a non-blocking socket does not get set to blocking."""
socket = self.DummySocket()
socket._sock = self.DummySocket(0.0)
self.client._disable_socket_timeout(socket)
self.assertEqual(socket.timeout, None)
self.assertEqual(socket._sock.timeout, 0.0)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment