hobo/tests/test_application.py

import io
import json
import tarfile

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

pytestmark = pytest.mark.django_db

WCS_AVAILABLE_OBJECTS = {
    "data": [
        {
            "id": "forms",
            "text": "Forms",
            "singular": "Form",
            "urls": {"list": "https://wcs.example.invalid/api/export-import/forms/"},
        },
        {
            "id": "cards",
            "text": "Card Models",
            "singular": "Card Model",
            "urls": {"list": "https://wcs.example.invalid/api/export-import/cards/"},
        },
        {
            "id": "workflows",
            "text": "Workflows",
            "singular": "Workflow",
            "urls": {"list": "https://wcs.example.invalid/api/export-import/workflows/"},
        },
        {
            "id": "blocks",
            "text": "Blocks",
            "singular": "Block of fields",
            "minor": True,
            "urls": {"list": "https://wcs.example.invalid/api/export-import/blocks/"},
        },
        {
            "id": "data-sources",
            "text": "Data Sources",
            "singular": "Data Source",
            "minor": True,
            "urls": {"list": "https://wcs.example.invalid/api/export-import/data-sources/"},
        },
        {
            "id": "mail-templates",
            "text": "Mail Templates",
            "singular": "Mail Template",
            "minor": True,
            "urls": {"list": "https://wcs.example.invalid/api/export-import/mail-templates/"},
        },
        {
            "id": "wscalls",
            "text": "Webservice Calls",
            "singular": "Webservice Call",
            "minor": True,
            "urls": {"list": "https://wcs.example.invalid/api/export-import/wscalls/"},
        },
    ]
}

WCS_AVAILABLE_FORMS = {
    "data": [
        {
            "id": "test-form",
            "text": "Test Form",
            "type": "forms",
            "urls": {
                "export": "https://wcs.example.invalid/api/export-import/forms/test-form/",
                "dependencies": "https://wcs.example.invalid/api/export-import/forms/test-form/dependencies/",
            },
        },
        {
            "id": "test2-form",
            "text": "Second Test Form",
            "type": "forms",
            "urls": {
                "export": "https://wcs.example.invalid/api/export-import/forms/test2-form/",
                "dependencies": "https://wcs.example.invalid/api/export-import/forms/test2-form/dependencies/",
            },
        },
    ]
}

WCS_FORM_DEPENDENCIES = {
    "data": [
        {
            "id": "test-card",
            "text": "Test Card",
            "type": "cards",
            "urls": {
                "export": "https://wcs.example.invalid/api/export-import/cards/test-card/",
                "dependencies": "https://wcs.example.invalid/api/export-import/cards/test-card/dependencies/",
            },
        }
    ]
}


def mocked_http(url, request):
    assert '&signature=' in url.query
    if url.netloc == 'wcs.example.invalid' and url.path == '/api/export-import/':
        return {'content': json.dumps(WCS_AVAILABLE_OBJECTS), 'status_code': 200}

    if url.path == '/api/export-import/forms/':
        return {'content': json.dumps(WCS_AVAILABLE_FORMS), 'status_code': 200}

    if url.path == '/api/export-import/forms/test-form/dependencies/':
        return {'content': json.dumps(WCS_FORM_DEPENDENCIES), 'status_code': 200}

    if url.path.endswith('/dependencies/'):
        return {'content': json.dumps({'data': []}), 'status_code': 200}

    if url.path == '/api/export-import/forms/test-form/':
        return {'content': '<formdef/>', 'status_code': 200, 'headers': {'content-length': '10'}}

    if url.path == '/api/export-import/cards/test-card/':
        return {'content': '<carddef/>', 'status_code': 200, 'headers': {'content-length': '10'}}

    if url.path == '/api/export-import/bundle-import/':
        return {'content': '{}', 'status_code': 200}


def test_create_application(app, admin_user, settings):
    Wcs.objects.create(base_url='https://wcs.example.invalid', slug='foobar', title='Foobar')

    settings.KNOWN_SERVICES = {
        'wcs': {
            'foobar': {
                'title': 'Foobar',
                'url': 'https://wcs.example.invalid/',
                'orig': 'example.org',
                'secret': 'xxx',
            }
        }
    }

    login(app)

    resp = app.get('/applications/')
    resp = resp.click('Create')
    resp.form['name'] = 'Test'
    resp = resp.form.submit()

    with HTTMock(mocked_http):
        resp = resp.follow()
        assert 'You should now assemble the different parts of your application.' in resp.text

        # edit metadata
        resp = resp.click('Metadata')
        resp.form['description'] = 'Lorem ipsum'
        resp = resp.form.submit().follow()

        # add forms
        assert '/add/forms/' in resp
        resp = resp.click('Forms')
        assert resp.form.fields['elements'][0]._value == 'test-form'
        assert resp.form.fields['elements'][1]._value == 'test2-form'
        resp.form.fields['elements'][0].checked = True
        resp = resp.form.submit().follow()
        assert Application.objects.get(slug='test').elements.count() == 1
        element = Application.objects.get(slug='test').elements.all()[0]
        assert element.slug == 'test-form'

        resp = resp.click('Scan dependencies').follow()
        assert Application.objects.get(slug='test').elements.count() == 2
        assert 'Test Card' in resp.text

        resp = resp.click('Generate application bundle').follow()
        resp = resp.click('Download')
        assert resp.content_type == 'application/x-tar'
        # uncompressed tar, primitive check of contents
        assert b'<formdef/>' in resp.content
        assert b'<carddef/>' in resp.content


def test_delete_application(app, admin_user, settings):
    Wcs.objects.create(base_url='https://wcs.example.invalid', slug='foobar', title='Foobar')

    settings.KNOWN_SERVICES = {
        'wcs': {
            'foobar': {
                'title': 'Foobar',
                'url': 'https://wcs.example.invalid/',
                'orig': 'example.org',
                'secret': 'xxx',
            }
        }
    }

    login(app)

    Application.objects.create(name='AppToDelete', slug='app_to_delete')
    Application.objects.create(name='OtherApp', slug='other_app')

    assert Application.objects.count() == 2

    resp = app.get('/applications/manifest/app_to_delete/delete/')

    resp = resp.forms[0].submit()
    resp = resp.follow()

    assert '/applications/' in resp
    assert 'AppToDelete' not in resp.text

    assert Application.objects.count() == 1
    assert Application.objects.first().name == 'OtherApp'


@pytest.fixture
def app_bundle():
    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': 'blocks', 'slug': 'test', 'name': 'test', 'auto-dependency': True},
                {'type': 'workflows', 'slug': 'test', 'name': 'test', '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)

    return tar_io.getvalue()


def test_deploy_application(app, admin_user, settings, app_bundle):
    Application.objects.all().delete()
    Wcs.objects.create(base_url='https://wcs.example.invalid', slug='foobar', title='Foobar')

    settings.KNOWN_SERVICES = {
        '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, 'application/x-tar')
        with HTTMock(mocked_http):
            resp = resp.form.submit().follow()

        assert Application.objects.count() == 1
        assert Application.objects.get(slug='test').name == 'Test'
        assert Application.objects.get(slug='test').elements.count() == 3