0001-datasource-collect-agendas-48282.patch

Lauréline Guérin, 16 février 2021 16:00

Voir les différences: en ligne côte à côte

Subject: [PATCH 1/5] datasource: collect agendas (#48282)

 tests/conftest.py               |   5 +
 tests/test_datasource_chrono.py | 328 ++++++++++++++++++++++++++++++++
 wcs/admin/data_sources.py       |  10 +-
 wcs/ctl/check_hobos.py          |   2 +
 wcs/data_sources.py             | 117 +++++++++++-
 5 files changed, 446 insertions(+), 16 deletions(-)
 create mode 100644 tests/test_datasource_chrono.py

         return value
     @pytest.fixture
     def chrono_url(request, pub):
         return site_options(request, pub, 'options', 'chrono_url', 'http://chrono.example.net/')
     @pytest.fixture
     def fargo_url(request, pub):
         return site_options(request, pub, 'options', 'fargo_url', 'http://fargo.example.net/')

     # -*- coding: utf-8 -*-
     import pytest
     import json
     import shutil
     from django.utils.six import StringIO
     from quixote import cleanup
     from wcs import fields
     from wcs.data_sources import NamedDataSource, build_agenda_datasources, collect_agenda_data
     from wcs.formdef import FormDef
     from wcs.qommon.misc import ConnectionError
     from wcs.qommon.http_request import HTTPRequest
     import mock
     from utilities import create_temporary_pub
     def setup_module(module):
         cleanup()
         global pub
         pub = create_temporary_pub()
         pub.cfg['debug'] = {'logger': True}
         pub.write_cfg()
         pub.set_config()
     def teardown_module(module):
         shutil.rmtree(pub.APP_DIR)
     @pytest.fixture
     def pub(request):
         req = HTTPRequest(None, {'SERVER_NAME': 'example.net', 'SCRIPT_NAME': ''})
         pub.set_app_dir(req)
         pub._set_request(req)
         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: 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: 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 = [
             StringIO(json.dumps({"data": AGENDA_MEETINGS_DATA})),
             StringIO(json.dumps({"data": AGENDA_MEETING_TYPES_DATA['meetings-A']})),
             StringIO(json.dumps({"data": AGENDA_MEETING_TYPES_DATA['virtual-B']})),
+        ]
         urlopen.reset_mock()
         assert collect_agenda_data(pub) == [
+            {
                 'text': 'Meetings A - Slot 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 - Slot 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 = [
             StringIO(json.dumps({"data": AGENDA_MEETINGS_DATA})),
             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 = [
             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

             if self.datasource is None:
                 self.datasource = NamedDataSource()
         def is_used(self):
             for formdef in get_formdefs_of_all_kinds():
                 if self.datasource.is_used_in_formdef(formdef):
                     return True
             return False
         def get_form(self):
             form = Form(enctype='multipart/form-data', advanced_label=_('Additional options'))
             form.add(StringWidget, 'name', title=_('Name'), required=True, size=30, value=self.datasource.name)
-...
                     'data-dynamic-display-value': 'geojson',
                 },
+            )
             if self.datasource.slug and not self.is_used():
             if self.datasource.slug and not self.datasource.is_used():
                 form.add(
                     StringWidget,
                     'slug',
-...
         def delete(self):
             form = Form(enctype='multipart/form-data')
             if not self.datasource_ui.is_used():
             if not self.datasource.is_used():
                 form.widgets.append(
                     HtmlWidget('<p>%s</p>' % _('You are about to irrevocably delete this data source.'))
+                )

                     continue
                 if service.get('service-id') == 'fargo':
                     config.set('options', 'fargo_url', service.get('base_url'))
                 elif service.get('service-id') == 'chrono':
                     config.set('options', 'chrono_url', service.get('base_url'))
             try:
                 portal_agent_url = config.get('variables', 'portal_agent_url')

     from django.utils import six
     from django.utils.encoding import force_text, force_bytes
     from django.utils.six.moves.urllib import parse as urllib
     from django.utils.six.moves.urllib import parse as urlparse
     from quixote import get_publisher, get_request, get_session
     from quixote.html import TemplateIO
     from .qommon import _, force_str
     from .qommon import misc
     from .qommon import get_logger
     from .qommon.cron import CronJob
     from .qommon.form import *
     from .qommon.humantime import seconds2humanduration
     from .qommon.misc import get_variadic_url
     from .qommon import misc
     from .qommon import get_logger
     from .qommon.publisher import get_publisher_class
     from .qommon.storage import StorableObject
     from .qommon.template import Template
     from .qommon.xml_storage import XmlStorableObject
-...
         return tupled_items
     def get_json_from_url(url, data_source):
     def get_json_from_url(url, data_source=None, log_message_part='JSON data source'):
         url = sign_url_auto_orig(url)
         data_source = data_source or {}
         data_key = data_source.get('data_attribute') or 'data'
         geojson = data_source.get('type') == 'geojson'
         try:
-...
                 if not isinstance(entries.get(data_key), list):
                     raise ValueError('not a json dict with a %s list attribute' % data_key)
         except misc.ConnectionError as e:
             get_logger().warning('Error loading JSON data source (%s)' % str(e))
             get_logger().warning('Error loading %s (%s)' % (log_message_part, str(e)))
             return None
         except (ValueError, TypeError) as e:
             get_logger().warning('Error reading JSON data source output (%s)' % str(e))
             get_logger().warning('Error reading %s output (%s)' % (log_message_part, str(e)))
             return None
         return entries
-...
         text_attribute = None
         id_property = None
         label_template_property = None
         external = None
         external_status = None
         # declarations for serialization
         XML_NODES = [
-...
             ('text_attribute', 'str'),
             ('id_property', 'str'),
             ('label_template_property', 'str'),
             ('external', 'str'),
             ('external_status', 'str'),
             ('data_source', 'data_source'),
+        ]
-...
                 url = get_variadic_url(url, vars)
             return url
         def is_used_in_formdef(self, formdef):
             from .fields import WidgetField
         def is_used(self):
             from wcs.formdef import get_formdefs_of_all_kinds
             for formdef in get_formdefs_of_all_kinds():
                 if self.is_used_in_formdef(formdef):
                     return True
             return False
         def is_used_in_formdef(self, formdef):
             for field in formdef.fields or []:
                 data_source = getattr(field, 'data_source', None)
                 if not data_source:
-...
         def inspect_keys(self):
             return []
     def has_chrono(publisher):
         return publisher.get_site_option('chrono_url') is not None
     def chrono_url(publisher, url):
         chrono_url = publisher.get_site_option('chrono_url')
         return urlparse.urljoin(chrono_url, url)
     def collect_agenda_data(publisher):
         agenda_url = chrono_url(publisher, 'api/agenda/')
         result = get_json_from_url(agenda_url, log_message_part='agenda')
         if result is None:
             return
         # build datasources from chrono
         agenda_data = []
         for agenda in result.get('data') or []:
             if agenda['kind'] == 'events':
                 agenda_data.append({'text': agenda['text'], 'url': agenda['api']['datetimes_url']})
             elif agenda['kind'] in ['meetings', 'virtual']:
                 agenda_data.append(
                     {'text': _('%s - Slot types') % agenda['text'], 'url': agenda['api']['meetings_url']}
+                )
                 # get also meeting types
                 mt_url = chrono_url(publisher, 'api/agenda/%s/meetings/' % agenda['id'])
                 mt_results = get_json_from_url(mt_url, log_message_part='agenda')
                 if mt_results is None:
                     return
                 for meetingtype in mt_results.get('data') or []:
                     agenda_data.append(
+                        {
                             'text': _('%s - Slots of type %s') % (agenda['text'], meetingtype['text']),
                             'url': meetingtype['api']['datetimes_url'],
+                        }
+                    )
         return agenda_data
     def build_agenda_datasources(publisher):
         if not has_chrono(publisher):
             return
         agenda_data = collect_agenda_data(publisher)
         if agenda_data is None:
             return
         # fetch existing datasources
         existing_datasources = {}
         for datasource in NamedDataSource.select():
             if datasource.external != 'agenda':
                 continue
             existing_datasources[datasource.data_source['value']] = datasource
         seen_datasources = []
         # build datasources from chrono
         for agenda in agenda_data:
             url = agenda['url']
             datasource = existing_datasources.get(url)
             if datasource is None:
                 datasource = NamedDataSource()
                 datasource.external = 'agenda'
                 datasource.data_source = {'type': 'json', 'value': url}
             datasource.external_status = None  # reset
             datasource.name = agenda['text']
             datasource.store()
             # maintain caches
             existing_datasources[url] = datasource
             seen_datasources.append(url)
         # now check outdated agenda datasources
         for url, datasource in existing_datasources.items():
             if url in seen_datasources:
                 continue
             if datasource.is_used():
                 datasource.external_status = 'not-found'
                 datasource.store()
                 continue
             datasource.remove_self()
     if get_publisher_class():
         # every hour: check for agenda datasources
         get_publisher_class().register_cronjob(
             CronJob(build_agenda_datasources, name='build_agenda_datasources', minutes=[0])
+        )
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0001-datasource-collect-agendas-48282.patch