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/'

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()

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()