| 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) |