test_kolla_docker.py

            stream=True)
        self.assertFalse(self.dw.changed)
        self.dw.module.fail_json.assert_called_once_with(
            msg="Unknown error message: unexpected error",
            failed=True)

    def test_remove_image(self):
        self.dw = get_DockerWorker(
            {'image': 'myregistrydomain.com:5000/ubuntu:16.04',
             'action': 'remove_image'})
        self.dw.dc.images.return_value = self.fake_data['images']

        self.dw.remove_image()
        self.assertTrue(self.dw.changed)
        self.dw.dc.remove_image.assert_called_once_with(
            image='myregistrydomain.com:5000/ubuntu:16.04')

    def test_remove_image_not_exists(self):
        self.dw = get_DockerWorker(
            {'image': 'myregistrydomain.com:5000/non_existing:16.04',
             'action': 'remove_image'})
        self.dw.dc.images.return_value = self.fake_data['images']

        self.dw.remove_image()
        self.assertFalse(self.dw.changed)

    def test_remove_image_exception_409(self):
        resp = mock.MagicMock()
        resp.status_code = 409
        docker_except = docker_error.APIError('test error', resp)
        self.dw = get_DockerWorker(
            {'image': 'myregistrydomain.com:5000/ubuntu:16.04',
             'action': 'remove_image'})
        self.dw.dc.images.return_value = self.fake_data['images']
        self.dw.dc.remove_image.side_effect = docker_except

        self.assertRaises(docker_error.APIError, self.dw.remove_image)
        self.assertTrue(self.dw.changed)
        self.dw.module.fail_json.assert_called_once_with(
            failed=True,
            msg=("Image 'myregistrydomain.com:5000/ubuntu:16.04' "
                 "is currently in-use")
        )

    def test_remove_image_exception_500(self):
        resp = mock.MagicMock()
        resp.status_code = 500
        docker_except = docker_error.APIError('test error', resp)
        self.dw = get_DockerWorker(
            {'image': 'myregistrydomain.com:5000/ubuntu:16.04',
             'action': 'remove_image'})
        self.dw.dc.images.return_value = self.fake_data['images']
        self.dw.dc.remove_image.side_effect = docker_except

        self.assertRaises(docker_error.APIError, self.dw.remove_image)
        self.assertTrue(self.dw.changed)
        self.dw.module.fail_json.assert_called_once_with(
            failed=True,
            msg=("Server error")
        )


class TestVolume(base.BaseTestCase):

    def setUp(self):
        super(TestVolume, self).setUp()
        self.fake_data = copy.deepcopy(FAKE_DATA)
        self.volumes = {
            'Volumes':
            [{'Driver': 'local',
              'Labels': None,
              'Mountpoint': '/var/lib/docker/volumes/nova_compute/_data',
              'Name': 'nova_compute'},
             {'Driver': 'local',
              'Labels': None,
              'Mountpoint': '/var/lib/docker/volumes/mariadb/_data',
              'Name': 'mariadb'}]
        }

    def test_create_volume(self):
        self.dw = get_DockerWorker({'name': 'rabbitmq',
                                    'action': 'create_volume'})
        self.dw.dc.volumes.return_value = self.volumes

        self.dw.create_volume()
        self.dw.dc.volumes.assert_called_once_with()
        self.assertTrue(self.dw.changed)
        self.dw.dc.create_volume.assert_called_once_with(
            name='rabbitmq',
            driver='local')

    def test_create_volume_exists(self):
        self.dw = get_DockerWorker({'name': 'nova_compute',
                                    'action': 'create_volume'})
        self.dw.dc.volumes.return_value = self.volumes

        self.dw.create_volume()
        self.dw.dc.volumes.assert_called_once_with()
        self.assertFalse(self.dw.changed)

    def test_remove_volume(self):
        self.dw = get_DockerWorker({'name': 'nova_compute',
                                    'action': 'remove_volume'})
        self.dw.dc.volumes.return_value = self.volumes

        self.dw.remove_volume()
        self.assertTrue(self.dw.changed)
        self.dw.dc.remove_volume.assert_called_once_with(name='nova_compute')

    def test_remove_volume_not_exists(self):
        self.dw = get_DockerWorker({'name': 'rabbitmq',
                                    'action': 'remove_volume'})
        self.dw.dc.volumes.return_value = self.volumes

        self.dw.remove_volume()
        self.assertFalse(self.dw.changed)

    def test_remove_volume_exception(self):
        resp = mock.MagicMock()
        resp.status_code = 409
        docker_except = docker_error.APIError('test error', resp)
        self.dw = get_DockerWorker({'name': 'nova_compute',
                                    'action': 'remove_volume'})
        self.dw.dc.volumes.return_value = self.volumes
        self.dw.dc.remove_volume.side_effect = docker_except

        self.assertRaises(docker_error.APIError, self.dw.remove_volume)
        self.assertTrue(self.dw.changed)
        self.dw.module.fail_json.assert_called_once_with(
            failed=True,
            msg="Volume named 'nova_compute' is currently in-use"
        )


class TestAttrComp(base.BaseTestCase):

    def setUp(self):
        super(TestAttrComp, self).setUp()
        self.fake_data = copy.deepcopy(FAKE_DATA)

    def test_compare_cap_add_neg(self):
        container_info = {'HostConfig': dict(CapAdd=['data'])}
        self.dw = get_DockerWorker({'cap_add': ['data']})
        self.assertFalse(self.dw.compare_cap_add(container_info))

    def test_compare_cap_add_pos(self):
        container_info = {'HostConfig': dict(CapAdd=['data1'])}
        self.dw = get_DockerWorker({'cap_add': ['data2']})
        self.assertTrue(self.dw.compare_cap_add(container_info))

    def test_compare_ipc_mode_neg(self):
        container_info = {'HostConfig': dict(IpcMode='data')}
        self.dw = get_DockerWorker({'ipc_mode': 'data'})
        self.assertFalse(self.dw.compare_ipc_mode(container_info))

    def test_compare_ipc_mode_pos(self):
        container_info = {'HostConfig': dict(IpcMode='data1')}
        self.dw = get_DockerWorker({'ipc_mode': 'data2'})
        self.assertTrue(self.dw.compare_ipc_mode(container_info))

    def test_compare_security_opt_neg(self):
        container_info = {'HostConfig': dict(SecurityOpt=['data'])}
        self.dw = get_DockerWorker({'security_opt': ['data']})
        self.assertFalse(self.dw.compare_security_opt(container_info))

    def test_compare_security_opt_pos(self):
        container_info = {'HostConfig': dict(SecurityOpt=['data1'])}
        self.dw = get_DockerWorker({'security_opt': ['data2']})
        self.assertTrue(self.dw.compare_security_opt(container_info))

    def test_compare_pid_mode_neg(self):
        container_info = {'HostConfig': dict(PidMode='host')}
        self.dw = get_DockerWorker({'pid_mode': 'host'})
        self.assertFalse(self.dw.compare_pid_mode(container_info))

    def test_compare_pid_mode_pos(self):
        container_info = {'HostConfig': dict(PidMode='host1')}
        self.dw = get_DockerWorker({'pid_mode': 'host2'})
        self.assertTrue(self.dw.compare_pid_mode(container_info))

    def test_compare_cgroupns_mode_neg(self):
        container_info = {'HostConfig': dict(CgroupnsMode='host')}
        self.dw = get_DockerWorker({'cgroupns_mode': 'host'},
                                   docker_api_version='1.41')
        self.assertFalse(self.dw.compare_cgroupns_mode(container_info))

    def test_compare_cgroupns_mode_neg_backward_compat(self):
        container_info = {'HostConfig': dict(CgroupnsMode='')}
        self.dw = get_DockerWorker({'cgroupns_mode': 'host'},
                                   docker_api_version='1.41')
        self.assertFalse(self.dw.compare_cgroupns_mode(container_info))

    def test_compare_cgroupns_mode_ignore(self):
        container_info = {'HostConfig': dict(CgroupnsMode='private')}
        self.dw = get_DockerWorker({}, docker_api_version='1.41')
        self.assertFalse(self.dw.compare_cgroupns_mode(container_info))

    def test_compare_cgroupns_mode_pos(self):
        container_info = {'HostConfig': dict(CgroupnsMode='private')}
        self.dw = get_DockerWorker({'cgroupns_mode': 'host'},
                                   docker_api_version='1.41')
        self.assertTrue(self.dw.compare_cgroupns_mode(container_info))

    def test_compare_cgroupns_mode_pos_backward_compat(self):
        container_info = {'HostConfig': dict(CgroupnsMode='')}
        self.dw = get_DockerWorker({'cgroupns_mode': 'private'},
                                   docker_api_version='1.41')
        self.assertTrue(self.dw.compare_cgroupns_mode(container_info))

    def test_compare_cgroupns_mode_unsupported(self):
        container_info = {'HostConfig': dict()}
        self.dw = get_DockerWorker({'cgroupns_mode': 'host'})
        self.assertFalse(self.dw.compare_cgroupns_mode(container_info))

    def test_compare_privileged_neg(self):
        container_info = {'HostConfig': dict(Privileged=True)}
        self.dw = get_DockerWorker({'privileged': True})
        self.assertFalse(self.dw.compare_privileged(container_info))

    def test_compare_privileged_pos(self):
        container_info = {'HostConfig': dict(Privileged=True)}
        self.dw = get_DockerWorker({'privileged': False})
        self.assertTrue(self.dw.compare_privileged(container_info))

    def test_compare_labels_neg(self):
        container_info = {'Config': dict(Labels={'kolla_version': '2.0.1'})}
        self.dw = get_DockerWorker({'labels': {'kolla_version': '2.0.1'}})
        self.dw.check_image = mock.MagicMock(return_value=dict(
            Labels={'kolla_version': '2.0.1'}))
        self.assertFalse(self.dw.compare_labels(container_info))

    def test_compare_labels_pos(self):
        container_info = {'Config': dict(Labels={'kolla_version': '1.0.1'})}
        self.dw = get_DockerWorker({'labels': {'kolla_version': '2.0.1'}})
        self.dw.check_image = mock.MagicMock(return_value=dict(
            Labels={'kolla_version': '1.0.1'}))
        self.assertTrue(self.dw.compare_labels(container_info))

    def test_compare_tmpfs_neg(self):
        container_info = {'HostConfig': dict(Tmpfs=['foo'])}
        self.dw = get_DockerWorker({'tmpfs': ['foo']})

        self.assertFalse(self.dw.compare_tmpfs(container_info))

    def test_compare_tmpfs_neg_empty_string(self):
        container_info = {'HostConfig': dict()}
        self.dw = get_DockerWorker({'tmpfs': ['']})

        self.assertFalse(self.dw.compare_tmpfs(container_info))

    def test_compare_tmpfs_pos_different(self):
        container_info = {'HostConfig': dict(Tmpfs=['foo'])}
        self.dw = get_DockerWorker({'tmpfs': ['bar']})

        self.assertTrue(self.dw.compare_tmpfs(container_info))

    def test_compare_tmpfs_pos_empty_new(self):
        container_info = {'HostConfig': dict(Tmpfs=['foo'])}
        self.dw = get_DockerWorker({})

        self.assertTrue(self.dw.compare_tmpfs(container_info))

    def test_compare_tmpfs_pos_empty_current(self):
        container_info = {'HostConfig': dict()}
        self.dw = get_DockerWorker({'tmpfs': ['bar']})

        self.assertTrue(self.dw.compare_tmpfs(container_info))

    def test_compare_volumes_from_neg(self):
        container_info = {'HostConfig': dict(VolumesFrom=['777f7dc92da7'])}
        self.dw = get_DockerWorker({'volumes_from': ['777f7dc92da7']})

        self.assertFalse(self.dw.compare_volumes_from(container_info))

    def test_compare_volumes_from_post(self):
        container_info = {'HostConfig': dict(VolumesFrom=['777f7dc92da7'])}
        self.dw = get_DockerWorker({'volumes_from': ['ba8c0c54f0f2']})

        self.assertTrue(self.dw.compare_volumes_from(container_info))

    def test_compare_volumes_neg(self):
        container_info = {
            'Config': dict(Volumes=['/var/log/kolla/']),
            'HostConfig': dict(Binds=['kolla_logs:/var/log/kolla/:rw'])}
        self.dw = get_DockerWorker(
            {'volumes': ['kolla_logs:/var/log/kolla/:rw']})

        self.assertFalse(self.dw.compare_volumes(container_info))

    def test_compare_volumes_pos(self):
        container_info = {
            'Config': dict(Volumes=['/var/log/kolla/']),
            'HostConfig': dict(Binds=['kolla_logs:/var/log/kolla/:rw'])}
        self.dw = get_DockerWorker(
            {'volumes': ['/dev/:/dev/:rw']})

        self.assertTrue(self.dw.compare_volumes(container_info))

    def test_compare_environment_neg(self):
        container_info = {'Config': dict(
            Env=['KOLLA_CONFIG_STRATEGY=COPY_ALWAYS',
                 'KOLLA_BASE_DISTRO=ubuntu']
        )}
        self.dw = get_DockerWorker({
            'environment': dict(KOLLA_CONFIG_STRATEGY='COPY_ALWAYS',
                                KOLLA_BASE_DISTRO='ubuntu')})

        self.assertFalse(self.dw.compare_environment(container_info))

    def test_compare_environment_pos(self):
        container_info = {'Config': dict(
            Env=['KOLLA_CONFIG_STRATEGY=COPY_ALWAYS',
                 'KOLLA_BASE_DISTRO=ubuntu']
        )}
        self.dw = get_DockerWorker({
            'environment': dict(KOLLA_CONFIG_STRATEGY='COPY_ALWAYS',
                                KOLLA_BASE_DISTRO='centos')})

        self.assertTrue(self.dw.compare_environment(container_info))

    def test_compare_container_state_neg(self):
        container_info = {'State': dict(Status='running')}
        self.dw = get_DockerWorker({'state': 'running'})
        self.assertFalse(self.dw.compare_container_state(container_info))

    def test_compare_dimensions_pos(self):
        self.fake_data['params']['dimensions'] = {
            'blkio_weight': 10, 'mem_limit': 30}
        container_info = dict()
        container_info['HostConfig'] = {
            'CpuPeriod': 0, 'KernelMemory': 0, 'Memory': 0, 'CpuQuota': 0,
            'CpusetCpus': '', 'CpuShares': 0, 'BlkioWeight': 0,
            'CpusetMems': '', 'MemorySwap': 0, 'MemoryReservation': 0,
            'Ulimits': []}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.assertTrue(self.dw.compare_dimensions(container_info))

    def test_compare_dimensions_neg(self):
        self.fake_data['params']['dimensions'] = {
            'blkio_weight': 10}
        container_info = dict()
        container_info['HostConfig'] = {
            'CpuPeriod': 0, 'KernelMemory': 0, 'Memory': 0, 'CpuQuota': 0,
            'CpusetCpus': '', 'CpuShares': 0, 'BlkioWeight': 10,
            'CpusetMems': '', 'MemorySwap': 0, 'MemoryReservation': 0,
            'Ulimits': []}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.assertFalse(self.dw.compare_dimensions(container_info))

    def test_compare_wrong_dimensions(self):
        self.fake_data['params']['dimensions'] = {
            'blki_weight': 0}
        container_info = dict()
        container_info['HostConfig'] = {
            'CpuPeriod': 0, 'KernelMemory': 0, 'Memory': 0, 'CpuQuota': 0,
            'CpusetCpus': '', 'CpuShares': 0, 'BlkioWeight': 0,
            'CpusetMems': '', 'MemorySwap': 0, 'MemoryReservation': 0,
            'Ulimits': []}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.dw.compare_dimensions(container_info)
        self.dw.module.exit_json.assert_called_once_with(
            failed=True, msg=repr("Unsupported dimensions"),
            unsupported_dimensions=set(['blki_weight']))

    def test_compare_empty_dimensions(self):
        self.fake_data['params']['dimensions'] = dict()
        container_info = dict()
        container_info['HostConfig'] = {
            'CpuPeriod': 0, 'KernelMemory': 0, 'Memory': 0, 'CpuQuota': 0,
            'CpusetCpus': '1', 'CpuShares': 0, 'BlkioWeight': 0,
            'CpusetMems': '', 'MemorySwap': 0, 'MemoryReservation': 0,
            'Ulimits': []}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.assertTrue(self.dw.compare_dimensions(container_info))

    def test_compare_dimensions_removed_and_changed(self):
        self.fake_data['params']['dimensions'] = {
            'mem_reservation': 10}
        container_info = dict()
        # Here mem_limit and mem_reservation are already present
        # Now we are updating only 'mem_reservation'.
        # Ideally it should return True stating that the docker
        # dimensions have been changed.
        container_info['HostConfig'] = {
            'CpuPeriod': 0, 'KernelMemory': 0, 'Memory': 10, 'CpuQuota': 0,
            'CpusetCpus': '', 'CpuShares': 0, 'BlkioWeight': 0,
            'CpusetMems': '', 'MemorySwap': 0, 'MemoryReservation': 10,
            'Ulimits': []}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.assertTrue(self.dw.compare_dimensions(container_info))

    def test_compare_dimensions_explicit_default(self):
        self.fake_data['params']['dimensions'] = {
            'mem_reservation': 0}
        container_info = dict()
        # Here mem_limit and mem_reservation are already present
        # Now we are updating only 'mem_reservation'.
        # Ideally it should return True stating that the docker
        # dimensions have been changed.
        container_info['HostConfig'] = {
            'CpuPeriod': 0, 'KernelMemory': 0, 'Memory': 0, 'CpuQuota': 0,
            'CpusetCpus': '', 'CpuShares': 0, 'BlkioWeight': 0,
            'CpusetMems': '', 'MemorySwap': 0, 'MemoryReservation': 0,
            'Ulimits': []}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.assertFalse(self.dw.compare_dimensions(container_info))

    def test_compare_dimensions_kernel_memory_1_42(self):
        self.fake_data['params']['dimensions'] = {
            'kernel_memory': '1024'}
        container_info = dict()
        container_info['HostConfig'] = {
            'CpuPeriod': 0, 'Memory': 0, 'CpuQuota': 0,
            'CpusetCpus': '', 'CpuShares': 0, 'BlkioWeight': 0,
            'CpusetMems': '', 'MemorySwap': 0, 'MemoryReservation': 0,
            'Ulimits': []}
        self.dw = get_DockerWorker(self.fake_data['params'],
                                   docker_api_version='1.42')
        self.dw.compare_dimensions(container_info)
        self.dw.module.exit_json.assert_called_once_with(
            failed=True, msg=repr("Unsupported dimensions"),
            unsupported_dimensions=set(['kernel_memory']))

    def test_compare_container_state_pos(self):
        container_info = {'State': dict(Status='running')}
        self.dw = get_DockerWorker({'state': 'exited'})
        self.assertTrue(self.dw.compare_container_state(container_info))

    def test_compare_ulimits_pos(self):
        self.fake_data['params']['dimensions'] = {
            'ulimits': {'nofile': {'soft': 131072, 'hard': 131072}}}
        container_info = dict()
        container_info['HostConfig'] = {
            'CpuPeriod': 0, 'KernelMemory': 0, 'Memory': 0, 'CpuQuota': 0,
            'CpusetCpus': '', 'CpuShares': 0, 'BlkioWeight': 0,
            'CpusetMems': '', 'MemorySwap': 0, 'MemoryReservation': 0,
            'Ulimits': []}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.assertTrue(self.dw.compare_dimensions(container_info))

    def test_compare_ulimits_neg(self):
        self.fake_data['params']['dimensions'] = {
            'ulimits': {'nofile': {'soft': 131072, 'hard': 131072}}}
        ulimits_nofile = Ulimit(name='nofile',
                                soft=131072, hard=131072)
        container_info = dict()
        container_info['HostConfig'] = {
            'CpuPeriod': 0, 'KernelMemory': 0, 'Memory': 0, 'CpuQuota': 0,
            'CpusetCpus': '', 'CpuShares': 0, 'BlkioWeight': 0,
            'CpusetMems': '', 'MemorySwap': 0, 'MemoryReservation': 0,
            'Ulimits': [ulimits_nofile]}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.assertFalse(self.dw.compare_dimensions(container_info))

    def test_compare_empty_new_healthcheck(self):
        container_info = dict()
        container_info['Config'] = {
            'Healthcheck': {
                'Test': [
                    "CMD-SHELL",
                    "/bin/check.sh"],
                "Interval": 30000000000,
                "Timeout": 30000000000,
                "StartPeriod": 5000000000,
                "Retries": 3}}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.assertTrue(self.dw.compare_healthcheck(container_info))

    def test_compare_empty_current_healthcheck(self):
        self.fake_data['params']['healthcheck'] = {
            'test': ['CMD-SHELL', '/bin/check.sh'],
            'interval': 30,
            'timeout': 30,
            'start_period': 5,
            'retries': 3}
        container_info = dict()
        container_info['Config'] = {}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.assertTrue(self.dw.compare_healthcheck(container_info))

    def test_compare_healthcheck_no_test(self):
        self.fake_data['params']['healthcheck'] = {
            'interval': 30,
            'timeout': 30,
            'start_period': 5,
            'retries': 3}
        container_info = dict()
        container_info['Config'] = {
            'Healthcheck': {
                'Test': [
                    "CMD-SHELL",
                    "/bin/check.sh"],
                "Interval": 30000000000,
                "Timeout": 30000000000,
                "StartPeriod": 5000000000,
                "Retries": 3}}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.dw.compare_healthcheck(container_info)
        self.dw.module.exit_json.assert_called_once_with(
            failed=True, msg=repr("Missing healthcheck option"),
            missing_healthcheck=set(['test']))

    def test_compare_healthcheck_pos(self):
        self.fake_data['params']['healthcheck'] = \
            {'test': ['CMD', '/bin/check']}
        container_info = dict()
        container_info['Config'] = {
            'Healthcheck': {
                'Test': [
                    "CMD-SHELL",
                    "/bin/check.sh"],
                "Interval": 30000000000,
                "Timeout": 30000000000,
                "StartPeriod": 5000000000,
                "Retries": 3}}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.assertTrue(self.dw.compare_healthcheck(container_info))

    def test_compare_healthcheck_neg(self):
        self.fake_data['params']['healthcheck'] = \
            {'test': ['CMD-SHELL', '/bin/check.sh'],
             'interval': 30,
             'timeout': 30,
             'start_period': 5,
             'retries': 3}
        container_info = dict()
        container_info['Config'] = {
            "Healthcheck": {
                "Test": [
                    "CMD-SHELL",
                    "/bin/check.sh"],
                "Interval": 30000000000,
                "Timeout": 30000000000,
                "StartPeriod": 5000000000,
                "Retries": 3}}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.assertFalse(self.dw.compare_healthcheck(container_info))

    def test_compare_healthcheck_time_zero(self):
        self.fake_data['params']['healthcheck'] = \
            {'test': ['CMD-SHELL', '/bin/check.sh'],
             'interval': 0,
             'timeout': 30,
             'start_period': 5,
             'retries': 3}
        container_info = dict()
        container_info['Config'] = {
            "Healthcheck": {
                "Test": [
                    "CMD-SHELL",
                    "/bin/check.sh"],
                "Interval": 30000000000,
                "Timeout": 30000000000,
                "StartPeriod": 5000000000,
                "Retries": 3}}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.assertTrue(self.dw.compare_healthcheck(container_info))

    def test_compare_healthcheck_time_wrong_type(self):
        self.fake_data['params']['healthcheck'] = \
            {'test': ['CMD-SHELL', '/bin/check.sh'],
             'timeout': 30,
             'start_period': 5,
             'retries': 3}
        self.fake_data['params']['healthcheck']['interval'] = \
            {"broken": {"interval": "True"}}
        container_info = dict()
        container_info['Config'] = {
            "Healthcheck": {
                "Test": [
                    "CMD-SHELL",
                    "/bin/check.sh"],
                "Interval": 30000000000,
                "Timeout": 30000000000,
                "StartPeriod": 5000000000,
                "Retries": 3}}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.assertRaises(TypeError,
                          lambda: self.dw.compare_healthcheck(container_info))

    def test_compare_healthcheck_time_wrong_value(self):
        self.fake_data['params']['healthcheck'] = \
            {'test': ['CMD-SHELL', '/bin/check.sh'],
             'timeout': 30,
             'start_period': 5,
             'retries': 3}
        self.fake_data['params']['healthcheck']['interval'] = "dog"
        container_info = dict()
        container_info['Config'] = {
            "Healthcheck": {
                "Test": [
                    "CMD-SHELL",
                    "/bin/check.sh"],
                "Interval": 30000000000,
                "Timeout": 30000000000,
                "StartPeriod": 5000000000,
                "Retries": 3}}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.assertRaises(ValueError,
                          lambda: self.dw.compare_healthcheck(container_info))

    def test_compare_healthcheck_opt_missing(self):
        self.fake_data['params']['healthcheck'] = \
            {'test': ['CMD-SHELL', '/bin/check.sh'],
             'interval': 30,
             'timeout': 30,
             'retries': 3}
        container_info = dict()
        container_info['Config'] = {
            "Healthcheck": {
                "Test": [
                    "CMD-SHELL",
                    "/bin/check.sh"],
                "Interval": 30000000000,
                "Timeout": 30000000000,
                "StartPeriod": 5000000000,
                "Retries": 3}}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.dw.compare_healthcheck(container_info)
        self.dw.module.exit_json.assert_called_once_with(
            failed=True, msg=repr("Missing healthcheck option"),
            missing_healthcheck=set(['start_period']))

    def test_compare_healthcheck_opt_extra(self):
        self.fake_data['params']['healthcheck'] = \
            {'test': ['CMD-SHELL', '/bin/check.sh'],
             'interval': 30,
             'start_period': 5,
             'extra_option': 1,
             'timeout': 30,
             'retries': 3}
        container_info = dict()
        container_info['Config'] = {
            "Healthcheck": {
                "Test": [
                    "CMD-SHELL",
                    "/bin/check.sh"],
                "Interval": 30000000000,
                "Timeout": 30000000000,
                "StartPeriod": 5000000000,
                "Retries": 3}}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.dw.compare_healthcheck(container_info)
        self.dw.module.exit_json.assert_called_once_with(
            failed=True, msg=repr("Unsupported healthcheck options"),
            unsupported_healthcheck=set(['extra_option']))

    def test_compare_healthcheck_value_false(self):
        self.fake_data['params']['healthcheck'] = \
            {'test': ['CMD-SHELL', '/bin/check.sh'],
             'interval': 30,
             'start_period': 5,
             'extra_option': 1,
             'timeout': 30,
             'retries': False}
        container_info = dict()
        container_info['Config'] = {
            "Healthcheck": {
                "Test": [
                    "CMD-SHELL",
                    "/bin/check.sh"],
                "Interval": 30000000000,
                "Timeout": 30000000000,
                "StartPeriod": 5000000000,
                "Retries": 3}}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.assertTrue(self.dw.compare_healthcheck(container_info))

    def test_parse_healthcheck_empty(self):
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.assertIsNone(self.dw.parse_healthcheck(
                          self.fake_data.get('params', {}).get('healthcheck')))

    def test_parse_healthcheck_test_none(self):
        self.fake_data['params']['healthcheck'] = \
            {'test': 'NONE'}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.assertIsNone(self.dw.parse_healthcheck(
                          self.fake_data['params']['healthcheck']))

    def test_parse_healthcheck_test_none_brackets(self):
        self.fake_data['params']['healthcheck'] = \
            {'test': ['NONE']}
        self.dw = get_DockerWorker(self.fake_data['params'])
        self.assertIsNone(self.dw.parse_healthcheck(
                          self.fake_data['params']['healthcheck']))


class TestSystemd(base.BaseTestCase):
    def setUp(self) -> None:
        super(TestSystemd, self).setUp()
        self.params_dict = dict(
            name='test',
            restart_policy='no',
            client_timeout=120,
            restart_retries=10,
            graceful_timeout=15
        )
        swm.sleep = mock.Mock()
        self.sw = swm.SystemdWorker(self.params_dict)

    def test_manager(self):
        self.assertIsNotNone(self.sw)
        self.assertIsNotNone(self.sw.manager)

    def test_start(self):
        self.sw.perform_action = mock.Mock(return_value=True)
        self.sw.wait_for_unit = mock.Mock(return_value=True)

        self.sw.start()
        self.sw.perform_action.assert_called_once_with(
            'StartUnit',
            'kolla-test-container.service',
            'replace'
        )

    def test_restart(self):
        self.sw.perform_action = mock.Mock(return_value=True)
        self.sw.wait_for_unit = mock.Mock(return_value=True)

        self.sw.restart()
        self.sw.perform_action.assert_called_once_with(
            'RestartUnit',
            'kolla-test-container.service',
            'replace'
        )

    def test_stop(self):
        self.sw.perform_action = mock.Mock(return_value=True)

        self.sw.stop()
        self.sw.perform_action.assert_called_once_with(
            'StopUnit',
            'kolla-test-container.service',
            'replace'
        )

    def test_reload(self):
        self.sw.perform_action = mock.Mock(return_value=True)

        self.sw.reload()
        self.sw.perform_action.assert_called_once_with(
            'Reload',
            'kolla-test-container.service',
            'replace'
        )

    def test_enable(self):
        self.sw.perform_action = mock.Mock(return_value=True)

        self.sw.enable()
        self.sw.perform_action.assert_called_once_with(
            'EnableUnitFiles',
            ['kolla-test-container.service'],
            False,
            True
        )

    def test_check_unit_change(self):
        self.sw.generate_unit_file = mock.Mock()
        self.sw.check_unit_file = mock.Mock(return_value=True)
        open_mock = mock.mock_open(read_data='test data')
        return_val = None

        with mock.patch('builtins.open', open_mock, create=True):
            return_val = self.sw.check_unit_change('test data')

        self.assertFalse(return_val)
        self.sw.generate_unit_file.assert_not_called()
        open_mock.assert_called_with(
            '/etc/systemd/system/kolla-test-container.service',
            'r'
        )
        open_mock.return_value.read.assert_called_once()

    def test_check_unit_change_diff(self):
        self.sw.generate_unit_file = mock.Mock()
        self.sw.check_unit_file = mock.Mock(return_value=True)
        open_mock = mock.mock_open(read_data='new data')
        return_val = None

        with mock.patch('builtins.open', open_mock, create=True):
            return_val = self.sw.check_unit_change('old data')

        self.assertTrue(return_val)
        self.sw.generate_unit_file.assert_not_called()
        open_mock.assert_called_with(
            '/etc/systemd/system/kolla-test-container.service',
            'r'
        )
        open_mock.return_value.read.assert_called_once()

    @mock.patch(
        'kolla_systemd_worker.TEMPLATE',
        """${name}, ${restart_policy},
        ${graceful_timeout}, ${restart_timeout},
        ${restart_retries}"""
    )
    def test_generate_unit_file(self):
        self.sw = swm.SystemdWorker(self.params_dict)
        p = self.params_dict
        ref_string = f"""{p.get('name')}, {p.get('restart_policy')},
        {p.get('graceful_timeout')}, {p.get('client_timeout')},
        {p.get('restart_retries')}"""

        ret_string = self.sw.generate_unit_file()

        self.assertEqual(ref_string, ret_string)

    def test_create_unit_file(self):
        self.sw.generate_unit_file = mock.Mock(return_value='test data')
        self.sw.check_unit_change = mock.Mock(return_value=True)
        self.sw.reload = mock.Mock()
        self.sw.enable = mock.Mock()
        open_mock = mock.mock_open()
        return_val = None

        with mock.patch('builtins.open', open_mock, create=True):
            return_val = self.sw.create_unit_file()

        self.assertTrue(return_val)
        open_mock.assert_called_with(
            '/etc/systemd/system/kolla-test-container.service',
            'w'
        )
        open_mock.return_value.write.assert_called_once_with('test data')
        self.sw.reload.assert_called_once()
        self.sw.enable.assert_called_once()

    def test_create_unit_file_no_change(self):
        self.sw.generate_unit_file = mock.Mock()
        self.sw.check_unit_change = mock.Mock(return_value=False)
        self.sw.reload = mock.Mock()
        self.sw.enable = mock.Mock()
        open_mock = mock.mock_open()

        return_val = self.sw.create_unit_file()

        self.assertFalse(return_val)
        open_mock.assert_not_called()
        self.sw.reload.assert_not_called()
        self.sw.enable.assert_not_called()

    def test_remove_unit_file(self):
        self.sw.check_unit_file = mock.Mock(return_value=True)
        os.remove = mock.Mock()
        self.sw.reload = mock.Mock()

        return_val = self.sw.remove_unit_file()

        self.assertTrue(return_val)
        os.remove.assert_called_once_with(
            '/etc/systemd/system/kolla-test-container.service'
        )
        self.sw.reload.assert_called_once()

    def test_get_unit_state(self):
        unit_list = [
            ('foo.service', '', 'loaded', 'active', 'exited'),
            ('kolla-test-container.service', '', 'loaded', 'active', 'running')
        ]
        self.sw.manager.ListUnits = mock.Mock(return_value=unit_list)

        state = self.sw.get_unit_state()

        self.sw.manager.ListUnits.assert_called_once()
        self.assertEqual('running', state)

    def test_get_unit_state_not_exist(self):
        unit_list = [
            ('foo.service', '', 'loaded', 'active', 'exited'),
            ('bar.service', '', 'loaded', 'active', 'running')
        ]
        self.sw.manager.ListUnits = mock.Mock(return_value=unit_list)

        state = self.sw.get_unit_state()

        self.sw.manager.ListUnits.assert_called_once()
        self.assertIsNone(state)

    def test_wait_for_unit(self):
        self.sw.get_unit_state = mock.Mock()
        self.sw.get_unit_state.side_effect = ['starting', 'running']

        result = self.sw.wait_for_unit(10)

        self.assertTrue(result)

    def test_wait_for_unit_timeout(self):
        self.sw.get_unit_state = mock.Mock()
        self.sw.get_unit_state.side_effect = [
            'starting', 'starting', 'failed', 'failed']

        result = self.sw.wait_for_unit(10)

        self.assertFalse(result)