diff options
Diffstat (limited to 'roles/openshift_health_checker/test/docker_image_availability_test.py')
| -rw-r--r-- | roles/openshift_health_checker/test/docker_image_availability_test.py | 187 |
1 files changed, 169 insertions, 18 deletions
diff --git a/roles/openshift_health_checker/test/docker_image_availability_test.py b/roles/openshift_health_checker/test/docker_image_availability_test.py index 2a9c32f77..f054b9ccc 100644 --- a/roles/openshift_health_checker/test/docker_image_availability_test.py +++ b/roles/openshift_health_checker/test/docker_image_availability_test.py @@ -3,26 +3,177 @@ import pytest from openshift_checks.docker_image_availability import DockerImageAvailability -@pytest.mark.xfail(strict=True) # TODO: remove this once this test is fully implemented. -@pytest.mark.parametrize('task_vars,expected_result', [ +@pytest.mark.parametrize('deployment_type,is_active', [ + ("origin", True), + ("openshift-enterprise", True), + ("enterprise", False), + ("online", False), + ("invalid", False), + ("", False), +]) +def test_is_active(deployment_type, is_active): + task_vars = dict( + openshift_deployment_type=deployment_type, + ) + assert DockerImageAvailability.is_active(task_vars=task_vars) == is_active + + +@pytest.mark.parametrize("is_containerized", [ + True, + False, +]) +def test_all_images_available_locally(is_containerized): + def execute_module(module_name, args, task_vars): + assert module_name == "docker_image_facts" + assert 'name' in args + assert args['name'] + return { + 'images': [args['name']], + } + + result = DockerImageAvailability(execute_module=execute_module).run(tmp=None, task_vars=dict( + openshift=dict(common=dict( + service_type='origin', + is_containerized=is_containerized, + )), + openshift_deployment_type='origin', + openshift_release='v3.4', + openshift_image_tag='3.4', + )) + + assert not result.get('failed', False) + + +@pytest.mark.parametrize("module_failure", [ + True, + False, +]) +def test_all_images_available_remotely(module_failure): + def execute_module(module_name, args, task_vars): + return { + 'docker_image_facts': {'images': [], 'failed': module_failure}, + 'docker_info': {'info': {'Registries': [{'Name': 'docker.io'}, {'Name': 'registry.access.redhat.com'}]}}, + }.get(module_name, {'changed': False}) + + result = DockerImageAvailability(execute_module=execute_module).run(tmp=None, task_vars=dict( + openshift=dict(common=dict( + service_type='origin', + is_containerized=False, + )), + openshift_deployment_type='origin', + openshift_release='3.4' + )) + + assert not result.get('failed', False) + + +def test_all_images_unavailable(): + def execute_module(module_name=None, module_args=None, tmp=None, task_vars=None): + if module_name == "docker_info": + return { + 'info': { + 'Registries': [ + { + 'Name': 'docker.io' + }, + { + 'Name': 'registry.access.redhat.com' + } + ] + } + } + + if module_name == "docker_container": + return { + 'failed': True, + } + + return { + 'changed': False, + } + + check = DockerImageAvailability(execute_module=execute_module) + actual = check.run(tmp=None, task_vars=dict( + openshift=dict(common=dict( + service_type='origin', + is_containerized=False, + )), + openshift_deployment_type="openshift-enterprise", + openshift_release=None, + )) + + assert actual['failed'] + assert "required images are not available" in actual['msg'] + + +@pytest.mark.parametrize("message,extra_words", [ ( - dict( - openshift=dict(common=dict( - service_type='origin', - is_containerized=False, - )), - openshift_release='v3.5', - deployment_type='origin', - openshift_image_tag='', # FIXME: should not be required - ), - {'changed': False}, + "docker image update failure", + ["docker image update failure"], ), - # TODO: add more parameters here to test the multiple possible inputs that affect behavior. + ( + "Error connecting: Error while fetching server API version", + ["Docker is not running"] + ), + ( + "Failed to import docker-py", + ["docker-py module is not installed", "install the python docker-py package"] + ) ]) -def test_docker_image_availability(task_vars, expected_result): +def test_skopeo_update_failure(message, extra_words): def execute_module(module_name=None, module_args=None, tmp=None, task_vars=None): - return {'info': {}} # TODO: this will vary depending on input parameters. + if module_name == "docker_image": + return { + "failed": True, + "msg": message, + "changed": False, + } - check = DockerImageAvailability(execute_module=execute_module) - result = check.run(tmp=None, task_vars=task_vars) - assert result == expected_result + return { + 'changed': False, + } + + actual = DockerImageAvailability(execute_module=execute_module).run(tmp=None, task_vars=dict( + openshift=dict(common=dict( + service_type='origin', + is_containerized=False, + )), + openshift_deployment_type="openshift-enterprise", + openshift_release='', + )) + + assert actual["failed"] + for word in extra_words: + assert word in actual["msg"] + + +@pytest.mark.parametrize("module_failure", [ + True, + False, +]) +def test_no_registries_available(module_failure): + def execute_module(module_name=None, module_args=None, tmp=None, task_vars=None): + if module_name == "docker_info": + return { + 'changed': False, + 'failed': module_failure, + 'info': { + 'Registries': [], + } + } + + return { + 'changed': False, + } + + actual = DockerImageAvailability(execute_module=execute_module).run(tmp=None, task_vars=dict( + openshift=dict(common=dict( + service_type='origin', + is_containerized=False, + )), + openshift_deployment_type="openshift-enterprise", + openshift_release='', + )) + + assert actual['failed'] + assert "docker registries" in actual['msg'] |