get_thread_id_from_url utility

misago/threads/tests/test_utils.py

@@ -2,7 +2,7 @@ from misago.categories.models import Category
 from misago.core.testutils import MisagoTestCase
 
 from .. import testutils
-from ..utils import add_categories_to_threads
+from ..utils import add_categories_to_threads, get_thread_id_from_url
 
 
 class AddCategoriesToThreadsTests(MisagoTestCase):
@@ -144,3 +144,117 @@ class AddCategoriesToThreadsTests(MisagoTestCase):
 
         self.assertEqual(thread.top_category, self.category_a)
         self.assertEqual(thread.category, self.category_d)
+
+
+class MockRequest(object):
+    def __init__(self, scheme, host, wsgialias=''):
+        self.scheme = scheme
+        self.host = host
+
+        self.path_info = '/api/threads/123/merge/'
+        self.path = '{}'.format(wsgialias.rstrip('/'), self.path_info)
+
+    def get_host(self):
+        return self.host
+
+    def is_secure(self):
+        return self.scheme == 'https'
+
+
+class GetThreadIdFromUrlTests(MisagoTestCase):
+    def test_get_thread_id_from_valid_urls(self):
+        """get_thread_id_from_url extracts thread pk from valid urls"""
+        TEST_CASES = (
+            {
+                # perfect match
+                'request': MockRequest('https', 'testforum.com', '/discuss/'),
+                'url': 'https://testforum.com/discuss/thread/test-thread-123/',
+                'pk': 123
+            },
+            {
+                # we don't validate scheme in case site recently moved to https
+                # but user still has old url's saved somewhere
+                'request': MockRequest('http', 'testforum.com', '/discuss/'),
+                'url': 'http://testforum.com/discuss/thread/test-thread-432/post/12321/',
+                'pk': 432
+            },
+            {
+                # we can extract thread id from other thread urls
+                'request': MockRequest('https', 'testforum.com', '/discuss/'),
+                'url': 'http://testforum.com/discuss/thread/test-thread-432/post/12321/',
+                'pk': 432
+            },
+            {
+                # we can extract thread id from thread page url
+                'request': MockRequest('http', 'testforum.com', '/discuss/'),
+                'url': 'http://testforum.com/discuss/thread/test-thread-432/123/',
+                'pk': 432
+            },
+            {
+                # we can extract thread id from thread last post url with relative schema
+                'request': MockRequest('http', 'testforum.com', '/discuss/'),
+                'url': '//testforum.com/discuss/thread/test-thread-18/last/',
+                'pk': 18
+            },
+            {
+                # we can extract thread id from url that lacks scheme
+                'request': MockRequest('http', 'testforum.com', ''),
+                'url': 'testforum.com/thread/test-thread-12/last/',
+                'pk': 12
+            },
+            {
+                # we can extract thread id from schemaless thread last post url
+                'request': MockRequest('http', 'testforum.com', '/discuss/'),
+                'url': 'testforum.com/discuss/thread/test-thread-18/last/',
+                'pk': 18
+            },
+            {
+                # we can extract thread id from url that lacks scheme and hostname
+                'request': MockRequest('http', 'testforum.com', ''),
+                'url': '/thread/test-thread-13/',
+                'pk': 13
+            }
+        )
+
+        for case in TEST_CASES:
+            pk = get_thread_id_from_url(case['request'], case['url'])
+            self.assertEqual(
+                pk, case['pk'], 'get_thread_id_from_url for {} should return {}'.format(case['url'], case['pk']))
+
+    def test_get_thread_id_from_invalid_urls(self):
+        TEST_CASES = (
+            {
+                # invalid wsgi alias
+                'request': MockRequest('https', 'testforum.com'),
+                'url': 'http://testforum.com/discuss/thread/test-thread-123/'
+            },
+            {
+                # invalid hostname
+                'request': MockRequest('http', 'misago-project.org', '/discuss/'),
+                'url': 'https://testforum.com/discuss/thread/test-thread-432/post/12321/'
+            },
+            {
+                # non-thread url
+                'request': MockRequest('http', 'testforum.com'),
+                'url': 'https://testforum.com/user/bobboberson-123/'
+            },
+            {
+                # rubbish url
+                'request': MockRequest('http', 'testforum.com'),
+                'url': 'asdsadsasadsaSA&das8as*S(A*sa'
+            },
+            {
+                # blank url
+                'request': MockRequest('http', 'testforum.com'),
+                'url': '/'
+            },
+            {
+                # empty url
+                'request': MockRequest('http', 'testforum.com'),
+                'url': ''
+            }
+        )
+
+        for case in TEST_CASES:
+            pk = get_thread_id_from_url(case['request'], case['url'])
+            self.assertIsNone(pk, 'get_thread_id_from_url for {} should fail'.format(case['url']))

misago/threads/utils.py

@@ -1,3 +1,7 @@
+from django.core.urlresolvers import resolve
+from django.utils.six.moves.urllib.parse import urlparse
+
+
 def add_categories_to_threads(root_category, categories, threads):
     categories_dict = {}
     for category in categories:
@@ -30,3 +34,53 @@ def add_categories_to_threads(root_category, categories, threads):
                         category.has_child(thread.category)):
                     top_categories_map[thread.category_id] = category
                     thread.top_category = category
+
+
+SUPPORTED_THREAD_ROUTES = {
+    'misago:thread': 'pk',
+    'misago:thread-post': 'pk',
+    'misago:thread-last': 'pk',
+    'misago:thread-new': 'pk',
+    'misago:thread-unapproved': 'pk',
+}
+
+
+def get_thread_id_from_url(request, url):
+    try:
+        bits = urlparse(url)
+    except:
+        return None
+
+    if bits.netloc:
+        if bits.port:
+            hostname = ':'.join((bits.netloc, bits.port))
+        else:
+            hostname = bits.netloc
+
+        if hostname != request.get_host():
+            return None
+
+    if bits.path.startswith(request.get_host()):
+        clean_path = bits.path.lstrip(request.get_host())
+    else:
+        clean_path = bits.path
+
+    try:
+        wsgi_alias = request.path[:len(request.path_info)]
+        resolution = resolve(clean_path[len(wsgi_alias):])
+    except:
+        return None
+
+    if not resolution.namespaces:
+        return None
+
+    url_name = '{}:{}'.format(':'.join(resolution.namespaces), resolution.url_name)
+    kwargname = SUPPORTED_THREAD_ROUTES.get(url_name)
+
+    if not kwargname:
+        return None
+
+    try:
+        return int(resolution.kwargs.get(kwargname))
+    except (TypeError, ValueError):
+        return None