Fix verification tool for later Python/Django versions and add tests (#…

…41)

hawkrest/management/commands/hawkrequest.py

@@ -1,47 +1,87 @@
-from optparse import make_option
+import logging
 
 from django.core.management.base import BaseCommand, CommandError
-
-import logging
 from mohawk import Sender
 
-from hawkrest import lookup_credentials
+from hawkrest import HawkAuthentication
+
+
+DEFAULT_HTTP_METHOD = 'GET'
+
+CMD_OPTIONS = {
+    '--url': {
+        'action': 'store',
+        'type': str,
+        'help': 'Absolute URL to request.'
+    },
+    '--creds': {
+        'action': 'store',
+        'type': str,
+        'help': 'ID for Hawk credentials.'
+    },
+    '-X': {
+        'action': 'store',
+        'type': str,
+        'help': 'Request method. Default: {}.'.format(DEFAULT_HTTP_METHOD),
+        'default': DEFAULT_HTTP_METHOD
+    },
+    '-d': {
+        'action': 'store',
+        'type': str,
+        'help': 'Query string parameters.'
+    }
+}
+
+
+def get_requests_module():
+    import requests
+    return requests
+
+
+def request(url, method, data, headers):
+    try:
+        requests = get_requests_module()
+    except ImportError:
+        raise CommandError('To use this command you first need to '
+                           'install the requests module')
+
+    do_request = getattr(requests, method.lower())
+    res = do_request(url, data=data, headers=headers)
+    return res
+
+
+def lookup_credentials(creds_key):
+    return HawkAuthentication().hawk_credentials_lookup(creds_key)
 
 
 class Command(BaseCommand):
     help = 'Make a Hawk authenticated request'
-    option_list = BaseCommand.option_list + (
-        make_option('--url', action='store', type=str,
-                    help='Absolute URL to request.'),
-        make_option('--creds', action='store', type=str,
-                    help='ID for Hawk credentials.'),
-        make_option('-X', action='store', type=str,
-                    help='Request method. Default: %default.',
-                    default='GET'),
-        make_option('-d', action='store', type=str,
-                    help='Query string parameters'),
-    )
+
+    def add_arguments(self, parser):
+        for opt, config in CMD_OPTIONS.items():
+            parser.add_argument(opt, **config)
 
     def handle(self, *args, **options):
+        # Configure the mohawk lib for debug logging so we can see inputs to
+        # the signature functions and other useful stuff.
         hawk_log = logging.getLogger('mohawk')
         hawk_log.setLevel(logging.DEBUG)
         hawk_log.addHandler(logging.StreamHandler())
 
-        try:
-            import requests
-        except ImportError:
-            raise CommandError('To use this command you first need to '
-                               'install the requests module')
         url = options['url']
         if not url:
             raise CommandError('Specify a URL to load with --url')
 
+        creds_key = options['creds']
+        if not creds_key:
+            raise CommandError('Specify ID for Hawk credentials with --creds')
+
+        method = options['X']
         qs = options['d'] or ''
         request_content_type = ('application/x-www-form-urlencoded'
                                 if qs else 'text/plain')
-        method = options['X']
 
-        credentials = lookup_credentials(options['creds'])
+        credentials = lookup_credentials(creds_key)
 
         sender = Sender(credentials,
                         url, method.upper(),
@@ -51,22 +91,21 @@ def handle(self, *args, **options):
         headers = {'Authorization': sender.request_header,
                    'Content-Type': request_content_type}
 
-        do_request = getattr(requests, method.lower())
-        res = do_request(url, data=qs, headers=headers)
+        res = request(url, method.lower(), data=qs, headers=headers)
 
-        print '{method} -d {qs} {url}'.format(method=method.upper(),
-                                              qs=qs or 'None',
-                                              url=url)
-        print res.text
+        self.stdout.write('{method} -d {qs} {url}'.format(method=method.upper(),
+                                                          qs=qs or 'None',
+                                                          url=url))
+        self.stdout.write(res.text)
 
         # Verify we're talking to our trusted server.
-        print res.headers
+        self.stdout.write(str(res.headers))
         auth_hdr = res.headers.get('Server-Authorization', None)
         if auth_hdr:
             sender.accept_response(auth_hdr,
                                    content=res.text,
                                    content_type=res.headers['Content-Type'])
-            print '<response was Hawk verified>'
+            self.stdout.write('<response was Hawk verified>')
         else:
-            print '** NO Server-Authorization header **'
-            print '<response was NOT Hawk verified>'
+            self.stdout.write('** NO Server-Authorization header **')
+            self.stdout.write('<response was NOT Hawk verified>')

tests/test_command.py

@@ -0,0 +1,52 @@
+import mock
+
+from django.core.management.base import CommandError
+from django.core.management import call_command
+from requests.models import Response
+
+from tests.base import BaseTest
+
+
+def exec_cmd(**kwargs):
+    call_command('hawkrequest', **kwargs)
+
+
+class TestManagementCommand(BaseTest):
+
+    @mock.patch('hawkrest.management.commands.hawkrequest.get_requests_module')
+    def test_error_raised_if_requests_not_imported(self, mk_import):
+        mk_import.side_effect = ImportError()
+        with self.assertRaises(CommandError):
+            exec_cmd(url=self.url, creds=self.credentials_id)
+
+    def test_error_raised_if_url_not_specified(self):
+        with self.assertRaises(CommandError):
+            exec_cmd(creds=self.credentials_id)
+
+    def test_error_raised_if_creds_missing(self):
+        with self.assertRaises(CommandError):
+            exec_cmd(url=self.url)
+
+    def test_error_raises_if_creds_not_found(self):
+        with self.assertRaises(CommandError):
+            exec_cmd(creds='nonexistent')
+
+    @mock.patch('hawkrest.management.commands.hawkrequest.request')
+    @mock.patch('mohawk.Sender.accept_response')
+    def test_response_unverified_without_auth_header(self, mk_accept, mk_resp):
+        response = Response()
+        response._content = b'Unauthorized'
+        mk_resp.return_value = Response()
+        exec_cmd(url=self.url, creds=self.credentials_id)
+        self.assertFalse(mk_accept.called)
+
+    @mock.patch('hawkrest.management.commands.hawkrequest.request')
+    @mock.patch('mohawk.Sender.accept_response')
+    def test_response_verified_with_auth_header(self, mk_accept, mk_resp):
+        response = Response()
+        response.headers['Server-Authorization'] = 'xyz'
+        response.headers['Content-Type'] = 'text/plain'
+        response._content = b'Authorized'
+        mk_resp.return_value = response
+        exec_cmd(url=self.url, creds=self.credentials_id)
+        self.assertTrue(mk_accept.called)