noop: move common.as2_request_type to activitypub

activitypub.py

@@ -1222,7 +1222,7 @@ def actor(handle_or_id):
     user = _load_user(handle_or_id, create=True)
     proto = user
 
-    as2_type = common.as2_request_type()
+    as2_type = as2_request_type()
     if not as2_type:
         return redirect(user.web_url(), code=302)
 
@@ -1668,6 +1668,33 @@ def instance_info():
     }
 
 
+def as2_request_type():
+    """If this request has conneg (ie the ``Accept`` header) for AS2, returns its type.
+
+    Specifically, returns either
+    ``application/ld+json; profile="https://www.w3.org/ns/activitystreams"`` or
+    ``application/activity+json``.
+
+    If the current request's conneg isn't asking for AS2, returns None.
+
+    https://www.w3.org/TR/activitypub/#retrieving-objects
+    https://snarfed.org/2023-03-24_49619-2
+    """
+    if accept := request.headers.get('Accept'):
+        try:
+            negotiated = common._negotiator.negotiate(accept)
+        except ValueError:
+            # work around https://github.com/CottageLabs/negotiator/issues/6
+            negotiated = None
+        if negotiated:
+            accept_type = str(negotiated.content_type)
+            if accept_type == as2.CONTENT_TYPE:
+                return as2.CONTENT_TYPE
+            elif accept_type in (as2.CONTENT_TYPE_LD, as2.CONTENT_TYPE_LD_PROFILE):
+                return as2.CONTENT_TYPE_LD_PROFILE
+            logger.debug(f'Conneg resolved {accept_type} for Accept: {accept}')
+
+
 #
 # OAuth
 #

common.py

@@ -474,33 +474,6 @@ NDB_CONTEXT_KWARGS = {
 }
 
 
-def as2_request_type():
-    """If this request has conneg (ie the ``Accept`` header) for AS2, returns its type.
-
-    Specifically, returns either
-    ``application/ld+json; profile="https://www.w3.org/ns/activitystreams"`` or
-    ``application/activity+json``.
-
-    If the current request's conneg isn't asking for AS2, returns None.
-
-    https://www.w3.org/TR/activitypub/#retrieving-objects
-    https://snarfed.org/2023-03-24_49619-2
-    """
-    if accept := request.headers.get('Accept'):
-        try:
-            negotiated = _negotiator.negotiate(accept)
-        except ValueError:
-            # work around https://github.com/CottageLabs/negotiator/issues/6
-            negotiated = None
-        if negotiated:
-            accept_type = str(negotiated.content_type)
-            if accept_type == as2.CONTENT_TYPE:
-                return as2.CONTENT_TYPE
-            elif accept_type in (as2.CONTENT_TYPE_LD, as2.CONTENT_TYPE_LD_PROFILE):
-                return as2.CONTENT_TYPE_LD_PROFILE
-            logger.debug(f'Conneg resolved {accept_type} for Accept: {accept}')
-
-
 def log_request():
     """Logs GET query params and POST form.
 

redirect.py

@@ -25,9 +25,8 @@ from oauth_dropins.webutil import flask_util, util
 from oauth_dropins.webutil.flask_util import error
 from oauth_dropins.webutil.util import json_dumps, json_loads
 
-from activitypub import ActivityPub
+from activitypub import ActivityPub, as2_request_type
 from common import (
-    as2_request_type,
     CACHE_CONTROL_VARY_ACCEPT,
     CONTENT_TYPE_HTML,
     SUPERDOMAIN,

tests/test_activitypub.py

@@ -3715,3 +3715,18 @@ class ActivityPubUtilsTest(TestCase):
     def test_instance_info(self):
         # just check that it doesn't crash
         self.client.get('/api/v1/instance')
+
+    def test_as2_request_type(self):
+        for accept, expected in (
+                (as2.CONTENT_TYPE_LD_PROFILE, as2.CONTENT_TYPE_LD_PROFILE),
+                (as2.CONTENT_TYPE_LD, as2.CONTENT_TYPE_LD_PROFILE),
+                (as2.CONTENT_TYPE, as2.CONTENT_TYPE),
+                # TODO: handle eventually; this should return non-None
+                (activitypub.CONNEG_HEADERS_AS2_HTML['Accept'], None),
+                ('', None),
+                ('*/*', None),
+                ('text/html', None),
+        ):
+            with (self.subTest(accept=accept),
+                  app.test_request_context('/', headers={'Accept': accept})):
+                self.assertEqual(expected, activitypub.as2_request_type())

tests/test_common.py

@@ -168,22 +168,6 @@ class CommonTest(TestCase):
 
         mock_client.report.assert_called_with('foo', http_context=None, bar='baz')
 
-    def test_as2_request_type(self):
-        for accept, expected in (
-                (as2.CONTENT_TYPE_LD_PROFILE, as2.CONTENT_TYPE_LD_PROFILE),
-                (as2.CONTENT_TYPE_LD, as2.CONTENT_TYPE_LD_PROFILE),
-                (as2.CONTENT_TYPE, as2.CONTENT_TYPE),
-                # TODO: handle eventually; this should return non-None
-                (CONNEG_HEADERS_AS2_HTML['Accept'], None),
-                ('', None),
-                ('*/*', None),
-                ('text/html', None),
-        ):
-            with (self.subTest(accept=accept),
-                  app.test_request_context('/', headers={'Accept': accept})):
-                self.assertEqual(expected, common.as2_request_type())
-
-
     @patch('oauth_dropins.webutil.appengine_config.tasks_client.create_task')
     def test_create_task_no_request_context(self, mock_create_task):
         common.RUN_TASKS_INLINE = False