passerelle/tests/test_bbb.py

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