import os
import signal
import tempfile

import docker
import pytest
import six

from ..base import requires_api_version
from .. import helpers

BUSYBOX = helpers.BUSYBOX


class ListContainersTest(helpers.BaseTestCase):
    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(helpers.BaseTestCase):

    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.response.text
        self.assertIn(
            'You cannot remove a running container', 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):
        if not helpers.exec_driver_is_native():
            pytest.skip('Exec driver not native')

        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.utils.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.utils.LogConfig(
            type='asdf-nope',
            config={}
        )

        container = self.client.create_container(
            BUSYBOX, ['true'],
            host_config=self.client.create_host_config(log_config=log_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
            self.client.start(container)

        assert expected_msg in str(excinfo.value)

    def test_valid_no_log_driver_specified(self):
        log_config = docker.utils.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.utils.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_host_config_exception_raising(self):
        self.assertRaises(TypeError,
                          self.client.create_host_config, mem_swappiness='40')

        self.assertRaises(ValueError,
                          self.client.create_host_config, pid_mode='40')


class VolumeBindTest(helpers.BaseTestCase):
    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)],
        )

    def test_create_with_binds_rw(self):

        container = self.run_with_volume(
            False,
            BUSYBOX,
            ['ls', self.mount_dest],
        )
        logs = self.client.logs(container)

        if six.PY3:
            logs = logs.decode('utf-8')
        self.assertIn(self.filename, logs)
        inspect_data = self.client.inspect_container(container)
        self.check_container_data(inspect_data, True)

    def test_create_with_binds_ro(self):
        self.run_with_volume(
            False,
            BUSYBOX,
            ['touch', os.path.join(self.mount_dest, self.filename)],
        )
        container = self.run_with_volume(
            True,
            BUSYBOX,
            ['ls', self.mount_dest],
        )
        logs = self.client.logs(container)

        if six.PY3:
            logs = logs.decode('utf-8')
        self.assertIn(self.filename, logs)

        inspect_data = self.client.inspect_container(container)
        self.check_container_data(inspect_data, False)

    def check_container_data(self, inspect_data, rw):
        if docker.utils.compare_version('1.20', self.client._version) < 0:
            self.assertIn('Volumes', inspect_data)
            self.assertIn(self.mount_dest, inspect_data['Volumes'])
            self.assertEqual(
                self.mount_origin, inspect_data['Volumes'][self.mount_dest]
            )
            self.assertIn(self.mount_dest, inspect_data['VolumesRW'])
            self.assertFalse(inspect_data['VolumesRW'][self.mount_dest])
        else:
            self.assertIn('Mounts', inspect_data)
            filtered = list(filter(
                lambda x: x['Destination'] == self.mount_dest,
                inspect_data['Mounts']
            ))
            self.assertEqual(len(filtered), 1)
            mount_data = filtered[0]
            self.assertEqual(mount_data['Source'], self.mount_origin)
            self.assertEqual(mount_data['RW'], rw)

    def run_with_volume(self, ro, *args, **kwargs):
        return self.run_container(
            *args,
            volumes={self.mount_dest: {}},
            host_config=self.client.create_host_config(
                binds={
                    self.mount_origin: {
                        'bind': self.mount_dest,
                        'ro': ro,
                    },
                },
                network_mode='none'
            ),
            **kwargs
        )


@requires_api_version('1.20')
class ArchiveTest(helpers.BaseTestCase):
    def test_get_file_archive_from_container(self):
        data = 'The Maid and the Pocket Watch of Blood'
        ctnr = self.client.create_container(
            BUSYBOX, 'sh -c "echo {0} > /vol1/data.txt"'.format(data),
            volumes=['/vol1']
        )
        self.tmp_containers.append(ctnr)
        self.client.start(ctnr)
        self.client.wait(ctnr)
        with tempfile.NamedTemporaryFile() as destination:
            strm, stat = self.client.get_archive(ctnr, '/vol1/data.txt')
            for d in strm:
                destination.write(d)
            destination.seek(0)
            retrieved_data = helpers.untar_file(destination, 'data.txt')
            if six.PY3:
                retrieved_data = retrieved_data.decode('utf-8')
            self.assertEqual(data, retrieved_data.strip())

    def test_get_file_stat_from_container(self):
        data = 'The Maid and the Pocket Watch of Blood'
        ctnr = self.client.create_container(
            BUSYBOX, 'sh -c "echo -n {0} > /vol1/data.txt"'.format(data),
            volumes=['/vol1']
        )
        self.tmp_containers.append(ctnr)
        self.client.start(ctnr)
        self.client.wait(ctnr)
        strm, stat = self.client.get_archive(ctnr, '/vol1/data.txt')
        self.assertIn('name', stat)
        self.assertEqual(stat['name'], 'data.txt')
        self.assertIn('size', stat)
        self.assertEqual(stat['size'], len(data))

    def test_copy_file_to_container(self):
        data = b'Deaf To All But The Song'
        with tempfile.NamedTemporaryFile() as test_file:
            test_file.write(data)
            test_file.seek(0)
            ctnr = self.client.create_container(
                BUSYBOX,
                'cat {0}'.format(
                    os.path.join('/vol1', os.path.basename(test_file.name))
                ),
                volumes=['/vol1']
            )
            self.tmp_containers.append(ctnr)
            with helpers.simple_tar(test_file.name) as test_tar:
                self.client.put_archive(ctnr, '/vol1', test_tar)
        self.client.start(ctnr)
        self.client.wait(ctnr)
        logs = self.client.logs(ctnr)
        if six.PY3:
            logs = logs.decode('utf-8')
            data = data.decode('utf-8')
        self.assertEqual(logs.strip(), data)

    def test_copy_directory_to_container(self):
        files = ['a.py', 'b.py', 'foo/b.py']
        dirs = ['foo', 'bar']
        base = helpers.make_tree(dirs, files)
        ctnr = self.client.create_container(
            BUSYBOX, 'ls -p /vol1', volumes=['/vol1']
        )
        self.tmp_containers.append(ctnr)
        with docker.utils.tar(base) as test_tar:
            self.client.put_archive(ctnr, '/vol1', test_tar)
        self.client.start(ctnr)
        self.client.wait(ctnr)
        logs = self.client.logs(ctnr)
        if six.PY3:
            logs = logs.decode('utf-8')
        results = logs.strip().split()
        self.assertIn('a.py', results)
        self.assertIn('b.py', results)
        self.assertIn('foo/', results)
        self.assertIn('bar/', results)


class RenameContainerTest(helpers.BaseTestCase):
    def test_rename_container(self):
        version = self.client.version()['Version']
        name = 'hong_meiling'
        res = self.client.create_container(BUSYBOX, 'true')
        self.assertIn('Id', res)
        self.tmp_containers.append(res['Id'])
        self.client.rename(res, name)
        inspect = self.client.inspect_container(res['Id'])
        self.assertIn('Name', inspect)
        if version == '1.5.0':
            self.assertEqual(name, inspect['Name'])
        else:
            self.assertEqual('/{0}'.format(name), inspect['Name'])


class StartContainerTest(helpers.BaseTestCase):
    def test_start_container(self):
        res = self.client.create_container(BUSYBOX, 'true')
        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)

    def test_start_container_with_dict_instead_of_id(self):
        res = self.client.create_container(BUSYBOX, 'true')
        self.assertIn('Id', res)
        self.tmp_containers.append(res['Id'])
        self.client.start(res)
        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)

    def test_run_shlex_commands(self):
        commands = [
            'true',
            'echo "The Young Descendant of Tepes & Septette for the '
            'Dead Princess"',
            'echo -n "The Young Descendant of Tepes & Septette for the '
            'Dead Princess"',
            '/bin/sh -c "echo Hello World"',
            '/bin/sh -c \'echo "Hello World"\'',
            'echo "\"Night of Nights\""',
            'true && echo "Night of Nights"'
        ]
        for cmd in commands:
            container = self.client.create_container(BUSYBOX, cmd)
            id = container['Id']
            self.client.start(id)
            self.tmp_containers.append(id)
            exitcode = self.client.wait(id)
            self.assertEqual(exitcode, 0, msg=cmd)


class WaitTest(helpers.BaseTestCase):
    def test_wait(self):
        res = self.client.create_container(BUSYBOX, ['sleep', '3'])
        id = res['Id']
        self.tmp_containers.append(id)
        self.client.start(id)
        exitcode = self.client.wait(id)
        self.assertEqual(exitcode, 0)
        inspect = self.client.inspect_container(id)
        self.assertIn('Running', inspect['State'])
        self.assertEqual(inspect['State']['Running'], False)
        self.assertIn('ExitCode', inspect['State'])
        self.assertEqual(inspect['State']['ExitCode'], exitcode)

    def test_wait_with_dict_instead_of_id(self):
        res = self.client.create_container(BUSYBOX, ['sleep', '3'])
        id = res['Id']
        self.tmp_containers.append(id)
        self.client.start(res)
        exitcode = self.client.wait(res)
        self.assertEqual(exitcode, 0)
        inspect = self.client.inspect_container(res)
        self.assertIn('Running', inspect['State'])
        self.assertEqual(inspect['State']['Running'], False)
        self.assertIn('ExitCode', inspect['State'])
        self.assertEqual(inspect['State']['ExitCode'], exitcode)


class LogsTest(helpers.BaseTestCase):
    def test_logs(self):
        snippet = 'Flowering Nights (Sakuya Iyazoi)'
        container = self.client.create_container(
            BUSYBOX, 'echo {0}'.format(snippet)
        )
        id = container['Id']
        self.tmp_containers.append(id)
        self.client.start(id)
        exitcode = self.client.wait(id)
        self.assertEqual(exitcode, 0)
        logs = self.client.logs(id)
        self.assertEqual(logs, (snippet + '\n').encode(encoding='ascii'))

    def test_logs_tail_option(self):
        snippet = '''Line1
Line2'''
        container = self.client.create_container(
            BUSYBOX, 'echo "{0}"'.format(snippet)
        )
        id = container['Id']
        self.tmp_containers.append(id)
        self.client.start(id)
        exitcode = self.client.wait(id)
        self.assertEqual(exitcode, 0)
        logs = self.client.logs(id, tail=1)
        self.assertEqual(logs, 'Line2\n'.encode(encoding='ascii'))

    def test_logs_streaming(self):
        snippet = 'Flowering Nights (Sakuya Iyazoi)'
        container = self.client.create_container(
            BUSYBOX, 'echo {0}'.format(snippet)
        )
        id = container['Id']
        self.tmp_containers.append(id)
        self.client.start(id)
        logs = six.binary_type()
        for chunk in self.client.logs(id, stream=True):
            logs += chunk

        exitcode = self.client.wait(id)
        self.assertEqual(exitcode, 0)

        self.assertEqual(logs, (snippet + '\n').encode(encoding='ascii'))

    def test_logs_with_dict_instead_of_id(self):
        snippet = 'Flowering Nights (Sakuya Iyazoi)'
        container = self.client.create_container(
            BUSYBOX, 'echo {0}'.format(snippet)
        )
        id = container['Id']
        self.tmp_containers.append(id)
        self.client.start(id)
        exitcode = self.client.wait(id)
        self.assertEqual(exitcode, 0)
        logs = self.client.logs(container)
        self.assertEqual(logs, (snippet + '\n').encode(encoding='ascii'))

    def test_logs_with_tail_0(self):
        snippet = 'Flowering Nights (Sakuya Iyazoi)'
        container = self.client.create_container(
            BUSYBOX, 'echo "{0}"'.format(snippet)
        )
        id = container['Id']
        self.tmp_containers.append(id)
        self.client.start(id)
        exitcode = self.client.wait(id)
        self.assertEqual(exitcode, 0)
        logs = self.client.logs(id, tail=0)
        self.assertEqual(logs, ''.encode(encoding='ascii'))


class DiffTest(helpers.BaseTestCase):
    def test_diff(self):
        container = self.client.create_container(BUSYBOX, ['touch', '/test'])
        id = container['Id']
        self.client.start(id)
        self.tmp_containers.append(id)
        exitcode = self.client.wait(id)
        self.assertEqual(exitcode, 0)
        diff = self.client.diff(id)
        test_diff = [x for x in diff if x.get('Path', None) == '/test']
        self.assertEqual(len(test_diff), 1)
        self.assertIn('Kind', test_diff[0])
        self.assertEqual(test_diff[0]['Kind'], 1)

    def test_diff_with_dict_instead_of_id(self):
        container = self.client.create_container(BUSYBOX, ['touch', '/test'])
        id = container['Id']
        self.client.start(id)
        self.tmp_containers.append(id)
        exitcode = self.client.wait(id)
        self.assertEqual(exitcode, 0)
        diff = self.client.diff(container)
        test_diff = [x for x in diff if x.get('Path', None) == '/test']
        self.assertEqual(len(test_diff), 1)
        self.assertIn('Kind', test_diff[0])
        self.assertEqual(test_diff[0]['Kind'], 1)


class StopTest(helpers.BaseTestCase):
    def test_stop(self):
        container = self.client.create_container(BUSYBOX, ['sleep', '9999'])
        id = container['Id']
        self.client.start(id)
        self.tmp_containers.append(id)
        self.client.stop(id, timeout=2)
        container_info = self.client.inspect_container(id)
        self.assertIn('State', container_info)
        state = container_info['State']
        self.assertIn('Running', state)
        self.assertEqual(state['Running'], False)

    def test_stop_with_dict_instead_of_id(self):
        container = self.client.create_container(BUSYBOX, ['sleep', '9999'])
        self.assertIn('Id', container)
        id = container['Id']
        self.client.start(container)
        self.tmp_containers.append(id)
        self.client.stop(container, timeout=2)
        container_info = self.client.inspect_container(id)
        self.assertIn('State', container_info)
        state = container_info['State']
        self.assertIn('Running', state)
        self.assertEqual(state['Running'], False)


class KillTest(helpers.BaseTestCase):
    def test_kill(self):
        container = self.client.create_container(BUSYBOX, ['sleep', '9999'])
        id = container['Id']
        self.client.start(id)
        self.tmp_containers.append(id)
        self.client.kill(id)
        container_info = self.client.inspect_container(id)
        self.assertIn('State', container_info)
        state = container_info['State']
        self.assertIn('ExitCode', state)
        if helpers.exec_driver_is_native():
            self.assertNotEqual(state['ExitCode'], 0)
        self.assertIn('Running', state)
        self.assertEqual(state['Running'], False)

    def test_kill_with_dict_instead_of_id(self):
        container = self.client.create_container(BUSYBOX, ['sleep', '9999'])
        id = container['Id']
        self.client.start(id)
        self.tmp_containers.append(id)
        self.client.kill(container)
        container_info = self.client.inspect_container(id)
        self.assertIn('State', container_info)
        state = container_info['State']
        self.assertIn('ExitCode', state)
        if helpers.exec_driver_is_native():
            self.assertNotEqual(state['ExitCode'], 0)
        self.assertIn('Running', state)
        self.assertEqual(state['Running'], False)

    def test_kill_with_signal(self):
        container = self.client.create_container(BUSYBOX, ['sleep', '60'])
        id = container['Id']
        self.client.start(id)
        self.tmp_containers.append(id)
        self.client.kill(id, signal=signal.SIGKILL)
        exitcode = self.client.wait(id)
        self.assertNotEqual(exitcode, 0)
        container_info = self.client.inspect_container(id)
        self.assertIn('State', container_info)
        state = container_info['State']
        self.assertIn('ExitCode', state)
        self.assertNotEqual(state['ExitCode'], 0)
        self.assertIn('Running', state)
        self.assertEqual(state['Running'], False, state)


class PortTest(helpers.BaseTestCase):
    def test_port(self):

        port_bindings = {
            '1111': ('127.0.0.1', '4567'),
            '2222': ('127.0.0.1', '4568')
        }

        container = self.client.create_container(
            BUSYBOX, ['sleep', '60'], ports=list(port_bindings.keys()),
            host_config=self.client.create_host_config(
                port_bindings=port_bindings, network_mode='bridge'
            )
        )
        id = container['Id']

        self.client.start(container)

        # Call the port function on each biding and compare expected vs actual
        for port in port_bindings:
            actual_bindings = self.client.port(container, port)
            port_binding = actual_bindings.pop()

            ip, host_port = port_binding['HostIp'], port_binding['HostPort']

            self.assertEqual(ip, port_bindings[port][0])
            self.assertEqual(host_port, port_bindings[port][1])

        self.client.kill(id)


class ContainerTopTest(helpers.BaseTestCase):
    def test_top(self):
        container = self.client.create_container(
            BUSYBOX, ['sleep', '60'])

        id = container['Id']

        self.client.start(container)
        res = self.client.top(container['Id'])
        self.assertEqual(
            res['Titles'],
            ['UID', 'PID', 'PPID', 'C', 'STIME', 'TTY', 'TIME', 'CMD']
        )
        self.assertEqual(len(res['Processes']), 1)
        self.assertEqual(res['Processes'][0][7], 'sleep 60')
        self.client.kill(id)

    def test_top_with_psargs(self):
        container = self.client.create_container(
            BUSYBOX, ['sleep', '60'])

        id = container['Id']

        self.client.start(container)
        res = self.client.top(container['Id'], 'waux')
        self.assertEqual(
            res['Titles'],
            ['USER', 'PID', '%CPU', '%MEM', 'VSZ', 'RSS',
                'TTY', 'STAT', 'START', 'TIME', 'COMMAND'],
        )
        self.assertEqual(len(res['Processes']), 1)
        self.assertEqual(res['Processes'][0][10], 'sleep 60')
        self.client.kill(id)


class RestartContainerTest(helpers.BaseTestCase):
    def test_restart(self):
        container = self.client.create_container(BUSYBOX, ['sleep', '9999'])
        id = container['Id']
        self.client.start(id)
        self.tmp_containers.append(id)
        info = self.client.inspect_container(id)
        self.assertIn('State', info)
        self.assertIn('StartedAt', info['State'])
        start_time1 = info['State']['StartedAt']
        self.client.restart(id, timeout=2)
        info2 = self.client.inspect_container(id)
        self.assertIn('State', info2)
        self.assertIn('StartedAt', info2['State'])
        start_time2 = info2['State']['StartedAt']
        self.assertNotEqual(start_time1, start_time2)
        self.assertIn('Running', info2['State'])
        self.assertEqual(info2['State']['Running'], True)
        self.client.kill(id)

    def test_restart_with_dict_instead_of_id(self):
        container = self.client.create_container(BUSYBOX, ['sleep', '9999'])
        self.assertIn('Id', container)
        id = container['Id']
        self.client.start(container)
        self.tmp_containers.append(id)
        info = self.client.inspect_container(id)
        self.assertIn('State', info)
        self.assertIn('StartedAt', info['State'])
        start_time1 = info['State']['StartedAt']
        self.client.restart(container, timeout=2)
        info2 = self.client.inspect_container(id)
        self.assertIn('State', info2)
        self.assertIn('StartedAt', info2['State'])
        start_time2 = info2['State']['StartedAt']
        self.assertNotEqual(start_time1, start_time2)
        self.assertIn('Running', info2['State'])
        self.assertEqual(info2['State']['Running'], True)
        self.client.kill(id)


class RemoveContainerTest(helpers.BaseTestCase):
    def test_remove(self):
        container = self.client.create_container(BUSYBOX, ['true'])
        id = container['Id']
        self.client.start(id)
        self.client.wait(id)
        self.client.remove_container(id)
        containers = self.client.containers(all=True)
        res = [x for x in containers if 'Id' in x and x['Id'].startswith(id)]
        self.assertEqual(len(res), 0)

    def test_remove_with_dict_instead_of_id(self):
        container = self.client.create_container(BUSYBOX, ['true'])
        id = container['Id']
        self.client.start(id)
        self.client.wait(id)
        self.client.remove_container(container)
        containers = self.client.containers(all=True)
        res = [x for x in containers if 'Id' in x and x['Id'].startswith(id)]
        self.assertEqual(len(res), 0)


class AttachContainerTest(helpers.BaseTestCase):
    def test_run_container_streaming(self):
        container = self.client.create_container(BUSYBOX, '/bin/sh',
                                                 detach=True, stdin_open=True)
        id = container['Id']
        self.tmp_containers.append(id)
        self.client.start(id)
        sock = self.client.attach_socket(container, ws=False)
        self.assertTrue(sock.fileno() > -1)

    def test_run_container_reading_socket(self):
        line = 'hi there and stuff and things, words!'
        # `echo` appends CRLF, `printf` doesn't
        command = "printf '{0}'".format(line)
        container = self.client.create_container(BUSYBOX, command,
                                                 detach=True, tty=False)
        ident = container['Id']
        self.tmp_containers.append(ident)

        opts = {"stdout": 1, "stream": 1, "logs": 1}
        pty_stdout = self.client.attach_socket(ident, opts)
        self.addCleanup(pty_stdout.close)

        self.client.start(ident)

        next_size = helpers.next_packet_size(pty_stdout)
        self.assertEqual(next_size, len(line))
        data = helpers.read_data(pty_stdout, next_size)
        self.assertEqual(data.decode('utf-8'), line)


class PauseTest(helpers.BaseTestCase):
    def test_pause_unpause(self):
        container = self.client.create_container(BUSYBOX, ['sleep', '9999'])
        id = container['Id']
        self.tmp_containers.append(id)
        self.client.start(container)
        self.client.pause(id)
        container_info = self.client.inspect_container(id)
        self.assertIn('State', container_info)
        state = container_info['State']
        self.assertIn('ExitCode', state)
        self.assertEqual(state['ExitCode'], 0)
        self.assertIn('Running', state)
        self.assertEqual(state['Running'], True)
        self.assertIn('Paused', state)
        self.assertEqual(state['Paused'], True)

        self.client.unpause(id)
        container_info = self.client.inspect_container(id)
        self.assertIn('State', container_info)
        state = container_info['State']
        self.assertIn('ExitCode', state)
        self.assertEqual(state['ExitCode'], 0)
        self.assertIn('Running', state)
        self.assertEqual(state['Running'], True)
        self.assertIn('Paused', state)
        self.assertEqual(state['Paused'], False)


class GetContainerStatsTest(helpers.BaseTestCase):
    @requires_api_version('1.19')
    def test_get_container_stats_no_stream(self):
        container = self.client.create_container(
            BUSYBOX, ['sleep', '60'],
        )
        self.tmp_containers.append(container)
        self.client.start(container)
        response = self.client.stats(container, stream=0)
        self.client.kill(container)

        self.assertEqual(type(response), dict)
        for key in ['read', 'networks', 'precpu_stats', 'cpu_stats',
                    'memory_stats', 'blkio_stats']:
            self.assertIn(key, response)

        @requires_api_version('1.17')
        def test_get_container_stats_stream(self):
            container = self.client.create_container(
                BUSYBOX, ['sleep', '60'],
            )
            self.tmp_containers.append(container)
            self.client.start(container)
            stream = self.client.stats(container)
            for chunk in stream:
                self.assertEqual(type(chunk), dict)
                for key in ['read', 'network', 'precpu_stats', 'cpu_stats',
                            'memory_stats', 'blkio_stats']:
                    self.assertIn(key, chunk)
