Produits Entr'ouvert » Combo

            return Response(response, status.HTTP_400_BAD_REQUEST)

        data = serializer.validated_data

        asset.asset = File(BytesIO(data['asset']['content']), name=data['asset'].get('filename'))

        asset.save()

        response = {'err': 0}

    @classmethod

    def load_serialized_object(cls, json_asset):

        json_asset['model'] = 'assets.asset'

        json_asset['pk'] = asset.id

        asset = next(serializers.deserialize('json', json.dumps([json_asset]), ignorenonexistent=True))

        asset.save()

def clean_assets_files():

    media_prefix = default_storage.path('')

        for filename in filenames:

            os.remove('%s/%s' % (basedir, filename))

def tar_assets_files(tar):

    media_prefix = default_storage.path('')

        for filename in filenames:

            tar.add(os.path.join(basedir, filename), os.path.join(basedir, filename)[len(media_prefix) :])

    export = {'assets': Asset.export_all_for_json()}

        context = super().get_context_data(**kwargs)

        media_prefix = default_storage.path('')

        computed_size = 0

            for filename in filenames:

                computed_size += os.stat(os.path.join(basedir, filename)).st_size

        context['size'] = computed_size

def get_chrono_events(agenda_reference, synchronous):

    chrono = get_chrono_service()

    response = requests.get(

        'api/agenda/%s/datetimes/' % chrono_slug,

def dashboard_reorder_tiles(request, *args, **kwargs):

    new_order = request.GET['order'].split(',')

    tiles = {str(x.id): x for x in Tile.objects.filter(id__in=new_order)}

    for i, tile_id in enumerate(new_order):

    def __init__(self, *args, **kwargs):

        super().__init__(*args, **kwargs)

        available_charts = []

            result = requests.get(

                '/visualization/json/',

                remote_service=site_dict,

import django

from django.conf import settings

from combo.utils import requests

        return

    results = []

    for provider in settings.STATISTICS_PROVIDERS:

        if isinstance(provider, dict):

            url = provider['url']

            width=int(request.GET['width']) if request.GET.get('width') else None,

            height=int(request.GET['height']) if request.GET.get('height') else int(cell.height),

        )

        error_text = _('Unsupported dataset.')

    except HTTPError as e:

        if e.response.status_code == 404:

    def create_paid_invoice_basket_item(self, item_id, remote_item):

        subject = _('Invoice #%s') % remote_item.display_id

            remote_item_id=item_id,

            defaults=dict(

                user=self.user,

    @classmethod

    def load_serialized_object(cls, json_layer):

        json_layer['model'] = 'maps.maplayer'

        json_layer['pk'] = layer.id

        layer = next(serializers.deserialize('json', json.dumps([json_layer]), ignorenonexistent=True))

        layer.save()

                raise ValueError('id must be convertible to unicode', e)

            if not re.match(cls.ID_RE, id):

                raise ValueError('id must match regular expression %s' % cls.ID_RE)

                user=user, external_id=id, defaults=defaults

            )

        else:

        json_entry['model'] = 'pwa.pwanavigationentry'

        # deserialize once to get link_page by natural key

        fake_entry = next(serializers.deserialize('json', json.dumps([json_entry]), ignorenonexistent=True))

            label=json_entry['fields']['label'],

            url=json_entry['fields']['url'],

            link_page=fake_entry.object.link_page,

    if subscription_data is None:

        PushSubscription.objects.filter(user=request.user).delete()

    else:

            user=request.user, subscription_info=subscription_data

        )

        subscription.save()

        return HttpResponse(status=204)

    engines = []

        try:

            new_order = int(request.GET.get('pos_' + str(engine_slug)))

        except TypeError:

def compile_sub_slug(sub_slug):

    # will raise re.error if wrong regexp

    re.compile(sub_slug)

        cells = CellBase.get_cells(page=self.object, prefetch_validity_info=True)

        self.object.prefetched_cells = cells

        placeholders = []

        optional_placeholders = []

        for placeholder in self.object.get_placeholders(request=self.request, traverse_cells=True):

            placeholder_dict = {

    slug = 'portal'

    if getattr(settings, 'KNOWN_SERVICES') and 'combo' in settings.KNOWN_SERVICES:

        # switch to custom slug if the site is the portal agent.

            if service.get('is-portal-agent') and not service.get('secret'):

                slug = 'portal-agent'

                break

        user_page_base_url = user_page.get_online_url()

        # return entry using first(&only) authentic (if it exists)

            return {

                'users': {

                    'url': service['url'] + 'api/users/?q=%(q)s',

@register.tag

def skeleton_extra_placeholder(parser, token):

    try:

    except ValueError:

        raise template.TemplateSyntaxError("%r tag requires exactly one argument" % token.contents.split()[0])

        if cell.slug not in cell_by_slugs:

            cell_by_slugs[cell.slug] = []

        cell_by_slugs[cell.slug].append(cell)

        for cell in slug_cells:

            cell.use_slug_as_id = bool(len(slug_cells) == 1)

        if cell.slug not in cell_by_slugs:

            cell_by_slugs[cell.slug] = []

        cell_by_slugs[cell.slug].append(cell)

        for cell in slug_cells:

            cell.use_slug_as_id = bool(len(slug_cells) == 1)

        self.page = Page(public=False, title=_('Style Demo'))

        self.page.template_name = self.template_name

        self.page.save()

        for i, placeholder in enumerate(self.page.get_placeholders(request=self.request)):

            cell = TextCell(page=self.page, placeholder=placeholder.key, order=0)

            if placeholder.key == 'footer':

        and settings.COMBO_INITIAL_LOGIN_PAGE_PATH

        and (request.user and not request.user.is_anonymous)

    ):

        if not profile.initial_login_view_timestamp:

            # first connection of user, record that and redirect to welcome URL

            profile.initial_login_view_timestamp = timezone.now()

            for services in settings.KNOWN_SERVICES.values():

                for service in services.values():

                    remote_url = service.get('url')

                    if remote_scheme == scheme and remote_netloc == netloc:

                        remote_service = service

                        break

                query_params['orig'] = remote_service.get('orig')

            remote_service_base_url = remote_service.get('url')

            query = urlencode(query_params)

            if '?' in url:

def count_asset_files():

    nb_assets = 0

    media_prefix = default_storage.path('')

        nb_assets += len(filenames)

    return nb_assets

                'combo-public-ajax-page-cell',

                kwargs={'page_pk': page.id, 'cell_reference': cell.get_reference()},

            )

            assert requests_get.call_count == 1

            assert requests_get.call_args[0][0] == 'http://foo?var=plop'

            assert requests_get.call_args[-1]['log_errors'] == False

    page = Page(title='example page', slug='example-page')

    page.save()

    with override_settings(JSON_CELL_TYPES={'foobar': {'name': 'Foobar', 'url': 'http://test/'}}):

        cell = ConfigJsonCell()

        cell.key = 'foobar'

def test_add_to_dashboard_order(app, site):

    test_add_to_dashboard(app, site)

    cell = TextCell.objects.get(order=101)

    assert Tile.objects.count() == 2

    assert Tile.objects.all()[0].cell.text == 'hello world (100)'

    assert Tile.objects.all()[1].cell.text == 'hello world (101)'

    cell = TextCell.objects.get(order=102)

    assert Tile.objects.count() == 3

    assert Tile.objects.all()[0].cell.text == 'hello world (100)'

    assert Tile.objects.all()[1].cell.text == 'hello world (101)'

    with override_settings(COMBO_DASHBOARD_NEW_TILE_POSITION='first'):

        cell = TextCell.objects.get(order=103)

        assert Tile.objects.count() == 4

        assert Tile.objects.all()[0].cell.text == 'hello world (103)'

        assert Tile.objects.all()[1].cell.text == 'hello world (100)'

    app.reset()  # logout

    tile = Tile.objects.all()[0]

    page = Page.objects.get(slug='two')

        reverse(

            'combo-public-ajax-page-cell',

            kwargs={'page_pk': page.id, 'cell_reference': tile.cell.get_reference()},

    )

    app = login(app)

        reverse(

            'combo-public-ajax-page-cell',

            kwargs={'page_pk': page.id, 'cell_reference': tile.cell.get_reference()},

        status=200,

    )

    app = login(app, username='plop', password='plop')

        reverse(

            'combo-public-ajax-page-cell',

            kwargs={'page_pk': page.id, 'cell_reference': tile.cell.get_reference()},

    test_add_to_dashboard(app, site)

    app.reset()  # logout

    tile = Tile.objects.all()[0]

        reverse('combo-dashboard-remove-tile', kwargs={'cell_reference': tile.cell.get_reference()}),

        status=403,

    )

    app = login(app)

        reverse('combo-dashboard-remove-tile', kwargs={'cell_reference': tile.cell.get_reference()}),

        status=302,

    )

    # add cells to dashboard

    app = login(app)

    for i in range(103, 108):

            reverse(

                'combo-dashboard-add-tile',

                kwargs={'cell_reference': TextCell.objects.get(order=i).get_reference()},

    app.reset()  # logout

    tiles = Tile.objects.filter(user=user).order_by('?')

    new_order = ','.join([str(x.id) for x in tiles])

        reverse('combo-dashboard-reorder-tiles', kwargs={'dashboard_id': dashboard.id}),

        params={'order': new_order},

        status=403,

    )

    app = login(app)

        reverse('combo-dashboard-reorder-tiles', kwargs={'dashboard_id': dashboard.id}),

        params={'order': new_order},

    )

    cell.statistic = Statistic.objects.get(slug='one-serie')

    cell.save()

    request = new_api_mock.call['requests'][0]

    assert 'time_interval=' not in request.url

    assert 'ou=' not in request.url

    cell.filter_params = {'time_interval': 'month', 'ou': 'default'}

    cell.save()

    request = new_api_mock.call['requests'][1]

    assert 'time_interval=month' in request.url

    assert 'ou=default' in request.url

    freezer.move_to(date)

    cell.time_range = 'previous-year'

    cell.save()

    request = new_api_mock.call['requests'][2]

    assert 'time_interval=month' in request.url

    assert 'ou=default' in request.url

    cell.time_range = 'current-week'

    cell.save()

    request = new_api_mock.call['requests'][-1]

    assert 'start=2020-03-02' in request.url and 'end=2020-03-09' in request.url

    cell.time_range = 'previous-week'

    cell.save()

    request = new_api_mock.call['requests'][-1]

    assert 'start=2020-02-24' in request.url and 'end=2020-03-02' in request.url

    cell.time_range = 'next-week'

    cell.save()

    request = new_api_mock.call['requests'][-1]

    assert 'start=2020-03-09' in request.url and 'end=2020-03-16' in request.url

    cell.time_range = 'range'

    cell.save()

    request = new_api_mock.call['requests'][-1]

    assert 'start' not in urllib.parse.parse_qs(urllib.parse.urlparse(request.url).query)

    assert 'end' not in urllib.parse.parse_qs(urllib.parse.urlparse(request.url).query)

    cell.time_range_start = '2020-10-01'

    cell.save()

    request = new_api_mock.call['requests'][-1]

    assert 'start=2020-10-01' in request.url

    cell.time_range_end = '2020-11-03'

    cell.save()

    request = new_api_mock.call['requests'][-1]

    assert 'start=2020-10-01' in request.url and 'end=2020-11-03' in request.url

    cell.save()

    freezer.move_to('2021-01-02')

    request = new_api_mock.call['requests'][0]

    assert 'start=2021-01-01' in request.url and 'end=2021-02-01' in request.url

    cell.time_range = 'previous-month'

    cell.save()

    request = new_api_mock.call['requests'][1]

    assert 'start=2020-12-01' in request.url and 'end=2021-01-01' in request.url

    freezer.move_to('2021-11-02')

    cell.time_range = 'next-month'

    cell.save()

    request = new_api_mock.call['requests'][2]

    assert 'start=2021-12-01' in request.url and 'end=2022-01-01' in request.url

    page.save()

    cell = LingoBasketLinkCell(page=page, placeholder='content', order=0)

    cell.save()

def test_tipi_cell():

        '/lingo/return-payment-backend/%s/' % regie.payment_backend.id

    )

    # simulate successful call to callback URL

        resp = app.get(get_url(with_payment_backend, 'lingo-callback', regie), params=args)

    assert resp.status_code == 200

    # simulate successful return URL

    url = '%s?NameId=%s&orig=wcs' % (reverse('api-add-basket-item'), user_name_id)

    url = sign_url(url, settings.LINGO_API_SIGN_KEY)

    resp = app.post_json(url, params=data)

    assert resp.status_code == 200

    response = json.loads(resp.text)

    assert response['result'] == 'success'

def test_payment_callback(app, basket_page, regie, user, with_payment_backend):

    page = Page(title='xxx', slug='index', template_name='standard')

    page.save()

        user=user, regie=regie, subject='test_item', amount='10.5', source_url='http://example.org/testitem/'

    )

    resp = login(app).get(basket_page.get_online_url())

    assert get_resp.status_code == 200

    assert Transaction.objects.get(order_id=transaction_id).status == 3

        user=user, regie=regie, subject='test_item', amount='11.5', source_url='http://example.org/testitem/'

    )

    resp = login(app).get(basket_page.get_online_url())

@pytest.mark.parametrize('with_payment_backend', [False, True])

def test_payment_callback_no_regie(app, basket_page, regie, user, with_payment_backend):

        user=user, regie=regie, subject='test_item', amount='10.5', source_url='http://example.org/testitem/'

    )

    resp = login(app).get(basket_page.get_online_url())

    assert get_resp.status_code == 200

    assert Transaction.objects.get(order_id=transaction_id).status == 3

        user=user, regie=regie, subject='test_item', amount='11.5', source_url='http://example.org/testitem/'

    )

    resp = login(app).get(basket_page.get_online_url())

def test_payment_return_without_query_string(app, basket_page, regie, user, with_payment_backend):

    page = Page(title='xxx', slug='index', template_name='standard')

    page.save()

        user=user, regie=regie, subject='test_item', amount='10.5', source_url='http://example.org/testitem/'

    )

    resp = login(app).get(basket_page.get_online_url())

    # payment status is obtained through callback

    callback_url = get_url(with_payment_backend, 'lingo-callback', regie)

        get_resp = app.get(callback_url, params=data)

    transaction = Transaction.objects.get(order_id=transaction_id)

    assert transaction.status == 3

@pytest.mark.parametrize('with_payment_backend', [False, True])

def test_payment_no_callback_just_return(caplog, app, basket_page, regie, user, with_payment_backend):

        user=user, regie=regie, subject='test_item', amount='10.5', source_url='http://example.org/testitem/'

    )

    resp = login(app).get(basket_page.get_online_url())

def test_request_payment_exception(app, basket_page, regie, user):

        user=user, regie=regie, subject='test_item', amount='10.5', source_url='http://example.org/testitem/'

    )

    # check if regie webservice has been correctly called

    assert mock_send.call_args[0][0].method == 'GET'

    url = mock_send.call_args[0][0].url

    assert scheme == 'http'

    assert netloc == 'example.org'

    assert path == '/regie/invoices/'

    # check if regie webservice has been correctly called

    assert mock_send.call_args[0][0].method == 'GET'

    url = mock_send.call_args[0][0].url

    assert scheme == 'http'

    assert netloc == 'example.org'

    assert path == '/regie/invoices/history/'

    mock_json = mock.Mock()

    mock_json.json.return_value = {'err': 0, 'data': INVOICES[0]}

    mock_get.return_value = mock_json

@mock.patch('combo.apps.lingo.models.requests.get')

    params = []

    new_order = [2, 3, 1, 4, 5]

        params.append(('ph_data_textcell-%s' % cell.id, 'content'))  # no placeholder change

        params.append(('pos_data_textcell-%s' % cell.id, str(new_order[i])))

    resp = app.get('/manage/assets/')

    assert 'have any asset yet.' in resp.text

        filepath = os.path.join(settings.CKEDITOR_UPLOAD_PATH, 'cc.png')

        pix = b'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABAQAAAAA3bvkkAAAACklEQVQI12NoAAAAggCB3UNq9AAAAABJRU5ErkJggg=='

        default_storage.save(filepath, BytesIO(base64.decodebytes(pix)))

    assert 'CKEditorFuncNum=0' in resp.request.url

    assert resp.text.count('<tr class="asset') == 2

        filepath = os.path.join(settings.CKEDITOR_UPLOAD_PATH, 'cc.png')

        pix = b'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABAQAAAAA3bvkkAAAACklEQVQI12NoAAAAggCB3UNq9AAAAABJRU5ErkJggg=='

        default_storage.save(filepath, BytesIO(base64.decodebytes(pix)))

    app = login(app)

        with pytest.raises(TemplateSyntaxError):

            cell.render({})

    params = []

    new_order = [2, 3, 1, 4, 5]

        params.append(('pos_%s' % item.pk, str(new_pos)))

    app = login(app)

@mock.patch('combo.utils.requests_wrapper.RequestsSession.request')

def test_notify_remote_items(mock_get, app, john_doe, jane_doe, regie, monkeypatch, freezer):

    freezer.move_to('2016-01-02')

    invoice_now = now()

    invoice_now = freezer()

    FAKE_PENDING_INVOICES = {

        page.parent_id = parent_id

        page.save()

        add_pages.num += 1

            add_pages(depth + 1, page.id)

    add_pages.num = 1

    page = Page(title='Home', slug='index', template_name='standard', public=False)

    page.save()

    app = login(app)

def test_page_skeleton(app):

    pwa_settings.push_notifications = False

    pwa_settings.save()

    with mock.patch('pywebpush.webpush') as webpush:

        assert webpush.call_count == 0

    pwa_settings.push_notifications = True

    pwa_settings.save()

    with mock.patch('pywebpush.webpush') as webpush:

        assert webpush.call_count == 1

        assert webpush.call_args[1]['subscription_info'] == {'sample': 'content'}

        PWA_VAPID_CLAIMS={'sub': 'mailto:admin@entrouvert.com'},

    ):

        with mock.patch('pywebpush.webpush') as webpush:

            assert webpush.call_count == 1

            assert webpush.call_args[1]['subscription_info'] == {'sample': 'content'}

            assert webpush.call_args[1]['vapid_private_key'] == settings.PWA_VAPID_PRIVATE_KEY

        assert len(resp.json['icons']) == 5

        # make sure largest icon is referenced in service worker

        assert resp.json['icons'][-1]['src'].split('/')[-1] in resp.text

        requests.get('/foo/bar/', remote_service=remote_service)

        url = send.call_args[0][0].url

        assert url.startswith('http://example.org/foo/bar/?')

        query = urlparse.parse_qs(querystring, keep_blank_values=True)

        assert query['orig'][0] == 'myself'

        assert query['email'][0] == ''

        requests.get('/foo/bar/', remote_service=remote_service, without_user=True)

        url = send.call_args[0][0].url

        assert url.startswith('http://example.org/foo/bar/?')

        query = urlparse.parse_qs(querystring, keep_blank_values=True)

        assert query['orig'][0] == 'myself'

        assert 'email' not in query

        requests.get('http://example.org/foo/bar/', remote_service='auto')

        url = send.call_args[0][0].url

        assert url.startswith('http://example.org/foo/bar/?')

        query = urlparse.parse_qs(querystring, keep_blank_values=True)

        assert query['orig'][0] == 'combo'

        assert check_query(querystring, 'combo') == True

        requests.get('/foo/bar/', remote_service=remote_service, user=user)

        url = send.call_args[0][0].url

        assert url.startswith('http://example.org/foo/bar/?')

        query = urlparse.parse_qs(querystring, keep_blank_values=True)

        assert query['NameID'][0] == 'r2d2'

        assert 'email' not in query

        requests.get('/foo/bar/', remote_service=remote_service, user=user, federation_key='email')

        url = send.call_args[0][0].url

        assert url.startswith('http://example.org/foo/bar/?')

        query = urlparse.parse_qs(querystring, keep_blank_values=True)

        assert query['email'][0] == 'foo@example.net'

        assert 'NameID' not in query

        requests.get('/foo/bar/', remote_service=remote_service, user=user)

        url = send.call_args[0][0].url

        assert url.startswith('http://example.org/foo/bar/?')

        query = urlparse.parse_qs(querystring, keep_blank_values=True)

        assert 'NameID' not in query

        assert query['email'][0] == 'foo@example.net'

        requests.get('/foo/bar/', remote_service=remote_service, user=user)

        url = send.call_args[0][0].url

        assert url.startswith('http://example.org/foo/bar/?')

        query = urlparse.parse_qs(querystring, keep_blank_values=True)

        assert query['NameID'][0] == ''

        assert query['email'][0] == ''

        with mock.patch('combo.utils.requests.get') as requests_get:

            data = {'data': []}

            requests_get.return_value = mock.Mock(json=lambda: data, status_code=200)

            assert requests_get.call_args[0][0] == 'http://www.example.net/search/foo/'

    params = []

    new_order = [2, 3, 1]

        params.append(('pos_%s' % service_slug, str(new_pos)))

    app = login(app)