diff --git a/tests/test_datasource_chrono.py b/tests/test_datasource_chrono.py index 732c17f63..d9362138d 100644 --- a/tests/test_datasource_chrono.py +++ b/tests/test_datasource_chrono.py @@ -1,10 +1,11 @@ +import os from unittest import mock import pytest import responses from wcs import fields -from wcs.data_sources import NamedDataSource, build_agenda_datasources, collect_agenda_data +from wcs.data_sources import NamedDataSource, build_agenda_datasources, collect_agenda_data, translate_url from wcs.formdef import FormDef from wcs.qommon.http_request import HTTPRequest from wcs.qommon.misc import ConnectionError @@ -97,6 +98,27 @@ AGENDA_MEETING_TYPES_DATA = { } +def test_translate_url(pub, chrono_url): + pub.load_site_options() + if not pub.site_options.has_section('variables'): + pub.site_options.add_section('variables') + pub.site_options.set('variables', 'foo_url', 'http://foo.bar') + with open(os.path.join(pub.app_dir, 'site-options.cfg'), 'w') as fd: + pub.site_options.write(fd) + assert translate_url(pub, 'http://chrono.example.net/foo/bar/') == 'http://chrono.example.net/foo/bar/' + + pub.site_options.set('variables', 'foo_url', 'http://chrono.example.net/') + with open(os.path.join(pub.app_dir, 'site-options.cfg'), 'w') as fd: + pub.site_options.write(fd) + assert translate_url(pub, 'http://chrono.example.net/foo/bar/') == '{{ foo_url }}foo/bar/' + + pub.site_options.set('variables', 'foo_url', 'http://foo.bar') + pub.site_options.set('variables', 'agendas_url', 'http://chrono.example.net/') + with open(os.path.join(pub.app_dir, 'site-options.cfg'), 'w') as fd: + pub.site_options.write(fd) + assert translate_url(pub, 'http://chrono.example.net/foo/bar/') == '{{ agendas_url }}foo/bar/' + + @responses.activate def test_collect_agenda_data(pub, chrono_url): pub.load_site_options() @@ -211,22 +233,27 @@ def test_build_agenda_datasources_without_chrono(mock_collect, pub): @mock.patch('wcs.data_sources.collect_agenda_data') def test_build_agenda_datasources(mock_collect, pub, chrono_url): pub.load_site_options() + if not pub.site_options.has_section('variables'): + pub.site_options.add_section('variables') + pub.site_options.set('variables', 'agendas_url', 'http://chrono.example.net/') + with open(os.path.join(pub.app_dir, 'site-options.cfg'), 'w') as fd: + pub.site_options.write(fd) NamedDataSource.wipe() # create some datasource, with same urls, but external != 'agenda' ds = NamedDataSource(name='Foo A') - ds.data_source = {'type': 'json', 'value': 'http://chrono.example.net/api/agenda/events-A/datetimes/'} + ds.data_source = {'type': 'json', 'value': '{{ agendas_url }}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.data_source = {'type': 'json', 'value': '{{ agendas_url }}api/agenda/events-B/datetimes/'} ds.store() ds = NamedDataSource(name='Foo A') - ds.data_source = {'type': 'json', 'value': 'http://chrono.example.net/api/agenda/events-A/datetimes/'} + ds.data_source = {'type': 'json', 'value': '{{ agendas_url }}api/agenda/events-A/datetimes/'} ds.external = 'agenda_manual' ds.store() ds = NamedDataSource(name='Foo B') ds.external = 'agenda_manual' - ds.data_source = {'type': 'json', 'value': 'http://chrono.example.net/api/agenda/events-B/datetimes/'} + ds.data_source = {'type': 'json', 'value': '{{ agendas_url }}api/agenda/events-B/datetimes/'} ds.store() # error during collect @@ -264,7 +291,7 @@ def test_build_agenda_datasources(mock_collect, pub, chrono_url): assert datasource1.external_status is None assert datasource1.data_source == { 'type': 'json', - 'value': 'http://chrono.example.net/api/agenda/events-A/datetimes/', + 'value': '{{ agendas_url }}api/agenda/events-A/datetimes/', } assert datasource2.name == 'Events B' assert datasource2.slug == 'chrono_ds_slugb' @@ -272,7 +299,7 @@ def test_build_agenda_datasources(mock_collect, pub, chrono_url): assert datasource2.external_status is None assert datasource2.data_source == { 'type': 'json', - 'value': 'http://chrono.example.net/api/agenda/events-B/datetimes/', + 'value': '{{ agendas_url }}api/agenda/events-B/datetimes/', } # again, datasources already exist, but name is wrong => change it @@ -292,7 +319,7 @@ def test_build_agenda_datasources(mock_collect, pub, chrono_url): assert datasource2.slug == 'wrong_again' # all datasources does not exist, one is unknown - datasource1.data_source['value'] = 'http://chrono.example.net/api/agenda/events-FOOBAR/datetimes/' + datasource1.data_source['value'] = '{{ agendas_url }}api/agenda/events-FOOBAR/datetimes/' datasource1.store() build_agenda_datasources(pub) @@ -305,14 +332,14 @@ def test_build_agenda_datasources(mock_collect, pub, chrono_url): assert datasource2.external_status is None assert datasource2.data_source == { 'type': 'json', - 'value': 'http://chrono.example.net/api/agenda/events-B/datetimes/', + 'value': '{{ agendas_url }}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/', + 'value': '{{ agendas_url }}api/agenda/events-A/datetimes/', } # all datasources does not exist, one is unknown but used @@ -324,7 +351,7 @@ def test_build_agenda_datasources(mock_collect, pub, chrono_url): ] formdef.store() assert datasource3.is_used_in_formdef(formdef) - datasource3.data_source['value'] = 'http://chrono.example.net/api/agenda/events-FOOBAR/datetimes/' + datasource3.data_source['value'] = '{{ agendas_url }}api/agenda/events-FOOBAR/datetimes/' datasource3.store() build_agenda_datasources(pub) assert NamedDataSource.count() == 4 + 3 @@ -337,7 +364,7 @@ def test_build_agenda_datasources(mock_collect, pub, chrono_url): assert datasource2.external_status is None assert datasource2.data_source == { 'type': 'json', - 'value': 'http://chrono.example.net/api/agenda/events-B/datetimes/', + 'value': '{{ agendas_url }}api/agenda/events-B/datetimes/', } assert datasource3.name == 'Events A' assert datasource3.slug == 'chrono_ds_sluga' @@ -345,7 +372,7 @@ def test_build_agenda_datasources(mock_collect, pub, chrono_url): assert datasource3.external_status == 'not-found' assert datasource3.data_source == { 'type': 'json', - 'value': 'http://chrono.example.net/api/agenda/events-FOOBAR/datetimes/', + 'value': '{{ agendas_url }}api/agenda/events-FOOBAR/datetimes/', } assert datasource4.name == 'Events A' assert datasource4.slug == 'chrono_ds_sluga_1' @@ -353,7 +380,7 @@ def test_build_agenda_datasources(mock_collect, pub, chrono_url): assert datasource4.external_status is None assert datasource4.data_source == { 'type': 'json', - 'value': 'http://chrono.example.net/api/agenda/events-A/datetimes/', + 'value': '{{ agendas_url }}api/agenda/events-A/datetimes/', } # a datasource was marked as unknown @@ -363,3 +390,26 @@ def test_build_agenda_datasources(mock_collect, pub, chrono_url): assert NamedDataSource.count() == 4 + 3 datasource4 = NamedDataSource.get(4 + 4) assert datasource4.external_status is None + + +def test_agenda_datasources_migration(pub, chrono_url): + pub.load_site_options() + if not pub.site_options.has_section('variables'): + pub.site_options.add_section('variables') + pub.site_options.set('variables', 'agendas_url', 'http://chrono.example.net/') + with open(os.path.join(pub.app_dir, 'site-options.cfg'), 'w') as fd: + pub.site_options.write(fd) + NamedDataSource.wipe() + + ds = NamedDataSource(name='Foo A') + ds.data_source = {'type': 'json', 'value': 'http://chrono.example.net/api/agenda/events-A/datetimes/'} + ds.external = 'agenda' + ds.store() + ds = NamedDataSource.get(ds.id) + assert ds.data_source['value'] == '{{ agendas_url }}api/agenda/events-A/datetimes/' + ds = NamedDataSource(name='Foo A') + ds.data_source = {'type': 'json', 'value': 'http://chrono.example.net/api/agenda/events-A/datetimes/'} + ds.external = 'agenda_manual' + ds.store() + ds = NamedDataSource.get(ds.id) + assert ds.data_source['value'] == '{{ agendas_url }}api/agenda/events-A/datetimes/' diff --git a/wcs/data_sources.py b/wcs/data_sources.py index 3e6b4f6de..d3586573b 100644 --- a/wcs/data_sources.py +++ b/wcs/data_sources.py @@ -661,6 +661,20 @@ class NamedDataSource(XmlStorableObject): StorableObject.__init__(self) self.name = name + def migrate(self): + changed = False + + # 2023-05-30 + publisher = get_publisher() + if self.agenda_ds and has_chrono(publisher): + url = (self.data_source or {}).get('value') + if url: + self.data_source['value'] = translate_url(publisher, url) + changed = True + + if changed: + self.store(comment=_('Automatic update'), snapshot_store_user=False) + @property def category(self): return DataSourceCategory.get(self.category_id, ignore_errors=True) @@ -1115,6 +1129,21 @@ def chrono_url(publisher, url): return urllib.parse.urljoin(chrono_url, url) +def chrono_variable(publisher): + chrono_url = publisher.get_site_option('chrono_url') + for key, value in publisher.get_site_options(section='variables').items(): + if value == chrono_url: + return key + + +def translate_url(publisher, url): + variable = chrono_variable(publisher) + if not variable: + return url + chrono_url = publisher.get_site_option('chrono_url') + return url.replace(chrono_url, '{{ %s }}' % variable) + + def collect_agenda_data(publisher): agenda_url = chrono_url(publisher, 'api/agenda/') result = get_json_from_url(agenda_url, log_message_part='agenda') @@ -1175,7 +1204,7 @@ def build_agenda_datasources(publisher, **kwargs): # build datasources from chrono for agenda in agenda_data: - url = agenda['url'] + url = translate_url(publisher, agenda['url']) datasource = existing_datasources.get(url) if datasource is None: datasource = NamedDataSource() diff --git a/wcs/qommon/publisher.py b/wcs/qommon/publisher.py index 58a377fde..8d4a594ac 100644 --- a/wcs/qommon/publisher.py +++ b/wcs/qommon/publisher.py @@ -448,6 +448,14 @@ class QommonPublisher(Publisher): except configparser.NoOptionError: return defaults.get(section, {}).get(option) + def get_site_options(self, section='options'): + if self.site_options is None: + self.load_site_options() + try: + return dict(self.site_options.items(section, raw=True)) + except configparser.NoSectionError: + return {} + def get_site_storages(self): if self.site_options is None: self.load_site_options()