micropub: convert form-encoded input, remove reserved keywords, refac…

…tor tests

micropub.py

@@ -20,6 +20,14 @@
 
 logger = logging.getLogger(__name__)
 
+RESERVED_PARAMS = ('access_token', 'action', 'q', 'url')
+RESERVED_PREFIX = 'mp-'
+
+
+def remove_reserved(params):
+  return {k: v for k, v in params.items()
+          if k not in RESERVED_PARAMS and not k.startswith(RESERVED_PREFIX)}
+
 
 class Micropub(PublishBase):
   """Micropub endpoint."""
@@ -39,21 +47,25 @@ def dispatch_request(self):
 
     # handle input
     if request.is_json:
-      pass
-    elif request.files:
-      pass
+      mf2 = request.json
     elif request.form:
+      mf2 = {
+        'h': request.values.get('h') or 'entry',
+        'properties': remove_reserved(request.form.to_dict()),
+      }
+    elif request.files:
       pass
     else:
       return self.error(error='invalid_request', extra_json={
         'error_description': f'Unsupported Content-Type {request.content_type}',
       })
 
-    obj = microformats2.json_to_object(request.json)
+    obj = microformats2.json_to_object(mf2)
     logging.debug(f'Converted to ActivityStreams object: {json_dumps(obj, indent=2)}')
 
     # TODO: is this the right idea to require mf2 url so I can de-dupe?
-    url = util.get_url(obj)
+    url = request.values.get('url') or util.get_url(obj)
+    assert url
 
     # done with the sanity checks, create the Publish entity
     self.entity = self.get_or_add_publish_entity(url)

tests/test_micropub.py

@@ -2,6 +2,7 @@
 import html
 
 import micropub
+from models import Publish, PublishedPage
 from . import testutil
 
 
@@ -16,6 +17,34 @@ def setUp(self):
       id='foo.com', features=['publish'], auth_entity=auth_key)
     self.source.put()
 
+  def assert_response(self, status=201, **kwargs):
+    resp = self.client.post('/micropub', **kwargs)
+    body = resp.get_data(as_text=True)
+    self.assertEqual(status, resp.status_code,
+                     f'{status} != {resp.status_code}: {body}')
+    self.assertEqual('http://fake/url', resp.headers['Location'])
+    return resp
+
+  def check_entity(self, url='http://foo', content='foo bar baz',
+                   html_content=None, expected_html=None):
+    # if html_content is None:
+    #   html_content = content
+    self.assertTrue(PublishedPage.get_by_id(url))
+    publish = Publish.query().get()
+    self.assertEqual(self.source.key, publish.source)
+    self.assertEqual('complete', publish.status)
+    self.assertEqual('post', publish.type)
+    self.assertEqual('FakeSource post label', publish.type_label())
+    # if expected_html is None:
+    #   expected_html = (self.post_html % html_content)
+    # self.assertEqual(expected_html + self.backlink, publish.html)
+    self.assertEqual({
+      'id': 'fake id',
+      'url': 'http://fake/url',
+      'content': content,
+      'granary_message': 'granary message',
+    }, publish.published)
+
   def test_query_config(self):
     resp = self.client.get('/micropub?q=config')
     self.assertEqual(200, resp.status_code)
@@ -40,17 +69,13 @@ def test_bad_content_type(self):
 
   # def test_already_published(self):
 
-  # def test_create_form_encoded(self):
-  #   resp = self.client.post('/micropub', data={
-  #     'h': 'entry',
-  #     'content': 'Micropub+test+of+creating+a+basic+h-entry',
-  #   })
-  #   body = html.unescape(resp.get_data(as_text=True))
-  #   self.assertEqual(201, resp.status_code,
-  #                    f'201 != {resp.status_code}: {body}')
-  #   self.assertEqual('xyz', resp.headers['Location'])
-  #
-  #   # TODO: check Publish entity, Fake send
+  def test_create_form_encoded(self):
+    resp = self.assert_response(data={
+      'h': 'entry',
+      'content': 'foo bar baz',
+      'url': 'http://foo',
+    })
+    self.check_entity()
 
   # def test_create_form_encoded_token_param(self):
   #   resp = self.client.post('/micropub', data={
@@ -105,19 +130,14 @@ def test_bad_content_type(self):
 # category=test1
 
   def test_create_json(self):
-    resp = self.client.post('/micropub', json={
+    resp = self.assert_response(json={
       'type': ['h-entry'],
       'properties': {
-        'content': ['Micropub test of creating an h-entry with a JSON request'],
+        'content': ['foo bar baz'],
         'url': ['http://foo'],
       },
     })
-    body = html.unescape(resp.get_data(as_text=True))
-    self.assertEqual(201, resp.status_code,
-                     f'201 != {resp.status_code}: {body}')
-    self.assertEqual('http://fake/url', resp.headers['Location'])
-
-    # TODO: check Publish entity, Fake send
+    self.check_entity()
 
 #   def test_create_json_multiple_categories(self):
 # {