diff --git a/.gitignore b/.gitignore index 6d7373f..71501ca 100644 --- a/.gitignore +++ b/.gitignore @@ -33,3 +33,6 @@ nosetests.xml .mr.developer.cfg .project .pydevproject + +# Vim +*.swp diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..66ecd94 --- /dev/null +++ b/Makefile @@ -0,0 +1,27 @@ +nosetests := nosetests \ + --no-byte-compile \ + --with-coverage \ + --cover-erase \ + --cover-package=codebase \ + --cover-branches \ + --cover-inclusive + +runtests-install: + @clear + @echo Preflight tasks... + @pip install -U pip >> /dev/null + @pip install -r tests/requirements.txt >> /dev/null + @find ./ -name "*.pyc" -delete >> /dev/null + + @echo Running tests: + @$(nosetests) --where=tests/unit $(specs) + +runtests: + @clear + @echo Running tests: + @$(nosetests) --where=tests/unit $(specs) + +runtests-all: + @clear + @echo Running tests: + @$(nosetests) $(specs) diff --git a/README.md b/README.md index 3d78b0b..24b7fee 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@ A Python client providing read/write access to the [Codebase API](http://support Install ------- - pip install -U git+git://github.com/igniteflow/codebase-python-api-client + pip install git+git://github.com/igniteflow/codebase-python-api-client CLI --- @@ -56,10 +56,3 @@ Use the client in your code }, } codebase.add_note(ticket_id=1, data=note_data) - -Debugging ---------- - -By default, data is given and returned as Python dicts. To get the raw requests.Response object, just set CodeBaseAPI.DEBUG to True. - - diff --git a/bin/codebase b/bin/codebase index ff3681b..09bd1f0 100755 --- a/bin/codebase +++ b/bin/codebase @@ -1,37 +1,81 @@ #!/usr/bin/env python - +import argparse +import logging import pprint -import sys + +from terminaltables import AsciiTable from codebase.client import CodeBaseAPI +logger = logging.getLogger(__name__) + +""" +Simple command-line interface to the Codebase API. Example usages: + + codebase myproject all_notes 6 + codebase myproject search "status:new" + +Arguments: +1. project name +2. api function name +3. args to pass to the api function + +Search: + +codebase my_project search "status:code complete" +""" + + +def get_search_table_data(response): + table_headers = [ + '#', + 'Summary', + 'Assignee', + ] + + return [table_headers] + [ + [ticket['ticket']['ticket_id'], ticket['ticket']['summary'], ticket['ticket']['assignee']] + for ticket in response + ] + + +def available_commands(): + return [ + name + for name in dir(CodeBaseAPI) + if name[0] != '_' and name[0].islower() + ] + + +def main(): + parser = argparse.ArgumentParser(description='Codebase command-line interface') + parser.add_argument('project', help='Codebase project name') + parser.add_argument('command', choices=available_commands(), help='A Codebase API command') + parser.add_argument('search_term', type=str, help='A Codebase API command') + args = parser.parse_args() + + project = args.project + command = args.command + search_term = args.search_term + + codebase = CodeBaseAPI(project=project) + + try: + if command == 'search' and ':' in search_term: + k, v = search_term.split(':') + response = getattr(codebase, command)(**{k: v}) + table_data = get_search_table_data(response) + table = AsciiTable(table_data) + print table.table + else: + response = getattr(codebase, command)() + pprint.pprint(response) + except Exception as e: + logger.error(e) + print('Error for [project] {} [command] {} [args] {}'.format( + project, command, args + )) + if __name__ == "__main__": - """ - Simple command-line interface to the Codebase API. Example usage: - - codebase myproject all_notes 6 - - Arguments: - 1. project name - 2. api function name - 3. args to pass to the api function - """ - if len(sys.argv) == 1: - # if no args then print available API methods - for name in dir(CodeBaseAPI): - if name[0] != '_' and name[0].islower(): - print name - else: - # make the API call - project = sys.argv[1] - command = sys.argv[2] - args = sys.argv[3:] - - codebase = CodeBaseAPI( - project=project, - debug=True - ) - - response = getattr(codebase, command)(*args) - pprint.pprint(response) + main() diff --git a/codebase/__init__.py b/codebase/__init__.py index 95e1be0..eea436a 100644 --- a/codebase/__init__.py +++ b/codebase/__init__.py @@ -1,3 +1,3 @@ import logging -logger = logging.getLogger(__name__) \ No newline at end of file +logger = logging.getLogger(__name__) diff --git a/codebase/client.py b/codebase/client.py old mode 100755 new mode 100644 index 8775632..84aa419 --- a/codebase/client.py +++ b/codebase/client.py @@ -1,71 +1,108 @@ import base64 import json import logging -import requests +import urllib import urllib2 +import urlparse + +import xmltodict from codebase import logger from codebase.settings import Settings class Auth(object): + CTYPE_JSON = 'json' + CTYPE_XML = 'xml' + API_ENDPOINT = 'https://api3.codebasehq.com' - API_ENDPOINT = 'http://api3.codebasehq.com' - - def _default_settings(self): - settings = Settings() - self.username = settings.CODEBASE_USERNAME - self.apikey = settings.CODEBASE_APIKEY - - def __init__(self, project, username=None, apikey=None, debug=False, **kwargs): + def __init__(self, project=None, username=None, apikey=None, **kwargs): super(Auth, self).__init__(**kwargs) - if username and apikey: - self.username = username - self.apikey = apikey - else: - self._default_settings() + if not (username or apikey): + settings = self._get_settings() + username = settings.username + apikey = settings.apikey + + self.username = username + self.apikey = apikey self.project = project - self.DEBUG = debug - self.HEADERS = { - "Content-type": "application/json", - "Accept": "application/json", - "Authorization": base64.encodestring("%s:%s" % (self.username, self.apikey))\ - .replace('\n', '') - } + def _get_settings(self): + settings = Settings() + settings.import_settings() + return settings - if debug: - logging.basicConfig( - level=logging.DEBUG, - format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s', - datefmt='%m-%d %H:%M', + def get_absolute_url(self, path): + return urlparse.urljoin(self.API_ENDPOINT, path) + + def get_headers(self, ctype): + return { + 'Content-type': 'application/{}'.format(ctype), + 'Accept': 'application/{}'.format(ctype), + 'Authorization': base64.b64encode( + '{}:{}'.format(self.username, self.apikey) ) + } - def get(self, url): - response = requests.get(self.get_absolute_url(url), headers=self.HEADERS) - return self.handle_response(response) + def get_data(self, raw_data, ctype): + if ctype == self.CTYPE_XML: + return xmltodict.unparse(raw_data) + + # Encodes the parameters by default (i.e. using json). + return urllib.urlencode(raw_data) + + def _handle_response(self, response, ctype): + try: + status_code = response.getcode() + content = response.read() + logger.debug('{} returned status code {}'.format( + response.url, + status_code + )) + + if ctype == self.CTYPE_XML: + return xmltodict.parse(content) + return json.loads(content) + except Exception as e: + logging.exception('%s: %s', e.__class__.__name__, e.message) + + def _send_request(self, url, ctype=None, data=None): + if not ctype: + ctype = self.CTYPE_JSON + + absolute_url = self.get_absolute_url(url) + headers = self.get_headers(ctype) + req_params = { + 'url': absolute_url, + 'headers': headers, + } + if data: + data = self.get_data(data, ctype) + req_params['data'] = data - def post(self, url, data): - response = requests.post(self.get_absolute_url(url), data=json.dumps(data), headers=self.HEADERS) - return self.handle_response(response) + request = urllib2.Request(**req_params) + logging.info('Making request to {} with headers {}'.format( + request.get_full_url(), + request.headers, + )) - def handle_response(self, response): - logger.debug('Response status code: {}'.format(response.status_code)) - if response.ok: - return json.loads(response.content) - else: - return response.content + response = urllib2.urlopen(request) + return self._handle_response(response, ctype) - def get_absolute_url(self, path): - absolute_url = self.API_ENDPOINT + path - logger.debug(absolute_url) - return absolute_url + def get(self, url, ctype=None): + return self._send_request(url, ctype=ctype) + + def post(self, url, data, ctype=None): + return self._send_request(url, data=data, ctype=ctype) class CodeBaseAPI(Auth): + def projects(self): + return self.get('/projects') + def statuses(self): return self.get('/%s/tickets/statuses' % self.project) @@ -75,16 +112,48 @@ def priorities(self): def categories(self): return self.get('/%s/tickets/categories' % self.project) - def milestones(self): - return self.get('/%s/tickets/milestones' % self.project) + def types(self): + return self.get('/%s/tickets/types' % self.project) - def search(self, term): - terms = term.split(':') - if len(terms) == 1: - escaped_term = urllib2.quote(terms[0]) - else: - escaped_term = '{}:"{}"'.format(terms[0], urllib2.quote(terms[1])) - return self.get('/%s/tickets?query=%s' % (self.project, escaped_term)) + def milestones(self): + return self.get('/%s/milestones' % self.project) + + def search(self, term=None, page=None, **kwargs): + queries = [] + if term: + queries.append(term.strip()) + + for term_name, term_value in kwargs.iteritems(): + if term_name.startswith('not_'): + term_name = term_name.replace('not_', 'not-') + queries.append('{}:"{}"'.format(term_name, term_value.strip())) + + params = {'query': ' '.join(queries)} + if page and page > 1 and type(page) is int: + params['page'] = page + + url = '/{}/tickets?{}'.format(self.project, urllib.urlencode(params)) + return self.get(url) + + def search_all(self, term=None, **kwargs): + page = 1 + tickets = [] + while True: + try: + tickets.extend(self.search(term=term, page=page, **kwargs)) + page += 1 + except urllib2.HTTPError: + page -= 1 + break + except Exception, e: + logger.error( + u'An error occured while searching for "%s" ' + u'(current page: %s):\n%s', term, page, e + ) + page -= 1 + break + + return tickets def watchers(self, ticket_id): return self.get('/%s/tickets/%s/watchers' % (self.project, ticket_id)) @@ -116,6 +185,15 @@ def discussions(self): def discussion_categories(self): return self.get('/%s/discussions/categories' % self.project) + def create_ticket(self, data): + # It looks like that only XML requests are accepted for creating new + # tickets. + return self.post( + '/%s/tickets' % self.project, + data, + ctype=self.CTYPE_XML, + ) + def create_discussion(self, data): return self.post('/%s/discussions' % self.project, data) @@ -151,5 +229,4 @@ def hooks(self, repository): return self.get('/%s/%s/hooks' % (self.project, repository)) def add_hook(self, repository, data): - return self.get('/%s/%s/hooks' % (self.project, repository), data) - + return self.post('/%s/%s/hooks' % (self.project, repository), data) diff --git a/codebase/settings.py b/codebase/settings.py old mode 100755 new mode 100644 index 0a7d0b2..c94746d --- a/codebase/settings.py +++ b/codebase/settings.py @@ -1,6 +1,6 @@ +import errno import json - -from os.path import expanduser +import os from codebase import logger @@ -17,19 +17,60 @@ class Settings(object): """ By default settings are expected to be in a file named `.codebase` in your home directory """ - SETTINGS_FILE_PATH = expanduser("~") + '/.codebase' + SETTINGS_FILE_PATH = '~/.codebase' + + def __init__(self, file_path=None): + # Initializing the credentials attributes. + self.username = None + self.apikey = None + + # Getting the settings path. + if not file_path or file_path.startswith('~'): + file_path = file_path or self.SETTINGS_FILE_PATH + try: + file_path = os.path.expanduser(file_path) + except ImportError: + file_name = os.path.basename(file_path) + file_path = os.path.join(os.getcwd(), file_name) + logger.warning( + u'Looks like you are running on GoogleAppEngine, which ' + u'means that "os.path.expanduser" is not supported. ' + u'Please override the settings logic in ' + u'`codebase.client.Auth._get_settings` or explicitely ' + u'pass the required credentials informations.' + ) + logger.warning( + u'The following path will be used in place of the default ' + u'value: {}'.format(file_path) + ) - def get_path(self): - home = expanduser("~") - return home + '.codebase' + self.file_path = file_path - def __init__(self): + def import_settings(self): try: - with open(self.SETTINGS_FILE_PATH) as f: - for k, v in json.loads(f.read()).iteritems(): - setattr(self, k, v) - except IOError: + with open(self.file_path, 'r') as f: + settings = json.loads(f.read()) + + self.username = settings.pop('CODEBASE_USERNAME', None) + self.apikey = settings.pop('CODEBASE_APIKEY', None) + assert self.username is not None, '"CODEBASE_USERNAME" is required' + assert self.apikey is not None, '"CODEBASE_APIKEY" is required' + + for k, v in settings.iteritems(): + setattr(self, k, v) + except IOError, io_err: logger.error( - 'Settings file "~/.codebase" not found. Please add your settings in JSON format e.g. %s' - % EXAMPLE_FORMAT + u'Settings file "{}" not found. ' + u'Please add your settings in JSON format e.g. {}'.format( + self.file_path, + EXAMPLE_FORMAT, + ) ) + if io_err.errno == errno.EACCES: + # EACCES = 13, which means "file not accessible" (permission + # denied). + logger.error( + 'The problem could be caused by the name of the file. ' + 'Try to use a non-hidden file name (i.e. a file which ' + 'does not start with a dot).' + ) diff --git a/codebase/tests.py b/codebase/tests.py deleted file mode 100644 index 04d895c..0000000 --- a/codebase/tests.py +++ /dev/null @@ -1,32 +0,0 @@ -import unittest - -from client import CodeBaseAPI - - -class CodebaseAPITest(unittest.TestCase): - - PROJECT = 'foo' - - def setUp(self): - self.codebase = CodeBaseAPI( - project=self.PROJECT, - debug=True - ) - - def test_statuses(self): - self.assertEqual(self.codebase.statuses().status_code, 200) - - def test_priorities(self): - self.assertEqual(self.codebase.priorities().status_code, 200) - - def test_categories(self): - self.assertEqual(self.codebase.categories().status_code, 200) - - def test_milestones(self): - self.assertEqual(self.codebase.milestones().status_code, 200) - - def test_search(self): - self.assertEqual(self.codebase.search('one').status_code, 200) - -if __name__ == '__main__': - unittest.main() diff --git a/codebase/utils.py b/codebase/utils.py index 7733a39..4a433e5 100644 --- a/codebase/utils.py +++ b/codebase/utils.py @@ -7,6 +7,20 @@ class CodeBaseAPIUtils(CodeBaseAPI): def bulk_update_ticket_statuses(self, current_status_name, target_status_name): + """ + Example usage to set all "Approved for Dev" tp "Deployed to Dev": + + STATUS_TRANSITIONS = { + 'dev': ('Approved for Dev', 'Deployed to Dev'), + 'uat': ('Approved for UAT', 'Deployed to UAT'), + 'prod': ('Approved for Prod', 'Deployed to Prod'), + } + + env = 'dev' + current_status_name = STATUS_TRANSITIONS[env][0] + target_status_name = STATUS_TRANSITIONS[env][1] + codebase_utils.bulk_update_ticket_statuses(current_status_name, target_status_name) + """ # get the status ids because Codebase search doesn't support searching on status id new_status_id = None @@ -17,13 +31,19 @@ def bulk_update_ticket_statuses(self, current_status_name, target_status_name): new_status_id = status['ticketing_status']['id'] # exit if the ticket status was not found - if target_status_name is None: - logger.info('Status {} not found in project statuses. Options are: {}'.format( - [status['ticketing_status']['name'] for status in statuses] - )) + if new_status_id is None: + status_names = ', '.join([ + status['ticketing_status']['name'] for status in statuses + ]) + logger.info( + u'Status "{}" not found in project statuses. ' + u'Options are: {}'.format(target_status_name, status_names) + ) + return # update the tickets - items = self.search('status:{}'.format(current_status_name)) + items = self.search_all(status=current_status_name) + updated = [] for item in items: ticket_id = item['ticket']['ticket_id'] data = { @@ -36,9 +56,12 @@ def bulk_update_ticket_statuses(self, current_status_name, target_status_name): self.add_note(ticket_id, data) # print output - sys.stdout.write('[{}] {} {} --> {}'.format( + sys.stdout.write('[{}] {} {} --> {}\n'.format( ticket_id, item['ticket']['summary'], current_status_name, target_status_name - )) \ No newline at end of file + )) + updated.append(ticket_id) + + return updated diff --git a/setup.py b/setup.py index e1a0e98..9907789 100644 --- a/setup.py +++ b/setup.py @@ -11,7 +11,8 @@ url='https://github.com/igniteflow/codebase-python-api-client', packages=['codebase'], install_requires=[ - 'requests>=2.0.1', + 'xmltodict>=0.9.2', + 'terminaltables>=3.1.0', ], license='MIT', scripts=[ diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/decorators.py b/tests/decorators.py new file mode 100644 index 0000000..dc14f68 --- /dev/null +++ b/tests/decorators.py @@ -0,0 +1,63 @@ +import sys +if sys.version_info.major == 2: + import __builtin__ as builtins +else: + import builtins +import functools + +from mock import patch, mock_open + + +class patch_open(object): + + def __init__(self, read_data): + self.open_patch = patch.object( + builtins, + 'open', + mock_open(read_data=read_data), + ) + + def __call__(self, fn): + if hasattr(fn, 'patchings'): + fn.patchings.append(self) + return fn + + @functools.wraps(fn) + def wrapped_func(*args, **kwargs): + extra_args = [] + entered_patchers = [] + exc_info = tuple() + + try: + for patching in wrapped_func.patchings: + arg = patching.__enter__() + entered_patchers.append(patching) + extra_args.append(arg) + + args += tuple(extra_args) + func_res = fn(*args, **kwargs) + return func_res + except: + if ( + patching not in entered_patchers and + hasattr(patching, 'is_local') + ): + # the patcher may have been started, but an exception + # raised whilst entering one of its additional_patchers + entered_patchers.append(patching) + # Pass the exception to __exit__ + exc_info = sys.exc_info() + # re-raise the exception + raise + finally: + for patching in reversed(entered_patchers): + patching.__exit__(*exc_info) + + wrapped_func.patchings = [self] + return wrapped_func + + def __enter__(self): + return self.open_patch.__enter__() + + def __exit__(self, *args): + self.open_patch.__exit__(*args) diff --git a/tests/functional/__init__.py b/tests/functional/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/functional/tests.py b/tests/functional/tests.py new file mode 100644 index 0000000..2812163 --- /dev/null +++ b/tests/functional/tests.py @@ -0,0 +1,25 @@ +from unittest import TestCase + +from codebase.client import CodeBaseAPI + + +class CodebaseAPITest(TestCase): + PROJECT = 'foo' + + def setUp(self): + self.codebase = CodeBaseAPI(project=self.PROJECT) + + def test_statuses(self): + self.assertIsNotNone(self.codebase.statuses()) + + def test_priorities(self): + self.assertIsNotNone(self.codebase.priorities()) + + def test_categories(self): + self.assertIsNotNone(self.codebase.categories()) + + def test_milestones(self): + self.assertIsNotNone(self.codebase.milestones()) + + def test_search(self): + self.assertIsNotNone(self.codebase.search('one')) diff --git a/tests/requirements.txt b/tests/requirements.txt new file mode 100644 index 0000000..14310e9 --- /dev/null +++ b/tests/requirements.txt @@ -0,0 +1,4 @@ +xmltodict +nose +coverage +mock diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/client_tests.py b/tests/unit/client_tests.py new file mode 100644 index 0000000..eb47219 --- /dev/null +++ b/tests/unit/client_tests.py @@ -0,0 +1,808 @@ +from unittest import TestCase +import base64 +import json +import urllib +import urllib2 +import urlparse + +from mock import Mock, call, patch +import xmltodict + +from codebase.client import Auth, CodeBaseAPI + + +class AuthInitTestCase(TestCase): + + @patch( + 'codebase.client.Auth._get_settings', + return_value=Mock(username='foo', apikey='bar') + ) + def test_init_credentials_ignores_get_settings(self, get_settings_mock): + auth = Auth(project='project', username='some/body', apikey='bees') + + get_settings_mock.assert_not_called() + self.assertEqual(auth.project, 'project') + self.assertEqual(auth.username, 'some/body') + self.assertEqual(auth.apikey, 'bees') + + @patch( + 'codebase.client.Auth._get_settings', + return_value=Mock(username='foo', apikey='bar') + ) + def test_init_credentials_calls_get_settings(self, get_settings_mock): + auth = Auth(project='project') + + get_settings_mock.assert_called_once_with() + self.assertEqual(auth.project, 'project') + self.assertEqual(auth.username, 'foo') + self.assertEqual(auth.apikey, 'bar') + + @patch('codebase.client.Settings', return_value=Mock(username='foo', apikey='bar')) + def test_settings_ignored_on_init(self, settings_mock): + instance = settings_mock.return_value + + # Sanity check. + settings_mock.assert_not_called() + instance.import_settings.assert_not_called() + + # Creating the object. + auth = Auth(project='project', username='some/body', apikey='bees') + + settings_mock.assert_not_called() + instance.import_settings.assert_not_called() + + self.assertEqual(auth.project, 'project') + self.assertEqual(auth.username, 'some/body') + self.assertEqual(auth.apikey, 'bees') + + @patch('codebase.client.Settings', return_value=Mock(username='foo', apikey='bar')) + def test_settings_used_on_init(self, settings_mock): + instance = settings_mock.return_value + + # Sanity check. + settings_mock.assert_not_called() + instance.import_settings.assert_not_called() + + # Creating the object. + auth = Auth(project='project') + + settings_mock.assert_called_once_with() + instance.import_settings.assert_called_once_with() + self.assertEqual(auth.project, 'project') + self.assertEqual(auth.username, 'foo') + self.assertEqual(auth.apikey, 'bar') + + +class AuthTestCase(TestCase): + + def setUp(self): + super(AuthTestCase, self).setUp() + + self.base_api_url = 'https://api3.codebasehq.com' + self.auth_client = Auth( + project='project', + username='some/body', + apikey='bees', + ) + + def _create_fake_response(self, data, ctype=None, code=200): + # Does not change the data if ctype is different than json or xml + # (useful if we want to throw an error). + content = data + if ctype == 'xml': + content = xmltodict.unparse(data) + elif ctype == 'json': + content = json.dumps(data) + + return Mock( + getcode=Mock(return_value=code), + read=Mock(return_value=content), + url='http://theurl.com', + ) + + def test_get_absolute_url(self): + path = '/the/resource/' + expected_url = urlparse.urljoin(self.base_api_url, path) + self.assertEqual(self.auth_client.get_absolute_url(path), expected_url) + + def test_get_headers(self): + for ctype in ['json', 'xml']: + expected_headers = { + 'Content-type': 'application/{}'.format(ctype), + 'Accept': 'application/{}'.format(ctype), + 'Authorization': base64.b64encode('{}:{}'.format( + self.auth_client.username, self.auth_client.apikey + )), + } + self.assertEqual( + self.auth_client.get_headers(ctype), + expected_headers, + ) + + def test_get_data_json(self): + data = {'something': 'to send', 'to': 'codebase'} + self.assertEqual( + self.auth_client.get_data(data, 'json'), + urllib.urlencode(data), + ) + self.assertEqual( + self.auth_client.get_data(data, 'blabla'), + urllib.urlencode(data), + ) + + def test_get_data_xml(self): + data = {'theroot': {'something': 'to send', 'to': 'codebase'}} + self.assertEqual( + self.auth_client.get_data(data, 'xml'), + xmltodict.unparse(data), + ) + + @patch('codebase.client.logger') + def test_handle_response_json(self, logger_mock): + data = {'something': 'to send', 'to': 'codebase'} + response = self._create_fake_response(data, ctype='json') + + handled_response = self.auth_client._handle_response(response, 'json') + + self.assertEqual(handled_response, data) + response.getcode.assert_called_once_with() + response.read.assert_called_once_with() + logger_mock.debug.assert_called_once_with( + 'http://theurl.com returned status code 200' + ) + + @patch('codebase.client.logger') + def test_handle_response_xml(self, logger_mock): + data = {'theroot': {'something': 'to send', 'to': 'codebase'}} + response = self._create_fake_response(data, ctype='xml') + + handled_response = self.auth_client._handle_response(response, 'xml') + + self.assertEqual(handled_response, data) + response.getcode.assert_called_once_with() + response.read.assert_called_once_with() + logger_mock.debug.assert_called_once_with( + 'http://theurl.com returned status code 200' + ) + + @patch('codebase.client.logging') + @patch('codebase.client.logger') + def test_handle_response_error(self, logger_mock, logging_mock): + data = 'this is not a valid json' + response = self._create_fake_response(data) + + handled_response = self.auth_client._handle_response(response, 'json') + + self.assertIsNone(handled_response) + response.getcode.assert_called_once_with() + response.read.assert_called_once_with() + logger_mock.debug.assert_called_once_with( + 'http://theurl.com returned status code 200' + ) + logging_mock.exception.assert_called_once_with( + '%s: %s', 'ValueError', 'No JSON object could be decoded' + ) + + @patch('codebase.client.urllib2.urlopen') + @patch('codebase.client.urllib2.Request') + def test_send_request_default(self, urllib_req_mock, urllib_urlopen_mock): + response_data = {'data': 1} + response_fake = self._create_fake_response(response_data, ctype='json') + urllib_urlopen_mock.return_value = response_fake + resource_path = 'the/url/to/the/resource' + + response = self.auth_client._send_request(resource_path) + + self.assertEqual(response, response_data) + urllib_urlopen_mock.assert_called_once_with(urllib_req_mock.return_value) + urllib_req_mock.assert_called_once_with( + url=urlparse.urljoin(self.base_api_url, resource_path), + headers=self.auth_client.get_headers('json'), + ) + + @patch('codebase.client.urllib2.urlopen') + @patch('codebase.client.urllib2.Request') + def test_send_request_json(self, urllib_req_mock, urllib_urlopen_mock): + response_data = {'data': 1} + response_fake = self._create_fake_response(response_data, ctype='json') + urllib_urlopen_mock.return_value = response_fake + resource_path = 'the/url/to/the/resource' + + response = self.auth_client._send_request(resource_path, ctype='json') + + self.assertEqual(response, response_data) + urllib_urlopen_mock.assert_called_once_with(urllib_req_mock.return_value) + urllib_req_mock.assert_called_once_with( + url=urlparse.urljoin(self.base_api_url, resource_path), + headers=self.auth_client.get_headers('json'), + ) + + @patch('codebase.client.urllib2.urlopen') + @patch('codebase.client.urllib2.Request') + def test_send_request_xml(self, urllib_req_mock, urllib_urlopen_mock): + response_data = {'root': 'the value'} + response_fake = self._create_fake_response(response_data, ctype='xml') + urllib_urlopen_mock.return_value = response_fake + resource_path = 'the/url/to/the/resource' + + response = self.auth_client._send_request(resource_path, ctype='xml') + + self.assertEqual(dict(response), response_data) + urllib_urlopen_mock.assert_called_once_with(urllib_req_mock.return_value) + urllib_req_mock.assert_called_once_with( + url=urlparse.urljoin(self.base_api_url, resource_path), + headers=self.auth_client.get_headers('xml'), + ) + + @patch('codebase.client.urllib2.urlopen') + @patch('codebase.client.urllib2.Request') + def test_send_request_error(self, urllib_req_mock, urllib_urlopen_mock): + response_data = 'invalid xml' + response_fake = self._create_fake_response(response_data) + urllib_urlopen_mock.return_value = response_fake + resource_path = 'the/url/to/the/resource' + + response = self.auth_client._send_request(resource_path, ctype='xml') + + self.assertIsNone(response) + urllib_urlopen_mock.assert_called_once_with(urllib_req_mock.return_value) + urllib_req_mock.assert_called_once_with( + url=urlparse.urljoin(self.base_api_url, resource_path), + headers=self.auth_client.get_headers('xml'), + ) + + @patch('codebase.client.urllib2.urlopen') + @patch('codebase.client.urllib2.Request') + def test_send_request_data(self, urllib_req_mock, urllib_urlopen_mock): + response_data = {'response': 1} + request_data = {'request': 2} + response_fake = self._create_fake_response(response_data, ctype='json') + urllib_urlopen_mock.return_value = response_fake + resource_path = 'the/url/to/the/resource' + + response = self.auth_client._send_request(resource_path, data=request_data) + + self.assertEqual(response, response_data) + urllib_urlopen_mock.assert_called_once_with(urllib_req_mock.return_value) + urllib_req_mock.assert_called_once_with( + url=urlparse.urljoin(self.base_api_url, resource_path), + headers=self.auth_client.get_headers('json'), + data=self.auth_client.get_data(request_data, 'json') + ) + + @patch('codebase.client.Auth._send_request') + def test_get_default(self, send_request_mock): + data = {'data': 1} + send_request_mock.return_value = data + resource_path = 'the/url/to/the/resource' + + response = self.auth_client.get(resource_path) + + self.assertEqual(response, data) + send_request_mock.assert_called_once_with(resource_path, ctype=None) + + @patch('codebase.client.Auth._send_request') + def test_get_ctype(self, send_request_mock): + data = {'data': 1} + send_request_mock.return_value = data + resource_path = 'the/url/to/the/resource' + + response = self.auth_client.get(resource_path, ctype='json') + + self.assertEqual(response, data) + send_request_mock.assert_called_once_with(resource_path, ctype='json') + + @patch('codebase.client.Auth._send_request') + def test_post_default(self, send_request_mock): + data_response = {'response': 1} + data_request = {'request': 2} + send_request_mock.return_value = data_response + resource_path = 'the/url/to/the/resource' + + response = self.auth_client.post(resource_path, data_request) + + self.assertEqual(response, data_response) + send_request_mock.assert_called_once_with( + resource_path, + data=data_request, + ctype=None, + ) + + @patch('codebase.client.Auth._send_request') + def test_post_ctype(self, send_request_mock): + data_response = {'response': 1} + data_request = {'request': 2} + send_request_mock.return_value = data_response + resource_path = 'the/url/to/the/resource' + + response = self.auth_client.post(resource_path, data_request, ctype='json') + + self.assertEqual(response, data_response) + send_request_mock.assert_called_once_with( + resource_path, + data=data_request, + ctype='json', + ) + + +@patch('codebase.client.CodeBaseAPI.post') +@patch('codebase.client.CodeBaseAPI.get') +class CodeBaseAPITestCase(TestCase): + + def setUp(self): + super(CodeBaseAPITestCase, self).setUp() + + self.api_client = CodeBaseAPI( + project='project', + username='some/body', + apikey='bees', + ) + + def test_get_projects(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + self.api_client.projects() + + get_mock.assert_called_once_with('/projects') + post_mock.assert_not_called() + + def test_statuses(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + expected_url = '/{}/tickets/statuses'.format(self.api_client.project) + self.api_client.statuses() + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + def test_priorities(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + expected_url = '/{}/tickets/priorities'.format(self.api_client.project) + self.api_client.priorities() + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + def test_categories(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + expected_url = '/{}/tickets/categories'.format(self.api_client.project) + self.api_client.categories() + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + def test_types(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + expected_url = '/{}/tickets/types'.format(self.api_client.project) + self.api_client.types() + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + def test_milestones(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + expected_url = '/{}/milestones'.format(self.api_client.project) + self.api_client.milestones() + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + def test_watchers(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + expected_url = '/{}/tickets/12/watchers'.format(self.api_client.project) + + self.api_client.watchers(12) + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + def test_project_groups(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + self.api_client.project_groups() + + get_mock.assert_called_once_with('/project_groups') + post_mock.assert_not_called() + + def test_get_project_users(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + expected_url = '/{}/assignments'.format(self.api_client.project) + self.api_client.get_project_users() + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + def test_set_project_users(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + data = {'some_data': 'some-value'} + expected_url = '/{}/assignments'.format(self.api_client.project) + self.api_client.set_project_users(data) + + get_mock.assert_not_called() + post_mock.assert_called_once_with(expected_url, data) + + def test_activity(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + self.api_client.activity() + + get_mock.assert_called_once_with('/activity') + post_mock.assert_not_called() + + def test_project_activity(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + expected_url = '/{}/activity'.format(self.api_client.project) + self.api_client.project_activity() + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + def test_users(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + self.api_client.users() + + get_mock.assert_called_once_with('/users') + post_mock.assert_not_called() + + def test_roles(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + self.api_client.roles() + + get_mock.assert_called_once_with('/roles') + post_mock.assert_not_called() + + def test_discussions(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + expected_url = '/{}/discussions'.format(self.api_client.project) + self.api_client.discussions() + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + def test_discussion_categories(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + expected_url = '/{}/discussions/categories'.format(self.api_client.project) + self.api_client.discussion_categories() + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + def test_create_ticket(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + data = {'some_data': 'some-value'} + expected_url = '/{}/tickets'.format(self.api_client.project) + self.api_client.create_ticket(data) + + get_mock.assert_not_called() + post_mock.assert_called_once_with( + expected_url, + data, + ctype='xml' + ) + + def test_create_discussion(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + data = {'some_data': 'some-value'} + expected_url = '/{}/discussions'.format(self.api_client.project) + self.api_client.create_discussion(data) + + get_mock.assert_not_called() + post_mock.assert_called_once_with(expected_url, data) + + def test_posts_in_discussion(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + permalink = 'something-to-point-to' + expected_url = '/{}/discussions/{}/posts'.format( + self.api_client.project, + permalink, + ) + self.api_client.posts_in_discussion(permalink) + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + def test_createpost_in_discussion(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + data = {'some_data': 'some-value'} + permalink = 'something-to-point-to' + expected_url = '/{}/discussions/{}/posts'.format( + self.api_client.project, + permalink, + ) + self.api_client.createpost_in_discussion(permalink, data) + + get_mock.assert_not_called() + post_mock.assert_called_once_with(expected_url, data) + + def test_notes(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + expected_url = '/{}/tickets/12/notes'.format(self.api_client.project) + self.api_client.notes(12) + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + def test_note(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + expected_url = '/{}/tickets/12/notes/94'.format(self.api_client.project) + self.api_client.note(12, 94) + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + def test_add_note(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + data = {'some_data': 'some-value'} + expected_url = '/{}/tickets/12/notes'.format(self.api_client.project) + self.api_client.add_note(12, data) + + get_mock.assert_not_called() + post_mock.assert_called_once_with(expected_url, data) + + def test_branches(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + expected_url = '/{}/repo-name/branches'.format(self.api_client.project) + self.api_client.branches('repo-name') + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + def test_hooks(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + expected_url = '/{}/repo-name/hooks'.format(self.api_client.project) + self.api_client.hooks('repo-name') + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + def test_add_hook(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + data = {'some_data': 'some-value'} + expected_url = '/{}/repo-name/hooks'.format(self.api_client.project) + self.api_client.add_hook('repo-name', data) + + get_mock.assert_not_called() + post_mock.assert_called_once_with(expected_url, data) + + +@patch('codebase.client.CodeBaseAPI.post') +@patch('codebase.client.CodeBaseAPI.get') +class CodeBaseAPISearchTestCase(TestCase): + + def setUp(self): + super(CodeBaseAPISearchTestCase, self).setUp() + + self.api_client = CodeBaseAPI( + project='project', + username='some/body', + apikey='bees', + ) + + def test_search_no_terms(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + expected_url = '/{}/tickets?query='.format(self.api_client.project) + self.api_client.search() + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + def test_search_only_term(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + term = 'something interesting' + expected_url = '/{}/tickets?query={}'.format( + self.api_client.project, + urllib.quote_plus(term), + ) + + self.api_client.search(term=term) + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + def test_search_only_kwargs(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + expected_url = '/{}/tickets?{}'.format( + self.api_client.project, + urllib.urlencode({'query': 'something:"more interesting"'}), + ) + + self.api_client.search(something='more interesting') + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + def test_search_exclude(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + expected_url = '/{}/tickets?{}'.format( + self.api_client.project, + urllib.urlencode({'query': 'not-something:"more interesting"'}), + ) + + self.api_client.search(not_something='more interesting') + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + def test_search_with_page_number(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + expected_url = '/{}/tickets?{}'.format( + self.api_client.project, + urllib.urlencode( + {'query': 'the term', 'page': 2} + ), + ) + + self.api_client.search(term='the term', page=2) + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + def test_search_with_bad_page_number(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + for page_number in (-1, 0, 1, 'a-string'): + expected_url = '/{}/tickets?{}'.format( + self.api_client.project, + urllib.urlencode({'query': 'the term'}), + ) + + self.api_client.search(term='the term', page=page_number) + + get_mock.assert_called_with(expected_url) + post_mock.assert_not_called() + + def test_search_with_everything(self, get_mock, post_mock): + get_mock.assert_not_called() + post_mock.assert_not_called() + + expected_url = '/{}/tickets?{}'.format( + self.api_client.project, + urllib.urlencode({ + 'query': 'title something:"more interesting" not-status:"New"', + 'page': 2, + }), + ) + + self.api_client.search( + term='title', + something='more interesting', + not_status='New', + page=2, + ) + + get_mock.assert_called_once_with(expected_url) + post_mock.assert_not_called() + + +@patch('codebase.client.CodeBaseAPI.post') +@patch('codebase.client.CodeBaseAPI.search') +class CodeBaseAPISearchAllTestCase(TestCase): + + def setUp(self): + super(CodeBaseAPISearchAllTestCase, self).setUp() + + self.api_client = CodeBaseAPI( + project='project', + username='some/body', + apikey='bees', + ) + + def test_search_all_no_res(self, search_mock, post_mock): + search_mock.assert_not_called() + post_mock.assert_not_called() + + search_mock.side_effect = urllib2.HTTPError + + res = self.api_client.search_all(term='title', status='New') + + self.assertEqual(res, []) + + search_mock.assert_called_once_with(term='title', page=1, status='New') + post_mock.assert_not_called() + + def test_search_all(self, search_mock, post_mock): + search_mock.assert_not_called() + post_mock.assert_not_called() + + first_page_results = [Mock(pk=i) for i in range(3)] + second_page_results = [Mock(pk=i) for i in range(3, 5)] + search_mock.side_effect = [ + first_page_results, + second_page_results, + urllib2.HTTPError(*[None] * 5), + ] + + res = self.api_client.search_all(term='title', status='New') + + self.assertEqual(res, first_page_results + second_page_results) + + search_mock.assert_has_calls([ + call(term='title', page=1, status='New'), + call(term='title', page=2, status='New'), + call(term='title', page=3, status='New'), # this will throw a urllib2.HTTPError + ]) + post_mock.assert_not_called() + + @patch('codebase.client.logger') + def test_search_all_throws_unexpected_error( + self, logger_mock, search_mock, post_mock + ): + search_mock.assert_not_called() + post_mock.assert_not_called() + + first_page_results = [Mock(pk=i) for i in range(3)] + error = ValueError() + search_mock.side_effect = [first_page_results, error] + + res = self.api_client.search_all(term='title', status='New') + + self.assertEqual(res, first_page_results) + + logger_mock.error.assert_called_once_with( + u'An error occured while searching for "%s" ' + u'(current page: %s):\n%s', 'title', 2, error + ) + search_mock.assert_has_calls([ + call(term='title', page=1, status='New'), + call(term='title', page=2, status='New'), # this will throw a ValueError + ]) + post_mock.assert_not_called() diff --git a/tests/unit/settings_tests.py b/tests/unit/settings_tests.py new file mode 100644 index 0000000..1a3f03f --- /dev/null +++ b/tests/unit/settings_tests.py @@ -0,0 +1,222 @@ +from unittest import TestCase +import errno +import json +import os + +from mock import call, patch + +from codebase.settings import Settings, EXAMPLE_FORMAT +from ..decorators import patch_open + + +class SettingsTestCase(TestCase): + + def setUp(self): + super(SettingsTestCase, self).setUp() + + self.home = os.path.expanduser('~') + self.default_file_path = os.path.join(self.home, '.codebase') + + def test_init_credentials(self): + settings = Settings() + self.assertIsNone(settings.username) + self.assertIsNone(settings.apikey) + + def test_file_path_default(self): + settings = Settings() + self.assertEqual(settings.file_path, self.default_file_path) + + def test_file_path_custom_relative_path(self): + settings = Settings(file_path='./custom.txt') + self.assertEqual(settings.file_path, './custom.txt') + + def test_file_path_custom_abs_path(self): + settings = Settings(file_path='/home/user/custom.txt') + self.assertEqual(settings.file_path, '/home/user/custom.txt') + + def test_file_path_custom_using_expand_user(self): + settings = Settings(file_path='~/custom.txt') + self.assertEqual( + settings.file_path, + os.path.join(self.home, 'custom.txt') + ) + + @patch('codebase.settings.logger') + @patch('codebase.settings.os.path.expanduser') + def test_file_path_default_no_expanduser(self, expanduser_fake, logger_mock): + # If expanduser is not available (Google AppEngine), the settings file + # path is set to the current working directory. + expanduser_fake.side_effect = ImportError + settings = Settings() + self.assertEqual( + settings.file_path, + os.path.join(os.getcwd(), '.codebase') + ) + self.assertNotEqual( + settings.file_path, + self.default_file_path + ) + self.assertEqual(logger_mock.warning.call_count, 2) + + @patch('codebase.settings.logger') + @patch('codebase.settings.os.path.expanduser') + def test_file_path_custom_no_expanduser( + self, expanduser_fake, logger_mock + ): + # If expanduser is not available (Google AppEngine), the settings file + # path is set to the current working directory. + expanduser_fake.side_effect = ImportError + settings = Settings(file_path='~/.custom.txt') + self.assertEqual( + settings.file_path, + os.path.join(os.getcwd(), '.custom.txt') + ) + self.assertNotEqual( + settings.file_path, + os.path.join(self.home, '.custom.txt') + ) + self.assertEqual(logger_mock.warning.call_count, 2) + + @patch_open( + json.dumps({'CODEBASE_USERNAME': 'foo', 'CODEBASE_APIKEY': 'bar'}) + ) + def test_import_settings(self, open_mock): + settings = Settings() + self.assertIsNone(settings.username) + self.assertIsNone(settings.apikey) + + settings.import_settings() + + self.assertEqual(settings.username, 'foo') + self.assertEqual(settings.apikey, 'bar') + open_mock.assert_called_with(self.default_file_path, 'r') + + @patch_open(json.dumps( + {'CODEBASE_USERNAME': 'foo', 'CODEBASE_APIKEY': 'bar', 'other': 'attr'} + )) + def test_import_settings_additional_attribute(self, open_mock): + settings = Settings() + self.assertIsNone(settings.username) + self.assertIsNone(settings.apikey) + self.assertFalse(hasattr(settings, 'other')) + + settings.import_settings() + + self.assertEqual(settings.username, 'foo') + self.assertEqual(settings.apikey, 'bar') + self.assertEqual(settings.other, 'attr') + open_mock.assert_called_with(self.default_file_path, 'r') + + @patch_open('this is not a valid json data') + def test_import_settings_invalid_json(self, open_mock): + settings = Settings() + self.assertIsNone(settings.username) + self.assertIsNone(settings.apikey) + + with self.assertRaises(ValueError) as err: + settings.import_settings() + + self.assertEqual( + err.exception.message, + 'No JSON object could be decoded' + ) + self.assertIsNone(settings.username) + self.assertIsNone(settings.apikey) + open_mock.assert_called_with(self.default_file_path, 'r') + + def test_import_settings_missing_required_attributes(self): + settings = Settings() + + # CODEBASE_USERNAME is required. + no_username = json.dumps({'CODEBASE_APIKEY': 'bar'}) + with patch_open(no_username) as open_mock: + with self.assertRaises(AssertionError) as err: + settings.import_settings() + + self.assertEqual( + err.exception.message, + '"CODEBASE_USERNAME" is required' + ) + open_mock.assert_called_with(self.default_file_path, 'r') + + # CODEBASE_APIKEY is required. + no_api_key = json.dumps({'CODEBASE_USERNAME': 'bar'}) + with patch_open(no_api_key) as open_mock: + with self.assertRaises(AssertionError) as err: + settings.import_settings() + + self.assertEqual( + err.exception.message, + '"CODEBASE_APIKEY" is required' + ) + open_mock.assert_called_with(self.default_file_path, 'r') + + # None of the required fields is in the json file. + with patch_open('{}') as open_mock: + with self.assertRaises(AssertionError) as err: + settings.import_settings() + + self.assertEqual( + err.exception.message, + '"CODEBASE_USERNAME" is required' + ) + open_mock.assert_called_with(self.default_file_path, 'r') + + @patch('codebase.settings.logger') + @patch_open( + json.dumps({'CODEBASE_USERNAME': 'foo', 'CODEBASE_APIKEY': 'bar'}), + ) + def test_import_settings_io_error(self, open_mock, logger_mock): + open_mock.side_effect = IOError + + settings = Settings() + self.assertIsNone(settings.username) + self.assertIsNone(settings.apikey) + + settings.import_settings() + + self.assertIsNone(settings.username) + self.assertIsNone(settings.apikey) + open_mock.assert_called_with(self.default_file_path, 'r') + logger_mock.error.assert_called_once_with( + u'Settings file "{}" not found. ' + u'Please add your settings in JSON format e.g. {}'.format( + settings.file_path, + EXAMPLE_FORMAT, + ) + ) + + @patch('codebase.settings.logger') + @patch_open( + json.dumps({'CODEBASE_USERNAME': 'foo', 'CODEBASE_APIKEY': 'bar'}), + ) + def test_import_settings_io_error_no_permission(self, open_mock, logger_mock): + err = IOError() + err.errno = errno.EACCES + open_mock.side_effect = err + + settings = Settings() + self.assertIsNone(settings.username) + self.assertIsNone(settings.apikey) + + settings.import_settings() + + expected_calls = [ + call( + u'Settings file "{}" not found. ' + u'Please add your settings in JSON format e.g. {}'.format( + settings.file_path, + EXAMPLE_FORMAT, + ) + ), + call( + 'The problem could be caused by the name of the file. ' + 'Try to use a non-hidden file name (i.e. a file which ' + 'does not start with a dot).' + ), + ] + + self.assertIsNone(settings.username) + self.assertIsNone(settings.apikey) + open_mock.assert_called_with(self.default_file_path, 'r') + logger_mock.error.assert_has_calls(expected_calls) diff --git a/tests/unit/utils_tests.py b/tests/unit/utils_tests.py new file mode 100644 index 0000000..5022618 --- /dev/null +++ b/tests/unit/utils_tests.py @@ -0,0 +1,98 @@ +from unittest import TestCase + +from mock import call, patch + +from codebase.utils import CodeBaseAPIUtils + + +@patch('codebase.client.CodeBaseAPI.post') +@patch('codebase.client.CodeBaseAPI.get') +@patch('codebase.client.CodeBaseAPI.search_all') +class CodeBaseAPIUtilsTestCase(TestCase): + + def setUp(self): + super(CodeBaseAPIUtilsTestCase, self).setUp() + + self.api_client = CodeBaseAPIUtils( + project='project', + username='some/body', + apikey='bees', + ) + + self.statuses = [ + {'ticketing_status': {'name': 'New', 'id': 1}}, + {'ticketing_status': {'name': 'Code Complete', 'id': 2}}, + {'ticketing_status': {'name': 'In Review', 'id': 3}}, + {'ticketing_status': {'name': 'Completed', 'id': 4}}, + {'ticketing_status': {'name': 'Invalid', 'id': 5}}, + ] + self.tickets_found = [ + {'ticket': {'ticket_id': 12, 'summary': 'ticket 12'}}, + {'ticket': {'ticket_id': 34, 'summary': 'ticket 34'}}, + {'ticket': {'ticket_id': 56, 'summary': 'ticket 56'}}, + {'ticket': {'ticket_id': 78, 'summary': 'ticket 78'}}, + {'ticket': {'ticket_id': 90, 'summary': 'ticket 90'}}, + ] + + def test_bulk_update_ticket_statuses(self, search_all_mock, get_mock, post_mock): + search_all_mock.return_value = self.tickets_found + get_mock.return_value = self.statuses + + # Bulk update the 5 tickets from 'New' to 'Completed'. + bulk_res = self.api_client.bulk_update_ticket_statuses('New', 'Completed') + self.assertEqual( + bulk_res, + [item['ticket']['ticket_id'] for item in self.tickets_found], + ) + + # Makes sure the first call retrieves the statuses, while the second + # one searches for the 5 tickets. + search_all_mock.assert_called_once_with(status='New') + get_mock.assert_called_once_with( + '/{}/tickets/statuses'.format(self.api_client.project) + ) + # Makes sure the 5 tickets are updated, setting the status to + # 'Completed' (status id = 4). + post_mock.assert_has_calls([ + call( + '/project/tickets/{}/notes'.format(item['ticket']['ticket_id']), + {'ticket_note': {'changes': {'status_id': u'4'}}} + ) + for item in self.tickets_found + ]) + + @patch('codebase.utils.logger') + def test_bulk_update_ticket_statuses_no_status_found( + self, logger_mock, search_all_mock, get_mock, post_mock + ): + search_all_mock.return_value = self.tickets_found + get_mock.return_value = self.statuses + + # Trying to bulk update the 5 tickets, but using a non-existing status. + bulk_res = self.api_client.bulk_update_ticket_statuses( + 'New', + 'Not a valid status', + ) + self.assertIsNone(bulk_res) + + # Makes sure the first call retrieves the statuses, and there will + # not be more calls because the status doesn't exit. + get_mock.assert_called_once_with( + '/{}/tickets/statuses'.format(self.api_client.project) + ) + search_all_mock.assert_not_called() + # No calling to the post because the status doesn't exist. + post_mock.assert_not_called() + + # Logger. + available_statuses = ', '.join([ + 'New', + 'Code Complete', + 'In Review', + 'Completed', + 'Invalid', + ]) + logger_mock.info.assert_called_once_with( + u'Status "Not a valid status" not found in project statuses. ' + u'Options are: {}'.format(available_statuses) + )