diff --git a/src/attributecode/api.py b/src/attributecode/api.py index b97c8193..32a7dea3 100644 --- a/src/attributecode/api.py +++ b/src/attributecode/api.py @@ -18,25 +18,15 @@ from __future__ import print_function from __future__ import unicode_literals -from collections import namedtuple import json -try: - from urllib import urlencode - from urllib import quote -except ImportError: - from urllib.parse import urlencode - from urllib.parse import quote - -try: - import httplib # Python 2 -except ImportError: - import http.client as httplib # Python 3 - -try: - import urllib2 # Python 2 -except ImportError: - import urllib as urllib2 # Python 3 +try: # Python 2 + from urllib import urlencode, quote + from urllib2 import urlopen, Request, HTTPError, URLError +except ImportError: # Python 3 + from urllib.parse import urlencode, quote + from urllib.request import urlopen, Request + from urllib.error import HTTPError, URLError from attributecode import ERROR from attributecode import Error @@ -46,80 +36,6 @@ API call helpers """ -def build_api_url(url, api_username, api_key, license_key): - """ - Return a URl suitable for making an API call. - """ - url = url.rstrip('/') - - payload = {'username': api_username, - 'api_key': api_key, - 'format': 'json'} - encoded_payload = urlencode(payload) - - api_url = '%(url)s/%(license_key)s/?%(encoded_payload)s' % locals() - - # handle special characters in URL such as space etc. - api_url = quote(api_url, safe="%/:=&?~#+!$,;'@()*[]") - return api_url - - -def get_license_data(self, url, api_username, api_key, license_key): - """ - Return a list of errors and a dictionary of license data, given a DejaCode - API url and a DejaCode license_key, send an API request to get license - data for the license_key, authenticating through an api_key and username. - """ - full_url = build_api_url(url, api_username, api_key, license_key) - - errors = [] - license_data = {} - - msg = 'Failed to collect license data for %(license_key)s.' - - try: - request = urllib2.Request(full_url) - response = urllib2.urlopen(request) - response_content = response.read() - license_data = json.loads(response_content) - - except urllib2.HTTPError as e: - if e.code == httplib.UNAUTHORIZED: - msg = msg + ('Authorization denied: ' - 'Invalid api_username: %(api_username)s ' - 'or api_key: %(api_key)s.') - # FIXME: what about 404 and other cases? - errors.append(Error(ERROR, msg % locals())) - - except urllib2.URLError as e: - msg = msg + 'Network problem. Check your internet connection.' - errors.append(Error(ERROR, msg % locals())) - - except Exception as e: - # only keep the first 100 char of the exception - emsg = repr(e)[:100] - msg = msg + ' Error: %(emsg)s' - errors.append(Error(ERROR, msg % locals())) - - return errors, license_data - - -LicenseInfo = namedtuple('LicenseInfo', ['key', 'name', 'text']) - - -def get_license_info(self, url, api_username, api_key, license_key, - caller=get_license_data): - """ - Return a list of errors and a tuple of key, name, text for a given - license_key using a DejaCode API request at url. caller is the function - used to call the API and is used mostly for mocking in tests. - """ - errors, data = get_license_data(url, api_username, api_key, license_key) - key = data.get('key') - name = data.get('name') - text = data.get('full_text') - return errors, LicenseInfo(key, name, text) - def request_license_data(url, api_key, license_key): """ @@ -127,6 +43,9 @@ def request_license_data(url, api_key, license_key): Send a request to a given API URL to gather license data for license_key, authenticating through an api_key. """ + headers = { + 'Authorization': 'Token %s' % api_key, + } payload = { 'api_key': api_key, 'key': license_key, @@ -137,40 +56,42 @@ def request_license_data(url, api_key, license_key): encoded_payload = urlencode(payload) full_url = '%(url)s/?%(encoded_payload)s' % locals() # handle special characters in URL such as space etc. - full_url = quote(full_url, safe="%/:=&?~#+!$,;'@()*[]") - headers = {'Authorization': 'Token %s' % api_key} + quoted_url = quote(full_url, safe="%/:=&?~#+!$,;'@()*[]") + license_data = {} errors = [] try: - request = urllib2.Request(full_url, headers=headers) - response = urllib2.urlopen(request) - response_content = response.read() + request = Request(quoted_url, headers=headers) + response = urlopen(request) + response_content = response.read().decode('utf-8') license_data = json.loads(response_content) if not license_data['results']: - msg = (u"Invalid 'license': " + license_key) + msg = u"Invalid 'license': %s" % license_key errors.append(Error(ERROR, msg)) - except urllib2.HTTPError as http_e: + except HTTPError as http_e: # some auth problem if http_e.code == 403: - msg = (u"Authorization denied. Invalid '--api_key'. License generation is skipped.") + msg = (u"Authorization denied. Invalid '--api_key'. " + u"License generation is skipped.") errors.append(Error(ERROR, msg)) else: # Since no api_url/api_key/network status have # problem detected, it yields 'license' is the cause of # this exception. - msg = (u"Invalid 'license': " + license_key) + msg = u"Invalid 'license': %s" % license_key errors.append(Error(ERROR, msg)) except Exception as e: errors.append(Error(ERROR, str(e))) finally: license_data = license_data.get('results')[0] if license_data.get('count') == 1 else {} + return license_data, errors def get_license_details_from_api(url, api_key, license_key): """ - Returns the license_text of a given license_key using an API request. - Returns an empty string if the text is not available. + Return the license_text of a given license_key using an API request. + Return an empty string if the text is not available. """ license_data, errors = request_license_data(url, api_key, license_key) license_name = license_data.get('name', '') diff --git a/src/attributecode/cmd.py b/src/attributecode/cmd.py index a829a179..41095ed6 100644 --- a/src/attributecode/cmd.py +++ b/src/attributecode/cmd.py @@ -18,7 +18,6 @@ from __future__ import print_function from __future__ import unicode_literals -import codecs import logging import os from os.path import exists, join @@ -28,19 +27,11 @@ # silence unicode literals warnings click.disable_unicode_literals_warning = True -import attributecode -from attributecode import CRITICAL -from attributecode import ERROR -from attributecode import INFO -from attributecode import NOTSET -from attributecode import WARNING from attributecode import __about_spec_version__ from attributecode import __version__ -from attributecode import attrib -from attributecode import Error -from attributecode import gen +from attributecode.attrib import generate_and_save as attrib_generate_and_save +from attributecode.gen import generate as gen_generate from attributecode import model -from attributecode.model import About from attributecode import severities from attributecode.util import extract_zip from attributecode.util import to_posix @@ -147,7 +138,7 @@ def inventory(location, output, quiet, format): # accept zipped ABOUT files as input location = extract_zip(location) - errors, abouts = attributecode.model.collect_inventory(location) + errors, abouts = model.collect_inventory(location) write_errors = model.write_output(abouts, output, format) for err in write_errors: @@ -209,7 +200,7 @@ def gen(location, output, mapping, license_notice_text_location, fetch_license, click.echo('Generating .ABOUT files...') - errors, abouts = attributecode.gen.generate( + errors, abouts = gen_generate( location=location, base_dir=output, use_mapping=mapping, license_notice_text_location=license_notice_text_location, fetch_license=fetch_license) @@ -274,7 +265,7 @@ def attrib(location, output, template, mapping, inventory, quiet): location = extract_zip(location) inv_errors, abouts = model.collect_inventory(location) - no_match_errors = attributecode.attrib.generate_and_save( + no_match_errors = attrib_generate_and_save( abouts=abouts, output_location=output, use_mapping=mapping, template_loc=template, inventory_location=inventory) @@ -316,7 +307,7 @@ def check(location, show_all): click.echo('Running aboutcode-toolkit version ' + __version__) click.echo('Checking ABOUT files...') - errors, abouts = attributecode.model.collect_inventory(location) + errors, abouts = model.collect_inventory(location) msg_format = '%(sever)s: %(message)s' print_errors = [] diff --git a/src/attributecode/model.py b/src/attributecode/model.py index d29395c4..f9475e82 100644 --- a/src/attributecode/model.py +++ b/src/attributecode/model.py @@ -37,30 +37,19 @@ import re import sys -if sys.version_info[0] < 3: - # Python 2 +if sys.version_info[0] < 3: # Python 2 import backports.csv as csv -else: - # Python 3 + from urlparse import urljoin, urlparse + from urllib2 import urlopen, Request, HTTPError +else: # Python 3 + basestring = str import csv + from urllib.parse import urljoin, urlparse + from urllib.request import urlopen, Request + from urllib.error import HTTPError from license_expression import Licensing -try: - import urllib2 # Python 2 -except ImportError: - import urllib as urllib2 # Python 3 - -try: - from urlparse import urljoin, urlparse # Python 2 -except ImportError: - from urllib.parse import urljoin, urlparse # Python 3 - -try: - basestring # Python 2 -except NameError: - basestring = str # Python 3 - from attributecode import CRITICAL from attributecode import ERROR from attributecode import INFO @@ -81,7 +70,6 @@ class Field(object): An ABOUT file field. The initial value is a string. Subclasses can and will alter the value type as needed. """ - def __init__(self, name=None, value=None, required=False, present=False): # normalized names are lowercased per specification self.name = name @@ -228,7 +216,6 @@ class StringField(Field): A field containing a string value possibly on multiple lines. The validated value is a string. """ - def _validate(self, *args, **kwargs): errors = super(StringField, self)._validate(*args, ** kwargs) return errors @@ -640,7 +627,12 @@ def validate_fields(fields, about_file_path, running_inventory, base_dir, """ errors = [] for f in fields: - val_err = f.validate(base_dir=base_dir, about_file_path=about_file_path, running_inventory=running_inventory, license_notice_text_location=license_notice_text_location) + val_err = f.validate( + base_dir=base_dir, + about_file_path=about_file_path, + running_inventory=running_inventory, + license_notice_text_location=license_notice_text_location, + ) errors.extend(val_err) return errors @@ -666,7 +658,6 @@ def create_fields(self): Create fields in an ordered mapping to keep a standard ordering. We could use a metaclass to track ordering django-like but this approach is simpler. - """ self.fields = OrderedDict([ ('about_resource', ListField(required=True)), @@ -746,7 +737,8 @@ def __eq__(self, other): and self.fields == other.fields and self.custom_fields == other.custom_fields) - def attribution_fields(self, fields): + @staticmethod + def attribution_fields(fields): """ Return attrib-only fields """ @@ -767,16 +759,15 @@ def attribution_fields(self, fields): 'owner', 'author'] - return OrderedDict([(n, o,) for n, o in fields.items() - if n in attrib_fields]) + return OrderedDict([(n, o) for n, o in fields.items() + if n in attrib_fields]) def same_attribution(self, other): """ Equality based on attribution-related fields. """ return (isinstance(other, self.__class__) - and self.attribution_fields(self.fields) - == self.attribution_fields(other.fields)) + and self.attribution_fields(self.fields) == self.attribution_fields(other.fields)) def resolved_resources_paths(self): """ @@ -1091,7 +1082,6 @@ def dump(self, location, with_absent=False, with_empty=True): errors.append(msg) return errors - def dump_lic(self, location, license_dict): """ Write LICENSE files and return the a list of key, name, context and the url @@ -1124,6 +1114,7 @@ def dump_lic(self, location, license_dict): pass return license_key_name_context_url + # valid field name field_name = r'(?P[a-z][0-9a-z_]*)' @@ -1406,13 +1397,6 @@ def by_license_content(abouts): return OrderedDict(sorted(grouped.items())) -def common_licenses(abouts): - """ - Return a ordered dictionary of repeated licenses sorted by key and update - the list of about objects with license references for repeated licenses. - """ - pass - def pre_process_and_fetch_license_dict(abouts, api_url, api_key): """ Modify a list of About data dictionaries by adding license information @@ -1432,7 +1416,7 @@ def pre_process_and_fetch_license_dict(abouts, api_url, api_key): msg = u'Network problem. Please check your Internet connection. License generation is skipped.' errors.append(Error(ERROR, msg)) for about in abouts: - # No need to go thru all the about objects for license extraction if we detected + # No need to go through all the about objects for license extraction if we detected # invalid '--api_key' auth_error = Error(ERROR, u"Authorization denied. Invalid '--api_key'. License generation is skipped.") if auth_error in errors: @@ -1460,6 +1444,7 @@ def pre_process_and_fetch_license_dict(abouts, api_url, api_key): key_text_dict[license_key] = detail_list return key_text_dict, errors + def parse_license_expression(lic_expression): licensing = Licensing() lic_list = [] @@ -1469,6 +1454,7 @@ def parse_license_expression(lic_expression): lic_list = licensing.license_keys(lic_expression) return special_char, lic_list + def special_char_in_license_expresion(lic_expression): not_support_char = [ '!', '@', '#', '$', '%', '^', '&', '*', '(', ')', @@ -1480,13 +1466,14 @@ def special_char_in_license_expresion(lic_expression): special_character.append(char) return special_character + def valid_api_url(api_url): try: - request = urllib2.Request(api_url) + request = Request(api_url) # This will always goes to exception as no key are provided. # The purpose of this code is to validate the provided api_url is correct - response = urllib2.urlopen(request) - except urllib2.HTTPError as http_e: + urlopen(request) + except HTTPError as http_e: # The 403 error code is refer to "Authentication credentials were not provided.". # This is correct as no key are provided. if http_e.code == 403: diff --git a/tests/test_api.py b/tests/test_api.py index 42f8936e..502170b3 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -20,32 +20,41 @@ import unittest -from mock import patch +import mock -import attributecode from attributecode import api -from attributecode.api import LicenseInfo +from attributecode import ERROR +from attributecode import Error +from testing_utils import FakeResponse class ApiTest(unittest.TestCase): - - def test_build_api_url(self): - url = 'http:/dejacode.org/' - api_username = 'phi' - api_key = 'ABCD' - license_key = 'apache' - expected = 'http:/dejacode.org/apache/?username=phi&api_key=ABCD&format=json' - result = api.build_api_url(url, api_username, api_key, license_key) + @mock.patch.object(api, 'request_license_data') + def test_api_get_license_details_from_api(self, request_license_data): + license_data = { + 'name': 'Apache License 2.0', + 'full_text': 'Apache License Version 2.0 ...', + 'key': 'apache-2.0', + } + errors = [] + request_license_data.return_value = license_data, errors + + expected = ('Apache License 2.0', 'apache-2.0', 'Apache License Version 2.0 ...', []) + result = api.get_license_details_from_api('url', 'api_key', 'license_key') assert expected == result - @patch.object(attributecode.api, 'get_license_data') - def test_get_license_info(self, mock_data): - mock_data.return_value = [], {'key': 'test', 'name': 'test_name', 'full_text': 'test_full_text' } - result = api.get_license_info(self, '', '', '', '') - assert result == ([], LicenseInfo(key='test', name='test_name', text='test_full_text')) - - @patch.object(attributecode.api, 'request_license_data') - def test_get_license_details_from_api(self, mock_data): - mock_data.return_value = {'name': 'test_name', 'full_text': 'test_full_text', 'key': 'test'}, [] - result = api.get_license_details_from_api('', '', '') - assert result == ('test_name', 'test', 'test_full_text', []) \ No newline at end of file + @mock.patch.object(api, 'urlopen') + def test_api_request_license_data(self, mock_data): + response_content = ( + b'{"count":1,"results":[{"name":"Apache 2.0","key":"apache-2.0","text":"Text"}]}' + ) + mock_data.return_value = FakeResponse(response_content) + license_data = api.request_license_data('http://fake.url/', 'api_key', 'apache-2.0') + expected = ({'name': 'Apache 2.0', 'key': 'apache-2.0', 'text': 'Text'}, []) + assert expected == license_data + + response_content = b'{"count":0,"results":[]}' + mock_data.return_value = FakeResponse(response_content) + license_data = api.request_license_data('http://fake.url/', 'api_key', 'apache-2.0') + expected = ({}, [Error(ERROR, "Invalid 'license': apache-2.0")]) + assert expected == license_data diff --git a/tests/test_model.py b/tests/test_model.py index 61561f0c..d4db513e 100644 --- a/tests/test_model.py +++ b/tests/test_model.py @@ -25,11 +25,7 @@ import unittest from unittest.case import expectedFailure -from testing_utils import extract_test_loc -from testing_utils import get_temp_file -from testing_utils import get_test_loc -from testing_utils import get_test_lines -from testing_utils import get_unicode_content +import mock import attributecode from attributecode import CRITICAL @@ -41,6 +37,11 @@ from attributecode import util from attributecode.util import add_unc from attributecode.util import load_csv +from testing_utils import extract_test_loc +from testing_utils import get_temp_file +from testing_utils import get_test_loc +from testing_utils import get_test_lines +from testing_utils import get_unicode_content def check_csv(expected, result): @@ -1313,3 +1314,25 @@ def test_by_name(self): ('eclipse', [d]), ]) assert expected == results + + +class FetchLicenseTest(unittest.TestCase): + @mock.patch.object(model, 'urlopen') + def test_valid_api_url(self, mock_data): + mock_data.return_value = '' + assert model.valid_api_url('non_valid_url') is False + + @mock.patch('attributecode.util.have_network_connection') + @mock.patch('attributecode.model.valid_api_url') + def test_pre_process_and_fetch_license_dict(self, have_network_connection, valid_api_url): + have_network_connection.return_value = True + + valid_api_url.return_value = False + error_msg = ('Network problem. Please check your Internet connection. ' + 'License generation is skipped.') + expected = ({}, [Error(ERROR, error_msg)]) + assert model.pre_process_and_fetch_license_dict([], '', '') == expected + + valid_api_url.return_value = True + expected = ({}, []) + assert model.pre_process_and_fetch_license_dict([], '', '') == expected diff --git a/tests/testing_utils.py b/tests/testing_utils.py index 6c77651d..ff60e16b 100644 --- a/tests/testing_utils.py +++ b/tests/testing_utils.py @@ -155,4 +155,14 @@ def extract_test_loc(path, extract_func=extract_zip): else: target_dir = get_temp_dir() extract_func(archive, target_dir) - return target_dir \ No newline at end of file + return target_dir + + +class FakeResponse(object): + response_content = None + + def __init__(self, response_content): + self.response_content = response_content + + def read(self): + return self.response_content