New Module: gitlab_deploy_key and related tests (#40097)

* Added module gitlab_deploy_key and related tests

* Fix sanity check issues

* Refactor to use common util method, add check_mode support

* Fix module shebang
This commit is contained in:
Marcus Watkins 2018-05-17 11:52:40 -06:00 committed by ansibot
parent 571c183f59
commit 7d87995207
3 changed files with 506 additions and 0 deletions

View file

@ -0,0 +1,37 @@
# -*- coding: utf-8 -*-
# (c) 2018, Marcus Watkins <marwatk@marcuswatkins.net>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
import json
from ansible.module_utils.urls import fetch_url
try:
from urllib import quote_plus # Python 2.X
except ImportError:
from urllib.parse import quote_plus # Python 3+
def request(module, api_url, project, path, access_token, private_token, rawdata='', method='GET'):
url = "%s/v4/projects/%s%s" % (api_url, quote_plus(project), path)
headers = {}
if access_token:
headers['Authorization'] = "Bearer %s" % access_token
else:
headers['Private-Token'] = private_token
headers['Accept'] = "application/json"
headers['Content-Type'] = "application/json"
response, info = fetch_url(module=module, url=url, headers=headers, data=rawdata, method=method)
status = info['status']
content = ""
if response:
content = response.read()
if status == 204:
return True, content
elif status == 200 or status == 201:
return True, json.loads(content)
else:
return False, str(status) + ": " + content

View file

@ -0,0 +1,233 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# (c) 2018, Marcus Watkins <marwatk@marcuswatkins.net>
# Based on code:
# (c) 2013, Phillip Gentry <phillip@cx.com>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
ANSIBLE_METADATA = {'metadata_version': '1.1',
'status': ['preview'],
'supported_by': 'community'}
DOCUMENTATION = '''
---
module: gitlab_deploy_key
short_description: Manages GitLab project deploy keys.
description:
- Adds, updates and removes project deploy keys
version_added: "2.6"
options:
api_url:
description:
- GitLab API url, e.g. https://gitlab.example.com/api
required: true
access_token:
description:
- The oauth key provided by GitLab. One of access_token or private_token is required. See https://docs.gitlab.com/ee/api/oauth2.html
required: false
private_token:
description:
- Personal access token to use. One of private_token or access_token is required. See https://docs.gitlab.com/ee/user/profile/personal_access_tokens.html
required: false
project:
description:
- Numeric project id or name of project in the form of group/name
required: true
title:
description:
- Deploy key's title
required: true
key:
description:
- Deploy key
required: true
can_push:
description:
- Whether this key can push to the project
type: bool
default: 'no'
state:
description:
- When C(present) the deploy key added to the project if it doesn't exist.
- When C(absent) it will be removed from the project if it exists
required: true
default: present
choices: [ "present", "absent" ]
author: "Marcus Watkins (@marwatk)"
'''
EXAMPLES = '''
# Example adding a project deploy key
- gitlab_deploy_key:
api_url: https://gitlab.example.com/api
access_token: "{{ access_token }}"
project: "my_group/my_project"
title: "Jenkins CI"
state: present
key: "ssh-rsa AAAAB3NzaC1yc2EAAAABJQAAAIEAiPWx6WM4lhHNedGfBpPJNPpZ7yKu+dnn1SJejgt4596k6YjzGGphH2TUxwKzxcKDKKezwkpfnxPkSMkuEspGRt/aZZ9w..."
# Update the above deploy key to add push access
- gitlab_deploy_key:
api_url: https://gitlab.example.com/api
access_token: "{{ access_token }}"
project: "my_group/my_project"
title: "Jenkins CI"
state: present
key: "ssh-rsa AAAAB3NzaC1yc2EAAAABJQAAAIEAiPWx6WM4lhHNedGfBpPJNPpZ7yKu+dnn1SJejgt4596k6YjzGGphH2TUxwKzxcKDKKezwkpfnxPkSMkuEspGRt/aZZ9w..."
can_push: yes
# Remove the previous deploy key from the project
- gitlab_deploy_key:
api_url: https://gitlab.example.com/api
access_token: "{{ access_token }}"
project: "my_group/my_project"
state: absent
key: "ssh-rsa AAAAB3NzaC1yc2EAAAABJQAAAIEAiPWx6WM4lhHNedGfBpPJNPpZ7yKu+dnn1SJejgt4596k6YjzGGphH2TUxwKzxcKDKKezwkpfnxPkSMkuEspGRt/aZZ9w..."
'''
RETURN = '''
msg:
description: Success or failure message
returned: always
type: string
sample: "Success"
result:
description: json parsed response from the server
returned: always
type: dict
error:
description: the error message returned by the Gitlab API
returned: failed
type: string
sample: "400: key is already in use"
previous_version:
description: object describing the state prior to this task
returned: changed
type: dict
'''
import json
from ansible.module_utils.basic import AnsibleModule
from copy import deepcopy
from ansible.module_utils.gitlab import request
def _list(module, api_url, project, access_token, private_token):
path = "/deploy_keys"
return request(module, api_url, project, path, access_token, private_token)
def _find(module, api_url, project, key, access_token, private_token):
success, data = _list(module, api_url, project, access_token, private_token)
if success:
for i in data:
if i["key"] == key:
return success, i
return success, None
return success, data
def _publish(module, api_url, project, data, access_token, private_token):
path = "/deploy_keys"
method = "POST"
if 'id' in data:
path += "/%s" % str(data["id"])
method = "PUT"
data = deepcopy(data)
data.pop('id', None)
return request(module, api_url, project, path, access_token, private_token, json.dumps(data, sort_keys=True), method)
def _delete(module, api_url, project, key_id, access_token, private_token):
path = "/deploy_keys/%s" % str(key_id)
return request(module, api_url, project, path, access_token, private_token, method='DELETE')
def _are_equivalent(input, existing):
for key in ['title', 'key', 'can_push']:
if key in input and key not in existing:
return False
if key not in input and key in existing:
return False
if not input[key] == existing[key]:
return False
return True
def main():
module = AnsibleModule(
argument_spec=dict(
api_url=dict(required=True),
access_token=dict(required=False, no_log=True),
private_token=dict(required=False, no_log=True),
project=dict(required=True),
key=dict(required=True),
state=dict(default='present', choices=['present', 'absent']),
can_push=dict(default='no', type='bool'),
title=dict(required=True),
),
mutually_exclusive=[
['access_token', 'private_token']
],
required_one_of=[
['access_token', 'private_token']
],
supports_check_mode=True,
)
api_url = module.params['api_url']
access_token = module.params['access_token']
private_token = module.params['private_token']
project = module.params['project']
state = module.params['state']
if not access_token and not private_token:
module.fail_json(msg="need either access_token or private_token")
input = {}
for key in ['title', 'key', 'can_push']:
input[key] = module.params[key]
success, existing = _find(module, api_url, project, input['key'], access_token, private_token)
if not success:
module.fail_json(msg="failed to list deploy keys", result=existing)
if existing:
input['id'] = existing['id']
changed = False
success = True
response = None
if state == 'present':
if not existing or not _are_equivalent(existing, input):
if not module.check_mode:
success, response = _publish(module, api_url, project, input, access_token, private_token)
changed = True
else:
if existing:
if not module.check_mode:
success, response = _delete(module, api_url, project, existing['id'], access_token, private_token)
changed = True
if success:
module.exit_json(changed=changed, msg='Success', result=response, previous_version=existing)
else:
module.fail_json(msg='Failure', error=response)
if __name__ == '__main__':
main()

View file

@ -0,0 +1,236 @@
# -*- coding: utf-8 -*-
# Copyright (c) 2018 Marcus Watkins <marwatk@marcuswatkins.net>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from ansible.compat.tests.mock import patch
from ansible.modules.source_control import gitlab_deploy_key
from ansible.module_utils._text import to_bytes
from ansible.module_utils import basic
import pytest
import json
fake_server_state = [
{
"id": 1,
"title": "Public key",
"key": 'ssh-rsa long/+base64//+string==',
"created_at": "2013-10-02T10:12:29Z",
"can_push": False
},
]
def set_module_args(args):
"""prepare arguments so that they will be picked up during module creation"""
args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
basic._ANSIBLE_ARGS = to_bytes(args)
class FakeReader:
def __init__(self, object):
self.content = json.dumps(object, sort_keys=True)
def read(self):
return self.content
class AnsibleExitJson(Exception):
"""Exception class to be raised by module.exit_json and caught by the test case"""
pass
class AnsibleFailJson(Exception):
"""Exception class to be raised by module.fail_json and caught by the test case"""
pass
def exit_json(*args, **kwargs):
"""function to patch over exit_json; package return data into an exception"""
if 'changed' not in kwargs:
kwargs['changed'] = False
raise AnsibleExitJson(kwargs)
def fail_json(*args, **kwargs):
"""function to patch over fail_json; package return data into an exception"""
kwargs['failed'] = True
raise AnsibleFailJson(kwargs)
@pytest.fixture
def fetch_url_mock(mocker):
return mocker.patch('ansible.module_utils.gitlab.fetch_url')
@pytest.fixture
def module_mock(mocker):
return mocker.patch.multiple(basic.AnsibleModule,
exit_json=exit_json,
fail_json=fail_json)
def test_access_token_output(capfd, fetch_url_mock, module_mock):
fetch_url_mock.return_value = [FakeReader(fake_server_state), {'status': 200}]
set_module_args({
'api_url': 'https://gitlab.example.com/api',
'access_token': 'test-access-token',
'project': '10',
'key': 'ssh-key foobar',
'title': 'a title',
'state': 'absent'
})
with pytest.raises(AnsibleExitJson) as result:
gitlab_deploy_key.main()
first_call = fetch_url_mock.call_args_list[0][1]
assert first_call['url'] == 'https://gitlab.example.com/api/v4/projects/10/deploy_keys'
assert first_call['headers']['Authorization'] == 'Bearer test-access-token'
assert 'Private-Token' not in first_call['headers']
assert first_call['method'] == 'GET'
def test_private_token_output(capfd, fetch_url_mock, module_mock):
fetch_url_mock.return_value = [FakeReader(fake_server_state), {'status': 200}]
set_module_args({
'api_url': 'https://gitlab.example.com/api',
'private_token': 'test-private-token',
'project': 'foo/bar',
'key': 'ssh-key foobar',
'title': 'a title',
'state': 'absent'
})
with pytest.raises(AnsibleExitJson) as result:
gitlab_deploy_key.main()
first_call = fetch_url_mock.call_args_list[0][1]
assert first_call['url'] == 'https://gitlab.example.com/api/v4/projects/foo%2Fbar/deploy_keys'
assert first_call['headers']['Private-Token'] == 'test-private-token'
assert 'Authorization' not in first_call['headers']
assert first_call['method'] == 'GET'
def test_bad_http_first_response(capfd, fetch_url_mock, module_mock):
fetch_url_mock.side_effect = [[FakeReader("Permission denied"), {'status': 403}], [FakeReader("Permission denied"), {'status': 403}]]
set_module_args({
'api_url': 'https://gitlab.example.com/api',
'access_token': 'test-access-token',
'project': '10',
'key': 'ssh-key foobar',
'title': 'a title',
'state': 'absent'
})
with pytest.raises(AnsibleFailJson):
gitlab_deploy_key.main()
def test_bad_http_second_response(capfd, fetch_url_mock, module_mock):
fetch_url_mock.side_effect = [[FakeReader(fake_server_state), {'status': 200}], [FakeReader("Permission denied"), {'status': 403}]]
set_module_args({
'api_url': 'https://gitlab.example.com/api',
'access_token': 'test-access-token',
'project': '10',
'key': 'ssh-key foobar',
'title': 'a title',
'state': 'present'
})
with pytest.raises(AnsibleFailJson):
gitlab_deploy_key.main()
def test_delete_non_existing(capfd, fetch_url_mock, module_mock):
fetch_url_mock.return_value = [FakeReader(fake_server_state), {'status': 200}]
set_module_args({
'api_url': 'https://gitlab.example.com/api',
'access_token': 'test-access-token',
'project': '10',
'key': 'ssh-key foobar',
'title': 'a title',
'state': 'absent'
})
with pytest.raises(AnsibleExitJson) as result:
gitlab_deploy_key.main()
assert result.value.args[0]['changed'] is False
def test_delete_existing(capfd, fetch_url_mock, module_mock):
fetch_url_mock.return_value = [FakeReader(fake_server_state), {'status': 200}]
set_module_args({
'api_url': 'https://gitlab.example.com/api',
'access_token': 'test-access-token',
'project': '10',
'key': 'ssh-rsa long/+base64//+string==',
'title': 'a title',
'state': 'absent'
})
with pytest.raises(AnsibleExitJson) as result:
gitlab_deploy_key.main()
second_call = fetch_url_mock.call_args_list[1][1]
assert second_call['url'] == 'https://gitlab.example.com/api/v4/projects/10/deploy_keys/1'
assert second_call['method'] == 'DELETE'
assert result.value.args[0]['changed'] is True
def test_add_new(capfd, fetch_url_mock, module_mock):
fetch_url_mock.return_value = [FakeReader(fake_server_state), {'status': 200}]
set_module_args({
'api_url': 'https://gitlab.example.com/api',
'access_token': 'test-access-token',
'project': '10',
'key': 'ssh-key foobar',
'title': 'a title',
'state': 'present'
})
with pytest.raises(AnsibleExitJson) as result:
gitlab_deploy_key.main()
second_call = fetch_url_mock.call_args_list[1][1]
assert second_call['url'] == 'https://gitlab.example.com/api/v4/projects/10/deploy_keys'
assert second_call['method'] == 'POST'
assert second_call['data'] == '{"can_push": false, "key": "ssh-key foobar", "title": "a title"}'
assert result.value.args[0]['changed'] is True
def test_update_existing(capfd, fetch_url_mock, module_mock):
fetch_url_mock.return_value = [FakeReader(fake_server_state), {'status': 200}]
set_module_args({
'api_url': 'https://gitlab.example.com/api',
'access_token': 'test-access-token',
'project': '10',
'title': 'Public key',
'key': 'ssh-rsa long/+base64//+string==',
'can_push': 'yes',
'state': 'present'
})
with pytest.raises(AnsibleExitJson) as result:
gitlab_deploy_key.main()
second_call = fetch_url_mock.call_args_list[1][1]
assert second_call['url'] == 'https://gitlab.example.com/api/v4/projects/10/deploy_keys/1'
assert second_call['method'] == 'PUT'
assert second_call['data'] == ('{"can_push": true, "key": "ssh-rsa long/+base64//+string==", "title": "Public key"}')
assert result.value.args[0]['changed'] is True
def test_unchanged_existing(capfd, fetch_url_mock, module_mock):
fetch_url_mock.return_value = [FakeReader(fake_server_state), {'status': 200}]
set_module_args({
'api_url': 'https://gitlab.example.com/api',
'access_token': 'test-access-token',
'project': '10',
'title': 'Public key',
'key': 'ssh-rsa long/+base64//+string==',
'can_push': 'no',
'state': 'present'
})
with pytest.raises(AnsibleExitJson) as result:
gitlab_deploy_key.main()
assert result.value.args[0]['changed'] is False
assert fetch_url_mock.call_count == 1