mirror of
https://github.com/ansible-collections/community.general.git
synced 2024-09-14 20:13:21 +02:00
70d8f02db7
* Extend env and add env_files support * Python 2.6 compat * Handle lists passed as string * Add changelog fragment * Use correct link formatting Co-Authored-By: hannseman <hannes@5monkeys.se> * Fix typo Co-Authored-By: hannseman <hannes@5monkeys.se> * Handle empty env and env_files values
100 lines
3.2 KiB
Python
100 lines
3.2 KiB
Python
import pytest
|
|
|
|
|
|
class APIErrorMock(Exception):
|
|
def __init__(self, message, response=None, explanation=None):
|
|
self.message = message
|
|
self.response = response
|
|
self.explanation = explanation
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def docker_module_mock(mocker):
|
|
docker_module_mock = mocker.MagicMock()
|
|
docker_utils_module_mock = mocker.MagicMock()
|
|
docker_errors_module_mock = mocker.MagicMock()
|
|
docker_errors_module_mock.APIError = APIErrorMock
|
|
mock_modules = {
|
|
'docker': docker_module_mock,
|
|
'docker.utils': docker_utils_module_mock,
|
|
'docker.errors': docker_errors_module_mock,
|
|
}
|
|
return mocker.patch.dict('sys.modules', **mock_modules)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def docker_swarm_service():
|
|
from ansible.modules.cloud.docker import docker_swarm_service
|
|
|
|
return docker_swarm_service
|
|
|
|
|
|
def test_retry_on_out_of_sequence_error(mocker, docker_swarm_service):
|
|
run_mock = mocker.MagicMock(
|
|
side_effect=APIErrorMock(
|
|
message='',
|
|
response=None,
|
|
explanation='rpc error: code = Unknown desc = update out of sequence',
|
|
)
|
|
)
|
|
manager = docker_swarm_service.DockerServiceManager(client=None)
|
|
manager.run = run_mock
|
|
with pytest.raises(APIErrorMock):
|
|
manager.run_safe()
|
|
assert run_mock.call_count == 3
|
|
|
|
|
|
def test_no_retry_on_general_api_error(mocker, docker_swarm_service):
|
|
run_mock = mocker.MagicMock(
|
|
side_effect=APIErrorMock(message='', response=None, explanation='some error')
|
|
)
|
|
manager = docker_swarm_service.DockerServiceManager(client=None)
|
|
manager.run = run_mock
|
|
with pytest.raises(APIErrorMock):
|
|
manager.run_safe()
|
|
assert run_mock.call_count == 1
|
|
|
|
|
|
def test_get_docker_environment(mocker, docker_swarm_service):
|
|
env_file_result = {'TEST1': 'A', 'TEST2': 'B', 'TEST3': 'C'}
|
|
env_dict = {'TEST3': 'CC', 'TEST4': 'D'}
|
|
env_string = "TEST3=CC,TEST4=D"
|
|
|
|
env_list = ['TEST3=CC', 'TEST4=D']
|
|
expected_result = sorted(['TEST1=A', 'TEST2=B', 'TEST3=CC', 'TEST4=D'])
|
|
mocker.patch.object(
|
|
docker_swarm_service, 'parse_env_file', return_value=env_file_result
|
|
)
|
|
mocker.patch.object(
|
|
docker_swarm_service,
|
|
'format_environment',
|
|
side_effect=lambda d: ['{0}={1}'.format(key, value) for key, value in d.items()],
|
|
)
|
|
# Test with env dict and file
|
|
result = docker_swarm_service.get_docker_environment(
|
|
env_dict, env_files=['dummypath']
|
|
)
|
|
assert result == expected_result
|
|
# Test with env list and file
|
|
result = docker_swarm_service.get_docker_environment(
|
|
env_list,
|
|
env_files=['dummypath']
|
|
)
|
|
assert result == expected_result
|
|
# Test with env string and file
|
|
result = docker_swarm_service.get_docker_environment(
|
|
env_string, env_files=['dummypath']
|
|
)
|
|
assert result == expected_result
|
|
|
|
assert result == expected_result
|
|
# Test with empty env
|
|
result = docker_swarm_service.get_docker_environment(
|
|
[], env_files=None
|
|
)
|
|
assert result == []
|
|
# Test with empty env_files
|
|
result = docker_swarm_service.get_docker_environment(
|
|
None, env_files=[]
|
|
)
|
|
assert result == []
|