618 lines
24 KiB
Python
618 lines
24 KiB
Python
|
# 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 uuid
|
||
|
from unittest 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 TestIsMeetingRunning:
|
||
|
def test_ok(self, bbb, mock):
|
||
|
mock.return_value.content = '''<response>
|
||
|
<returncode>SUCCESS</returncode>
|
||
|
<running>true</running>
|
||
|
</response>'''
|
||
|
assert bbb.is_meeting_running(meeting_id=MEETING_ID)
|
||
|
mock.return_value.content = '''<response>
|
||
|
<returncode>SUCCESS</returncode>
|
||
|
<running>false</running>
|
||
|
</response>'''
|
||
|
assert not bbb.is_meeting_running(meeting_id=MEETING_ID)
|
||
|
|
||
|
def test_not_found(self, bbb, mock):
|
||
|
mock.return_value.content = '''<response>
|
||
|
<returncode>FAILED</returncode>
|
||
|
<messageKey>notFound</messageKey>
|
||
|
</response>'''
|
||
|
assert not bbb.is_meeting_running(meeting_id=MEETING_ID)
|
||
|
|
||
|
def test_invalid_response(self, bbb, mock):
|
||
|
mock.return_value.content = '''<response>
|
||
|
<returncode>SUCCESS</returncode>
|
||
|
</response>'''
|
||
|
with pytest.raises(bbb.InvalidResponseError):
|
||
|
assert not bbb.is_meeting_running(meeting_id=MEETING_ID)
|
||
|
|
||
|
|
||
|
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 meeting_end(self):
|
||
|
with mock.patch('passerelle.apps.bbb.utils.BBB.Meeting.end') as end:
|
||
|
yield end
|
||
|
|
||
|
@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_normal(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,
|
||
|
'metadata': {
|
||
|
'formdata_url': 'https://wcs/form/1',
|
||
|
'start_date': '2022-01-10T10:30:00',
|
||
|
},
|
||
|
},
|
||
|
)
|
||
|
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'http://testserver/bbb/test/meeting/{MEETING_ID}/',
|
||
|
'is_running_url': f'http://testserver/bbb/test/meeting/{MEETING_ID}/is-running/',
|
||
|
'join_agent_url': f'http://testserver/bbb/test/meeting/{MEETING_ID}/join/agent/2b2111/',
|
||
|
'join_user_url': f'http://testserver/bbb/test/meeting/{MEETING_ID}/join/user/fb918f/',
|
||
|
'last_time_running': None,
|
||
|
'name': 'RDV',
|
||
|
'running': False,
|
||
|
'bbb_meeting_info': {
|
||
|
'logout_url': LOGOUT_URL,
|
||
|
},
|
||
|
'create_parameters': {'logoutUrl': 'https://portal/'},
|
||
|
'join_agent_parameters': None,
|
||
|
'join_user_parameters': None,
|
||
|
'metadata': {
|
||
|
'formdata_url': 'https://wcs/form/1',
|
||
|
'start_date': '2022-01-10T10:30:00',
|
||
|
},
|
||
|
}
|
||
|
meeting = connector.meetings.get()
|
||
|
assert meeting.meeting_id == MEETING_ID
|
||
|
assert meeting.name == MEETING_NAME
|
||
|
assert meeting.create_parameters == {'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={'logoutUrl': LOGOUT_URL},
|
||
|
metadata={
|
||
|
'formdata_url': 'https://wcs/form/1',
|
||
|
'start_date': '2022-01-10T10:30:00',
|
||
|
},
|
||
|
)
|
||
|
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'http://testserver/bbb/test/meeting/{MEETING_ID}/',
|
||
|
'is_running_url': f'http://testserver/bbb/test/meeting/{MEETING_ID}/is-running/',
|
||
|
'join_agent_url': f'http://testserver/bbb/test/meeting/{MEETING_ID}/join/agent/2b2111/',
|
||
|
'join_user_url': f'http://testserver/bbb/test/meeting/{MEETING_ID}/join/user/fb918f/',
|
||
|
'last_time_running': None,
|
||
|
'name': 'RDV',
|
||
|
'running': False,
|
||
|
'bbb_meeting_info': {
|
||
|
'create_time': 1234,
|
||
|
'logout_url': LOGOUT_URL,
|
||
|
},
|
||
|
'create_parameters': {'logoutUrl': 'https://portal/'},
|
||
|
'join_agent_parameters': None,
|
||
|
'join_user_parameters': None,
|
||
|
'metadata': {
|
||
|
'formdata_url': 'https://wcs/form/1',
|
||
|
'start_date': '2022-01-10T10:30:00',
|
||
|
},
|
||
|
}
|
||
|
# test meeting info 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']['bbb_meeting_info']
|
||
|
|
||
|
class TestDelete:
|
||
|
def test_ok(self, app, connector, meetings_create, meeting_end, freezer):
|
||
|
connector.meetings.create(
|
||
|
guid=UUID,
|
||
|
idempotent_id=IDEMPOTENT_ID,
|
||
|
name=MEETING_NAME,
|
||
|
create_parameters={'logoutUrl': LOGOUT_URL},
|
||
|
metadata={
|
||
|
'formdata_url': 'https://wcs/form/1',
|
||
|
'start_date': '2022-01-10T10:30:00',
|
||
|
},
|
||
|
)
|
||
|
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
|
||
|
assert connector.meetings.count() == 1
|
||
|
assert meeting_end.call_count == 0
|
||
|
response = app.delete(f'/bbb/{SLUG}/meeting/{MEETING_ID}/')
|
||
|
assert response.json['err'] == 0
|
||
|
assert connector.meetings.count() == 0
|
||
|
assert meeting_end.call_count == 1
|
||
|
|
||
|
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={'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 = True
|
||
|
response = app.get(f'/bbb/{SLUG}/meeting/{MEETING_ID}/is-running/')
|
||
|
assert response['Access-Control-Allow-Origin'] == '*'
|
||
|
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 = 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):
|
||
|
model_meeting = connector.meetings.create(
|
||
|
guid=UUID, name=MEETING_NAME, create_parameters={'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/{model_meeting.user_key}/',
|
||
|
)
|
||
|
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=User&meetingID=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'
|
||
|
'&role=VIEWER&createTime=1234&checksum=d8d1fb7bd4369146c71e697fba5f6104ec7a353e'
|
||
|
)
|
||
|
|
||
|
# full name through query string
|
||
|
response = app.get(
|
||
|
f'/bbb/{SLUG}/meeting/{MEETING_ID}/join/user/{model_meeting.user_key}/',
|
||
|
params={'full_name': 'John Doe'},
|
||
|
)
|
||
|
assert meetings_create.call_args == mock.call(
|
||
|
logoutUrl=LOGOUT_URL, meeting_id=MEETING_ID, name=MEETING_NAME
|
||
|
)
|
||
|
assert 'John+Doe' in response.location
|
||
|
|
||
|
# data from create call
|
||
|
model_meeting.user_full_name = 'Jane Doe'
|
||
|
model_meeting.join_user_parameters = {'guest': 'true'}
|
||
|
model_meeting.save()
|
||
|
response = app.get(
|
||
|
f'/bbb/{SLUG}/meeting/{MEETING_ID}/join/user/{model_meeting.user_key}/',
|
||
|
)
|
||
|
assert 'Jane+Doe' in response.location
|
||
|
|
||
|
def test_join_agent(self, app, connector, meetings_create):
|
||
|
model_meeting = connector.meetings.create(
|
||
|
guid=UUID, name=MEETING_NAME, create_parameters={'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/{model_meeting.agent_key}/',
|
||
|
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'
|
||
|
)
|
||
|
|
||
|
# full name through query string
|
||
|
response = app.get(
|
||
|
f'/bbb/{SLUG}/meeting/{MEETING_ID}/join/agent/{model_meeting.agent_key}/',
|
||
|
params={'full_name': 'John Doe'},
|
||
|
)
|
||
|
assert meetings_create.call_args == mock.call(
|
||
|
logoutUrl=LOGOUT_URL, meeting_id=MEETING_ID, name=MEETING_NAME
|
||
|
)
|
||
|
assert 'John+Doe' in response.location
|
||
|
|
||
|
# data from create call
|
||
|
model_meeting.agent_full_name = 'Jane Doe'
|
||
|
model_meeting.join_agent_parameters = {'guest': 'true'}
|
||
|
model_meeting.save()
|
||
|
response = app.get(
|
||
|
f'/bbb/{SLUG}/meeting/{MEETING_ID}/join/agent/{model_meeting.agent_key}/',
|
||
|
)
|
||
|
assert 'Jane+Doe' in response.location
|
||
|
assert 'guest=true' in response.location
|