<commit_before>import os
import signal
import tempfile

import docker
from docker.constants import IS_WINDOWS_PLATFORM
from docker.utils.socket import next_frame_size
from docker.utils.socket import read_exactly

import pytest

import six

from .base import BUSYBOX, BaseAPIIntegrationTest
from .. import helpers
from ..helpers import requires_api_version


class ListContainersTest(BaseAPIIntegrationTest):
    def test_list_containers(self):
        res0 = self.client.containers(all=True)
        size = len(res0)
        res1 = self.client.create_container(BUSYBOX, 'true')
        self.assertIn('Id', res1)
        self.client.start(res1['Id'])
        self.tmp_containers.append(res1['Id'])
        res2 = self.client.containers(all=True)
        self.assertEqual(size + 1, len(res2))
        retrieved = [x for x in res2 if x['Id'].startswith(res1['Id'])]
        self.assertEqual(len(retrieved), 1)
        retrieved = retrieved[0]
        self.assertIn('Command', retrieved)
        self.assertEqual(retrieved['Command'], six.text_type('true'))
        self.assertIn('Image', retrieved)
        self.assertRegex(retrieved['Image'], r'busybox:.*')
        self.assertIn('Status', retrieved)


class CreateContainerTest(BaseAPIIntegrationTest):

    def test_create(self):
        res = self.client.create_container(BUSYBOX, 'true')
        self.assertIn('Id', res)
        self.tmp_containers.append(res['Id'])

    def test_create_with_host_pid_mode(self):
        ctnr = self.client.create_container(
            BUSYBOX, 'true', host_config=self.client.create_host_config(
                pid_mode='host', network_mode='none'
            )
        )
        self.assertIn('Id', ctnr)
        self.tmp_containers.append(ctnr['Id'])
        self.client.start(ctnr)
        inspect = self.client.inspect_container(ctnr)
        self.assertIn('HostConfig', inspect)
        host_config = inspect['HostConfig']
        self.assertIn('PidMode', host_config)
        self.assertEqual(host_config['PidMode'], 'host')

    def test_create_with_links(self):
        res0 = self.client.create_container(
            BUSYBOX, 'cat',
            detach=True, stdin_open=True,
            environment={'FOO': '1'})

        container1_id = res0['Id']
        self.tmp_containers.append(container1_id)

        self.client.start(container1_id)

        res1 = self.client.create_container(
            BUSYBOX, 'cat',
            detach=True, stdin_open=True,
            environment={'FOO': '1'})

        container2_id = res1['Id']
        self.tmp_containers.append(container2_id)

        self.client.start(container2_id)

        # we don't want the first /
        link_path1 = self.client.inspect_container(container1_id)['Name'][1:]
        link_alias1 = 'mylink1'
        link_env_prefix1 = link_alias1.upper()

        link_path2 = self.client.inspect_container(container2_id)['Name'][1:]
        link_alias2 = 'mylink2'
        link_env_prefix2 = link_alias2.upper()

        res2 = self.client.create_container(
            BUSYBOX, 'env', host_config=self.client.create_host_config(
                links={link_path1: link_alias1, link_path2: link_alias2},
                network_mode='bridge'
            )
        )
        container3_id = res2['Id']
        self.tmp_containers.append(container3_id)
        self.client.start(container3_id)
        self.assertEqual(self.client.wait(container3_id), 0)

        logs = self.client.logs(container3_id)
        if six.PY3:
            logs = logs.decode('utf-8')
        self.assertIn('{0}_NAME='.format(link_env_prefix1), logs)
        self.assertIn('{0}_ENV_FOO=1'.format(link_env_prefix1), logs)
        self.assertIn('{0}_NAME='.format(link_env_prefix2), logs)
        self.assertIn('{0}_ENV_FOO=1'.format(link_env_prefix2), logs)

    def test_create_with_restart_policy(self):
        container = self.client.create_container(
            BUSYBOX, ['sleep', '2'],
            host_config=self.client.create_host_config(
                restart_policy={"Name": "always", "MaximumRetryCount": 0},
                network_mode='none'
            )
        )
        id = container['Id']
        self.client.start(id)
        self.client.wait(id)
        with self.assertRaises(docker.errors.APIError) as exc:
            self.client.remove_container(id)
        err = exc.exception.explanation
        self.assertIn(
            'You cannot remove ', err
        )
        self.client.remove_container(id, force=True)

    def test_create_container_with_volumes_from(self):
        vol_names = ['foobar_vol0', 'foobar_vol1']

        res0 = self.client.create_container(
            BUSYBOX, 'true', name=vol_names[0]
        )
        container1_id = res0['Id']
        self.tmp_containers.append(container1_id)
        self.client.start(container1_id)

        res1 = self.client.create_container(
            BUSYBOX, 'true', name=vol_names[1]
        )
        container2_id = res1['Id']
        self.tmp_containers.append(container2_id)
        self.client.start(container2_id)
        with self.assertRaises(docker.errors.DockerException):
            self.client.create_container(
                BUSYBOX, 'cat', detach=True, stdin_open=True,
                volumes_from=vol_names
            )
        res2 = self.client.create_container(
            BUSYBOX, 'cat', detach=True, stdin_open=True,
            host_config=self.client.create_host_config(
                volumes_from=vol_names, network_mode='none'
            )
        )
        container3_id = res2['Id']
        self.tmp_containers.append(container3_id)
        self.client.start(container3_id)

        info = self.client.inspect_container(res2['Id'])
        self.assertCountEqual(info['HostConfig']['VolumesFrom'], vol_names)

    def create_container_readonly_fs(self):
        ctnr = self.client.create_container(
            BUSYBOX, ['mkdir', '/shrine'],
            host_config=self.client.create_host_config(
                read_only=True, network_mode='none'
            )
        )
        self.assertIn('Id', ctnr)
        self.tmp_containers.append(ctnr['Id'])
        self.client.start(ctnr)
        res = self.client.wait(ctnr)
        self.assertNotEqual(res, 0)

    def create_container_with_name(self):
        res = self.client.create_container(BUSYBOX, 'true', name='foobar')
        self.assertIn('Id', res)
        self.tmp_containers.append(res['Id'])
        inspect = self.client.inspect_container(res['Id'])
        self.assertIn('Name', inspect)
        self.assertEqual('/foobar', inspect['Name'])

    def create_container_privileged(self):
        res = self.client.create_container(
            BUSYBOX, 'true', host_config=self.client.create_host_config(
                privileged=True, network_mode='none'
            )
        )
        self.assertIn('Id', res)
        self.tmp_containers.append(res['Id'])
        self.client.start(res['Id'])
        inspect = self.client.inspect_container(res['Id'])
        self.assertIn('Config', inspect)
        self.assertIn('Id', inspect)
        self.assertTrue(inspect['Id'].startswith(res['Id']))
        self.assertIn('Image', inspect)
        self.assertIn('State', inspect)
        self.assertIn('Running', inspect['State'])
        if not inspect['State']['Running']:
            self.assertIn('ExitCode', inspect['State'])
            self.assertEqual(inspect['State']['ExitCode'], 0)
        # Since Nov 2013, the Privileged flag is no longer part of the
        # container's config exposed via the API (safety concerns?).
        #
        if 'Privileged' in inspect['Config']:
            self.assertEqual(inspect['Config']['Privileged'], True)

    def test_create_with_mac_address(self):
        mac_address_expected = "02:42:ac:11:00:0a"
        container = self.client.create_container(
            BUSYBOX, ['sleep', '60'], mac_address=mac_address_expected)

        id = container['Id']

        self.client.start(container)
        res = self.client.inspect_container(container['Id'])
        self.assertEqual(mac_address_expected,
                         res['NetworkSettings']['MacAddress'])

        self.client.kill(id)

    @requires_api_version('1.20')
    def test_group_id_ints(self):
        container = self.client.create_container(
            BUSYBOX, 'id -G',
            host_config=self.client.create_host_config(group_add=[1000, 1001])
        )
        self.tmp_containers.append(container)
        self.client.start(container)
        self.client.wait(container)

        logs = self.client.logs(container)
        if six.PY3:
            logs = logs.decode('utf-8')
        groups = logs.strip().split(' ')
        self.assertIn('1000', groups)
        self.assertIn('1001', groups)

    @requires_api_version('1.20')
    def test_group_id_strings(self):
        container = self.client.create_container(
            BUSYBOX, 'id -G', host_config=self.client.create_host_config(
                group_add=['1000', '1001']
            )
        )
        self.tmp_containers.append(container)
        self.client.start(container)
        self.client.wait(container)

        logs = self.client.logs(container)
        if six.PY3:
            logs = logs.decode('utf-8')

        groups = logs.strip().split(' ')
        self.assertIn('1000', groups)
        self.assertIn('1001', groups)

    def test_valid_log_driver_and_log_opt(self):
        log_config = docker.types.LogConfig(
            type='json-file',
            config={'max-file': '100'}
        )

        container = self.client.create_container(
            BUSYBOX, ['true'],
            host_config=self.client.create_host_config(log_config=log_config)
        )
        self.tmp_containers.append(container['Id'])
        self.client.start(container)

        info = self.client.inspect_container(container)
        container_log_config = info['HostConfig']['LogConfig']

        self.assertEqual(container_log_config['Type'], log_config.type)
        self.assertEqual(container_log_config['Config'], log_config.config)

    def test_invalid_log_driver_raises_exception(self):
        log_config = docker.types.LogConfig(
            type='asdf-nope',
            config={}
        )

        expected_msg = "logger: no log driver named 'asdf-nope' is registered"
        with pytest.raises(docker.errors.APIError) as excinfo:
            # raises an internal server error 500
            container = self.client.create_container(
                BUSYBOX, ['true'], host_config=self.client.create_host_config(
                    log_config=log_config
                )
            )
            self.client.start(container)

        assert excinfo.value.explanation == expected_msg

    def test_valid_no_log_driver_specified(self):
        log_config = docker.types.LogConfig(
            type="",
            config={'max-file': '100'}
        )

        container = self.client.create_container(
            BUSYBOX, ['true'],
            host_config=self.client.create_host_config(log_config=log_config)
        )
        self.tmp_containers.append(container['Id'])
        self.client.start(container)

        info = self.client.inspect_container(container)
        container_log_config = info['HostConfig']['LogConfig']

        self.assertEqual(container_log_config['Type'], "json-file")
        self.assertEqual(container_log_config['Config'], log_config.config)

    def test_valid_no_config_specified(self):
        log_config = docker.types.LogConfig(
            type="json-file",
            config=None
        )

        container = self.client.create_container(
            BUSYBOX, ['true'],
            host_config=self.client.create_host_config(log_config=log_config)
        )
        self.tmp_containers.append(container['Id'])
        self.client.start(container)

        info = self.client.inspect_container(container)
        container_log_config = info['HostConfig']['LogConfig']

        self.assertEqual(container_log_config['Type'], "json-file")
        self.assertEqual(container_log_config['Config'], {})

    def test_create_with_memory_constraints_with_str(self):
        ctnr = self.client.create_container(
            BUSYBOX, 'true',
            host_config=self.client.create_host_config(
                memswap_limit='1G',
                mem_limit='700M'
            )
        )
        self.assertIn('Id', ctnr)
        self.tmp_containers.append(ctnr['Id'])
        self.client.start(ctnr)
        inspect = self.client.inspect_container(ctnr)

        self.assertIn('HostConfig', inspect)
        host_config = inspect['HostConfig']
        for limit in ['Memory', 'MemorySwap']:
            self.assertIn(limit, host_config)

    def test_create_with_memory_constraints_with_int(self):
        ctnr = self.client.create_container(
            BUSYBOX, 'true',
            host_config=self.client.create_host_config(mem_swappiness=40)
        )
        self.assertIn('Id', ctnr)
        self.tmp_containers.append(ctnr['Id'])
        self.client.start(ctnr)
        inspect = self.client.inspect_container(ctnr)

        self.assertIn('HostConfig', inspect)
        host_config = inspect['HostConfig']
        self.assertIn('MemorySwappiness', host_config)

    def test_create_with_environment_variable_no_value(self):
        container = self.client.create_container(
            BUSYBOX,
            ['echo'],
            environment={'Foo': None, 'Other': 'one', 'Blank': ''},
        )
        self.tmp_containers.append(container['Id'])
        config = self.client.inspect_container(container['Id'])
        assert (
            sorted(config['Config']['Env']) ==
            sorted(['Foo', 'Other=one', 'Blank='])
        )

    @requires_api_version('1.22')
    def test_create_with_tmpfs(self):
        tmpfs = {
            '/tmp1': 'size=3M'
        }

        container = self.client.create_container(
            BUSYBOX,
            ['echo'],
            host_config=self.client.create_host_config(
                tmpfs=tmpfs))

        self.tmp_containers.append(container['Id'])
        config = self.client.inspect_container(container)
        assert config['HostConfig']['Tmpfs'] == tmpfs

    @requires_api_version('1.24')
    def test_create_with_isolation(self):
        container = self.client.create_container(
            BUSYBOX, ['echo'], host_config=self.client.create_host_config(
                isolation='default'
            )
        )
        self.tmp_containers.append(container['Id'])
        config = self.client.inspect_container(container)
        assert config['HostConfig']['Isolation'] == 'default'

    @requires_api_version('1.25')
    def test_create_with_auto_remove(self):
        host_config = self.client.create_host_config(
            auto_remove=True
        )
        container = self.client.create_container(
            BUSYBOX, ['echo', 'test'], host_config=host_config
        )
        self.tmp_containers.append(container['Id'])
        config = self.client.inspect_container(container)
        assert config['HostConfig']['AutoRemove'] is True

    @requires_api_version('1.25')
    def test_create_with_stop_timeout(self):
        container = self.client.create_container(
            BUSYBOX, ['echo', 'test'], stop_timeout=25
        )
        self.tmp_containers.append(container['Id'])
        config = self.client.inspect_container(container)
        assert config['Config']['StopTimeout'] == 25

    @requires_api_version('1.24')
    @pytest.mark.xfail(True, reason='Not supported on most drivers')
    def test_create_with_storage_opt(self):
        host_config = self.client.create_host_config(
            storage_opt={'size': '120G'}
        )
        container = self.client.create_container(
            BUSYBOX, ['echo', 'test'], host_config=host_config
        )
        self.tmp_containers.append(container)
        config = self.client.inspect_container(container)
        assert config['HostConfig']['StorageOpt'] == {
            'size': '120G'
        }

    @requires_api_version('1.25')
    def test_create_with_init(self):
        ctnr = self.client.create_container(
            BUSYBOX, 'true',
            host_config=self.client.create_host_config(
                init=True
            )
        )
        self.tmp_containers.append(ctnr['Id'])
        config = self.client.inspect_container(ctnr)
        assert config['HostConfig']['Init'] is True

    @pytest.mark.xfail(True, reason='init-path removed in 17.05.0')
    @requires_api_version('1.25')
    def test_create_with_init_path(self):
        ctnr = self.client.create_container(
            BUSYBOX, 'true',
            host_config=self.client.create_host_config(
                init_path="/usr/libexec/docker-init"
            )
        )
        self.tmp_containers.append(ctnr['Id'])
        config = self.client.inspect_container(ctnr)
        assert config['HostConfig']['InitPath'] == "/usr/libexec/docker-init"


class VolumeBindTest(BaseAPIIntegrationTest):
    def setUp(self):
        super(VolumeBindTest, self).setUp()

        self.mount_dest = '/mnt'

        # Get a random pathname - we don't need it to exist locally
        self.mount_origin = tempfile.mkdtemp()
        self.filename = 'shared.txt'

        self.run_with_volume(
            False,
            BUSYBOX,
            ['touch', os.path.join(self.mount_dest, self.filename)],
        )
<commit_msg>Add integration test for CPU realtime options<commit_after>import os
import signal
import tempfile

import docker
from docker.constants import IS_WINDOWS_PLATFORM
from docker.utils.socket import next_frame_size
from docker.utils.socket import read_exactly

import pytest

import six

from .base import BUSYBOX, BaseAPIIntegrationTest
from .. import helpers
from ..helpers import requires_api_version


class ListContainersTest(BaseAPIIntegrationTest):
    def test_list_containers(self):
        res0 = self.client.containers(all=True)
        size = len(res0)
        res1 = self.client.create_container(BUSYBOX, 'true')
        self.assertIn('Id', res1)
        self.client.start(res1['Id'])
        self.tmp_containers.append(res1['Id'])
        res2 = self.client.containers(all=True)
        self.assertEqual(size + 1, len(res2))
        retrieved = [x for x in res2 if x['Id'].startswith(res1['Id'])]
        self.assertEqual(len(retrieved), 1)
        retrieved = retrieved[0]
        self.assertIn('Command', retrieved)
        self.assertEqual(retrieved['Command'], six.text_type('true'))
        self.assertIn('Image', retrieved)
        self.assertRegex(retrieved['Image'], r'busybox:.*')
        self.assertIn('Status', retrieved)


class CreateContainerTest(BaseAPIIntegrationTest):

    def test_create(self):
        res = self.client.create_container(BUSYBOX, 'true')
        self.assertIn('Id', res)
        self.tmp_containers.append(res['Id'])

    def test_create_with_host_pid_mode(self):
        ctnr = self.client.create_container(
            BUSYBOX, 'true', host_config=self.client.create_host_config(
                pid_mode='host', network_mode='none'
            )
        )
        self.assertIn('Id', ctnr)
        self.tmp_containers.append(ctnr['Id'])
        self.client.start(ctnr)
        inspect = self.client.inspect_container(ctnr)
        self.assertIn('HostConfig', inspect)
        host_config = inspect['HostConfig']
        self.assertIn('PidMode', host_config)
        self.assertEqual(host_config['PidMode'], 'host')

    def test_create_with_links(self):
        res0 = self.client.create_container(
            BUSYBOX, 'cat',
            detach=True, stdin_open=True,
            environment={'FOO': '1'})

        container1_id = res0['Id']
        self.tmp_containers.append(container1_id)

        self.client.start(container1_id)

        res1 = self.client.create_container(
            BUSYBOX, 'cat',
            detach=True, stdin_open=True,
            environment={'FOO': '1'})

        container2_id = res1['Id']
        self.tmp_containers.append(container2_id)

        self.client.start(container2_id)

        # we don't want the first /
        link_path1 = self.client.inspect_container(container1_id)['Name'][1:]
        link_alias1 = 'mylink1'
        link_env_prefix1 = link_alias1.upper()

        link_path2 = self.client.inspect_container(container2_id)['Name'][1:]
        link_alias2 = 'mylink2'
        link_env_prefix2 = link_alias2.upper()

        res2 = self.client.create_container(
            BUSYBOX, 'env', host_config=self.client.create_host_config(
                links={link_path1: link_alias1, link_path2: link_alias2},
                network_mode='bridge'
            )
        )
        container3_id = res2['Id']
        self.tmp_containers.append(container3_id)
        self.client.start(container3_id)
        self.assertEqual(self.client.wait(container3_id), 0)

        logs = self.client.logs(container3_id)
        if six.PY3:
            logs = logs.decode('utf-8')
        self.assertIn('{0}_NAME='.format(link_env_prefix1), logs)
        self.assertIn('{0}_ENV_FOO=1'.format(link_env_prefix1), logs)
        self.assertIn('{0}_NAME='.format(link_env_prefix2), logs)
        self.assertIn('{0}_ENV_FOO=1'.format(link_env_prefix2), logs)

    def test_create_with_restart_policy(self):
        container = self.client.create_container(
            BUSYBOX, ['sleep', '2'],
            host_config=self.client.create_host_config(
                restart_policy={"Name": "always", "MaximumRetryCount": 0},
                network_mode='none'
            )
        )
        id = container['Id']
        self.client.start(id)
        self.client.wait(id)
        with self.assertRaises(docker.errors.APIError) as exc:
            self.client.remove_container(id)
        err = exc.exception.explanation
        self.assertIn(
            'You cannot remove ', err
        )
        self.client.remove_container(id, force=True)

    def test_create_container_with_volumes_from(self):
        vol_names = ['foobar_vol0', 'foobar_vol1']

        res0 = self.client.create_container(
            BUSYBOX, 'true', name=vol_names[0]
        )
        container1_id = res0['Id']
        self.tmp_containers.append(container1_id)
        self.client.start(container1_id)

        res1 = self.client.create_container(
            BUSYBOX, 'true', name=vol_names[1]
        )
        container2_id = res1['Id']
        self.tmp_containers.append(container2_id)
        self.client.start(container2_id)
        with self.assertRaises(docker.errors.DockerException):
            self.client.create_container(
                BUSYBOX, 'cat', detach=True, stdin_open=True,
                volumes_from=vol_names
            )
        res2 = self.client.create_container(
            BUSYBOX, 'cat', detach=True, stdin_open=True,
            host_config=self.client.create_host_config(
                volumes_from=vol_names, network_mode='none'
            )
        )
        container3_id = res2['Id']
        self.tmp_containers.append(container3_id)
        self.client.start(container3_id)

        info = self.client.inspect_container(res2['Id'])
        self.assertCountEqual(info['HostConfig']['VolumesFrom'], vol_names)

    def create_container_readonly_fs(self):
        ctnr = self.client.create_container(
            BUSYBOX, ['mkdir', '/shrine'],
            host_config=self.client.create_host_config(
                read_only=True, network_mode='none'
            )
        )
        self.assertIn('Id', ctnr)
        self.tmp_containers.append(ctnr['Id'])
        self.client.start(ctnr)
        res = self.client.wait(ctnr)
        self.assertNotEqual(res, 0)

    def create_container_with_name(self):
        res = self.client.create_container(BUSYBOX, 'true', name='foobar')
        self.assertIn('Id', res)
        self.tmp_containers.append(res['Id'])
        inspect = self.client.inspect_container(res['Id'])
        self.assertIn('Name', inspect)
        self.assertEqual('/foobar', inspect['Name'])

    def create_container_privileged(self):
        res = self.client.create_container(
            BUSYBOX, 'true', host_config=self.client.create_host_config(
                privileged=True, network_mode='none'
            )
        )
        self.assertIn('Id', res)
        self.tmp_containers.append(res['Id'])
        self.client.start(res['Id'])
        inspect = self.client.inspect_container(res['Id'])
        self.assertIn('Config', inspect)
        self.assertIn('Id', inspect)
        self.assertTrue(inspect['Id'].startswith(res['Id']))
        self.assertIn('Image', inspect)
        self.assertIn('State', inspect)
        self.assertIn('Running', inspect['State'])
        if not inspect['State']['Running']:
            self.assertIn('ExitCode', inspect['State'])
            self.assertEqual(inspect['State']['ExitCode'], 0)
        # Since Nov 2013, the Privileged flag is no longer part of the
        # container's config exposed via the API (safety concerns?).
        #
        if 'Privileged' in inspect['Config']:
            self.assertEqual(inspect['Config']['Privileged'], True)

    def test_create_with_mac_address(self):
        mac_address_expected = "02:42:ac:11:00:0a"
        container = self.client.create_container(
            BUSYBOX, ['sleep', '60'], mac_address=mac_address_expected)

        id = container['Id']

        self.client.start(container)
        res = self.client.inspect_container(container['Id'])
        self.assertEqual(mac_address_expected,
                         res['NetworkSettings']['MacAddress'])

        self.client.kill(id)

    @requires_api_version('1.20')
    def test_group_id_ints(self):
        container = self.client.create_container(
            BUSYBOX, 'id -G',
            host_config=self.client.create_host_config(group_add=[1000, 1001])
        )
        self.tmp_containers.append(container)
        self.client.start(container)
        self.client.wait(container)

        logs = self.client.logs(container)
        if six.PY3:
            logs = logs.decode('utf-8')
        groups = logs.strip().split(' ')
        self.assertIn('1000', groups)
        self.assertIn('1001', groups)

    @requires_api_version('1.20')
    def test_group_id_strings(self):
        container = self.client.create_container(
            BUSYBOX, 'id -G', host_config=self.client.create_host_config(
                group_add=['1000', '1001']
            )
        )
        self.tmp_containers.append(container)
        self.client.start(container)
        self.client.wait(container)

        logs = self.client.logs(container)
        if six.PY3:
            logs = logs.decode('utf-8')

        groups = logs.strip().split(' ')
        self.assertIn('1000', groups)
        self.assertIn('1001', groups)

    def test_valid_log_driver_and_log_opt(self):
        log_config = docker.types.LogConfig(
            type='json-file',
            config={'max-file': '100'}
        )

        container = self.client.create_container(
            BUSYBOX, ['true'],
            host_config=self.client.create_host_config(log_config=log_config)
        )
        self.tmp_containers.append(container['Id'])
        self.client.start(container)

        info = self.client.inspect_container(container)
        container_log_config = info['HostConfig']['LogConfig']

        self.assertEqual(container_log_config['Type'], log_config.type)
        self.assertEqual(container_log_config['Config'], log_config.config)

    def test_invalid_log_driver_raises_exception(self):
        log_config = docker.types.LogConfig(
            type='asdf-nope',
            config={}
        )

        expected_msg = "logger: no log driver named 'asdf-nope' is registered"
        with pytest.raises(docker.errors.APIError) as excinfo:
            # raises an internal server error 500
            container = self.client.create_container(
                BUSYBOX, ['true'], host_config=self.client.create_host_config(
                    log_config=log_config
                )
            )
            self.client.start(container)

        assert excinfo.value.explanation == expected_msg

    def test_valid_no_log_driver_specified(self):
        log_config = docker.types.LogConfig(
            type="",
            config={'max-file': '100'}
        )

        container = self.client.create_container(
            BUSYBOX, ['true'],
            host_config=self.client.create_host_config(log_config=log_config)
        )
        self.tmp_containers.append(container['Id'])
        self.client.start(container)

        info = self.client.inspect_container(container)
        container_log_config = info['HostConfig']['LogConfig']

        self.assertEqual(container_log_config['Type'], "json-file")
        self.assertEqual(container_log_config['Config'], log_config.config)

    def test_valid_no_config_specified(self):
        log_config = docker.types.LogConfig(
            type="json-file",
            config=None
        )

        container = self.client.create_container(
            BUSYBOX, ['true'],
            host_config=self.client.create_host_config(log_config=log_config)
        )
        self.tmp_containers.append(container['Id'])
        self.client.start(container)

        info = self.client.inspect_container(container)
        container_log_config = info['HostConfig']['LogConfig']

        self.assertEqual(container_log_config['Type'], "json-file")
        self.assertEqual(container_log_config['Config'], {})

    def test_create_with_memory_constraints_with_str(self):
        ctnr = self.client.create_container(
            BUSYBOX, 'true',
            host_config=self.client.create_host_config(
                memswap_limit='1G',
                mem_limit='700M'
            )
        )
        self.assertIn('Id', ctnr)
        self.tmp_containers.append(ctnr['Id'])
        self.client.start(ctnr)
        inspect = self.client.inspect_container(ctnr)

        self.assertIn('HostConfig', inspect)
        host_config = inspect['HostConfig']
        for limit in ['Memory', 'MemorySwap']:
            self.assertIn(limit, host_config)

    def test_create_with_memory_constraints_with_int(self):
        ctnr = self.client.create_container(
            BUSYBOX, 'true',
            host_config=self.client.create_host_config(mem_swappiness=40)
        )
        self.assertIn('Id', ctnr)
        self.tmp_containers.append(ctnr['Id'])
        self.client.start(ctnr)
        inspect = self.client.inspect_container(ctnr)

        self.assertIn('HostConfig', inspect)
        host_config = inspect['HostConfig']
        self.assertIn('MemorySwappiness', host_config)

    def test_create_with_environment_variable_no_value(self):
        container = self.client.create_container(
            BUSYBOX,
            ['echo'],
            environment={'Foo': None, 'Other': 'one', 'Blank': ''},
        )
        self.tmp_containers.append(container['Id'])
        config = self.client.inspect_container(container['Id'])
        assert (
            sorted(config['Config']['Env']) ==
            sorted(['Foo', 'Other=one', 'Blank='])
        )

    @requires_api_version('1.22')
    def test_create_with_tmpfs(self):
        tmpfs = {
            '/tmp1': 'size=3M'
        }

        container = self.client.create_container(
            BUSYBOX,
            ['echo'],
            host_config=self.client.create_host_config(
                tmpfs=tmpfs))

        self.tmp_containers.append(container['Id'])
        config = self.client.inspect_container(container)
        assert config['HostConfig']['Tmpfs'] == tmpfs

    @requires_api_version('1.24')
    def test_create_with_isolation(self):
        container = self.client.create_container(
            BUSYBOX, ['echo'], host_config=self.client.create_host_config(
                isolation='default'
            )
        )
        self.tmp_containers.append(container['Id'])
        config = self.client.inspect_container(container)
        assert config['HostConfig']['Isolation'] == 'default'

    @requires_api_version('1.25')
    def test_create_with_auto_remove(self):
        host_config = self.client.create_host_config(
            auto_remove=True
        )
        container = self.client.create_container(
            BUSYBOX, ['echo', 'test'], host_config=host_config
        )
        self.tmp_containers.append(container['Id'])
        config = self.client.inspect_container(container)
        assert config['HostConfig']['AutoRemove'] is True

    @requires_api_version('1.25')
    def test_create_with_stop_timeout(self):
        container = self.client.create_container(
            BUSYBOX, ['echo', 'test'], stop_timeout=25
        )
        self.tmp_containers.append(container['Id'])
        config = self.client.inspect_container(container)
        assert config['Config']['StopTimeout'] == 25

    @requires_api_version('1.24')
    @pytest.mark.xfail(True, reason='Not supported on most drivers')
    def test_create_with_storage_opt(self):
        host_config = self.client.create_host_config(
            storage_opt={'size': '120G'}
        )
        container = self.client.create_container(
            BUSYBOX, ['echo', 'test'], host_config=host_config
        )
        self.tmp_containers.append(container)
        config = self.client.inspect_container(container)
        assert config['HostConfig']['StorageOpt'] == {
            'size': '120G'
        }

    @requires_api_version('1.25')
    def test_create_with_init(self):
        ctnr = self.client.create_container(
            BUSYBOX, 'true',
            host_config=self.client.create_host_config(
                init=True
            )
        )
        self.tmp_containers.append(ctnr['Id'])
        config = self.client.inspect_container(ctnr)
        assert config['HostConfig']['Init'] is True

    @pytest.mark.xfail(True, reason='init-path removed in 17.05.0')
    @requires_api_version('1.25')
    def test_create_with_init_path(self):
        ctnr = self.client.create_container(
            BUSYBOX, 'true',
            host_config=self.client.create_host_config(
                init_path="/usr/libexec/docker-init"
            )
        )
        self.tmp_containers.append(ctnr['Id'])
        config = self.client.inspect_container(ctnr)
        assert config['HostConfig']['InitPath'] == "/usr/libexec/docker-init"

    @requires_api_version('1.24')
    @pytest.mark.xfail(not os.path.exists('/sys/fs/cgroup/cpu.rt_runtime_us'),
                       reason='CONFIG_RT_GROUP_SCHED isn\'t enabled')
    def test_create_with_cpu_rt_options(self):
        ctnr = self.client.create_container(
            BUSYBOX, 'true', host_config=self.client.create_host_config(
                cpu_rt_period=1000, cpu_rt_runtime=500
            )
        )
        self.tmp_containers.append(ctnr)
        config = self.client.inspect_container(ctnr)
        assert config['HostConfig']['CpuRealtimeRuntime'] == 500
        assert config['HostConfig']['CpuRealtimePeriod'] == 1000


class VolumeBindTest(BaseAPIIntegrationTest):
    def setUp(self):
        super(VolumeBindTest, self).setUp()

        self.mount_dest = '/mnt'

        # Get a random pathname - we don't need it to exist locally
        self.mount_origin = tempfile.mkdtemp()
        self.filename = 'shared.txt'

        self.run_with_volume(
            False,
            BUSYBOX,
            ['touch', os.path.join(self.mount_dest, self.filename)],
        )

{'fraction_non_alphanumeric': 0.12099954924498535, 'fraction_numerical': 0.011578769438810007, 'mean_word_length': 2.517340467697186, 'pattern_counts': {'":': 4, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3386097', 'n_tokens_mistral': 10984, 'n_tokens_neox': 10541, 'n_words': 1522}