Fix verification tool for later Py/Dj versions and add tests

hawkrest/management/commands/hawkrequest.py

@@ -1,47 +1,69 @@
-from optparse import make_option
-
 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 request(url, method, data, headers):
+    import requests
+    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 handle(self, *args, **options):
-        hawk_log = logging.getLogger('mohawk')
-        hawk_log.setLevel(logging.DEBUG)
-        hawk_log.addHandler(logging.StreamHandler())
+    def add_arguments(self, parser):
+        for opt, config in CMD_OPTIONS.items():
+            parser.add_argument(opt, **config)
 
-        try:
-            import requests
-        except ImportError:
-            raise CommandError('To use this command you first need to '
-                               'install the requests module')
+    def handle(self, *args, **options):
         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 +73,25 @@ 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)
+        try:
+            res = request(url, method.lower(), data=qs, headers=headers)
+        except ImportError:
+            raise CommandError('To use this command you first need to '
+                               'install the requests module')
 
-        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,60 @@
+import mock
+
+from django.core.management.base import CommandError
+from django.core.management import call_command
+
+from tests.base import BaseTest
+
+
+def exec_cmd(**kwargs):
+    call_command('hawkrequest', **kwargs)
+
+
+class UnauthorizedResponse:
+    def __init__(self):
+        self.headers = {}
+        self.text = 'Unauthorized'
+
+
+class AuthorizedResponse:
+    def __init__(self):
+        self.headers = {
+            'Server-Authorization': 'xyz',
+            'Content-Type': 'text/plain'
+        }
+        self.text = 'Authorized'
+
+
+cmd_request = 'hawkrest.management.commands.hawkrequest.request'
+
+
+class TestManagementCommand(BaseTest):
+
+    @mock.patch(cmd_request, mock.Mock(side_effect=ImportError))
+    def test_error_raised_if_requests_not_imported(self):
+        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(cmd_request, mock.Mock(return_value=UnauthorizedResponse()))
+    @mock.patch('mohawk.Sender.accept_response')
+    def test_response_unverified_without_auth_header(self, mock_mohawk):
+        exec_cmd(url=self.url, creds=self.credentials_id)
+        self.assertFalse(mock_mohawk.called)
+
+    @mock.patch(cmd_request, mock.Mock(return_value=AuthorizedResponse()))
+    @mock.patch('mohawk.Sender.accept_response')
+    def test_response_verified_with_auth_header(self, mock_mohawk):
+        exec_cmd(url=self.url, creds=self.credentials_id)
+        self.assertTrue(mock_mohawk.called)