Added new module github_repo (#1683)

* Added new module github_repo

* Fixed sanity errors

* Fixed sanity errors

* Unit tests for github_repo module

* Fixed import-before-documentation

* Added PyGithub dependency for unit tests

* Fixed errata

* Require Python >= 2.7

* Support for check_mode and other improvements

* Fixed import-before-documentation

* Improved module parameter requirements, check mode and docs

* Code improvements

* Fixed version tag
This commit is contained in:
Álvaro Torres Cogollo 2021-03-06 13:30:43 +01:00 committed by GitHub
parent 7425e9840d
commit 53c6b49673
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 497 additions and 1 deletions

View file

@ -0,0 +1,252 @@
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import requests
import re
import json
import sys
from httmock import with_httmock, urlmatch, response
from ansible_collections.community.general.tests.unit.compat import unittest
from ansible_collections.community.general.plugins.modules.source_control.github import github_repo
GITHUB_MINIMUM_PYTHON_VERSION = (2, 7)
@urlmatch(netloc=r'.*')
def debug_mock(url, request):
print(request.original.__dict__)
@urlmatch(netloc=r'api\.github\.com:443$', path=r'/orgs/.*', method="get")
def get_orgs_mock(url, request):
match = re.search(r"api\.github\.com:443/orgs/(?P<org>[^/]+)", request.url)
org = match.group("org")
# https://docs.github.com/en/rest/reference/orgs#get-an-organization
headers = {'content-type': 'application/json'}
content = {
"login": org,
"url": "https://api.github.com:443/orgs/{0}".format(org)
}
content = json.dumps(content).encode("utf-8")
return response(200, content, headers, None, 5, request)
@urlmatch(netloc=r'api\.github\.com:443$', path=r'/user', method="get")
def get_user_mock(url, request):
# https://docs.github.com/en/rest/reference/users#get-the-authenticated-user
headers = {'content-type': 'application/json'}
content = {
"login": "octocat",
"url": "https://api.github.com:443/users/octocat"
}
content = json.dumps(content).encode("utf-8")
return response(200, content, headers, None, 5, request)
@urlmatch(netloc=r'api\.github\.com:443$', path=r'/repos/.*/.*', method="get")
def get_repo_notfound_mock(url, request):
return response(404, "{\"message\": \"Not Found\"}", "", "Not Found", 5, request)
@urlmatch(netloc=r'api\.github\.com:443$', path=r'/repos/.*/.*', method="get")
def get_repo_mock(url, request):
match = re.search(
r"api\.github\.com:443/repos/(?P<org>[^/]+)/(?P<repo>[^/]+)", request.url)
org = match.group("org")
repo = match.group("repo")
# https://docs.github.com/en/rest/reference/repos#get-a-repository
headers = {'content-type': 'application/json'}
content = {
"name": repo,
"full_name": "{0}/{1}".format(org, repo),
"url": "https://api.github.com:443/repos/{0}/{1}".format(org, repo),
"private": False,
"description": "This your first repo!",
"default_branch": "master",
"allow_rebase_merge": True
}
content = json.dumps(content).encode("utf-8")
return response(200, content, headers, None, 5, request)
@urlmatch(netloc=r'api\.github\.com:443$', path=r'/orgs/.*/repos', method="post")
def create_new_org_repo_mock(url, request):
match = re.search(
r"api\.github\.com:443/orgs/(?P<org>[^/]+)/repos", request.url)
org = match.group("org")
repo = json.loads(request.body)
headers = {'content-type': 'application/json'}
# https://docs.github.com/en/rest/reference/repos#create-an-organization-repository
content = {
"name": repo['name'],
"full_name": "{0}/{1}".format(org, repo['name']),
"private": repo['private'],
"description": repo['description']
}
content = json.dumps(content).encode("utf-8")
return response(201, content, headers, None, 5, request)
@urlmatch(netloc=r'api\.github\.com:443$', path=r'/user/repos', method="post")
def create_new_user_repo_mock(url, request):
repo = json.loads(request.body)
headers = {'content-type': 'application/json'}
# https://docs.github.com/en/rest/reference/repos#create-a-repository-for-the-authenticated-user
content = {
"name": repo['name'],
"full_name": "{0}/{1}".format("octocat", repo['name']),
"private": repo['private'],
"description": repo['description']
}
content = json.dumps(content).encode("utf-8")
return response(201, content, headers, None, 5, request)
@urlmatch(netloc=r'api\.github\.com:443$', path=r'/repos/.*/.*', method="patch")
def patch_repo_mock(url, request):
match = re.search(
r"api\.github\.com:443/repos/(?P<org>[^/]+)/(?P<repo>[^/]+)", request.url)
org = match.group("org")
repo = match.group("repo")
body = json.loads(request.body)
headers = {'content-type': 'application/json'}
# https://docs.github.com/en/rest/reference/repos#update-a-repository
content = {
"name": repo,
"full_name": "{0}/{1}".format(org, repo),
"url": "https://api.github.com:443/repos/{0}/{1}".format(org, repo),
"private": body['private'],
"description": body['description'],
"default_branch": "master",
"allow_rebase_merge": True
}
content = json.dumps(content).encode("utf-8")
return response(200, content, headers, None, 5, request)
@urlmatch(netloc=r'api\.github\.com:443$', path=r'/repos/.*/.*', method="delete")
def delete_repo_mock(url, request):
# https://docs.github.com/en/rest/reference/repos#delete-a-repository
return response(204, None, None, None, 5, request)
@urlmatch(netloc=r'api\.github\.com:443$', path=r'/repos/.*/.*', method="delete")
def delete_repo_notfound_mock(url, request):
# https://docs.github.com/en/rest/reference/repos#delete-a-repository
return response(404, "{\"message\": \"Not Found\"}", "", "Not Found", 5, request)
class TestGithubRepo(unittest.TestCase):
def setUp(self):
if sys.version_info < GITHUB_MINIMUM_PYTHON_VERSION:
self.skipTest("Python %s+ is needed for PyGithub" %
",".join(map(str, GITHUB_MINIMUM_PYTHON_VERSION)))
@with_httmock(get_orgs_mock)
@with_httmock(get_repo_notfound_mock)
@with_httmock(create_new_org_repo_mock)
def test_create_new_org_repo(self):
result = github_repo.run_module({
'username': None,
'password': None,
"access_token": "mytoken",
"organization": "MyOrganization",
"name": "myrepo",
"description": "Just for fun",
"private": False,
"state": "present"
})
self.assertEqual(result['changed'], True)
self.assertEqual(result['repo']['private'], False)
@with_httmock(get_user_mock)
@with_httmock(get_repo_notfound_mock)
@with_httmock(create_new_user_repo_mock)
def test_create_new_user_repo(self):
result = github_repo.run_module({
'username': None,
'password': None,
"access_token": "mytoken",
"organization": None,
"name": "myrepo",
"description": "Just for fun",
"private": True,
"state": "present"
})
self.assertEqual(result['changed'], True)
self.assertEqual(result['repo']['private'], True)
@with_httmock(get_orgs_mock)
@with_httmock(get_repo_mock)
@with_httmock(patch_repo_mock)
def test_patch_existing_org_repo(self):
result = github_repo.run_module({
'username': None,
'password': None,
"access_token": "mytoken",
"organization": "MyOrganization",
"name": "myrepo",
"description": "Just for fun",
"private": True,
"state": "present"
})
self.assertEqual(result['changed'], True)
self.assertEqual(result['repo']['private'], True)
@with_httmock(get_orgs_mock)
@with_httmock(get_repo_mock)
@with_httmock(delete_repo_mock)
def test_delete_org_repo(self):
result = github_repo.run_module({
'username': None,
'password': None,
"access_token": "mytoken",
"organization": "MyOrganization",
"name": "myrepo",
"description": "Just for fun",
"private": False,
"state": "absent"
})
self.assertEqual(result['changed'], True)
@with_httmock(get_user_mock)
@with_httmock(get_repo_mock)
@with_httmock(delete_repo_mock)
def test_delete_user_repo(self):
result = github_repo.run_module({
'username': None,
'password': None,
"access_token": "mytoken",
"organization": None,
"name": "myrepo",
"description": "Just for fun",
"private": False,
"state": "absent"
})
self.assertEqual(result['changed'], True)
@with_httmock(get_orgs_mock)
@with_httmock(get_repo_notfound_mock)
@with_httmock(delete_repo_notfound_mock)
def test_delete_org_repo_notfound(self):
result = github_repo.run_module({
'username': None,
'password': None,
"access_token": "mytoken",
"organization": "MyOrganization",
"name": "myrepo",
"description": "Just for fun",
"private": True,
"state": "absent"
})
self.assertEqual(result['changed'], False)
if __name__ == "__main__":
unittest.main()