Web API: file upload & proposal_pipeline support, tests, and some cleanup

regparser/web/jobs/utils.py

@@ -83,9 +83,9 @@ def add_redis_data_to_job_data(job_data):
     return job_data
 
 
-def status_url(job_id, sub_path=""):
+def get_host():
     """
-    We want to give users a URL for checking the status of a job.
+    We want to provide users with status URLs, and so need host:port.
     While I can't think of an exploit resulting from relying on the host data
     from the request if the request were spoofed, we'll be cautious and define
     the canonical host data ourselves.
@@ -95,38 +95,56 @@ def status_url(job_id, sub_path=""):
     Impure
         Pulls information from settings.
 
-    :arg uuid4 job_id: The UUID of the job.
-    :arg str sub_path: The part of the path indicating the type of job.
-
     :rtype: str
-    :returns: The URL for checking on the status of the job.
+    :returns: The URL for host in the form host:port, for example:
+        http://domain.tld:2323
+
+        Note that the schema is not supplied (we assume it's included in the
+        string provided to settings) and no trailing slash is provided.
+
+        We assume that the port from setigns is the bare number, with no
+        trailing colon, so we add that here.
     """
     hostname = getattr(settings, "CANONICAL_HOSTNAME", "")
     hostport = getattr(settings, "CANONICAL_PORT", "")
-    if hostport:
+    if hostport and hostport is not "80":
         hostport = ":%s" % hostport
-    return "%s%s/rp/job/%s%s/" % (hostname, hostport, sub_path, job_id)
+    elif hostport is "80":
+        hostport = ""
+    return "%s%s" % (hostname, hostport)
+
+
+def status_url(job_id, sub_path=""):
+    """
+    Returns a URL for checking the status of a job.
+
+    Impure
+        Via get_host(), pulls information from settings.
+
+    :arg uuid4 job_id: The UUID of the job.
+    :arg str sub_path: The part of the path indicating the type of job. Must
+        include a trailing slash.
+
+    :rtype: str
+    :returns: The URL for checking on the status of the job.
+    """
+    if sub_path and not sub_path.endswith("/"):
+        raise ValueError
+    host = get_host()
+    return "%s/rp/job/%s%s/" % (host, sub_path, job_id)
 
 
 def file_url(file_hash):
     """
-    We want to give users a URL for retrieving an uploaded file.
-    While I can't think of an exploit resulting from relying on the host data
-    from the request if the request were spoofed, we'll be cautious and define
-    the canonical host data ourselves.
-    This helper function puts the lookup and the string manipulation in one
-    place.
+    Returns a URL for retrieving an uploaded file.
 
     Impure
-        Pulls information from settings.
+        Via get_host(), pulls information from settings.
 
     :arg str file_hash: The MD5 hexstring of the file contents.
 
     :rtype: str
     :returns: The URL for checking on the status of the job.
     """
-    hostname = getattr(settings, "CANONICAL_HOSTNAME", "")
-    hostport = getattr(settings, "CANONICAL_PORT", "")
-    if hostport:
-        hostport = ":%s" % hostport
-    return "%s%s/rp/job/upload/%s/" % (hostname, hostport, file_hash)
+    host = get_host()
+    return "%s/rp/job/upload/%s/" % (host, file_hash)

tests/test_web_api.py

@@ -1,6 +1,7 @@
 from hashlib import md5
 from mock import patch
 from os import path as ospath
+from random import choice
 from regparser.web.jobs.models import job_status_values
 from regparser.web.jobs.utils import (
     eregs_site_api_url,
@@ -9,7 +10,12 @@
 )
 from regparser.web.jobs.views import FileUploadView as PatchedFileUploadView
 from rest_framework.test import APITestCase
+from string import hexdigits
 from tempfile import NamedTemporaryFile
+
+import pytest
+import settings
+
 try:
     from urllib.parse import urlparse
 except ImportError:
@@ -19,6 +25,10 @@
 fake_email_id = "4774f0f6-b53e-4b34-821e-fc8ae5e113fe"
 
 
+def fake_getter(attrname):
+    return attrname
+
+
 def fake_add_redis_data_to_job_data(job_data):
     return job_data
 
@@ -305,3 +315,61 @@ def test_create_and_read_and_delete(self):
 
         response = self.client.get("/rp/job/proposal-pipeline/", format="json")
         self.assertEqual(0, len(response.data))
+
+
+@patch.object(settings, "CANONICAL_HOSTNAME", "http://domain.tld")
+def test_status_url():
+    domain = "http://domain.tld"
+    urlpath = "/rp/job/"
+    hexes = ["".join([choice(hexdigits) for i in range(32)]) for j in range(6)]
+
+    def _check(port=None):
+        for hx in hexes:
+            url = urlparse(status_url(hx))
+            assert domain == "%s://%s" % (url.scheme, url.hostname)
+            if port is None:
+                assert url.port is port
+            else:
+                assert url.port == port
+            assert "%s%s/" % (urlpath, hx) == url.path
+
+            url = urlparse(status_url(hx, sub_path="%s/" % hx[:10]))
+            assert domain == "%s://%s" % (url.scheme, url.hostname)
+            if port is None:
+                assert url.port is port
+            else:
+                assert url.port == port
+            assert "%s%s%s/" % (urlpath, "%s/" % hx[:10], hx) == url.path
+
+    with patch.object(settings, "CANONICAL_PORT", "2323"):
+        _check(port=2323)
+
+    with patch.object(settings, "CANONICAL_PORT", "80"):
+        _check()
+
+    with patch.object(settings, "CANONICAL_PORT", ""):
+        _check()
+
+    with pytest.raises(ValueError) as err:
+        status_url("something", "something-without-a-slash")
+
+    assert isinstance(err.value, ValueError)
+
+
+@patch.object(settings, "CANONICAL_HOSTNAME", "http://domain.tld")
+def test_file_url():
+    urlpath = "/rp/job/upload/"
+    domain = "http://domain.tld"
+    hexes = ["".join([choice(hexdigits) for i in range(32)]) for j in range(6)]
+
+    with patch.object(settings, "CANONICAL_PORT", "2323"):
+        for hx in hexes:
+            assert file_url(hx) == "%s:2323%s%s/" % (domain, urlpath, hx)
+
+    with patch.object(settings, "CANONICAL_PORT", "80"):
+        for hx in hexes:
+            assert file_url(hx) == "%s%s%s/" % (domain, urlpath, hx)
+
+    with patch.object(settings, "CANONICAL_PORT", ""):
+        for hx in hexes:
+            assert file_url(hx) == "%s%s%s/" % (domain, urlpath, hx)