Make named url work with optional url prefix

awx/main/middleware.py

@@ -6,7 +6,7 @@
 import threading
 import time
 import urllib.parse
-from pathlib import Path
+from pathlib import Path, PurePosixPath
 
 from django.conf import settings
 from django.contrib.auth import logout
@@ -138,14 +138,36 @@ def _named_url_to_pk(cls, node, resource, named_url):
 
  @classmethod
  def _convert_named_url(cls, url_path):
- url_units = url_path.split('/')
- # If the identifier is an empty string, it is always invalid.
- if len(url_units) < 6 or url_units[1] != 'api' or url_units[2] not in ['v2'] or not url_units[4]:
- return url_path
- resource = url_units[3]
+ default_prefix = PurePosixPath('/api/v2/')
+ optional_prefix = PurePosixPath(f'/api/{settings.OPTIONAL_API_URLPATTERN_PREFIX}/v2/')
+
+ url_path_original = url_path
+ url_path = PurePosixPath(url_path)
+
+ if set(optional_prefix.parts).issubset(set(url_path.parts)):
+ url_prefix = optional_prefix
+ elif set(default_prefix.parts).issubset(set(url_path.parts)):
+ url_prefix = default_prefix
+ else:
+ return url_path_original
+
+ # Remove prefix
+ url_path = PurePosixPath(*url_path.parts[len(url_prefix.parts) :])
+ try:
+ resource_path = PurePosixPath(url_path.parts[0])
+ name = url_path.parts[1]
+ url_suffix = PurePosixPath(*url_path.parts[2:]) # remove name and resource
+ except IndexError:
+ return url_path_original
+
+ resource = resource_path.parts[0]
  if resource in settings.NAMED_URL_MAPPINGS:
- url_units[4] = cls._named_url_to_pk(settings.NAMED_URL_GRAPH[settings.NAMED_URL_MAPPINGS[resource]], resource, url_units[4])
- return '/'.join(url_units)
+ pk = PurePosixPath(cls._named_url_to_pk(settings.NAMED_URL_GRAPH[settings.NAMED_URL_MAPPINGS[resource]], resource, name))
+ else:
+ return url_path_original
+
+ parts = url_prefix.parts + resource_path.parts + pk.parts + url_suffix.parts
+ return PurePosixPath(*parts).as_posix() + '/'
 
  def process_request(self, request):
  old_path = request.path_info

awx/main/tests/functional/test_named_url.py

@@ -205,3 +205,65 @@ def test_403_vs_404(get):
 
  get(f'/api/v2/users/{cindy.pk}/', expect=401)
  get('/api/v2/users/cindy/', expect=404)
+
+
+@pytest.mark.django_db
+class TestConvertNamedUrl:
+ @pytest.mark.parametrize(
+ "url",
+ (
+ "/api/",
+ "/api/v2/",
+ "/api/v2/hosts/",
+ "/api/v2/hosts/1/",
+ "/api/v2/organizations/1/inventories/",
+ "/api/foo/",
+ "/api/foo/v2/",
+ "/api/foo/v2/organizations/",
+ "/api/foo/v2/organizations/1/",
+ "/api/foo/v2/organizations/1/inventories/",
+ "/api/foobar/",
+ "/api/foobar/v2/",
+ "/api/foobar/v2/organizations/",
+ "/api/foobar/v2/organizations/1/",
+ "/api/foobar/v2/organizations/1/inventories/",
+ "/api/foobar/v2/organizations/1/inventories/",
+ ),
+ )
+ def test_noop(self, url):
+ settings.OPTIONAL_API_URLPATTERN_PREFIX = ''
+ assert URLModificationMiddleware._convert_named_url(url) == url
+
+ settings.OPTIONAL_API_URLPATTERN_PREFIX = 'foo'
+ assert URLModificationMiddleware._convert_named_url(url) == url
+
+ def test_named_org(self):
+ test_org = Organization.objects.create(name='test_org')
+
+ assert URLModificationMiddleware._convert_named_url('/api/v2/organizations/test_org/') == f'/api/v2/organizations/{test_org.pk}/'
+
+ def test_named_org_optional_api_urlpattern_prefix_interaction(self, settings):
+ settings.OPTIONAL_API_URLPATTERN_PREFIX = 'bar'
+ test_org = Organization.objects.create(name='test_org')
+
+ assert URLModificationMiddleware._convert_named_url('/api/bar/v2/organizations/test_org/') == f'/api/bar/v2/organizations/{test_org.pk}/'
+
+ @pytest.mark.parametrize("prefix", ['', 'bar'])
+ def test_named_org_not_found(self, prefix, settings):
+ settings.OPTIONAL_API_URLPATTERN_PREFIX = prefix
+ if prefix:
+ prefix += '/'
+
+ assert URLModificationMiddleware._convert_named_url(f'/api/{prefix}v2/organizations/does-not-exist/') == f'/api/{prefix}v2/organizations/0/'
+
+ @pytest.mark.parametrize("prefix", ['', 'bar'])
+ def test_named_sub_resource(self, prefix, settings):
+ settings.OPTIONAL_API_URLPATTERN_PREFIX = prefix
+ test_org = Organization.objects.create(name='test_org')
+ if prefix:
+ prefix += '/'
+
+ assert (
+ URLModificationMiddleware._convert_named_url(f'/api/{prefix}v2/organizations/test_org/inventories/')
+ == f'/api/{prefix}v2/organizations/{test_org.pk}/inventories/'
+ )