Produits Entr'ouvert » Passerelle

# Generated by Django 2.2.28 on 2022-06-15 08:22

import uuid

import django.db.models.deletion

from django.db import migrations, models

class Migration(migrations.Migration):

    initial = True

    dependencies = [

        ('base', '0029_auto_20210202_1627'),

    ]

    operations = [

        migrations.CreateModel(

            name='Resource',

            fields=[

                (

                    'id',

                    models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),

                ),

                ('title', models.CharField(max_length=50, verbose_name='Title')),

                ('slug', models.SlugField(unique=True, verbose_name='Identifier')),

                ('description', models.TextField(verbose_name='Description')),

                (

                    'basic_auth_username',

                    models.CharField(

                        blank=True, max_length=128, verbose_name='Basic authentication username'

                    ),

                ),

                (

                    'basic_auth_password',

                    models.CharField(

                        blank=True, max_length=128, verbose_name='Basic authentication password'

                    ),

                ),

                (

                    'client_certificate',

                    models.FileField(

                        blank=True, null=True, upload_to='', verbose_name='TLS client certificate'

                    ),

                ),

                (

                    'trusted_certificate_authorities',

                    models.FileField(blank=True, null=True, upload_to='', verbose_name='TLS trusted CAs'),

                ),

                (

                    'verify_cert',

                    models.BooleanField(blank=True, default=True, verbose_name='TLS verify certificates'),

                ),

                (

                    'http_proxy',

                    models.CharField(blank=True, max_length=128, verbose_name='HTTP and HTTPS proxy'),

                ),

                (

                    'bbb_url',

                    models.URLField(

                        help_text='Base URL of Big Blue Button (use "bbb-conf --secret" to get it)',

                        max_length=400,

                        verbose_name='BBB URL',

                    ),

                ),

                (

                    'shared_secret',

                    models.CharField(

                        help_text='Shared ecret (use "bbb-conf --secret" to get it)', max_length=128

                    ),

                ),

                (

                    'users',

                    models.ManyToManyField(

                        blank=True,

                        related_name='_resource_users_+',

                        related_query_name='+',

                        to='base.ApiUser',

                    ),

                ),

            ],

            options={

                'verbose_name': 'Big Blue Button',

            },

        ),

        migrations.CreateModel(

            name='Meeting',

            fields=[

                (

                    'id',

                    models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),

                ),

                ('created', models.DateTimeField(auto_now_add=True, verbose_name='Created')),

                ('updated', models.DateTimeField(auto_now=True, verbose_name='Updated')),

                ('guid', models.UUIDField(default=uuid.uuid4, unique=True, verbose_name='UUID')),

                ('name', models.TextField(verbose_name='Name')),

                ('idempotent_id', models.TextField(unique=True, verbose_name='Idempotent ID')),

                ('running', models.BooleanField(default=False, verbose_name='Is running?')),

                ('last_time_running', models.DateTimeField(null=True, verbose_name='Last time running')),

                ('create_parameters', models.TextField(null=True, verbose_name='Create parameters')),

                (

                    'resource',

                    models.ForeignKey(

                        on_delete=django.db.models.deletion.CASCADE,

                        related_name='meetings',

                        to='bbb.Resource',

                        verbose_name='Resource',

                    ),

                ),

            ],

        ),

    ]

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2019  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# 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 json

import time

import uuid

from django.core.cache import cache

from django.db import models, transaction

from django.http import HttpResponseRedirect

from django.shortcuts import get_object_or_404

from django.urls import reverse

from django.utils.functional import cached_property

from django.utils.timezone import now

from django.utils.translation import gettext_lazy as _

from passerelle.base.models import BaseResource, HTTPResource

from passerelle.utils.api import endpoint

from passerelle.utils.jsonresponse import APIError

from . import utils

class Resource(BaseResource, HTTPResource):

    bbb_url = models.URLField(

        max_length=400,

        verbose_name=_('BBB URL'),

        help_text=_('Base URL of Big Blue Button (use "bbb-conf --secret" to get it)'),

    )

    shared_secret = models.CharField(

        max_length=128, help_text=_('Shared ecret (use "bbb-conf --secret" to get it)')

    )

    category = _('Business Process Connectors')

    class Meta:

        verbose_name = _('Big Blue Button')

    @cached_property

    def bbb(self):

        return utils.BBB(url=self.bbb_url, shared_secret=self.shared_secret, session=self.requests)

    def check_status(self):

        try:

            self.bbb.get_meetings()

        except utils.BBB.BBBError as e:

            raise APIError(e)

    @endpoint(

        methods=['post'],

        name='meeting',

        perm='can_access',

        description_post=_('Create a meeting'),

        post={

            'request_body': {

                'schema': {

                    'application/json': {

                        'type': 'object',

                        'properties': {

                            'name': {

                                'type': 'string',

                            },

                            'idempotent_id': {

                                'type': 'string',

                            },

                            'create_parameters': {

                                'type': 'object',

                                'properties': {

                                    'logoutUrl': {'type': 'string'},

                                },

                                'additionalProperties': True,

                            },

                        },

                        'required': ['name', 'idempotent_id'],

                        'unflattent': True,

                    }

                }

            }

        },

    )

    @transaction.atomic

    def meetings_endpoint(self, request, post_data):

        create_parameters = (

            json.dumps(post_data['create_parameters']) if post_data.get('create_parameters') else None

        )

        meeting, created = self.meetings.get_or_create(

            name=post_data['name'],

            idempotent_id=post_data['idempotent_id'],

            create_parameters=create_parameters,

        )

        try:

            meeting.create()

        except utils.BBB.BBBError as e:

            raise APIError(e)

        return {'data': meeting.to_json()}

    @endpoint(

        methods=['get'],

        name='meeting',

        perm='can_access',

        pattern=r'^(?P<guid>[0-9a-f]{32})/?$',

        example_pattern='{guid}/',

        description_post=_('Get a meeting'),

        parameters={

            'guid': {

                'description': _('Meeting guid'),

                'example_value': '7edb43abf2004f55a8a526ac4b1403e4',

            },

        },

    )

    def meeting_endpoint(self, request, guid):

        meeting = get_object_or_404(Meeting.objects.select_for_update(), guid=guid)

        return {'data': meeting.to_json()}

    @endpoint(

        methods=['get'],

        name='meeting',

        perm='can_access',

        pattern=r'^(?P<guid>[0-9a-f]{32})/is-running/?$',

        example_pattern='{guid}/is-running/',

        description_post=_('Report if meeting is running'),

        parameters={

            'guid': {

                'description': _('Meeting guid'),

                'example_value': '7edb43abf2004f55a8a526ac4b1403e4',

            },

        },

    )

    @transaction.atomic

    def is_running(self, request, guid):

        meeting = get_object_or_404(Meeting.objects.select_for_update(), guid=guid)

        if (now() - meeting.updated).total_seconds() > 5:

            meeting.update_is_running()

        return {'data': meeting.running}

    @endpoint(

        methods=['get'],

        name='meeting',

        perm='can_access',

        pattern=r'^(?P<guid>[0-9a-f]{32})/join/agent/?$',

        example_pattern='{guid}/join/agent/',

        description_post=_('Get a meeting'),

        parameters={

            'guid': {

                'description': _('Meeting guid'),

                'example_value': '7edb43abf2004f55a8a526ac4b1403e4',

            },

            'full_name': {

                'description': _('Agent full name'),

                'example_value': 'John Doe',

            },

        },

    )

    def join_agent(self, request, guid, full_name):

        meeting = get_object_or_404(Meeting, guid=guid)

        url = meeting.create().join_url(full_name, self.bbb.ROLE_MODERATOR)

        return HttpResponseRedirect(url)

    @endpoint(

        methods=['get'],

        name='meeting',

        perm='can_access',

        pattern=r'^(?P<guid>[0-9a-f]{32})/join/user/?$',

        example_pattern='{guid}/join/user/',

        description_post=_('Get a meeting'),

        parameters={

            'guid': {

                'description': _('Meeting guid'),

                'example_value': '7edb43abf2004f55a8a526ac4b1403e4',

            },

            'full_name': {

                'description': _('User full name'),

                'example_value': 'John Doe',

            },

        },

    )

    def join_user(self, request, guid, full_name):

        meeting = get_object_or_404(Meeting, guid=guid)

        url = meeting.create().join_url(full_name, self.bbb.ROLE_VIEWER)

        return HttpResponseRedirect(url)

    def _make_endpoint_url(self, endpoint, rest):

        return reverse(

            'generic-endpoint',

            kwargs={'connector': 'bbb', 'endpoint': endpoint, 'slug': self.slug, 'rest': rest},

        )

class Meeting(models.Model):

    created = models.DateTimeField(verbose_name=_('Created'), auto_now_add=True)

    updated = models.DateTimeField(verbose_name=_('Updated'), auto_now=True)

    resource = models.ForeignKey(

        Resource, verbose_name=_('Resource'), on_delete=models.CASCADE, related_name='meetings'

    )

    guid = models.UUIDField(verbose_name=_('UUID'), unique=True, default=uuid.uuid4)

    name = models.TextField(verbose_name=_('Name'))

    idempotent_id = models.TextField(verbose_name=_('Idempotent ID'), unique=True)

    running = models.BooleanField(verbose_name=_('Is running?'), default=False)

    last_time_running = models.DateTimeField(verbose_name=_('Last time running'), null=True)

    create_parameters = models.TextField('Create parameters', null=True)

    @property

    def join_user_url(self):

        return self.resource._make_endpoint_url(endpoint='meeting', rest=f'{self.guid.hex}/join/user/')

    @property

    def join_agent_url(self):

        return self.resource._make_endpoint_url(endpoint='meeting', rest=f'{self.guid.hex}/join/agent/')

    @property

    def url(self):

        return self.resource._make_endpoint_url(endpoint='meeting', rest=f'{self.guid.hex}/')

    @property

    def is_running_url(self):

        return self.resource._make_endpoint_url(endpoint='meeting', rest=f'{self.guid.hex}/')

    def to_json(self):

        return {

            'guid': self.guid.hex,

            'created': self.created,

            'updated': self.updated,

            'name': self.name,

            'idempotent_id': self.idempotent_id,

            'running': self.running,

            'last_time_running': self.last_time_running,

            'url': self.url,

            'join_user_url': self.join_user_url,

            'join_agent_url': self.join_agent_url,

            'is_running_url': self.is_running_url,

            'bbb_meeting_info': self.meeting_info(),

        }

    @property

    def meeting_id(self):

        return self.guid.hex

    def update_is_running(self):

        try:

            running = self.resource.bbb.is_meeting_running(self.meeting_id)['running'] == 'true'

        except self.resource.bbb.BBBError:

            return

        if self.running != running:

            self.running = running

            if running:

                self.last_time_running = now()

                self.save(update_fields=['updated', 'last_time_running', 'running'])

            else:

                self.save(update_fields=['updated', 'running'])

    @property

    def create_kwargs(self):

        return json.loads(self.create_parameters) if self.create_parameters else {}

    def create(self):

        return self.resource.bbb.meetings.create(

            name=self.name, meeting_id=self.meeting_id, **self.create_kwargs

        )

    @property

    def cache_key(self):

        return f'bbb_{self.resource.slug}_{self.guid.hex}'

    def meeting_info(self):

        start = time.time()

        try:

            data, timestamp = cache.get(self.cache_key)

        except TypeError:

            data = None

        if data is None or (start - timestamp) > 30:

            try:

                data = self.resource.bbb.meetings.get(meeting_id=self.meeting_id).attributes

            except utils.BBB.BBBError:

                data = data or {}

            cache.set(self.cache_key, (data, start))

        return data

{% extends "passerelle/manage/service_view.html" %}

{% load i18n passerelle %}

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2019  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# 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 hashlib

import re

import xml.etree.ElementTree as ET

import requests

from django.utils.http import urlencode

class BBB:

    CALL_CREATE = 'create'

    CALL_END = 'end'

    CALL_GET_MEETING_INFO = 'getMeetingInfo'

    CALL_GET_MEETINGS = 'getMeetings'

    CALL_IS_MEETING_RUNNING = 'isMeetingRunning'

    CALL_JOIN = 'join'

    PARAM_ATTENDEE_PW = 'attendeePW'

    PARAM_CREATE_TIME = 'createTime'

    PARAM_FULL_NAME = 'fullName'

    PARAM_MEETING_ID = 'meetingID'

    PARAM_MODERATOR_PW = 'moderatorPW'

    PARAM_NAME = 'name'

    PARAM_ROLE = 'role'

    TAG_RETURN_CODE = 'returncode'

    TAG_MESSAGE_KEY = 'messageKey'

    TAG_MESSAGE = 'message'

    RETURN_CODE_SUCCESS = 'SUCCESS'

    RETURN_CODE_FAILED = 'FAILED'

    RETURN_CODES = [RETURN_CODE_SUCCESS, RETURN_CODE_FAILED]

    ROLE_MODERATOR = 'MODERATOR'

    ROLE_VIEWER = 'VIEWER'

    ROLES = [ROLE_MODERATOR, ROLE_VIEWER]

    MESSAGE_KEY_ID_NOT_UNIQUE = 'idNotUnique'

    MESSAGE_KEY_NOT_FOUND = 'notFound'

    MESSAGE_KEY_CHECKSUM_ERROR = 'checksumError'

    MESSAGE_KEY_SENT_END_MEETING_REQUEST = 'sentEndMeetingRequest'

    class BBBError(Exception):

        pass

    class FailedError(BBBError):

        def __init__(self, message, message_key=None):

            self.message_key = message_key

            super().__init__(message)

        def __repr__(self):

            return f'<{self.__class__.__name__} "{self}" message_key={self.message_key}>'

    class NotFoundError(FailedError):

        pass

    class IdNotUniqueError(FailedError):

        pass

    class ChecksumError(FailedError):

        pass

    MESSAGE_KEY_TO_CLASS = {

        MESSAGE_KEY_ID_NOT_UNIQUE: IdNotUniqueError,

        MESSAGE_KEY_NOT_FOUND: NotFoundError,

        MESSAGE_KEY_CHECKSUM_ERROR: ChecksumError,

    }

    @staticmethod

    def camel_to_snake(string):

        string = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', string)

        string = re.sub('(.)([0-9]+)', r'\1_\2', string)

        return re.sub('([a-z0-9])([A-Z])', r'\1_\2', string).lower()

    def __init__(self, *, url, shared_secret, session=None):

        if not url.endswith('/'):

            raise self.BBBError(f'expected url ending with a slash, got {url!r}')

        self.url = url

        self.shared_secret = shared_secret

        self.http_session = session or requests.Session()

        self.meetings = self.Meetings(bbb=self)

    def _build_query(self, name, parameters):

        """Method to create valid API query

        to call on BigBlueButton. Because each query

        should have a encrypted checksum based on request Data.

        """

        query = urlencode(parameters)

        prepared = f'{name}{query}{self.shared_secret}'

        checksum = hashlib.sha1(prepared.encode('utf-8')).hexdigest()

        sep = '&' if query else ''

        result = f'{query}{sep}checksum={checksum}'

        return result

    def _http_get(self, url):

        try:

            raw_response = self.http_session.get(url)

            raw_response.raise_for_status()

        except requests.RequestException as e:

            raise self.BBBError('transport error', e)

        return raw_response

    @classmethod

    def _parse_xml_doc(cls, root, data):

        for child in root:

            if len(child) == 0:

                data[child.tag.split('}')[-1]] = child.text

            elif max(len(sub) for sub in child) == 0:

                # subobject case

                data[child.tag] = cls._parse_xml_doc(child, {})

            else:

                # array case

                data[child.tag] = [cls._parse_xml_doc(sub, {}) for sub in child]

        return data

    @classmethod

    def _parse_response(cls, response):

        try:

            root = ET.fromstring(response.content)

            if root.tag != 'response':

                raise ValueError('root tag is not response')

            data = {}

            cls._parse_xml_doc(root, data)

            if cls.TAG_RETURN_CODE not in data:

                raise ValueError(f'expected a return code, got {data!r}')

            return_code = data.get(cls.TAG_RETURN_CODE)

            if return_code not in cls.RETURN_CODES:

                raise ValueError(f'expected a return code, got {data!r}')

            data.pop(cls.TAG_RETURN_CODE)

            return return_code, data

        except Exception as e:

            raise cls.BBBError('invalid response', e)

    def _make_url(self, name, parameters):

        query = self._build_query(name, parameters)

        return f'{self.url}api/{name}?{query}'

    def _api_call(self, name, parameters):

        url = self._make_url(name, parameters)

        raw_response = self._http_get(url)

        return_code, data = self._parse_response(raw_response)

        if return_code == self.RETURN_CODE_FAILED:

            exception_class = self.MESSAGE_KEY_TO_CLASS.get(data.get(self.TAG_MESSAGE_KEY), self.FailedError)

            raise exception_class(

                message=data.get(self.TAG_MESSAGE, 'NO_MESSAGE_KEY'),

                message_key=data.get(self.TAG_MESSAGE_KEY),

            )

        return data

    def create_meeting(self, name, meeting_id, attendee_pw=None, moderator_pw=None, **kwargs):

        parameters = {

            self.PARAM_NAME: name,

            self.PARAM_MEETING_ID: meeting_id,

            self.PARAM_ATTENDEE_PW: (

                attendee_pw

                or hashlib.sha1((self.shared_secret + meeting_id + 'attendee').encode()).hexdigest()

            ),

            self.PARAM_MODERATOR_PW: (

                moderator_pw

                or hashlib.sha1((self.shared_secret + meeting_id + 'moderator').encode()).hexdigest()

            ),

        }

        return self._api_call(self.CALL_CREATE, dict(parameters, **kwargs))

    def get_meeting_info(self, meeting_id):

        parameters = {

            self.PARAM_MEETING_ID: meeting_id,

        }

        return self._api_call(self.CALL_GET_MEETING_INFO, parameters)

    def is_meeting_running(self, meeting_id):

        parameters = {

            self.PARAM_MEETING_ID: meeting_id,

        }

        try:

            return self._api_call(self.CALL_IS_MEETING_RUNNING, parameters)

        except self.NotFoundError:

            return False

    def end_meeting(self, meeting_id):

        parameters = {

            self.PARAM_MEETING_ID: meeting_id,

        }

        try:

            response = self._api_call(self.CALL_END, parameters)

        except self.NotFoundError:

            return False

        if response[self.TAG_MESSAGE_KEY] == self.MESSAGE_KEY_SENT_END_MEETING_REQUEST:

            return True

        raise self.BBBError(f'expected notFound or sentEndMeetingRequest, got {response!r}')

    def get_meetings(self):

        return self._api_call(self.CALL_GET_MEETINGS, {})

    def make_join_url(self, full_name, meeting_id, role, create_time=None, **kwargs):

        if role not in self.ROLES:

            raise self.BBBError(f'expected a role value, got {role!r}')

        parameters = {

            self.PARAM_FULL_NAME: full_name,

            self.PARAM_MEETING_ID: meeting_id,

            self.PARAM_ROLE: role,

        }

        if create_time is not None:

            parameters[self.PARAM_CREATE_TIME] = create_time

        return self._make_url(self.CALL_JOIN, parameters)

    class Meeting:

        is_running = False

        meeting_name = None

        meeting_id = None

        def __init__(self, bbb: 'BBB', **kwargs):

            self.bbb = bbb

            # came to snake case

            parameters = {BBB.camel_to_snake(key): value for key, value in kwargs.items()}

            self.message = {k: v for k, v in parameters.items() if k.startswith('message')}

            self.meeting_id = parameters.pop('meeting_id')

            self.meeting_name = parameters.pop('meeting_name')

            self.attributes = {k: v for k, v in parameters.items() if not k.startswith('message')}

        def update(self):

            data = self.bbb.get_meeting_info(self.meeting_id)

            parameters = {

                BBB.camel_to_snake(key): value

                for key, value in data.items()

                if key not in ['meetingID', 'meetingName']

            }

            self.message = {k: v for k, v in parameters.items() if k.startswith('message')}

            self.attributes = {k: v for k, v in parameters.items() if not k.startswith('message')}

        def update_is_running(self):

            self.is_running = self.bbb.is_meeting_running(self.meeting_id)['running'] == 'true'

            return self.is_running

        def end(self):

            try:

                return self.bbb.end_meeting(self.meeting_id)

            finally:

                self.is_running = False

        def join_url(self, full_name, role):

            return self.bbb.make_join_url(

                full_name=full_name,

                role=role,

                meeting_id=self.meeting_id,

                create_time=self.attributes['create_time'],

            )

        def to_dict(self):

            return {k: v for k, v in self.__dict__.items() if k not in ['bbb']}

        def __repr__(self):

            return f'<Meeting name={self.meeting_name} id={self.meeting_id}>'

    class Meetings:

        def __init__(self, bbb: 'BBB'):

            self.bbb = bbb

        def create(self, name, meeting_id, **kwargs):

            return BBB.Meeting(

                self.bbb,

                meeting_name=name,

                **self.bbb.create_meeting(name=name, meeting_id=meeting_id, **kwargs),

            )

        def get(self, meeting_id):

            return BBB.Meeting(self.bbb, **self.bbb.get_meeting_info(meeting_id=meeting_id))

        def all(self):

            response = self.bbb.get_meetings()

            for response_meeting in response.get('meetings', []):

                yield BBB.Meeting(self.bbb, **response_meeting)

    'passerelle.apps.bbb',

# Copyright (C) 2021  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# 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 datetime

import json

import uuid

import mock

import pytest

import passerelle.apps.bbb.utils as bbb_utils

from . import utils

BBB_URL = 'https://example.com/bigbluebutton/'

SHARED_SECRET = 'ABCD'

SLUG = 'test'

MEETING_NAME = 'RDV'

MEETING_ID = 'a' * 32

UUID = uuid.UUID(MEETING_ID)

IDEMPOTENT_ID = '10-1999'

LOGOUT_URL = 'https://portal/'

@pytest.fixture

def connector(db):

    from passerelle.apps.bbb.models import Resource

    return utils.setup_access_rights(

        Resource.objects.create(

            slug=SLUG,

            bbb_url=BBB_URL,

            shared_secret=SHARED_SECRET,

        )

    )

class TestManage:

    pytestmark = pytest.mark.django_db

    @pytest.fixture

    def app(self, app, admin_user):

        from .test_manager import login

        login(app)

        return app

    def test_homepage(self, app, connector):

        app.get(f'/bbb/{SLUG}/')

class TestBBB:

    @pytest.fixture

    def bbb(self):

        return bbb_utils.BBB(url=BBB_URL, shared_secret=SHARED_SECRET)

    @pytest.fixture

    def mock(self):

        with mock.patch('requests.Session.send') as requests_send:

            requests_send.return_value = mock.Mock()

            yield requests_send

    def test_create_failure(self, bbb, mock):

        mock.return_value.status_code = 200

        mock.return_value.content = '<response/>'

        with pytest.raises(bbb.BBBError):

            bbb.create_meeting(name=MEETING_NAME, meeting_id=MEETING_ID)

    class TestCreate:

        @pytest.fixture

        def mock(self, mock):

            mock.return_value.content = '''<response>

    <returncode>SUCCESS</returncode>

    <message>ok</message>

    <meetingID>aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa</meetingID>

</response>'''

            return mock

        def test_create(self, bbb, mock):

            result = bbb.create_meeting(name=MEETING_NAME, meeting_id=MEETING_ID)

            prepared_request = mock.call_args[0][0]

            assert prepared_request.url == (

                'https://example.com/bigbluebutton/api/create?'

                'name=RDV&meetingID=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa&attendeePW=6256459f2baead794d599a1ad21a07fd4bc5743a'

                '&moderatorPW=2ebb4732fcbdd5a9f0751c3ffa7a5f51f755f980&checksum=4ac7d65fe6beb6f86fa5a13d6d8e0d1fadee0b50'

            )

            assert result == {'meetingID': MEETING_ID, 'message': 'ok'}

        def test_meetings_create(self, bbb, mock):

            meeting = bbb.meetings.create(name=MEETING_NAME, meeting_id=MEETING_ID)

            prepared_request = mock.call_args[0][0]

            assert prepared_request.url == (

                'https://example.com/bigbluebutton/api/create?'

                'name=RDV&meetingID=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa&attendeePW=6256459f2baead794d599a1ad21a07fd4bc5743a'

                '&moderatorPW=2ebb4732fcbdd5a9f0751c3ffa7a5f51f755f980&checksum=4ac7d65fe6beb6f86fa5a13d6d8e0d1fadee0b50'

            )

            assert meeting.meeting_name == MEETING_NAME

            assert meeting.meeting_id == MEETING_ID

            assert meeting.attributes == {}

            assert meeting.message == {'message': 'ok'}

    class TestGetMeetingInfo:

        @pytest.fixture

        def mock(self, mock):

            mock.return_value.content = '''<response>

    <returncode>SUCCESS</returncode>

    <meetingName>RDV</meetingName>

    <message>ok</message>

    <meetingID>aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa</meetingID>

    <metadata>

      <coin>1</coin>

    </metadata>

    <others>

      <other>

        <a>1</a>

      </other>

    </others>

</response>'''

            return mock

        def test_get_meeting_info(self, bbb, mock):

            result = bbb.get_meeting_info(meeting_id=MEETING_ID)

            prepared_request = mock.call_args[0][0]

            assert (

                prepared_request.url == 'https://example.com/bigbluebutton/api/getMeetingInfo'

                '?meetingID=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa&checksum=4d71906c30e6e62d8b9f1a3e57233bb64126e0da'

            )

            assert result == {

                'meetingName': MEETING_NAME,

                'meetingID': MEETING_ID,

                'message': 'ok',

                'metadata': {'coin': '1'},

                'others': [{'a': '1'}],

            }

        def test_meetings_get(self, bbb, mock):

            meeting = bbb.meetings.get(meeting_id=MEETING_ID)

            prepared_request = mock.call_args[0][0]

            assert (

                prepared_request.url == 'https://example.com/bigbluebutton/api/getMeetingInfo'

                '?meetingID=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa&checksum=4d71906c30e6e62d8b9f1a3e57233bb64126e0da'

            )

            assert meeting.meeting_name == MEETING_NAME

            assert meeting.meeting_id == MEETING_ID

            assert meeting.attributes == {

                'metadata': {'coin': '1'},

                'others': [{'a': '1'}],

            }

            assert meeting.message == {'message': 'ok'}

    class TestMakeJoinURL:

        def test_make_join_url(self, bbb):

            assert bbb.make_join_url(

                meeting_id=MEETING_ID, full_name='John Doe', role=bbb.ROLE_MODERATOR

            ) == (

                'https://example.com/bigbluebutton/api/join?fullName=John+Doe'

                '&meetingID=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa&role=MODERATOR'

                '&checksum=838762e70a6fe3257ea81f7b27acbde9945acaf9'

            )

        def test_meeting_join_url(self, bbb):

            meeting = bbb.Meeting(bbb, meeting_name=None, meeting_id=MEETING_ID, create_time=1234)

            assert meeting.join_url(full_name='John Doe', role=bbb.ROLE_MODERATOR) == (

                'https://example.com/bigbluebutton/api/join?fullName=John+Doe'

                '&meetingID=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa&role=MODERATOR'

                '&createTime=1234&checksum=07a7e7fd233dd213bd6dfc5332cc842afac56ba5'

            )

    class TestEndMeeting:

        @pytest.fixture

        def mock(self, mock):

            mock.return_value.content = '''<response>

    <returncode>SUCCESS</returncode>

    <messageKey>sentEndMeetingRequest</messageKey>

</response>'''

            return mock

        def test_end_meeting(self, bbb, mock):

            result = bbb.end_meeting(meeting_id=MEETING_ID)

            prepared_request = mock.call_args[0][0]

            assert (

                prepared_request.url == 'https://example.com/bigbluebutton/api/end'

                '?meetingID=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa&checksum=ed94e9d091f0a105e47d91f7f0633b6f7b881698'

            )

            assert result is True

        def test_meeting_end(self, bbb, mock):

            meeting = bbb.Meeting(bbb, meeting_name=None, meeting_id=MEETING_ID)

            result = meeting.end()

            prepared_request = mock.call_args[0][0]

            assert (

                prepared_request.url == 'https://example.com/bigbluebutton/api/end'

                '?meetingID=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa&checksum=ed94e9d091f0a105e47d91f7f0633b6f7b881698'

            )

            assert result is True

    class TestEndMeetingIfNotFound:

        @pytest.fixture

        def mock(self, mock):

            mock.return_value.content = '''<response>

    <returncode>FAILED</returncode>

    <messageKey>notFound</messageKey>

</response>'''

            return mock

        def test_end_meeting(self, bbb, mock):

            result = bbb.end_meeting(meeting_id=MEETING_ID)

            prepared_request = mock.call_args[0][0]

            assert (

                prepared_request.url == 'https://example.com/bigbluebutton/api/end'

                '?meetingID=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa&checksum=ed94e9d091f0a105e47d91f7f0633b6f7b881698'

            )

            assert result is False

    class TestGetMeetings:

        @pytest.fixture

        def mock(self, mock):

            mock.return_value.content = '''<response>

    <returncode>SUCCESS</returncode>

    <meetings>

      <meeting>

        <meetingName>RDV</meetingName>

        <message>ok</message>

        <meetingID>aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa</meetingID>

        <metadata>

          <coin>1</coin>

        </metadata>

        <others>

          <other>

            <a>1</a>

          </other>

        </others>

      </meeting>

    </meetings>

</response>'''

            return mock

        def test_get_meetings(self, bbb, mock):

            result = bbb.get_meetings()

            prepared_request = mock.call_args[0][0]

            assert (

                prepared_request.url

                == 'https://example.com/bigbluebutton/api/getMeetings?checksum=64bf453f633be1f9d2c7b0f6bed4a4702dbcc41e'

            )

            assert result == {

                'meetings': [

                    {

                        'meetingID': 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa',

                        'meetingName': 'RDV',

                        'message': 'ok',

                        'metadata': {'coin': '1'},

                        'others': [{'a': '1'}],

                    }

                ]

            }

        def test_meetings_all(self, bbb, mock):

            result = list(bbb.meetings.all())

            prepared_request = mock.call_args[0][0]

            assert (

                prepared_request.url

                == 'https://example.com/bigbluebutton/api/getMeetings?checksum=64bf453f633be1f9d2c7b0f6bed4a4702dbcc41e'

            )

            assert len(result) == 1

            meeting = result[0]

            assert meeting.meeting_name == MEETING_NAME

            assert meeting.meeting_id == MEETING_ID

            assert meeting.attributes == {

                'metadata': {'coin': '1'},

                'others': [{'a': '1'}],

            }

            assert meeting.message == {'message': 'ok'}

class TestAPI:

    @pytest.fixture(autouse=True)

    def stable(self, freezer):

        from passerelle.apps.bbb.models import Meeting

        freezer.move_to('2022-01-01T12:00:00Z')

        with mock.patch.object(Meeting._meta.get_field('guid'), 'default', UUID):

            yield None

    @pytest.fixture

    def meetings_create(self):

        with mock.patch('passerelle.apps.bbb.utils.BBB.Meetings.create') as create:

            yield create

    @pytest.fixture

    def meetings_get(self):

        with mock.patch('passerelle.apps.bbb.utils.BBB.Meetings.get') as get:

            yield get

    @pytest.fixture

    def is_meeting_running(self):

        with mock.patch('passerelle.apps.bbb.utils.BBB.is_meeting_running') as is_running:

            yield is_running

    class TestCreate:

        def test_ok(self, app, connector, meetings_create, meetings_get):

            meeting = bbb_utils.BBB.Meeting(

                None, meeting_id=MEETING_ID, meeting_name=MEETING_NAME, logout_url=LOGOUT_URL

            )

            meetings_create.return_value = meeting

            meetings_get.return_value = meeting

            response = app.post_json(

                f'/bbb/{SLUG}/meeting',

                params={

                    'name': MEETING_NAME,

                    'idempotent_id': IDEMPOTENT_ID,

                    'create_parameters': {

                        'logoutUrl': LOGOUT_URL,

                    },

                },

            )

            assert meetings_create.call_args == mock.call(

                logoutUrl=LOGOUT_URL, meeting_id=MEETING_ID, name=MEETING_NAME

            )

            assert response.json['err'] == 0

            assert response.json['data'] == {

                'created': '2022-01-01T12:00:00Z',

                'updated': '2022-01-01T12:00:00Z',

                'guid': MEETING_ID,

                'idempotent_id': IDEMPOTENT_ID,

                'url': f'/bbb/test/meeting/{MEETING_ID}/',

                'is_running_url': f'/bbb/test/meeting/{MEETING_ID}/',

                'join_agent_url': f'/bbb/test/meeting/{MEETING_ID}/join/agent/',

                'join_user_url': f'/bbb/test/meeting/{MEETING_ID}/join/user/',

                'last_time_running': None,

                'name': 'RDV',

                'running': False,

                'bbb_meeting_info': {

                    'logout_url': LOGOUT_URL,

                },

            }

            meeting = connector.meetings.get()

            assert meeting.meeting_id == MEETING_ID

            assert meeting.name == MEETING_NAME

            assert meeting.create_kwargs == {'logoutUrl': LOGOUT_URL}

        def test_error(self, app, connector, meetings_create):

            meetings_create.side_effect = bbb_utils.BBB.BBBError('coin')

            response = app.post_json(

                f'/bbb/{SLUG}/meeting',

                params={

                    'name': MEETING_NAME,

                    'idempotent_id': IDEMPOTENT_ID,

                    'create_parameters': {

                        'logoutUrl': LOGOUT_URL,

                    },

                },

            )

            assert response.json['err'] == 1

            assert response.json['err_desc'] == 'coin'

    class TestGet:

        def test_ok(self, app, connector, meetings_get, freezer):

            connector.meetings.create(

                guid=UUID,

                idempotent_id=IDEMPOTENT_ID,

                name=MEETING_NAME,

                create_parameters=json.dumps({'logoutUrl': LOGOUT_URL}),

            )

            meeting = bbb_utils.BBB.Meeting(

                bbb=connector.bbb,

                meeting_id=MEETING_ID,

                meeting_name=MEETING_NAME,

                logout_url=LOGOUT_URL,

                create_time=1234,

            )

            meetings_get.return_value = meeting

            response = app.get(f'/bbb/{SLUG}/meeting/{MEETING_ID}/')

            assert response.json['err'] == 0

            assert response.json['data'] == {

                'created': '2022-01-01T12:00:00Z',

                'updated': '2022-01-01T12:00:00Z',

                'guid': MEETING_ID,

                'idempotent_id': IDEMPOTENT_ID,

                'url': f'/bbb/test/meeting/{MEETING_ID}/',

                'is_running_url': f'/bbb/test/meeting/{MEETING_ID}/',

                'join_agent_url': f'/bbb/test/meeting/{MEETING_ID}/join/agent/',

                'join_user_url': f'/bbb/test/meeting/{MEETING_ID}/join/user/',

                'last_time_running': None,

                'name': 'RDV',

                'running': False,

                'bbb_meeting_info': {

                    'create_time': 1234,

                    'logout_url': LOGOUT_URL,

                },

            }

            # test cache

            freezer.move_to(datetime.timedelta(seconds=100))

            meetings_get.side_effect = bbb_utils.BBB.BBBError('boom!')

            response = app.get(f'/bbb/{SLUG}/meeting/{MEETING_ID}/')

            assert response.json['err'] == 0

            assert response.json['data'] == {

                'created': '2022-01-01T12:00:00Z',

                'updated': '2022-01-01T12:00:00Z',

                'guid': MEETING_ID,

                'idempotent_id': IDEMPOTENT_ID,

                'url': f'/bbb/test/meeting/{MEETING_ID}/',

                'is_running_url': f'/bbb/test/meeting/{MEETING_ID}/',

                'join_agent_url': f'/bbb/test/meeting/{MEETING_ID}/join/agent/',

                'join_user_url': f'/bbb/test/meeting/{MEETING_ID}/join/user/',

                'last_time_running': None,

                'name': 'RDV',

                'running': False,

                'bbb_meeting_info': {

                    'create_time': 1234,

                    'logout_url': LOGOUT_URL,

                },

            }

    class TestIsRunning:

        def test_ok(self, app, connector, meetings_get, is_meeting_running, freezer):

            connector.meetings.create(

                guid=UUID,

                idempotent_id=IDEMPOTENT_ID,

                name=MEETING_NAME,

                create_parameters=json.dumps({'logoutUrl': LOGOUT_URL}),

            )

            meeting = bbb_utils.BBB.Meeting(

                bbb=connector.bbb,

                meeting_id=MEETING_ID,

                meeting_name=MEETING_NAME,

                logout_url=LOGOUT_URL,

                create_time=1234,

            )

            freezer.move_to(datetime.timedelta(seconds=6))

            meetings_get.return_value = meeting

            is_meeting_running.return_value = {'running': 'true'}

            response = app.get(f'/bbb/{SLUG}/meeting/{MEETING_ID}/is-running/')

            assert response.json['err'] == 0

            assert response.json['data'] is True

            assert connector.meetings.get().running is True

            # test cache

            is_meeting_running.return_value = {'running': 'false'}

            is_meeting_running.side_effect = bbb_utils.BBB.BBBError('boom!')

            response = app.get(f'/bbb/{SLUG}/meeting/{MEETING_ID}/is-running/')

            assert response.json['err'] == 0

            assert response.json['data'] is True

            # test cache expired

            freezer.move_to(datetime.timedelta(seconds=6))

            response = app.get(f'/bbb/{SLUG}/meeting/{MEETING_ID}/is-running/')

            assert response.json['err'] == 0

            assert response.json['data'] is True

            is_meeting_running.side_effect = None

            response = app.get(f'/bbb/{SLUG}/meeting/{MEETING_ID}/is-running/')

            assert response.json['err'] == 0

            assert response.json['data'] is False

    class TestJoin:

        def test_join_user(self, app, connector, meetings_create):

            connector.meetings.create(

                guid=UUID, name=MEETING_NAME, create_parameters=json.dumps({'logoutUrl': LOGOUT_URL})

            )

            meeting = bbb_utils.BBB.Meeting(

                bbb=connector.bbb,

                meeting_id=MEETING_ID,

                meeting_name=MEETING_NAME,

                logout_url=LOGOUT_URL,

                create_time=1234,

            )

            meetings_create.return_value = meeting

            response = app.get(

                f'/bbb/{SLUG}/meeting/{MEETING_ID}/join/user/', params={'full_name': 'John Doe'}

            )

            assert meetings_create.call_args == mock.call(

                logoutUrl=LOGOUT_URL, meeting_id=MEETING_ID, name=MEETING_NAME

            )

            assert response.location == (

                'https://example.com/bigbluebutton/api/join'

                '?fullName=John+Doe&meetingID=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'

                '&role=VIEWER&createTime=1234&checksum=1851e34e269a5063605bc64589ea53ded2f9744d'

            )

        def test_join_agent(self, app, connector, meetings_create):

            connector.meetings.create(

                guid=UUID, name=MEETING_NAME, create_parameters=json.dumps({'logoutUrl': LOGOUT_URL})

            )

            meeting = bbb_utils.BBB.Meeting(

                bbb=connector.bbb,

                meeting_id=MEETING_ID,

                meeting_name=MEETING_NAME,

                logout_url=LOGOUT_URL,

                create_time=1234,

            )

            meetings_create.return_value = meeting

            response = app.get(

                f'/bbb/{SLUG}/meeting/{MEETING_ID}/join/agent/', params={'full_name': 'John Doe'}

            )

            assert meetings_create.call_args == mock.call(

                logoutUrl=LOGOUT_URL, meeting_id=MEETING_ID, name=MEETING_NAME

            )

            assert response.location == (

                'https://example.com/bigbluebutton/api/join'

                '?fullName=John+Doe&meetingID=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'

                '&role=MODERATOR&createTime=1234&checksum=07a7e7fd233dd213bd6dfc5332cc842afac56ba5'

            )