0001-applications-handle-roles-in-app-bundles-68061.patch

Frédéric Péters, 12 août 2022 12:09

Voir les différences: en ligne côte à côte

Subject: [PATCH] applications: handle roles in app bundles (#68061)

 hobo/applications/models.py |  35 +++++++++++++
 hobo/applications/utils.py  |   3 +-
 tests/test_application.py   | 101 +++++++++++++++++++++++++++++++++++-
 3 files changed, 137 insertions(+), 2 deletions(-)

     # You should have received a copy of the GNU Affero General Public License
     # along with this program.  If not, see <http://www.gnu.org/licenses/>.
     import io
     import json
     import tarfile
     import urllib.parse
     from django.conf import settings
-...
         def deploy(self):
             bundle_content = self.bundle.read()
             self.deploy_roles(bundle_content)
             for service_id, services in getattr(settings, 'KNOWN_SERVICES', {}).items():
                 if service_id not in Application.SUPPORTED_MODULES:
                     continue
-...
                         continue
                     # TODO: look at response content for afterjob URLs to display a progress bar
                     pass
         def get_authentic_service(self):
             for service_id, services in getattr(settings, 'KNOWN_SERVICES', {}).items():
                 if service_id == 'authentic':
                     for service in services.values():
                         return service
             return None
         def deploy_roles(self, bundle):
             tar_io = io.BytesIO(bundle)
             service = self.get_authentic_service()
             if not service:
                 return
             roles_api_url = urllib.parse.urljoin(service['url'], 'api/roles/?update_or_create=slug')
             provision_api_url = urllib.parse.urljoin(service['url'], 'api/provision/')
             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())
                     # create or update
                     response = requests.post(roles_api_url, json=role_info)
                     if not response.ok:
                         # TODO: report failures
                         continue
                     # then force provisionning
                     response = requests.post(provision_api_url, json={'role_uuid': response.json()['uuid']})
                     if not response.ok:
                         # TODO: report failures
                         continue

             scheme, netloc, path, params, query, fragment = urllib.parse.urlparse(url)
             url = urllib.parse.urlunparse(('', '', path, params, query, fragment))
             query_params = {'orig': remote_service.get('orig')}
             query_params = dict(urllib.parse.parse_qsl(query))
             query_params['orig'] = remote_service.get('orig')
             remote_service_base_url = remote_service.get('url')
             scheme, netloc, dummy, params, old_query, fragment = urllib.parse.urlparse(remote_service_base_url)

     import json
     import tarfile
     import httmock
     import pytest
     from httmock import HTTMock
     from test_manager import login
     from webtest import Upload
     from hobo.applications.models import Application
     from hobo.environment.models import Wcs
     from hobo.environment.models import Authentic, Wcs
     pytestmark = pytest.mark.django_db
-...
     def mocked_http(url, request):
         assert '&signature=' in url.query
         if url.netloc == 'idp.example.invalid':
             if url.path == '/api/roles/':
                 return {
                     'content': json.dumps({'name': 'test', 'uuid': '123', 'slug': 'test'}),
                     'status_code': 200,
+                }
             if url.path == '/api/provision/':
                 return {
                     'content': json.dumps({'name': 'test', 'uuid': '123', 'slug': 'test'}),
                     'status_code': 200,
+                }
             return {'content': '{}', 'status_code': 500}
         if url.netloc == 'wcs.example.invalid' and url.path == '/api/export-import/':
             return {'content': json.dumps(WCS_AVAILABLE_OBJECTS), 'status_code': 200}
-...
             assert Application.objects.count() == 1
             assert Application.objects.get(slug='test').name == 'Test'
             assert Application.objects.get(slug='test').elements.count() == 3
     @pytest.fixture
     def app_bundle_roles():
         tar_io = io.BytesIO()
         with tarfile.open(mode='w', fileobj=tar_io) as tar:
             manifest_json = {
                 'application': 'Test',
                 'slug': 'test',
                 'description': '',
                 'elements': [
                     {'type': 'forms', 'slug': 'test', 'name': 'test', 'auto-dependency': False},
                     {'type': 'roles', 'slug': 'test-role', 'name': 'test', 'auto-dependency': True},
                     {'type': 'roles', 'slug': 'test-role2', 'name': 'test2', 'auto-dependency': True},
                 ],
+            }
             manifest_fd = io.BytesIO(json.dumps(manifest_json, indent=2).encode())
             tarinfo = tarfile.TarInfo('manifest.json')
             tarinfo.size = len(manifest_fd.getvalue())
             tar.addfile(tarinfo, fileobj=manifest_fd)
             role_json = {'name': 'Test', 'slug': 'test-role', 'uuid': '061e5de7023946c79a2f7f1273afc5a2'}
             role_json_fd = io.BytesIO(json.dumps(role_json, indent=2).encode())
             tarinfo = tarfile.TarInfo('roles/test-role')
             tarinfo.size = len(role_json_fd.getvalue())
             tar.addfile(tarinfo, fileobj=role_json_fd)
             role_json = {'name': 'Test', 'slug': 'test-role2', 'uuid': '061e5de7023946c79a2f7f1273afc5a3'}
             role_json_fd = io.BytesIO(json.dumps(role_json, indent=2).encode())
             tarinfo = tarfile.TarInfo('roles/test-role2')
             tarinfo.size = len(role_json_fd.getvalue())
             tar.addfile(tarinfo, fileobj=role_json_fd)
         return tar_io.getvalue()
     def test_deploy_application_roles(app, admin_user, settings, app_bundle_roles):
         Application.objects.all().delete()
         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')
         settings.KNOWN_SERVICES = {
             'authentic': {
                 'idp': {
                     'title': 'Foobar',
                     'url': 'https://idp.example.invalid/',
                     'orig': 'example.org',
                     'secret': 'xxx',
+                }
             },
             'wcs': {
                 'foobar': {
                     'title': 'Foobar',
                     'url': 'https://wcs.example.invalid/',
                     'orig': 'example.org',
                     'secret': 'xxx',
+                }
             },
+        }
         login(app)
         resp = app.get('/applications/')
         for i in range(2):
             resp = resp.click('Install')
             resp.form['bundle'] = Upload('app.tar', app_bundle_roles, 'application/x-tar')
             with HTTMock(httmock.remember_called(mocked_http)):
                 resp = resp.form.submit().follow()
                 # roles
                 assert mocked_http.call['requests'][0].url.startswith(
                     'https://idp.example.invalid/api/roles/?update'
+                )
                 assert mocked_http.call['requests'][1].url.startswith(
                     'https://idp.example.invalid/api/provision/'
+                )
                 assert mocked_http.call['requests'][2].url.startswith(
                     'https://idp.example.invalid/api/roles/?update'
+                )
                 assert mocked_http.call['requests'][3].url.startswith(
                     'https://idp.example.invalid/api/provision/'
+                )
                 # then form
                 assert 'wcs.example.invalid' in mocked_http.call['requests'][4].url
                 assert mocked_http.call['count'] == 5
+    -

Projet

Général

Profil

Produits Entr'ouvert » Hobo

0001-applications-handle-roles-in-app-bundles-68061.patch