Add a provider filter for legacy namespaces. (#2009)

No-Issue

Signed-off-by: James Tanner <[email protected]>

galaxy_ng/app/api/v1/filtersets.py

@@ -14,6 +14,7 @@ class LegacyNamespaceFilter(filterset.FilterSet):
 
     keywords = filters.CharFilter(method='keywords_filter')
     owner = filters.CharFilter(method='owner_filter')
+    provider = filters.CharFilter(method='provider_filter')
 
     sort = filters.OrderingFilter(
         fields=(
@@ -52,6 +53,9 @@ def owner_filter(self, queryset, name, value):
 
         return queryset
 
+    def provider_filter(self, queryset, name, value):
+        return queryset.filter(namespace__name=value)
+
 
 class LegacyUserFilter(filterset.FilterSet):
 

galaxy_ng/tests/integration/community/test_v1_namespaces.py

@@ -6,10 +6,12 @@
 from ..utils import (
     ansible_galaxy,
     SocialGithubClient,
+    get_client,
+    generate_unused_namespace,
 )
-from ..utils.legacy import (
-    cleanup_social_user,
-)
+
+from ..utils.legacy import cleanup_social_user
+from ..utils.legacy import generate_unused_legacy_namespace
 
 pytestmark = pytest.mark.qa  # noqa: F821
 
@@ -77,3 +79,44 @@ def test_social_auth_creates_v3_namespace_as_v1_provider(ansible_config):
         resp = client.get(f'v1/namespaces/?name={github_user}')
         res = resp.json()
         assert res['results'][0]['avatar_url'] == avatar_url
+
+
+@pytest.mark.deployment_community
+def test_v1_namespace_provider_filter(ansible_config):
+
+    admin_config = ansible_config('admin')
+    admin_client = get_client(config=admin_config, request_token=False, require_auth=True)
+
+    # 2 v1 namespaces
+    v1_a_name = generate_unused_legacy_namespace(admin_client)
+    v1_a = admin_client('/api/v1/namespaces/', method='POST', args={'name': v1_a_name})
+    v1_a_id = v1_a['id']
+    v1_b_name = generate_unused_legacy_namespace(admin_client)
+    v1_b = admin_client('/api/v1/namespaces/', method='POST', args={'name': v1_b_name})
+    v1_b_id = v1_b['id']
+
+    # make 1 v3 namespace
+    v3_a_name = generate_unused_namespace(admin_client)
+    v3_a = admin_client(
+        '/api/v3/namespaces/', method='POST', args={'name': v3_a_name, 'groups': []}
+    )
+    v3_a_id = v3_a['id']
+    v3_b_name = generate_unused_namespace(admin_client)
+    v3_b = admin_client(
+        '/api/v3/namespaces/', method='POST', args={'name': v3_b_name, 'groups': []}
+    )
+    v3_b_id = v3_b['id']
+
+    # bind v1_a to v3 and v1_b to v3_b
+    admin_client(f'/api/v1/namespaces/{v1_a_id}/providers/', method='POST', args={'id': v3_a_id})
+    admin_client(f'/api/v1/namespaces/{v1_b_id}/providers/', method='POST', args={'id': v3_b_id})
+
+    # check the filter ..
+    resp_a = admin_client(f'/api/v1/namespaces/?provider={v3_a_name}')
+    resp_b = admin_client(f'/api/v1/namespaces/?provider={v3_b_name}')
+
+    assert resp_a['count'] == 1
+    assert resp_a['results'][0]['id'] == v1_a_id, resp_a
+
+    assert resp_b['count'] == 1
+    assert resp_b['results'][0]['id'] == v1_b_id, resp_b

galaxy_ng/tests/integration/utils/legacy.py

@@ -1,3 +1,5 @@
+import random
+import string
 import time
 
 from galaxykit.users import delete_user as delete_user_gk
@@ -160,3 +162,59 @@ def cleanup_social_user_gk(username, galaxy_client):
 
     # cleanup the user
     delete_user_gk(gc_admin, username)
+
+
+def generate_legacy_namespace(exclude=None):
+    """ Create a valid random legacy namespace string """
+
+    # This should be a list of pre-existing namespaces
+    if exclude is None:
+        exclude = []
+
+    def is_valid(ns):
+        """ Assert namespace meets backend requirements """
+        if ns is None:
+            return False
+        if ns in exclude:
+            return False
+        if len(namespace) < 3:
+            return False
+        if len(namespace) > 64:
+            return False
+        for char in namespace:
+            if char not in string.ascii_lowercase + string.ascii_uppercase + string.digits:
+                return False
+
+        return True
+
+    namespace = None
+    while not is_valid(namespace):
+        namespace = ''
+        namespace += random.choice(string.ascii_lowercase + string.ascii_uppercase + string.digits)
+        for x in range(0, random.choice(range(3, 63))):
+            namespace += random.choice(string.ascii_lowercase + string.digits + '_')
+
+    return namespace
+
+
+def get_all_legacy_namespaces(api_client=None):
+    namespaces = []
+    next_page = '/api/v1/namespaces/'
+    while next_page:
+        resp = api_client(next_page)
+        namespaces.extend(resp['results'])
+        next_page = resp.get('next')
+        if next_page:
+            # trim the proto+host+port ...
+            ix = next_page.index('/api')
+            next_page = next_page[ix:]
+    return namespaces
+
+
+def generate_unused_legacy_namespace(api_client=None):
+    """ Make a random legacy_namespace string that does not exist """
+
+    assert api_client is not None, "api_client is a required param"
+    existing = get_all_legacy_namespaces(api_client=api_client)
+    existing = dict((x['name'], x) for x in existing)
+    return generate_legacy_namespace(exclude=list(existing.keys()))