Test to demonstrate failure discussed in #4.

pmr2/oauth/tests/base.py

@@ -69,7 +69,7 @@ class TestRequest(pmr2.z3cform.tests.base.TestRequest):
 
     zope.interface.implements(IOAuthTestLayer, IAttributeAnnotatable)
 
-    def __init__(self, oauth_keys=None, url=None, *a, **kw):
+    def __init__(self, oauth_keys=None, url=None, method=None, *a, **kw):
         super(TestRequest, self).__init__(*a, **kw)
         if url:
             parts = url.split('/')
@@ -87,6 +87,8 @@ def __init__(self, oauth_keys=None, url=None, *a, **kw):
             self._auth = self.to_header(oauth_keys)
 
         self.stdin = StringIO()
+        if method:
+            self.method = method
 
     def to_header(self, oauth_keys, realm=''):
         # copied from oauth2 (for now)
@@ -105,7 +107,7 @@ def to_header(self, oauth_keys, realm=''):
 
 def SignedTestRequest(form=None, consumer=None, token=None, method=None,
         url=None, callback=None, timestamp=None, verifier=None,
-        signature_type='AUTH_HEADER',
+        signature_type='AUTH_HEADER', raw_body=None,
         *a, **kw):
     """\
     Creates a signed TestRequest
@@ -125,7 +127,10 @@ def safe_unicode(s):
     if form is None:
         form = {}
 
-    result = TestRequest(form=form, url=url, *a, **kw)
+    result = TestRequest(form=form, url=url, method=method, *a, **kw)
+    if raw_body:
+        result.stdin.write(raw_body)
+
     url = url or result.getURL()
     url = safe_unicode(url)
     method = method and safe_unicode(method) or safe_unicode(result.method)
@@ -145,15 +150,22 @@ def safe_unicode(s):
         signature_type=signature_type,
     )
 
-    url_signed, headers, body = client.sign(url, method)
+    if result.getHeader('Content-Type') == 'application/x-www-form-urlencoded':
+        headers = {'Content-Type': 'application/x-www-form-urlencoded'}
+        url_signed, headers, body = client.sign(url, method,
+            body=raw_body, headers=headers)
+    else:
+        url_signed, headers, body = client.sign(url, method)
 
     # lazy not importing oauthlib tokens.
     if signature_type == 'AUTH_HEADER':
         result._auth = headers['Authorization']
         return result
     elif signature_type == 'QUERY':
         qs = urlparse.urlsplit(url_signed).query
-        result = TestRequest(form=form, url=url, QUERY_STRING=qs)
+        result = TestRequest(form=form, url=url, QUERY_STRING=qs, *a, **kw)
+        if raw_body:
+            result.stdin.write(raw_body)
         return result
 
 def makeToken(qsstr):

pmr2/oauth/tests/test_extraction.py

@@ -257,6 +257,16 @@ def test_1001_success_access_token_3_5_3(self):
         credentials = plugin.extractCredentials(request)
         self.assertEqual(credentials['userid'], self.default_user_id)
 
+    def test_1050_success_with_www_form_body(self):
+        # use request token
+        plugin = self.plugin
+        consumer, token = self.save_consumer_and_token()
+        request = SignedTestRequest(consumer=consumer, token=token,
+            method='POST', raw_body='title=test&value=1',
+            CONTENT_TYPE='application/x-www-form-urlencoded')
+        credentials = plugin.extractCredentials(request)
+        self.assertEqual(credentials['userid'], self.default_user_id)
+
     def test_1100_missing_token_ignored(self):
         # Should not forbid cases where the oauth_token is missing (it
         # could be a RequestToken, let that page handle it).