applications: handle roles in app bundles (#68061)
This commit is contained in:
parent
2d04e39274
commit
128c9fa79e
|
|
@ -14,6 +14,9 @@
|
|||
# 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
|
||||
|
|
@ -88,6 +91,7 @@ class Version(models.Model):
|
|||
|
||||
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
|
||||
|
|
@ -102,3 +106,34 @@ class Version(models.Model):
|
|||
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
|
||||
|
|
|
|||
|
|
@ -55,7 +55,8 @@ class Requests(RequestsSession):
|
|||
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)
|
||||
|
|
|
|||
|
|
@ -2,13 +2,14 @@ import io
|
|||
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
|
||||
|
||||
|
|
@ -103,6 +104,20 @@ WCS_FORM_DEPENDENCIES = {
|
|||
|
||||
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}
|
||||
|
||||
|
|
@ -260,3 +275,87 @@ def test_deploy_application(app, admin_user, settings, app_bundle):
|
|||
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
|
||||
|
|
|
|||
Loading…
Reference in New Issue