--- a/autoland/webapi/tests/test_repo_check.py
+++ b/autoland/webapi/tests/test_repo_check.py
@@ -1,15 +1,18 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
"""
Test repository handling logic.
"""
+import copy
+import urllib.parse
+
from autolandweb.server import make_app
from autolandweb.testing import MountebankClient
import pytest
# FIXME: need to get pycharm test runner to add /app to the container
# PYTHONPATH
import sys
@@ -64,16 +67,378 @@ class FakeReviewBoard:
"is": {
"statusCode": 404
}
}]
}
]
)
+ def create_repositories(self, repositories, n_per_page=25):
+ """Create repositories in the fake reviewboard server.
+
+ For reviewboard documentation on the repository list resource
+ see: https://www.reviewboard.org/docs/manual/2.5/webapi/2.0/resources/repository-list/#webapi2.0-repository-list-resource
+
+ Args:
+ repositories: A list where each entry is a description
+ of a repository. Each repository description is a dict of
+ with the following keys:
+ * bug_tracker(string): optional url of the bug tracker.
+ * id(integer): unique id number for the repository.
+ * mirror_path(string): optional mirror path/url for
+ the repo.
+ * name(string): unique name for the repo.
+ * path(string): path/url for the repo.
+ * tool(string): the type of scm tool used for the repo.
+ * visible(boolean): whether the repo is visible.
+
+ Example repositories list:
+ [
+ {
+ "bug_tracker": "https://bugzilla.mozilla.org",
+ "id": 1,
+ "mirror_path": "",
+ "name": "gecko",
+ "path": "https://reviewboard-hg.mozilla.org/gecko",
+ "tool": "Mercurial",
+ "visible": True,
+ },
+ {
+ "bug_tracker": "",
+ "id": 2,
+ "mirror_path": "",
+ "name": "Navi SVN",
+ "path": "http://svn.navi.cx/misc",
+ "tool": "Subversion",
+ "visible": True,
+ },
+ ]
+ n_per_page: The number of entries that should appear per
+ page on the paginated list resource. For mocking
+ simplicity the query arg used by the client to specify
+ this on request is ignored.
+ """ # noqa
+ base_url = self.url + '/api/repositories/'
+ items = []
+
+ for repo in repositories:
+ item = copy.copy(repo)
+ item['links'] = self._links_for_repository(item['id'])
+ items.append({
+ 'repository': item,
+ 'stat': 'ok',
+ })
+
+ pages = self._paginate_list_resource(
+ items,
+ 'repository',
+ base_url,
+ 'repositories',
+ list_links={
+ 'create': {
+ 'href': base_url,
+ 'method': 'POST',
+ },
+ },
+ n_per_page=n_per_page
+ )
+
+ stubs = []
+ stubs.extend(
+ [self._stub_item('repository', item) for item in items]
+ )
+ stubs.extend(
+ self._stub_paginated_list('repositories', 'repository', pages)
+ )
+ stubs.append(self._404_stub()) # 400 Everything else.
+ self.mountebank.create_stub(stubs)
+
+ def _links_for_repository(self, id):
+ url = self.url + '/api/repositories/{id}/'.format(id=id)
+ return {
+ 'branches': {
+ 'href': url + 'branches/',
+ 'method': 'GET'
+ },
+ 'commits': {
+ 'href': url + 'commits/',
+ 'method': 'GET'
+ },
+ 'delete': {
+ 'href': url,
+ 'method': 'DELETE'
+ },
+ 'diff_file_attachments': {
+ 'href': url + 'diff-file-attachments/',
+ 'method': 'GET'
+ },
+ 'info': {
+ 'href': url + 'info/',
+ 'method': 'GET'
+ },
+ 'self': {
+ 'href': url,
+ 'method': 'GET'
+ },
+ 'update': {
+ 'href': url,
+ 'method': 'PUT'
+ },
+ }
+
+ def _404_stub(self):
+ return {
+ 'responses': [
+ {
+ 'is': {
+ 'statusCode': 400,
+ 'headers': {
+ 'Content-Type': 'application/json'
+ },
+ 'body': {
+ 'err': {
+ 'code': 100,
+ 'msg': 'Object does not exist'
+ },
+ 'stat': 'fail'
+ },
+ },
+ },
+ ],
+ }
+
+ def _rb_contenttype(self, key):
+ """Return a content-type string for a resource key.
+
+ Review Board uses vendored content-types to differentiate
+ resources.
+
+ Args:
+ key: The resource data key (for either a list or item) which
+ will be used as part of the content-type.
+ Returns:
+ A content-type string.
+ """
+ return 'application/vnd.reviewboard.org.{key}+json'.format(key=key)
+
+ def _stub_paginated_list(self, list_key, item_key, pages):
+ """Return a mountebank stub for a paginated list resource.
+
+ Args:
+ list_key: The key list data is kept under in the resource.
+ item_key: The string key which the item data is found
+ under in the item.
+ pages: A list of pages to be turned into stubs.
+ Returns:
+ A list of dictionaries conforming to the spec
+ for a mountebank stub.
+ """
+ stubs = []
+ stubs.extend(
+ [self._stub_page(list_key, item_key, page) for page in pages]
+ )
+
+ # Add the bare list resource.
+ parsed_base_url = urllib.parse.urlparse(
+ pages[0]['links']['self']['href'])
+ stubs.append(
+ {
+ 'predicates': [
+ {
+ 'equals': {
+ 'method': 'GET',
+ 'path': parsed_base_url.path,
+ }
+ },
+ ],
+ 'responses': [
+ {
+ 'is': {
+ 'statusCode': 200,
+ 'headers': {
+ 'Content-Type':
+ self._rb_contenttype(list_key),
+ },
+ 'body': pages[0],
+ },
+ },
+ ],
+ }
+ )
+
+ return stubs
+
+ def _stub_page(self, list_key, item_key, page):
+ """Return a mountebank stub for a list resource page.
+
+ Args:
+ list_key: The string key which the list data is found
+ under in the page.
+ item_key: The string key which the item data is found
+ under in the item.
+ page: The response for this page.
+ Returns:
+ A dictionary conforming to the spec for a
+ mountebank stub.
+ """
+ parsed_url = urllib.parse.urlparse(page['links']['self']['href'])
+ parsed_query = urllib.parse.parse_qs(parsed_url.query)
+ query = {
+ 'start': parsed_query.get('start', [0])[0],
+ 'max-results': parsed_query.get('max-results', [0])[0],
+ },
+
+ return {
+ 'predicates': [
+ {
+ 'equals': {
+ 'method': 'GET',
+ 'path': parsed_url.path,
+ 'query': query,
+ }
+ },
+ ],
+ 'responses': [
+ {
+ 'is': {
+ 'statusCode': 200,
+ 'headers': {
+ 'Content-Type':
+ self._rb_contenttype(list_key),
+ 'Item-Content-Type':
+ self._rb_contenttype(item_key),
+ },
+ 'body': page,
+ },
+ },
+ ],
+ }
+
+ def _stub_item(self, item_key, item):
+ """Return a mountebank stub for an item.
+
+ Args:
+ item_key: The string key which the item data is found
+ under in the item.
+ item: The response for this item.
+ Returns:
+ A dictionary conforming to the spec for a
+ mountebank stub.
+ """
+ parsed_url = urllib.parse.urlparse(
+ item[item_key]['links']['self']['href']
+ )
+
+ return {
+ 'predicates': [
+ {
+ 'equals': {
+ 'method': 'GET',
+ 'path': parsed_url.path,
+ }
+ },
+ ],
+ 'responses': [
+ {
+ 'is': {
+ 'statusCode': 200,
+ 'headers': {
+ 'Content-Type': self._rb_contenttype(item_key)
+ },
+ 'body': item,
+ },
+ },
+ ],
+ }
+
+ def _paginate_list_resource(
+ self,
+ items,
+ item_key,
+ list_url,
+ list_key,
+ list_links={},
+ n_per_page=25
+ ):
+ """Paginate a Review Board list resource.
+
+ Args:
+ items: A list of item resources to be paginated.
+ item_key: A string key which the item data lives under
+ in each item.
+ list_url: The url to the list resource for the `items`.
+ list_key: A string key that items will live under on the
+ list resource.
+ list_links: A dict of links to be added to every page of
+ the list resource. The "self", "previous", and "next"
+ links are not required and will be generated by this
+ method.
+ n_per_page: The number of entries that should appear per
+ page on the paginated list resource.
+ Returns:
+ A list of pages each containing the item resources for
+ that page.
+ """
+ base_url = list_url + '?max-results={n}'.format(n=n_per_page)
+ total_results = len(items)
+ pages = []
+
+ # DO (DO WHILE LOOP)
+ while True:
+ # Item resources have the data under a key in the
+ # response, but we only want that data for in the
+ # list, so we strip off everything else.
+ page_items = [item[item_key] for item in items[:n_per_page]]
+ items = items[n_per_page:]
+
+ self_start = n_per_page * len(pages)
+ prev_start = n_per_page * (len(pages) - 1)
+ next_start = n_per_page * (len(pages) + 1)
+ self_url = base_url + '&start={i}'.format(i=self_start)
+ prev_url = None
+ next_url = None
+ page = {
+ 'stat': 'ok',
+ 'total_results': total_results,
+ 'links': copy.copy(list_links),
+ list_key: page_items,
+ }
+
+ if len(pages) > 0:
+ prev_url = base_url + '&start={i}'.format(i=prev_start)
+
+ if items:
+ next_url = base_url + '&start={i}'.format(i=next_start)
+
+ if prev_url is not None:
+ page['links']['prev'] = {
+ 'href': prev_url,
+ 'method': 'GET',
+ }
+
+ if next_url is not None:
+ page['links']['next'] = {
+ 'href': next_url,
+ 'method': 'GET',
+ }
+
+ page['links']['self'] = {
+ 'href': self_url,
+ 'method': 'GET',
+ }
+
+ pages.append(page)
+
+ # WHILE (DO WHILE LOOP)
+ if not items:
+ break
+
+ return pages
+
@pytest.fixture
def app(reviewboard):
"""Returns the tornado.Application instance we'll be testing against.
Required for pytest-tornado to function.
"""
return make_app(