зеркало из https://github.com/docker/docker-py.git
825 строки
28 KiB
Python
825 строки
28 KiB
Python
import unittest
|
|
|
|
import pytest
|
|
|
|
import docker
|
|
from docker.constants import DEFAULT_DATA_CHUNK_SIZE, DEFAULT_DOCKER_API_VERSION
|
|
from docker.models.containers import Container, _create_container_args
|
|
from docker.models.images import Image
|
|
from docker.types import EndpointConfig
|
|
|
|
from .fake_api import FAKE_CONTAINER_ID, FAKE_EXEC_ID, FAKE_IMAGE_ID
|
|
from .fake_api_client import make_fake_client
|
|
|
|
|
|
class ContainerCollectionTest(unittest.TestCase):
|
|
def test_run(self):
|
|
client = make_fake_client()
|
|
out = client.containers.run("alpine", "echo hello world")
|
|
|
|
assert out == b'hello world\n'
|
|
|
|
client.api.create_container.assert_called_with(
|
|
image="alpine",
|
|
command="echo hello world",
|
|
detach=False,
|
|
host_config={'NetworkMode': 'default'}
|
|
)
|
|
client.api.inspect_container.assert_called_with(FAKE_CONTAINER_ID)
|
|
client.api.start.assert_called_with(FAKE_CONTAINER_ID)
|
|
client.api.wait.assert_called_with(FAKE_CONTAINER_ID)
|
|
client.api.logs.assert_called_with(
|
|
FAKE_CONTAINER_ID, stderr=False, stdout=True, stream=True,
|
|
follow=True
|
|
)
|
|
|
|
def test_create_container_args(self):
|
|
networking_config = {
|
|
'foo': EndpointConfig(
|
|
DEFAULT_DOCKER_API_VERSION, aliases=['test'],
|
|
driver_opt={'key1': 'a'}
|
|
)
|
|
}
|
|
|
|
create_kwargs = _create_container_args({
|
|
'image': 'alpine',
|
|
'command': 'echo hello world',
|
|
'blkio_weight_device': [{'Path': 'foo', 'Weight': 3}],
|
|
'blkio_weight': 2,
|
|
'cap_add': ['foo'],
|
|
'cap_drop': ['bar'],
|
|
'cgroup_parent': 'foobar',
|
|
'cgroupns': 'host',
|
|
'cpu_period': 1,
|
|
'cpu_quota': 2,
|
|
'cpu_shares': 5,
|
|
'cpuset_cpus': '0-3',
|
|
'detach': False,
|
|
'device_read_bps': [{'Path': 'foo', 'Rate': 3}],
|
|
'device_read_iops': [{'Path': 'foo', 'Rate': 3}],
|
|
'device_write_bps': [{'Path': 'foo', 'Rate': 3}],
|
|
'device_write_iops': [{'Path': 'foo', 'Rate': 3}],
|
|
'devices': ['/dev/sda:/dev/xvda:rwm'],
|
|
'dns': ['8.8.8.8'],
|
|
'domainname': 'example.com',
|
|
'dns_opt': ['foo'],
|
|
'dns_search': ['example.com'],
|
|
'entrypoint': '/bin/sh',
|
|
'environment': {'FOO': 'BAR'},
|
|
'extra_hosts': {'foo': '1.2.3.4'},
|
|
'group_add': ['blah'],
|
|
'ipc_mode': 'foo',
|
|
'kernel_memory': 123,
|
|
'labels': {'key': 'value'},
|
|
'links': {'foo': 'bar'},
|
|
'log_config': {'Type': 'json-file', 'Config': {}},
|
|
'lxc_conf': {'foo': 'bar'},
|
|
'healthcheck': {'test': 'true'},
|
|
'hostname': 'somehost',
|
|
'mac_address': 'abc123',
|
|
'mem_limit': 123,
|
|
'mem_reservation': 123,
|
|
'mem_swappiness': 2,
|
|
'memswap_limit': 456,
|
|
'name': 'somename',
|
|
'network_disabled': False,
|
|
'network': 'foo',
|
|
'networking_config': networking_config,
|
|
'oom_kill_disable': True,
|
|
'oom_score_adj': 5,
|
|
'pid_mode': 'host',
|
|
'pids_limit': 500,
|
|
'platform': 'linux',
|
|
'ports': {1111: 4567, 2222: None},
|
|
'privileged': True,
|
|
'publish_all_ports': True,
|
|
'read_only': True,
|
|
'restart_policy': {'Name': 'always'},
|
|
'security_opt': ['blah'],
|
|
'shm_size': 123,
|
|
'stdin_open': True,
|
|
'stop_signal': 9,
|
|
'sysctls': {'foo': 'bar'},
|
|
'tmpfs': {'/blah': ''},
|
|
'tty': True,
|
|
'ulimits': [{"Name": "nofile", "Soft": 1024, "Hard": 2048}],
|
|
'user': 'bob',
|
|
'userns_mode': 'host',
|
|
'uts_mode': 'host',
|
|
'version': DEFAULT_DOCKER_API_VERSION,
|
|
'volume_driver': 'some_driver', 'volumes': [
|
|
'/home/user1/:/mnt/vol2',
|
|
'/var/www:/mnt/vol1:ro',
|
|
'volumename:/mnt/vol3r',
|
|
'/volumewithnohostpath',
|
|
'/anothervolumewithnohostpath:ro',
|
|
'C:\\windows\\path:D:\\hello\\world:rw'
|
|
],
|
|
'volumes_from': ['container'],
|
|
'working_dir': '/code',
|
|
})
|
|
|
|
expected = {
|
|
'image': 'alpine',
|
|
'command': 'echo hello world',
|
|
'domainname': 'example.com',
|
|
'detach': False,
|
|
'entrypoint': '/bin/sh',
|
|
'environment': {'FOO': 'BAR'},
|
|
'host_config': {
|
|
'Binds': [
|
|
'/home/user1/:/mnt/vol2',
|
|
'/var/www:/mnt/vol1:ro',
|
|
'volumename:/mnt/vol3r',
|
|
'/volumewithnohostpath',
|
|
'/anothervolumewithnohostpath:ro',
|
|
'C:\\windows\\path:D:\\hello\\world:rw'
|
|
],
|
|
'BlkioDeviceReadBps': [{'Path': 'foo', 'Rate': 3}],
|
|
'BlkioDeviceReadIOps': [{'Path': 'foo', 'Rate': 3}],
|
|
'BlkioDeviceWriteBps': [{'Path': 'foo', 'Rate': 3}],
|
|
'BlkioDeviceWriteIOps': [{'Path': 'foo', 'Rate': 3}],
|
|
'BlkioWeightDevice': [{'Path': 'foo', 'Weight': 3}],
|
|
'BlkioWeight': 2,
|
|
'CapAdd': ['foo'],
|
|
'CapDrop': ['bar'],
|
|
'CgroupnsMode': 'host',
|
|
'CgroupParent': 'foobar',
|
|
'CpuPeriod': 1,
|
|
'CpuQuota': 2,
|
|
'CpuShares': 5,
|
|
'CpusetCpus': '0-3',
|
|
'Devices': [
|
|
{
|
|
'PathOnHost': '/dev/sda',
|
|
'CgroupPermissions': 'rwm',
|
|
'PathInContainer': '/dev/xvda',
|
|
},
|
|
],
|
|
'Dns': ['8.8.8.8'],
|
|
'DnsOptions': ['foo'],
|
|
'DnsSearch': ['example.com'],
|
|
'ExtraHosts': ['foo:1.2.3.4'],
|
|
'GroupAdd': ['blah'],
|
|
'IpcMode': 'foo',
|
|
'KernelMemory': 123,
|
|
'Links': ['foo:bar'],
|
|
'LogConfig': {'Type': 'json-file', 'Config': {}},
|
|
'LxcConf': [{'Key': 'foo', 'Value': 'bar'}],
|
|
'Memory': 123,
|
|
'MemoryReservation': 123,
|
|
'MemorySwap': 456,
|
|
'MemorySwappiness': 2,
|
|
'NetworkMode': 'foo',
|
|
'OomKillDisable': True,
|
|
'OomScoreAdj': 5,
|
|
'PidMode': 'host',
|
|
'PidsLimit': 500,
|
|
'PortBindings': {
|
|
'1111/tcp': [{'HostIp': '', 'HostPort': '4567'}],
|
|
'2222/tcp': [{'HostIp': '', 'HostPort': ''}]
|
|
},
|
|
'Privileged': True,
|
|
'PublishAllPorts': True,
|
|
'ReadonlyRootfs': True,
|
|
'RestartPolicy': {'Name': 'always'},
|
|
'SecurityOpt': ['blah'],
|
|
'ShmSize': 123,
|
|
'Sysctls': {'foo': 'bar'},
|
|
'Tmpfs': {'/blah': ''},
|
|
'Ulimits': [
|
|
{"Name": "nofile", "Soft": 1024, "Hard": 2048},
|
|
],
|
|
'UsernsMode': 'host',
|
|
'UTSMode': 'host',
|
|
'VolumeDriver': 'some_driver',
|
|
'VolumesFrom': ['container'],
|
|
},
|
|
'healthcheck': {'test': 'true'},
|
|
'hostname': 'somehost',
|
|
'labels': {'key': 'value'},
|
|
'mac_address': 'abc123',
|
|
'name': 'somename',
|
|
'network_disabled': False,
|
|
'networking_config': {
|
|
'EndpointsConfig': {
|
|
'foo': {
|
|
'Aliases': ['test'],
|
|
'DriverOpts': {'key1': 'a'},
|
|
},
|
|
}
|
|
},
|
|
'platform': 'linux',
|
|
'ports': [('1111', 'tcp'), ('2222', 'tcp')],
|
|
'stdin_open': True,
|
|
'stop_signal': 9,
|
|
'tty': True,
|
|
'user': 'bob',
|
|
'volumes': [
|
|
'/mnt/vol2',
|
|
'/mnt/vol1',
|
|
'/mnt/vol3r',
|
|
'/volumewithnohostpath',
|
|
'/anothervolumewithnohostpath',
|
|
'D:\\hello\\world'
|
|
],
|
|
'working_dir': '/code',
|
|
}
|
|
|
|
assert create_kwargs == expected
|
|
|
|
def test_run_detach(self):
|
|
client = make_fake_client()
|
|
container = client.containers.run('alpine', 'sleep 300', detach=True)
|
|
assert isinstance(container, Container)
|
|
assert container.id == FAKE_CONTAINER_ID
|
|
client.api.create_container.assert_called_with(
|
|
image='alpine',
|
|
command='sleep 300',
|
|
detach=True,
|
|
host_config={
|
|
'NetworkMode': 'default',
|
|
}
|
|
)
|
|
client.api.inspect_container.assert_called_with(FAKE_CONTAINER_ID)
|
|
client.api.start.assert_called_with(FAKE_CONTAINER_ID)
|
|
|
|
def test_run_pull(self):
|
|
client = make_fake_client()
|
|
|
|
# raise exception on first call, then return normal value
|
|
client.api.create_container.side_effect = [
|
|
docker.errors.ImageNotFound(""),
|
|
client.api.create_container.return_value
|
|
]
|
|
|
|
container = client.containers.run('alpine', 'sleep 300', detach=True)
|
|
|
|
assert container.id == FAKE_CONTAINER_ID
|
|
client.api.pull.assert_called_with(
|
|
'alpine', platform=None, tag='latest', all_tags=False, stream=True
|
|
)
|
|
|
|
def test_run_with_error(self):
|
|
client = make_fake_client()
|
|
client.api.logs.return_value = "some error"
|
|
client.api.wait.return_value = {'StatusCode': 1}
|
|
|
|
with pytest.raises(docker.errors.ContainerError) as cm:
|
|
client.containers.run('alpine', 'echo hello world')
|
|
assert cm.value.exit_status == 1
|
|
assert "some error" in cm.exconly()
|
|
|
|
def test_run_with_image_object(self):
|
|
client = make_fake_client()
|
|
image = client.images.get(FAKE_IMAGE_ID)
|
|
client.containers.run(image)
|
|
client.api.create_container.assert_called_with(
|
|
image=image.id,
|
|
command=None,
|
|
detach=False,
|
|
host_config={
|
|
'NetworkMode': 'default',
|
|
}
|
|
)
|
|
|
|
def test_run_remove(self):
|
|
client = make_fake_client()
|
|
client.containers.run("alpine")
|
|
client.api.remove_container.assert_not_called()
|
|
|
|
client = make_fake_client()
|
|
client.api.wait.return_value = {'StatusCode': 1}
|
|
with pytest.raises(docker.errors.ContainerError):
|
|
client.containers.run("alpine")
|
|
client.api.remove_container.assert_not_called()
|
|
|
|
client = make_fake_client()
|
|
client.containers.run("alpine", remove=True)
|
|
client.api.remove_container.assert_called_with(FAKE_CONTAINER_ID)
|
|
|
|
client = make_fake_client()
|
|
client.api.wait.return_value = {'StatusCode': 1}
|
|
with pytest.raises(docker.errors.ContainerError):
|
|
client.containers.run("alpine", remove=True)
|
|
client.api.remove_container.assert_called_with(FAKE_CONTAINER_ID)
|
|
|
|
client = make_fake_client()
|
|
client.api._version = '1.24'
|
|
with pytest.raises(RuntimeError):
|
|
client.containers.run("alpine", detach=True, remove=True)
|
|
|
|
client = make_fake_client()
|
|
client.api._version = '1.23'
|
|
with pytest.raises(RuntimeError):
|
|
client.containers.run("alpine", detach=True, remove=True)
|
|
|
|
client = make_fake_client()
|
|
client.api._version = '1.25'
|
|
client.containers.run("alpine", detach=True, remove=True)
|
|
client.api.remove_container.assert_not_called()
|
|
client.api.create_container.assert_called_with(
|
|
command=None,
|
|
image='alpine',
|
|
detach=True,
|
|
host_config={'AutoRemove': True,
|
|
'NetworkMode': 'default'}
|
|
)
|
|
|
|
client = make_fake_client()
|
|
client.api._version = '1.26'
|
|
client.containers.run("alpine", detach=True, remove=True)
|
|
client.api.remove_container.assert_not_called()
|
|
client.api.create_container.assert_called_with(
|
|
command=None,
|
|
image='alpine',
|
|
detach=True,
|
|
host_config={'AutoRemove': True,
|
|
'NetworkMode': 'default'}
|
|
)
|
|
|
|
def test_run_platform(self):
|
|
client = make_fake_client()
|
|
|
|
# raise exception on first call, then return normal value
|
|
client.api.create_container.side_effect = [
|
|
docker.errors.ImageNotFound(""),
|
|
client.api.create_container.return_value
|
|
]
|
|
|
|
client.containers.run(image='alpine', platform='linux/arm64')
|
|
|
|
client.api.pull.assert_called_with(
|
|
'alpine',
|
|
tag='latest',
|
|
all_tags=False,
|
|
stream=True,
|
|
platform='linux/arm64',
|
|
)
|
|
|
|
client.api.create_container.assert_called_with(
|
|
detach=False,
|
|
platform='linux/arm64',
|
|
image='alpine',
|
|
command=None,
|
|
host_config={'NetworkMode': 'default'},
|
|
)
|
|
|
|
def test_run_networking_config_without_network(self):
|
|
client = make_fake_client()
|
|
|
|
with pytest.raises(RuntimeError):
|
|
client.containers.run(
|
|
image='alpine',
|
|
networking_config={'aliases': ['test'],
|
|
'driver_opt': {'key1': 'a'}}
|
|
)
|
|
|
|
def test_run_networking_config_with_network_mode(self):
|
|
client = make_fake_client()
|
|
|
|
with pytest.raises(RuntimeError):
|
|
client.containers.run(
|
|
image='alpine',
|
|
network_mode='none',
|
|
networking_config={'aliases': ['test'],
|
|
'driver_opt': {'key1': 'a'}}
|
|
)
|
|
|
|
def test_run_networking_config(self):
|
|
client = make_fake_client()
|
|
|
|
networking_config = {
|
|
'foo': EndpointConfig(
|
|
DEFAULT_DOCKER_API_VERSION, aliases=['test'],
|
|
driver_opt={'key1': 'a'}
|
|
)
|
|
}
|
|
|
|
client.containers.run(
|
|
image='alpine',
|
|
network='foo',
|
|
networking_config=networking_config
|
|
)
|
|
|
|
client.api.create_container.assert_called_with(
|
|
detach=False,
|
|
image='alpine',
|
|
command=None,
|
|
networking_config={'EndpointsConfig': {
|
|
'foo': {'Aliases': ['test'], 'DriverOpts': {'key1': 'a'}}}
|
|
},
|
|
host_config={'NetworkMode': 'foo'}
|
|
)
|
|
|
|
def test_run_networking_config_with_undeclared_network(self):
|
|
client = make_fake_client()
|
|
|
|
networking_config = {
|
|
'foo': EndpointConfig(
|
|
DEFAULT_DOCKER_API_VERSION, aliases=['test_foo'],
|
|
driver_opt={'key2': 'b'}
|
|
),
|
|
'bar': EndpointConfig(
|
|
DEFAULT_DOCKER_API_VERSION, aliases=['test'],
|
|
driver_opt={'key1': 'a'}
|
|
)
|
|
}
|
|
|
|
client.containers.run(
|
|
image='alpine',
|
|
network='foo',
|
|
networking_config=networking_config
|
|
)
|
|
|
|
client.api.create_container.assert_called_with(
|
|
detach=False,
|
|
image='alpine',
|
|
command=None,
|
|
networking_config={'EndpointsConfig': {
|
|
'foo': {'Aliases': ['test_foo'], 'DriverOpts': {'key2': 'b'}},
|
|
'bar': {'Aliases': ['test'], 'DriverOpts': {'key1': 'a'}},
|
|
}},
|
|
host_config={'NetworkMode': 'foo'}
|
|
)
|
|
|
|
def test_run_networking_config_only_undeclared_network(self):
|
|
client = make_fake_client()
|
|
|
|
networking_config = {
|
|
'bar': EndpointConfig(
|
|
DEFAULT_DOCKER_API_VERSION, aliases=['test'],
|
|
driver_opt={'key1': 'a'}
|
|
)
|
|
}
|
|
|
|
client.containers.run(
|
|
image='alpine',
|
|
network='foo',
|
|
networking_config=networking_config
|
|
)
|
|
|
|
client.api.create_container.assert_called_with(
|
|
detach=False,
|
|
image='alpine',
|
|
command=None,
|
|
networking_config={'foo': None},
|
|
host_config={'NetworkMode': 'foo'}
|
|
)
|
|
|
|
def test_create(self):
|
|
client = make_fake_client()
|
|
container = client.containers.create(
|
|
'alpine',
|
|
'echo hello world',
|
|
environment={'FOO': 'BAR'}
|
|
)
|
|
assert isinstance(container, Container)
|
|
assert container.id == FAKE_CONTAINER_ID
|
|
client.api.create_container.assert_called_with(
|
|
image='alpine',
|
|
command='echo hello world',
|
|
environment={'FOO': 'BAR'},
|
|
host_config={'NetworkMode': 'default'}
|
|
)
|
|
client.api.inspect_container.assert_called_with(FAKE_CONTAINER_ID)
|
|
|
|
def test_create_with_image_object(self):
|
|
client = make_fake_client()
|
|
image = client.images.get(FAKE_IMAGE_ID)
|
|
client.containers.create(image)
|
|
client.api.create_container.assert_called_with(
|
|
image=image.id,
|
|
command=None,
|
|
host_config={'NetworkMode': 'default'}
|
|
)
|
|
|
|
def test_create_networking_config_without_network(self):
|
|
client = make_fake_client()
|
|
|
|
client.containers.create(
|
|
image='alpine',
|
|
networking_config={'aliases': ['test'],
|
|
'driver_opt': {'key1': 'a'}}
|
|
)
|
|
|
|
client.api.create_container.assert_called_with(
|
|
image='alpine',
|
|
command=None,
|
|
host_config={'NetworkMode': 'default'}
|
|
)
|
|
|
|
def test_create_networking_config_with_network_mode(self):
|
|
client = make_fake_client()
|
|
|
|
client.containers.create(
|
|
image='alpine',
|
|
network_mode='none',
|
|
networking_config={'aliases': ['test'],
|
|
'driver_opt': {'key1': 'a'}}
|
|
)
|
|
|
|
client.api.create_container.assert_called_with(
|
|
image='alpine',
|
|
command=None,
|
|
host_config={'NetworkMode': 'none'}
|
|
)
|
|
|
|
def test_create_networking_config(self):
|
|
client = make_fake_client()
|
|
|
|
networking_config = {
|
|
'foo': EndpointConfig(
|
|
DEFAULT_DOCKER_API_VERSION, aliases=['test'],
|
|
driver_opt={'key1': 'a'}
|
|
)
|
|
}
|
|
|
|
client.containers.create(
|
|
image='alpine',
|
|
network='foo',
|
|
networking_config=networking_config
|
|
)
|
|
|
|
client.api.create_container.assert_called_with(
|
|
image='alpine',
|
|
command=None,
|
|
networking_config={'EndpointsConfig': {
|
|
'foo': {'Aliases': ['test'], 'DriverOpts': {'key1': 'a'}}}
|
|
},
|
|
host_config={'NetworkMode': 'foo'}
|
|
)
|
|
|
|
def test_create_networking_config_with_undeclared_network(self):
|
|
client = make_fake_client()
|
|
|
|
networking_config = {
|
|
'foo': EndpointConfig(
|
|
DEFAULT_DOCKER_API_VERSION, aliases=['test_foo'],
|
|
driver_opt={'key2': 'b'}
|
|
),
|
|
'bar': EndpointConfig(
|
|
DEFAULT_DOCKER_API_VERSION, aliases=['test'],
|
|
driver_opt={'key1': 'a'}
|
|
)
|
|
}
|
|
|
|
client.containers.create(
|
|
image='alpine',
|
|
network='foo',
|
|
networking_config=networking_config
|
|
)
|
|
|
|
client.api.create_container.assert_called_with(
|
|
image='alpine',
|
|
command=None,
|
|
networking_config={'EndpointsConfig': {
|
|
'foo': {'Aliases': ['test_foo'], 'DriverOpts': {'key2': 'b'}},
|
|
'bar': {'Aliases': ['test'], 'DriverOpts': {'key1': 'a'}},
|
|
}},
|
|
host_config={'NetworkMode': 'foo'}
|
|
)
|
|
|
|
def test_create_networking_config_only_undeclared_network(self):
|
|
client = make_fake_client()
|
|
|
|
networking_config = {
|
|
'bar': EndpointConfig(
|
|
DEFAULT_DOCKER_API_VERSION, aliases=['test'],
|
|
driver_opt={'key1': 'a'}
|
|
)
|
|
}
|
|
|
|
client.containers.create(
|
|
image='alpine',
|
|
network='foo',
|
|
networking_config=networking_config
|
|
)
|
|
|
|
client.api.create_container.assert_called_with(
|
|
image='alpine',
|
|
command=None,
|
|
networking_config={'foo': None},
|
|
host_config={'NetworkMode': 'foo'}
|
|
)
|
|
|
|
def test_get(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
assert isinstance(container, Container)
|
|
assert container.id == FAKE_CONTAINER_ID
|
|
client.api.inspect_container.assert_called_with(FAKE_CONTAINER_ID)
|
|
|
|
def test_list(self):
|
|
client = make_fake_client()
|
|
containers = client.containers.list(all=True)
|
|
client.api.containers.assert_called_with(
|
|
all=True,
|
|
before=None,
|
|
filters=None,
|
|
limit=-1,
|
|
since=None
|
|
)
|
|
client.api.inspect_container.assert_called_with(FAKE_CONTAINER_ID)
|
|
assert len(containers) == 1
|
|
assert isinstance(containers[0], Container)
|
|
assert containers[0].id == FAKE_CONTAINER_ID
|
|
|
|
def test_list_ignore_removed(self):
|
|
def side_effect(*args, **kwargs):
|
|
raise docker.errors.NotFound('Container not found')
|
|
|
|
client = make_fake_client({
|
|
'inspect_container.side_effect': side_effect
|
|
})
|
|
|
|
with pytest.raises(docker.errors.NotFound):
|
|
client.containers.list(all=True, ignore_removed=False)
|
|
|
|
assert client.containers.list(all=True, ignore_removed=True) == []
|
|
|
|
|
|
class ContainerTest(unittest.TestCase):
|
|
def test_short_id(self):
|
|
container = Container(attrs={'Id': '8497fe9244dd45cac543eb3c37d8605077'
|
|
'6800eebef1f3ec2ee111e8ccf12db6'})
|
|
assert container.short_id == '8497fe9244dd'
|
|
|
|
def test_name(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
assert container.name == 'foobar'
|
|
|
|
def test_status(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
assert container.status == "running"
|
|
|
|
def test_attach(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.attach(stream=True)
|
|
client.api.attach.assert_called_with(FAKE_CONTAINER_ID, stream=True)
|
|
|
|
def test_commit(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
image = container.commit()
|
|
client.api.commit.assert_called_with(FAKE_CONTAINER_ID,
|
|
repository=None,
|
|
tag=None)
|
|
assert isinstance(image, Image)
|
|
assert image.id == FAKE_IMAGE_ID
|
|
|
|
def test_diff(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.diff()
|
|
client.api.diff.assert_called_with(FAKE_CONTAINER_ID)
|
|
|
|
def test_exec_run(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.exec_run("echo hello world", privileged=True, stream=True)
|
|
client.api.exec_create.assert_called_with(
|
|
FAKE_CONTAINER_ID, "echo hello world", stdout=True, stderr=True,
|
|
stdin=False, tty=False, privileged=True, user='', environment=None,
|
|
workdir=None,
|
|
)
|
|
client.api.exec_start.assert_called_with(
|
|
FAKE_EXEC_ID, detach=False, tty=False, stream=True, socket=False,
|
|
demux=False,
|
|
)
|
|
|
|
def test_exec_run_failure(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.exec_run("docker ps", privileged=True, stream=False)
|
|
client.api.exec_create.assert_called_with(
|
|
FAKE_CONTAINER_ID, "docker ps", stdout=True, stderr=True,
|
|
stdin=False, tty=False, privileged=True, user='', environment=None,
|
|
workdir=None,
|
|
)
|
|
client.api.exec_start.assert_called_with(
|
|
FAKE_EXEC_ID, detach=False, tty=False, stream=False, socket=False,
|
|
demux=False,
|
|
)
|
|
|
|
def test_export(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.export()
|
|
client.api.export.assert_called_with(
|
|
FAKE_CONTAINER_ID, DEFAULT_DATA_CHUNK_SIZE
|
|
)
|
|
|
|
def test_get_archive(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.get_archive('foo')
|
|
client.api.get_archive.assert_called_with(
|
|
FAKE_CONTAINER_ID, 'foo', DEFAULT_DATA_CHUNK_SIZE, False
|
|
)
|
|
|
|
def test_image(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
assert container.image.id == FAKE_IMAGE_ID
|
|
|
|
def test_kill(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.kill(signal=5)
|
|
client.api.kill.assert_called_with(FAKE_CONTAINER_ID, signal=5)
|
|
|
|
def test_labels(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
assert container.labels == {'foo': 'bar'}
|
|
|
|
def test_logs(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.logs()
|
|
client.api.logs.assert_called_with(FAKE_CONTAINER_ID)
|
|
|
|
def test_pause(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.pause()
|
|
client.api.pause.assert_called_with(FAKE_CONTAINER_ID)
|
|
|
|
def test_put_archive(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.put_archive('path', 'foo')
|
|
client.api.put_archive.assert_called_with(FAKE_CONTAINER_ID,
|
|
'path', 'foo')
|
|
|
|
def test_remove(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.remove()
|
|
client.api.remove_container.assert_called_with(FAKE_CONTAINER_ID)
|
|
|
|
def test_rename(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.rename("foo")
|
|
client.api.rename.assert_called_with(FAKE_CONTAINER_ID, "foo")
|
|
|
|
def test_resize(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.resize(1, 2)
|
|
client.api.resize.assert_called_with(FAKE_CONTAINER_ID, 1, 2)
|
|
|
|
def test_restart(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.restart()
|
|
client.api.restart.assert_called_with(FAKE_CONTAINER_ID)
|
|
|
|
def test_start(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.start()
|
|
client.api.start.assert_called_with(FAKE_CONTAINER_ID)
|
|
|
|
def test_stats(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.stats()
|
|
client.api.stats.assert_called_with(FAKE_CONTAINER_ID)
|
|
|
|
def test_stop(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.stop()
|
|
client.api.stop.assert_called_with(FAKE_CONTAINER_ID)
|
|
|
|
def test_top(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.top()
|
|
client.api.top.assert_called_with(FAKE_CONTAINER_ID)
|
|
|
|
def test_unpause(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.unpause()
|
|
client.api.unpause.assert_called_with(FAKE_CONTAINER_ID)
|
|
|
|
def test_update(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.update(cpu_shares=2)
|
|
client.api.update_container.assert_called_with(FAKE_CONTAINER_ID,
|
|
cpu_shares=2)
|
|
|
|
def test_wait(self):
|
|
client = make_fake_client()
|
|
container = client.containers.get(FAKE_CONTAINER_ID)
|
|
container.wait()
|
|
client.api.wait.assert_called_with(FAKE_CONTAINER_ID)
|