applications: create roles in instance OU (#76273)

hobo/applications/models.py

@@ -458,13 +458,23 @@ class Version(models.Model):
         if not service:
             return
         roles_api_url = urllib.parse.urljoin(service['url'], 'api/roles/?update_or_create=slug')
+        role_ou = None
         provision_api_url = urllib.parse.urljoin(service['url'], 'api/provision/')
+        ou_slug_variable = Variable.objects.filter(name='ou-slug').first()
+        if ou_slug_variable:
+            roles_api_url += '&update_or_create=ou'
+            if ou_slug_variable.service:  # variable on main instance has service defined
+                role_ou = 'default'
+            else:
+                role_ou = ou_slug_variable.value
         with tarfile.open(fileobj=tar_io) as tar:
             manifest = json.loads(tar.extractfile('manifest.json').read().decode())
             for element in manifest.get('elements'):
                 if element.get('type') != 'roles':
                     continue
                 role_info = json.loads(tar.extractfile('%s/%s' % (element['type'], element['slug'])).read())
+                if role_ou:
+                    role_info['ou'] = role_ou
                 # create or update
                 response = requests.post(roles_api_url, json=role_info)
                 if not response.ok:

hobo/applications/utils.py

@@ -53,13 +53,13 @@ class Requests(RequestsSession):
         scheme, netloc, path, params, query, fragment = urllib.parse.urlparse(url)
         url = urllib.parse.urlunparse(('', '', path, params, query, fragment))
 
-        query_params = dict(urllib.parse.parse_qsl(query))
+        query_params = urllib.parse.parse_qs(query)
         query_params['orig'] = remote_service.get('orig')
 
         remote_service_base_url = remote_service.get('url')
         scheme, netloc, dummy, params, _, fragment = urllib.parse.urlparse(remote_service_base_url)
 
-        query = urlencode(query_params)
+        query = urlencode(query_params, doseq=True)
         url = urllib.parse.urlunparse((scheme, netloc, path, params, query, fragment))
 
         return super().request(method, url, **kwargs)

tests/test_application.py

@@ -7,6 +7,7 @@ import tarfile
 
 import httmock
 import pytest
+from django.contrib.contenttypes.models import ContentType
 from httmock import HTTMock
 from pyquery import PyQuery
 from test_manager import login
@@ -1321,7 +1322,7 @@ def app_bundle_roles():
 
 def test_deploy_application_roles(app, admin_user, settings, app_bundle_roles):
     Authentic.objects.create(base_url='https://idp.example.invalid', slug='idp', title='Foobar')
-    Wcs.objects.create(base_url='https://wcs.example.invalid', slug='foobar', title='Foobar')
+    wcs = Wcs.objects.create(base_url='https://wcs.example.invalid', slug='foobar', title='Foobar')
 
     settings.KNOWN_SERVICES = {
         'authentic': {
@@ -1410,6 +1411,32 @@ def test_deploy_application_roles(app, admin_user, settings, app_bundle_roles):
         assert job.status == 'failed'
         assert job.exception == 'Failed to provision role test-role (500)'
 
+    # test import on a specific service OU
+    v = Variable.objects.create(name='ou-slug', label='OU', value='service-ou')
+    Application.objects.all().delete()
+    resp = app.get('/applications/')
+    resp = resp.click('Install')
+    resp.form['bundle'] = Upload('app.tar', app_bundle_roles, 'application/x-tar')
+    with HTTMock(httmock.remember_called(mocked_http)):
+        resp.form.submit().follow()
+        # ou must be specified when calling authentic API
+        assert mocked_http.call['requests'][0].url.startswith(
+            'https://idp.example.invalid/api/roles/?update_or_create=slug&update_or_create=ou'
+        )
+        assert b'"ou": "service-ou"' in mocked_http.call['requests'][0].body
+
+    # test import on a default OU
+    Application.objects.all().delete()
+    resp = app.get('/applications/')
+    resp = resp.click('Install')
+    resp.form['bundle'] = Upload('app.tar', app_bundle_roles, 'application/x-tar')
+    v.service_type = ContentType.objects.get_for_model(Wcs)
+    v.service_pk = wcs.id
+    v.save()
+    with HTTMock(httmock.remember_called(mocked_http)):
+        resp.form.submit().follow()
+        assert b'"ou": "default"' in mocked_http.call['requests'][0].body
+
 
 def test_job_status_page(app, admin_user, settings):
     Wcs.objects.create(base_url='https://wcs.example.invalid', slug='foobar', title='Foobar')