commit-index: bugzilla.py module for accessing BMO's API to add attachments (
bug 1347930) r?mars
- Created commitindex/reviews/bugzilla.py module containing Bugzilla class
that has methods for attaching patches to bugs, validating API keys, and
checking if a bug is confidential.
- Created tests/test_bmo_attachments.py to test adding an attachment to a
bug in BMO using the REST API.
- Copied over testing.py from autoland to aid in Mountebank testing.
MozReview-Commit-ID: CxGF85ioNsX
--- a/.hgignore
+++ b/.hgignore
@@ -1,14 +1,16 @@
# use glob syntax.
syntax: glob
*.pyc
*~
*.egg-info
.DS_Store
.cache/
+venv
+.ropeproject
autoland/webapi/.cache
autoland/ui/build/
autoland/ui/node_modules/
autoland/ui/coverage/
autoland/ui/.coverage/
new file mode 100644
--- /dev/null
+++ b/commitindex/commitindex/reviews/bugzilla.py
@@ -0,0 +1,117 @@
+# 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/.
+
+# TODO:
+# 1. Load REST URL from system wide config
+# 2. New content_type for conduit attachments?
+# 3. Add comment_tags for conduit attachments?
+
+"""Interface to a Bugzilla system."""
+
+
+from urllib.parse import quote
+import json
+import logging
+import requests
+
+
+class Bugzilla(object):
+ """
+ Interface to a Bugzilla system.
+ """
+
+ def __init__(self, rest_url=None):
+ self.rest_url = rest_url
+ self.session = requests.Session()
+ self.logger = logging.getLogger(__name__)
+
+ def call(self, method, path, data=None):
+ """Perform REST API call and decode JSON"""
+ headers = {
+ 'Accept': 'application/json',
+ 'Content-Type': 'application/json'
+ }
+
+ if method == 'GET':
+ response = self.session.get(self.rest_url + path, headers=headers)
+
+ if method == 'POST':
+ response = self.session.post(self.rest_url + path, json=data,
+ headers=headers)
+
+ try:
+ data = json.loads(response.content.decode('utf-8'))
+ except:
+ raise BugzillaError(400, "Error decoding JSON data")
+
+ if isinstance(data, dict) and 'error' in data and data['error']:
+ raise BugzillaError(data['code'], data['message'])
+
+ return data
+
+ def is_bug_confidential(self, bug_id):
+ """Check if bug is confidential"""
+ self.logger.info('Checking if bug %d is confidential.', bug_id)
+ try:
+ self.call('GET', '/bug/' + quote(str(bug_id)))
+ except BugzillaError as error:
+ if error.fault_code == 102:
+ return True
+ except:
+ raise BugzillaError(error.fault_code, error.fault_string)
+
+ return False
+
+ def valid_api_key(self, username, api_key):
+ """Check if API key is valid for specific username"""
+ self.logger.info('Checking valid API key for %s.', username)
+ try:
+ self.call('GET', '/valid_login?login=' + quote(username) +
+ '&api_key=' + quote(api_key))
+ except BugzillaError as error:
+ if error.fault_code == 306:
+ return False
+ except:
+ raise BugzillaError(error.fault_code, error.fault_string)
+
+ return True
+
+ def create_attachment(self, bug_id, attach_data, api_key=None):
+ """Create the attachment using the provided flags.
+
+ The `flags` parameter is an array of flags to set/update/clear. This
+ array matches the Bugzilla flag API:
+ Setting:
+ {
+ 'id': flag.id
+ 'name': 'review',
+ 'status': '?',
+ 'requestee': reviewer.email
+ }
+ Clearing:
+ {
+ 'id': flag.id,
+ 'status': 'X'
+ }
+ """
+
+ self.logger.info('Posting attachment to bug %d.', bug_id)
+
+ try:
+ result = self.call('POST', '/bug/' + quote(str(bug_id)) +
+ '/attachment?api_key=' + quote(str(api_key)),
+ attach_data)
+ except BugzillaError as error:
+ print('code: %d string: %s' % (error.fault_code, error.fault_string))
+ return None
+
+ return int(list(result['attachments'].keys())[0])
+
+
+class BugzillaError(Exception):
+ """Generic Bugzilla Exception"""
+ def __init__(self, msg, code=None):
+ super(BugzillaError, self).__init__(msg)
+ self.fault_code = code
+ self.fault_string = msg
new file mode 100644
--- /dev/null
+++ b/commitindex/testing.py
@@ -0,0 +1,64 @@
+# 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 helpers.
+"""
+
+from collections import namedtuple
+
+import requests
+
+
+class MountebankClient:
+ def __init__(self, host, port=2525, imposter_port=4000):
+ self.host = host
+ self.port = port
+ self.imposter_port = imposter_port
+
+ @property
+ def imposters_admin_url(self):
+ return self.get_endpoint_with_port(self.port, '/imposters')
+
+ @property
+ def stub_baseurl(self):
+ return self.get_endpoint_with_port(self.imposter_port)
+
+ def get_endpoint(self, path=''):
+ """Construct a URL for the imposter service with optional path."""
+ return self.get_endpoint_with_port(self.imposter_port, path)
+
+ def get_endpoint_with_port(self, port, path=''):
+ """Construct a service endpoint URL with port and optional path."""
+ return 'http://{0}:{1}{2}'.format(self.host, port, path)
+
+ def create_imposter(self, imposter_json):
+ """Take a dict and turn it into a service stub."""
+ response = requests.post(self.imposters_admin_url, json=imposter_json)
+ if response.status_code != 201:
+ raise RuntimeError(
+ "mountebank imposter creation failed: {0} {1}".
+ format(response.status_code, response.content)
+ )
+
+ def create_stub(self, stub_json):
+ """Create a http stub using the default imposter port."""
+ self.create_imposter(
+ {
+ 'port': self.imposter_port,
+ 'protocol': 'http',
+ 'stubs': stub_json
+ }
+ )
+
+ def reset_imposters(self):
+ """Delete all imposters."""
+ requests.delete(self.imposters_admin_url)
+
+ def get_requests(self):
+ """Return a list of requests made to the imposter."""
+ url = self.imposters_admin_url + '/' + str(self.imposter_port)
+ return requests.get(url).json().get('requests')
+
+
+MBHostInfo = namedtuple('MBHostInfo', 'ip adminport imposterport')
new file mode 100644
--- /dev/null
+++ b/commitindex/tests/test_bmo_attachments.py
@@ -0,0 +1,124 @@
+# 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.
+"""
+
+from commitindex.reviews.bugzilla import Bugzilla
+from testing import MountebankClient
+
+import pytest
+
+
+class FakeBugzilla:
+ """Setups up the imposter test double emulating Bugzilla"""
+ def __init__(self, mountebank_client):
+ self.mountebank = mountebank_client
+
+ @property
+ def url(self):
+ """Return fully qualified url for server"""
+ # Copied from the project docker-compose.yml file.
+ return 'http://172.17.0.2:' + str(self.mountebank.imposter_port)
+
+ def create_attachment(self, bug_id):
+ """Create a attachment in the fake bugzilla server."""
+ path = '/bug/' + str(bug_id) + '/attachment'
+ self.mountebank.create_stub(
+ [
+ {
+ "predicates": [{
+ "equals": {
+ "method": "POST",
+ "headers": {
+ "Content-Type": "application/json"
+ },
+ "path": path
+ }
+ }],
+ "responses": [{
+ "is": {
+ "statusCode": 200,
+ "headers": {
+ "Content-Type": "application/json"
+ },
+ "body": {
+ "attachments": {
+ 12345: {}
+ }
+ }
+ }
+ }]
+ },
+ # 404 everything else
+ {
+ "predicates": [{
+ "not": {
+ "equals": {
+ "path": path
+ }
+ }
+ }],
+ "responses": [{
+ "is": {
+ "statusCode": 404
+ }
+ }]
+ }
+ ]
+ )
+
+
+@pytest.fixture(scope='session')
+def mountebank():
+ """Returns configured Mounteback client instance"""
+ # The docker-compose internal DNS entry for the mountebank container
+ mountebank_host = "172.17.0.2"
+ # Lifted from the docker-compose file
+ mountebank_admin_port = 2525
+ mountebank_imposter_port = 4000
+
+ return MountebankClient(
+ mountebank_host, mountebank_admin_port, mountebank_imposter_port
+ )
+
+
+@pytest.fixture
+def bugzilla(request, mountebank):
+ """Returns emulated Bugzilla service methods"""
+ # NOTE: comment out the line below if you want mountebank to save your
+ # requests and responses for inspection after the test suite completes.
+ # You can manually clean up the imposters afterwards by sending HTTP
+ # DELETE to the exposed mountebank admin port, documented in
+ # docker-compose.yml, or by restarting the mountebank container. See
+ # http://www.mbtest.org/docs/api/stubs for details.
+ request.addfinalizer(mountebank.reset_imposters)
+ return FakeBugzilla(mountebank)
+
+
+@pytest.mark.bugzilla
+def test_create_valid_attachment(bugzilla):
+ """Tests adding an attachment to the Bugzilla service"""
+ attach_data = {
+ "is_patch" : False,
+ "comment" : "This is a new attachment comment",
+ "summary" : "Test Attachment",
+ "content_type" : "text/plain",
+ "data" : "data to be encoded",
+ "file_name" : "test_attachment.patch",
+ "is_private" : False,
+ "flags" : [
+ {
+ "name" : "review",
+ "status" : "?",
+ "requestee" : "dkl@mozilla.com",
+ "new" : True
+ }
+ ]
+ }
+
+ bugzilla.create_attachment(1234)
+ bug_test = Bugzilla(rest_url=bugzilla.url)
+ result = bug_test.create_attachment(1234, attach_data, '12345')
+ assert isinstance(result, int)