331 lines
11 KiB
Python
331 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import io
|
|
import json
|
|
import os
|
|
|
|
import mock
|
|
import pytest
|
|
|
|
from wcs import fields
|
|
from wcs.data_sources import NamedDataSource, build_agenda_datasources, collect_agenda_data
|
|
from wcs.formdef import FormDef
|
|
from wcs.qommon.http_request import HTTPRequest
|
|
from wcs.qommon.misc import ConnectionError
|
|
|
|
from .utilities import clean_temporary_pub, create_temporary_pub
|
|
|
|
|
|
def teardown_module(module):
|
|
clean_temporary_pub()
|
|
|
|
|
|
@pytest.fixture
|
|
def pub(request):
|
|
pub = create_temporary_pub()
|
|
req = HTTPRequest(None, {'SCRIPT_NAME': '/', 'SERVER_NAME': 'example.net'})
|
|
pub.set_app_dir(req)
|
|
pub._set_request(req)
|
|
|
|
open(os.path.join(pub.app_dir, 'site-options.cfg'), 'w').write(
|
|
'''
|
|
[debug]
|
|
logger=true
|
|
'''
|
|
)
|
|
|
|
pub.load_site_options()
|
|
|
|
return pub
|
|
|
|
|
|
AGENDA_EVENTS_DATA = [
|
|
{
|
|
"api": {
|
|
"datetimes_url": "http://chrono.example.net/api/agenda/events-A/datetimes/",
|
|
},
|
|
"id": "events-A",
|
|
"kind": "events",
|
|
"text": "Events A",
|
|
},
|
|
{
|
|
"api": {
|
|
"datetimes_url": "http://chrono.example.net/api/agenda/events-B/datetimes/",
|
|
},
|
|
"id": "events-B",
|
|
"kind": "events",
|
|
"text": "Events B",
|
|
},
|
|
]
|
|
|
|
|
|
AGENDA_MEETINGS_DATA = [
|
|
{
|
|
"api": {"meetings_url": "http://chrono.example.net/api/agenda/meetings-A/meetings/"},
|
|
"id": "meetings-A",
|
|
"kind": "meetings",
|
|
"text": "Meetings A",
|
|
},
|
|
{
|
|
"api": {
|
|
"meetings_url": "http://chrono.example.net/api/agenda/virtual-B/meetings/",
|
|
},
|
|
"id": "virtual-B",
|
|
"kind": "virtual",
|
|
"text": "Virtual B",
|
|
},
|
|
]
|
|
|
|
|
|
AGENDA_MEETING_TYPES_DATA = {
|
|
'meetings-A': [
|
|
{
|
|
"api": {
|
|
"datetimes_url": "http://chrono.example.net/api/agenda/meetings-A/meetings/mt-1/datetimes/"
|
|
},
|
|
"id": "mt-1",
|
|
"text": "MT 1",
|
|
},
|
|
{
|
|
"api": {
|
|
"datetimes_url": "http://chrono.example.net/api/agenda/meetings-A/meetings/mt-2/datetimes/"
|
|
},
|
|
"id": "mt-2",
|
|
"text": "MT 2",
|
|
},
|
|
],
|
|
'virtual-B': [
|
|
{
|
|
"api": {
|
|
"datetimes_url": "http://chrono.example.net/api/agenda/virtual-B/meetings/mt-3/datetimes/"
|
|
},
|
|
"id": "mt-3",
|
|
"text": "MT 3",
|
|
},
|
|
],
|
|
}
|
|
|
|
|
|
@mock.patch('wcs.qommon.misc.urlopen')
|
|
def test_collect_agenda_data(urlopen, pub, chrono_url):
|
|
pub.load_site_options()
|
|
NamedDataSource.wipe()
|
|
|
|
urlopen.side_effect = lambda *args: io.StringIO('{"data": []}')
|
|
assert collect_agenda_data(pub) == []
|
|
assert urlopen.call_args_list == [mock.call('http://chrono.example.net/api/agenda/')]
|
|
|
|
urlopen.side_effect = ConnectionError
|
|
urlopen.reset_mock()
|
|
assert collect_agenda_data(pub) is None
|
|
assert urlopen.call_args_list == [mock.call('http://chrono.example.net/api/agenda/')]
|
|
|
|
# events agenda
|
|
urlopen.side_effect = lambda *args: io.StringIO(json.dumps({"data": AGENDA_EVENTS_DATA}))
|
|
urlopen.reset_mock()
|
|
assert collect_agenda_data(pub) == [
|
|
{'text': 'Events A', 'url': 'http://chrono.example.net/api/agenda/events-A/datetimes/'},
|
|
{'text': 'Events B', 'url': 'http://chrono.example.net/api/agenda/events-B/datetimes/'},
|
|
]
|
|
assert urlopen.call_args_list == [mock.call('http://chrono.example.net/api/agenda/')]
|
|
|
|
# meetings agenda
|
|
urlopen.side_effect = [
|
|
io.StringIO(json.dumps({"data": AGENDA_MEETINGS_DATA})),
|
|
io.StringIO(json.dumps({"data": AGENDA_MEETING_TYPES_DATA['meetings-A']})),
|
|
io.StringIO(json.dumps({"data": AGENDA_MEETING_TYPES_DATA['virtual-B']})),
|
|
]
|
|
urlopen.reset_mock()
|
|
assert collect_agenda_data(pub) == [
|
|
{
|
|
'text': 'Meetings A - Meeting types',
|
|
'url': 'http://chrono.example.net/api/agenda/meetings-A/meetings/',
|
|
},
|
|
{
|
|
'text': 'Meetings A - Slots of type MT 1',
|
|
'url': 'http://chrono.example.net/api/agenda/meetings-A/meetings/mt-1/datetimes/',
|
|
},
|
|
{
|
|
'text': 'Meetings A - Slots of type MT 2',
|
|
'url': 'http://chrono.example.net/api/agenda/meetings-A/meetings/mt-2/datetimes/',
|
|
},
|
|
{
|
|
'text': 'Virtual B - Meeting types',
|
|
'url': 'http://chrono.example.net/api/agenda/virtual-B/meetings/',
|
|
},
|
|
{
|
|
'text': 'Virtual B - Slots of type MT 3',
|
|
'url': 'http://chrono.example.net/api/agenda/virtual-B/meetings/mt-3/datetimes/',
|
|
},
|
|
]
|
|
assert urlopen.call_args_list == [
|
|
mock.call('http://chrono.example.net/api/agenda/'),
|
|
mock.call('http://chrono.example.net/api/agenda/meetings-A/meetings/'),
|
|
mock.call('http://chrono.example.net/api/agenda/virtual-B/meetings/'),
|
|
]
|
|
|
|
# if meeting types could not be collected
|
|
urlopen.side_effect = [
|
|
io.StringIO(json.dumps({"data": AGENDA_MEETINGS_DATA})),
|
|
io.StringIO(json.dumps({"data": AGENDA_MEETING_TYPES_DATA['meetings-A']})),
|
|
ConnectionError,
|
|
]
|
|
urlopen.reset_mock()
|
|
assert collect_agenda_data(pub) is None
|
|
assert urlopen.call_args_list == [
|
|
mock.call('http://chrono.example.net/api/agenda/'),
|
|
mock.call('http://chrono.example.net/api/agenda/meetings-A/meetings/'),
|
|
mock.call('http://chrono.example.net/api/agenda/virtual-B/meetings/'),
|
|
]
|
|
|
|
urlopen.side_effect = [
|
|
io.StringIO(json.dumps({"data": AGENDA_MEETINGS_DATA})),
|
|
ConnectionError,
|
|
]
|
|
urlopen.reset_mock()
|
|
assert collect_agenda_data(pub) is None
|
|
assert urlopen.call_args_list == [
|
|
mock.call('http://chrono.example.net/api/agenda/'),
|
|
mock.call('http://chrono.example.net/api/agenda/meetings-A/meetings/'),
|
|
]
|
|
|
|
|
|
@mock.patch('wcs.data_sources.collect_agenda_data')
|
|
def test_build_agenda_datasources_without_chrono(mock_collect, pub):
|
|
NamedDataSource.wipe()
|
|
build_agenda_datasources(pub)
|
|
assert mock_collect.call_args_list == []
|
|
assert NamedDataSource.count() == 0
|
|
|
|
|
|
@mock.patch('wcs.data_sources.collect_agenda_data')
|
|
def test_build_agenda_datasources(mock_collect, pub, chrono_url):
|
|
pub.load_site_options()
|
|
NamedDataSource.wipe()
|
|
|
|
# create some datasource, with same urls, but not external
|
|
ds = NamedDataSource(name='Foo A')
|
|
ds.data_source = {'type': 'json', 'value': 'http://chrono.example.net/api/agenda/events-A/datetimes/'}
|
|
ds.store()
|
|
ds = NamedDataSource(name='Foo B')
|
|
ds.data_source = {'type': 'json', 'value': 'http://chrono.example.net/api/agenda/events-B/datetimes/'}
|
|
ds.store()
|
|
|
|
# error during collect
|
|
mock_collect.return_value = None
|
|
build_agenda_datasources(pub)
|
|
assert NamedDataSource.count() == 2 # no changes
|
|
|
|
# no agenda datasource found in chrono
|
|
mock_collect.return_value = []
|
|
build_agenda_datasources(pub)
|
|
assert NamedDataSource.count() == 2 # no changes
|
|
|
|
# 2 agenda datasources found
|
|
mock_collect.return_value = [
|
|
{'text': 'Events A', 'url': 'http://chrono.example.net/api/agenda/events-A/datetimes/'},
|
|
{'text': 'Events B', 'url': 'http://chrono.example.net/api/agenda/events-B/datetimes/'},
|
|
]
|
|
|
|
# agenda datasources does not exist, create them
|
|
build_agenda_datasources(pub)
|
|
assert NamedDataSource.count() == 2 + 2
|
|
datasource1 = NamedDataSource.get(2 + 1)
|
|
datasource2 = NamedDataSource.get(2 + 2)
|
|
assert datasource1.name == 'Events A'
|
|
assert datasource1.external == 'agenda'
|
|
assert datasource1.external_status is None
|
|
assert datasource1.data_source == {
|
|
'type': 'json',
|
|
'value': 'http://chrono.example.net/api/agenda/events-A/datetimes/',
|
|
}
|
|
assert datasource2.name == 'Events B'
|
|
assert datasource2.external == 'agenda'
|
|
assert datasource2.external_status is None
|
|
assert datasource2.data_source == {
|
|
'type': 'json',
|
|
'value': 'http://chrono.example.net/api/agenda/events-B/datetimes/',
|
|
}
|
|
|
|
# again, datasources already exist, but name is wrong => change it
|
|
datasource1.name = 'wrong'
|
|
datasource1.store()
|
|
datasource2.name = 'wrong again'
|
|
datasource2.store()
|
|
build_agenda_datasources(pub)
|
|
assert NamedDataSource.count() == 2 + 2
|
|
datasource1 = NamedDataSource.get(2 + 1)
|
|
datasource2 = NamedDataSource.get(2 + 2)
|
|
assert datasource1.name == 'Events A'
|
|
assert datasource2.name == 'Events B'
|
|
|
|
# all datasources does not exist, one is unknown
|
|
datasource1.data_source['value'] = 'http://chrono.example.net/api/agenda/events-FOOBAR/datetimes/'
|
|
datasource1.store()
|
|
|
|
build_agenda_datasources(pub)
|
|
assert NamedDataSource.count() == 2 + 2
|
|
# first datasource was deleted, because not found and not used
|
|
datasource2 = NamedDataSource.get(2 + 2)
|
|
datasource3 = NamedDataSource.get(2 + 3)
|
|
assert datasource2.name == 'Events B'
|
|
assert datasource2.external == 'agenda'
|
|
assert datasource2.external_status is None
|
|
assert datasource2.data_source == {
|
|
'type': 'json',
|
|
'value': 'http://chrono.example.net/api/agenda/events-B/datetimes/',
|
|
}
|
|
assert datasource3.name == 'Events A'
|
|
assert datasource3.external == 'agenda'
|
|
assert datasource3.external_status is None
|
|
assert datasource3.data_source == {
|
|
'type': 'json',
|
|
'value': 'http://chrono.example.net/api/agenda/events-A/datetimes/',
|
|
}
|
|
|
|
# all datasources does not exist, one is unknown but used
|
|
FormDef.wipe()
|
|
formdef = FormDef()
|
|
formdef.name = 'foobar'
|
|
formdef.fields = [
|
|
fields.ItemField(id='0', label='string', type='item', data_source={'type': datasource3.slug}),
|
|
]
|
|
formdef.store()
|
|
assert datasource3.is_used_in_formdef(formdef)
|
|
datasource3.data_source['value'] = 'http://chrono.example.net/api/agenda/events-FOOBAR/datetimes/'
|
|
datasource3.store()
|
|
build_agenda_datasources(pub)
|
|
assert NamedDataSource.count() == 2 + 3
|
|
datasource2 = NamedDataSource.get(2 + 2)
|
|
datasource3 = NamedDataSource.get(2 + 3)
|
|
datasource4 = NamedDataSource.get(2 + 4)
|
|
assert datasource2.name == 'Events B'
|
|
assert datasource2.external == 'agenda'
|
|
assert datasource2.external_status is None
|
|
assert datasource2.data_source == {
|
|
'type': 'json',
|
|
'value': 'http://chrono.example.net/api/agenda/events-B/datetimes/',
|
|
}
|
|
assert datasource3.name == 'Events A'
|
|
assert datasource3.external == 'agenda'
|
|
assert datasource3.external_status == 'not-found'
|
|
assert datasource3.data_source == {
|
|
'type': 'json',
|
|
'value': 'http://chrono.example.net/api/agenda/events-FOOBAR/datetimes/',
|
|
}
|
|
assert datasource4.name == 'Events A'
|
|
assert datasource4.external == 'agenda'
|
|
assert datasource4.external_status is None
|
|
assert datasource4.data_source == {
|
|
'type': 'json',
|
|
'value': 'http://chrono.example.net/api/agenda/events-A/datetimes/',
|
|
}
|
|
|
|
# a datasource was marked as unknown
|
|
datasource4.external_status = 'not-found'
|
|
datasource4.store()
|
|
build_agenda_datasources(pub)
|
|
assert NamedDataSource.count() == 2 + 3
|
|
datasource4 = NamedDataSource.get(2 + 4)
|
|
assert datasource4.external_status is None
|