277 lines
9.9 KiB
Python
277 lines
9.9 KiB
Python
import json
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
from requests.exceptions import ConnectionError
|
|
from requests.models import Response
|
|
|
|
from lingo.agendas.chrono import (
|
|
ChronoError,
|
|
collect_agenda_data,
|
|
get_event,
|
|
get_subscriptions,
|
|
refresh_agendas,
|
|
)
|
|
from lingo.agendas.models import Agenda
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
AGENDA_DATA = [
|
|
{
|
|
"slug": "events-a",
|
|
"kind": "events",
|
|
"text": "Events A",
|
|
"category": None,
|
|
"category_label": None,
|
|
},
|
|
{
|
|
"slug": "events-b",
|
|
"kind": "events",
|
|
"text": "Events B",
|
|
"category": "foo",
|
|
"category_label": "Foo",
|
|
},
|
|
{
|
|
"slug": "meetings-a",
|
|
"kind": "meetings",
|
|
"text": "Meetings A",
|
|
"category": None,
|
|
"category_label": None,
|
|
},
|
|
{
|
|
"slug": "virtual-b",
|
|
"kind": "virtual",
|
|
"text": "Virtual B",
|
|
"category": "foo",
|
|
"category_label": "Foo",
|
|
},
|
|
]
|
|
|
|
|
|
class MockedRequestResponse(mock.Mock):
|
|
status_code = 200
|
|
|
|
def json(self):
|
|
return json.loads(self.content)
|
|
|
|
|
|
def test_collect_agenda_data_no_service(settings):
|
|
settings.KNOWN_SERVICES = {}
|
|
assert collect_agenda_data() is None
|
|
|
|
settings.KNOWN_SERVICES = {'other': []}
|
|
assert collect_agenda_data() is None
|
|
|
|
|
|
def test_collect_agenda_data():
|
|
with mock.patch('requests.Session.get') as requests_get:
|
|
requests_get.side_effect = ConnectionError()
|
|
assert collect_agenda_data() is None
|
|
|
|
with mock.patch('requests.Session.get') as requests_get:
|
|
mock_resp = Response()
|
|
mock_resp.status_code = 500
|
|
requests_get.return_value = mock_resp
|
|
assert collect_agenda_data() is None
|
|
|
|
with mock.patch('requests.Session.get') as requests_get:
|
|
mock_resp = Response()
|
|
mock_resp.status_code = 404
|
|
requests_get.return_value = mock_resp
|
|
assert collect_agenda_data() is None
|
|
|
|
with mock.patch('requests.Session.get') as requests_get:
|
|
requests_get.return_value = MockedRequestResponse(content=json.dumps({'foo': 'bar'}))
|
|
assert collect_agenda_data() is None
|
|
|
|
data = {'data': []}
|
|
with mock.patch('requests.Session.get') as requests_get:
|
|
requests_get.return_value = MockedRequestResponse(content=json.dumps(data))
|
|
assert collect_agenda_data() == []
|
|
assert requests_get.call_args_list[0][0] == ('api/agenda/',)
|
|
assert requests_get.call_args_list[0][1]['remote_service']['url'] == 'http://chrono.example.org'
|
|
|
|
data = {'data': AGENDA_DATA}
|
|
with mock.patch('requests.Session.get') as requests_get:
|
|
requests_get.return_value = MockedRequestResponse(content=json.dumps(data))
|
|
assert collect_agenda_data() == [
|
|
{'category_label': None, 'category_slug': None, 'label': 'Events A', 'slug': 'events-a'},
|
|
{'category_label': 'Foo', 'category_slug': 'foo', 'label': 'Events B', 'slug': 'events-b'},
|
|
]
|
|
|
|
|
|
@mock.patch('lingo.agendas.chrono.collect_agenda_data')
|
|
def test_refresh_agendas(mock_collect):
|
|
Agenda.objects.create(label='foo')
|
|
|
|
# error during collect
|
|
mock_collect.return_value = None
|
|
refresh_agendas()
|
|
assert Agenda.objects.count() == 1 # no changes
|
|
|
|
# 2 agendas found
|
|
mock_collect.return_value = [
|
|
{'category_label': None, 'category_slug': None, 'label': 'Events A', 'slug': 'events-a'},
|
|
{'category_label': 'Foo', 'category_slug': 'foo', 'label': 'Events B', 'slug': 'events-b'},
|
|
]
|
|
|
|
# agendas don't exist, create them
|
|
refresh_agendas()
|
|
assert Agenda.objects.count() == 2
|
|
agenda1 = Agenda.objects.all().order_by('pk')[0]
|
|
agenda2 = Agenda.objects.all().order_by('pk')[1]
|
|
assert agenda1.label == "Events A"
|
|
assert agenda1.slug == "events-a"
|
|
assert agenda1.category_label is None
|
|
assert agenda1.category_slug is None
|
|
assert agenda2.label == "Events B"
|
|
assert agenda2.slug == "events-b"
|
|
assert agenda2.category_label == 'Foo'
|
|
assert agenda2.category_slug == 'foo'
|
|
|
|
# again, but some attributes are wrong
|
|
agenda1.label = "Wrong"
|
|
agenda1.category_label = 'Foo'
|
|
agenda1.category_slug = 'foo'
|
|
agenda1.save()
|
|
agenda2.label = "Wrong"
|
|
agenda2.category_label is None
|
|
agenda2.category_slug is None
|
|
agenda2.save()
|
|
refresh_agendas()
|
|
assert Agenda.objects.count() == 2
|
|
new_agenda1 = Agenda.objects.all().order_by('pk')[0]
|
|
new_agenda2 = Agenda.objects.all().order_by('pk')[1]
|
|
assert new_agenda1.pk == agenda1.pk
|
|
assert new_agenda1.label == "Events A"
|
|
assert new_agenda1.slug == "events-a"
|
|
assert new_agenda1.category_label is None
|
|
assert new_agenda1.category_slug is None
|
|
assert new_agenda2.pk == agenda2.pk
|
|
assert new_agenda2.label == "Events B"
|
|
assert new_agenda2.slug == "events-b"
|
|
assert new_agenda2.category_label == 'Foo'
|
|
assert new_agenda2.category_slug == 'foo'
|
|
|
|
# no agenda in chrono
|
|
mock_collect.return_value = []
|
|
refresh_agendas()
|
|
assert Agenda.objects.count() == 0
|
|
|
|
|
|
def test_get_event_no_service(settings):
|
|
settings.KNOWN_SERVICES = {}
|
|
with pytest.raises(ChronoError) as e:
|
|
get_event('foo')
|
|
assert str(e.value) == 'Unable to get event details'
|
|
|
|
settings.KNOWN_SERVICES = {'other': []}
|
|
with pytest.raises(ChronoError) as e:
|
|
get_event('foo')
|
|
assert str(e.value) == 'Unable to get event details'
|
|
|
|
|
|
def test_get_event():
|
|
with mock.patch('requests.Session.get') as requests_get:
|
|
requests_get.side_effect = ConnectionError()
|
|
with pytest.raises(ChronoError) as e:
|
|
get_event('foo')
|
|
assert str(e.value) == 'Unable to get event details'
|
|
|
|
with mock.patch('requests.Session.get') as requests_get:
|
|
mock_resp = Response()
|
|
mock_resp.status_code = 500
|
|
requests_get.return_value = mock_resp
|
|
with pytest.raises(ChronoError) as e:
|
|
get_event('foo')
|
|
assert str(e.value) == 'Unable to get event details'
|
|
|
|
with mock.patch('requests.Session.get') as requests_get:
|
|
mock_resp = Response()
|
|
mock_resp.status_code = 404
|
|
requests_get.return_value = mock_resp
|
|
with pytest.raises(ChronoError) as e:
|
|
get_event('foo')
|
|
assert str(e.value) == 'Unable to get event details'
|
|
|
|
with mock.patch('requests.Session.get') as requests_get:
|
|
requests_get.return_value = MockedRequestResponse(content=json.dumps({'foo': 'bar'}))
|
|
with pytest.raises(ChronoError) as e:
|
|
get_event('foo')
|
|
assert str(e.value) == 'Unable to get event details'
|
|
|
|
data = {'data': []}
|
|
with mock.patch('requests.Session.get') as requests_get:
|
|
requests_get.return_value = MockedRequestResponse(content=json.dumps(data))
|
|
with pytest.raises(ChronoError) as e:
|
|
get_event('foo')
|
|
assert str(e.value) == 'Unable to get event details'
|
|
assert requests_get.call_args_list[0][0] == ('api/agendas/events/?slots=foo',)
|
|
assert requests_get.call_args_list[0][1]['remote_service']['url'] == 'http://chrono.example.org'
|
|
|
|
data = {'data': ['foo']}
|
|
with mock.patch('requests.Session.get') as requests_get:
|
|
requests_get.return_value = MockedRequestResponse(content=json.dumps(data))
|
|
assert get_event('foo') == 'foo'
|
|
|
|
data = {'data': ['foo', 'bar']} # should not happen
|
|
with mock.patch('requests.Session.get') as requests_get:
|
|
requests_get.return_value = MockedRequestResponse(content=json.dumps(data))
|
|
assert get_event('foo') == 'foo'
|
|
|
|
|
|
def test_get_subscriptions_no_service(settings):
|
|
settings.KNOWN_SERVICES = {}
|
|
with pytest.raises(ChronoError) as e:
|
|
get_subscriptions('foo', 'user:1')
|
|
assert str(e.value) == 'Unable to get subscription details'
|
|
|
|
settings.KNOWN_SERVICES = {'other': []}
|
|
with pytest.raises(ChronoError) as e:
|
|
get_subscriptions('foo', 'user:1')
|
|
assert str(e.value) == 'Unable to get subscription details'
|
|
|
|
|
|
def test_get_subscriptions():
|
|
with mock.patch('requests.Session.get') as requests_get:
|
|
requests_get.side_effect = ConnectionError()
|
|
with pytest.raises(ChronoError) as e:
|
|
get_subscriptions('foo', 'user:1')
|
|
assert str(e.value) == 'Unable to get subscription details'
|
|
|
|
with mock.patch('requests.Session.get') as requests_get:
|
|
mock_resp = Response()
|
|
mock_resp.status_code = 500
|
|
requests_get.return_value = mock_resp
|
|
with pytest.raises(ChronoError) as e:
|
|
get_subscriptions('foo', 'user:1')
|
|
assert str(e.value) == 'Unable to get subscription details'
|
|
|
|
with mock.patch('requests.Session.get') as requests_get:
|
|
mock_resp = Response()
|
|
mock_resp.status_code = 404
|
|
requests_get.return_value = mock_resp
|
|
with pytest.raises(ChronoError) as e:
|
|
get_subscriptions('foo', 'user:1')
|
|
assert str(e.value) == 'Unable to get subscription details'
|
|
|
|
with mock.patch('requests.Session.get') as requests_get:
|
|
requests_get.return_value = MockedRequestResponse(content=json.dumps({'foo': 'bar'}))
|
|
with pytest.raises(ChronoError) as e:
|
|
get_subscriptions('foo', 'user:1')
|
|
assert str(e.value) == 'Unable to get subscription details'
|
|
|
|
data = {'data': []}
|
|
with mock.patch('requests.Session.get') as requests_get:
|
|
requests_get.return_value = MockedRequestResponse(content=json.dumps(data))
|
|
with pytest.raises(ChronoError) as e:
|
|
get_subscriptions('foo', 'user:1')
|
|
assert str(e.value) == 'Unable to get subscription details'
|
|
assert requests_get.call_args_list[0][0] == ('api/agenda/foo/subscription/?user_external_id=user:1',)
|
|
assert requests_get.call_args_list[0][1]['remote_service']['url'] == 'http://chrono.example.org'
|
|
|
|
data = {'data': ['foo', 'bar']}
|
|
with mock.patch('requests.Session.get') as requests_get:
|
|
requests_get.return_value = MockedRequestResponse(content=json.dumps(data))
|
|
assert get_subscriptions('foo', 'user:1') == ['foo', 'bar']
|