Export caldav with logs
This commit is contained in:
Laurent Chedanne committed
1 parent
af94d397f5
commit
c05fc432eb
5 files changed
+836
-49
No files matched your search
@@ -0,0 +1,364 @@
|
||||
import contextlib
|
||||
import csv
|
||||
import importlib.util
|
||||
import io
|
||||
import re
|
||||
import sys
|
||||
from datetime import date, datetime, time, timezone
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
from types import ModuleType, SimpleNamespace
|
||||
import unittest
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
|
||||
# The regression tests use real iCalendar parsing, with no CalDAV connection.
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
'crealead_caldav', Path(__file__).resolve().parents[1] / 'crealead-caldav.py')
|
||||
exporter = importlib.util.module_from_spec(spec)
|
||||
with patch.dict('sys.modules', {'caldav': ModuleType('caldav')}):
|
||||
spec.loader.exec_module(exporter)
|
||||
|
||||
|
||||
def event_data(properties):
|
||||
return SimpleNamespace(data=(
|
||||
'BEGIN:VCALENDAR\r\nVERSION:2.0\r\nBEGIN:VEVENT\r\n'
|
||||
+ properties.replace('\n', '\r\n')
|
||||
+ '\r\nEND:VEVENT\r\nEND:VCALENDAR\r\n'))
|
||||
|
||||
|
||||
class ExportEventsTests(unittest.TestCase):
|
||||
def export(self, *events):
|
||||
server = SimpleNamespace(name='Agenda de test', date_search=Mock(return_value=events))
|
||||
config = dict(year_start=2026, month_start=10, day_start=1,
|
||||
year_end=2026, month_end=11, day_end=1)
|
||||
errors = io.StringIO()
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
|
||||
contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(errors):
|
||||
with exporter.log_output():
|
||||
result = exporter.get_all_events(config, [
|
||||
{'csv_name': 'test.csv', 'calendar_server': server}])
|
||||
self.log = (Path(directory) / 'logs/export.log').read_text(encoding='utf-8')
|
||||
self.assertTrue(server.date_search.call_args.kwargs['expand'])
|
||||
return result[0]['line'], errors.getvalue()
|
||||
|
||||
def test_timezone_parameters_and_chronological_sort(self):
|
||||
rows, errors = self.export(
|
||||
event_data('SUMMARY:Later\nDTSTART;TZID=Europe/Paris:20261015T140000\n'
|
||||
'DTEND;TZID=Europe/Paris:20261015T150000'),
|
||||
event_data('SUMMARY:Earlier\nDTSTART:20261015T080000Z\n'
|
||||
'DTEND:20261015T090000Z'))
|
||||
self.assertEqual(errors, '')
|
||||
self.assertEqual([row.title for row in rows], ['Earlier', 'Later'])
|
||||
self.assertEqual([row.start_time for row in rows], [time(10), time(14)])
|
||||
self.assertEqual([row.end_time for row in rows], [time(11), time(15)])
|
||||
|
||||
def test_utc_conversion_in_winter(self):
|
||||
rows, errors = self.export(event_data(
|
||||
'SUMMARY:Winter\nDTSTART:20260115T080000Z\nDTEND:20260115T090000Z'))
|
||||
self.assertEqual(errors, '')
|
||||
self.assertEqual(rows[0].start_time, time(9))
|
||||
self.assertEqual(rows[0].end_time, time(10))
|
||||
|
||||
def test_other_timezone_is_converted_to_paris(self):
|
||||
rows, errors = self.export(event_data(
|
||||
'SUMMARY:Call\nDTSTART;TZID=America/New_York:20261015T090000\n'
|
||||
'DTEND;TZID=America/New_York:20261015T100000'))
|
||||
self.assertEqual(errors, '')
|
||||
self.assertEqual(rows[0].start_time, time(15))
|
||||
self.assertEqual(rows[0].end_time, time(16))
|
||||
|
||||
def test_date_only_values_keep_their_dates_at_midnight(self):
|
||||
rows, errors = self.export(event_data(
|
||||
'SUMMARY:Day\nDTSTART;VALUE=DATE:20261015\nDTEND;VALUE=DATE:20261016'))
|
||||
self.assertEqual(errors, '')
|
||||
self.assertEqual(rows[0].start_date, date(2026, 10, 15))
|
||||
self.assertEqual(rows[0].end_date, date(2026, 10, 16))
|
||||
self.assertEqual(rows[0].start_time, time(0))
|
||||
self.assertEqual(rows[0].end_time, time(0))
|
||||
|
||||
def test_floating_times_are_local(self):
|
||||
rows, errors = self.export(event_data(
|
||||
'SUMMARY:Local\nDTSTART:20261015T090000\nDTEND:20261015T100000'))
|
||||
self.assertEqual(errors, '')
|
||||
self.assertEqual(rows[0].start_time, time(9))
|
||||
self.assertEqual(rows[0].end_time, time(10))
|
||||
|
||||
def test_duration_can_supply_end_time(self):
|
||||
rows, errors = self.export(event_data(
|
||||
'SUMMARY:Duration\nDTSTART:20261015T080000Z\nDURATION:PT1H30M'))
|
||||
self.assertEqual(errors, '')
|
||||
self.assertEqual(rows[0].end_time, time(11, 30))
|
||||
|
||||
def test_missing_or_invalid_dates_are_logged_and_skipped(self):
|
||||
for dates in (
|
||||
'DTEND:20261015T090000Z',
|
||||
'DTSTART:20261015T080000Z',
|
||||
'DTSTART:invalid\nDTEND:20261015T090000Z',
|
||||
'DTSTART:20261015T080000Z\nDTEND:invalid',
|
||||
):
|
||||
with self.subTest(dates=dates):
|
||||
rows, errors = self.export(
|
||||
event_data('UID:bad-event\nSUMMARY:Invalid\n' + dates),
|
||||
event_data('SUMMARY:Valid\nDTSTART:20261015T080000Z\n'
|
||||
'DTEND:20261015T090000Z'))
|
||||
self.assertEqual([row.title for row in rows], ['Valid'])
|
||||
self.assertIn('bad-event', errors)
|
||||
self.assertIn('test.csv', errors)
|
||||
self.assertIn('événement non exporté', errors)
|
||||
self.assertIn("Calendrier='Agenda de test'", self.log)
|
||||
self.assertIn("titre='Invalid'", self.log)
|
||||
self.assertIn('début=', self.log)
|
||||
self.assertIn('fin=', self.log)
|
||||
self.assertIn('dates de début/fin absentes ou invalides', self.log)
|
||||
self.assertEqual(self.log.count('[warning][event] - '), 1)
|
||||
|
||||
def test_event_without_summary_is_still_skipped(self):
|
||||
rows, errors = self.export(event_data(
|
||||
'DTSTART:20261015T080000Z\nDTEND:20261015T090000Z'))
|
||||
self.assertEqual(rows, [])
|
||||
self.assertIn('propriété SUMMARY absente', errors)
|
||||
self.assertIn("titre='indisponible'", self.log)
|
||||
self.assertIn("début='2026-10-15T08:00:00Z'", self.log)
|
||||
self.assertIn("Calendrier='Agenda de test'", self.log)
|
||||
self.assertEqual(self.log.count('[warning][event] - '), 1)
|
||||
|
||||
def test_malformed_calendar_does_not_block_other_events(self):
|
||||
rows, errors = self.export(
|
||||
SimpleNamespace(data='not an iCalendar document'),
|
||||
event_data('SUMMARY:Valid\nDTSTART:20261015T080000Z\n'
|
||||
'DTEND:20261015T090000Z'))
|
||||
self.assertEqual([row.title for row in rows], ['Valid'])
|
||||
self.assertIn('document iCalendar illisible', errors)
|
||||
self.assertIn("titre='indisponible'", self.log)
|
||||
self.assertIn("début='indisponible'", self.log)
|
||||
|
||||
def test_malformed_document_recovers_each_event_context(self):
|
||||
rows, errors = self.export(SimpleNamespace(data=(
|
||||
'BEGIN:VEVENT\nUID:first\nSUMMARY;LANGUAGE=fr:Réunion:\n suite\n'
|
||||
'DTSTART;TZID=Europe/Paris:20261015T090000\nEND:VEVENT\n'
|
||||
'BEGIN:VEVENT\nUID:second\nSUMMARY:Autre\n'
|
||||
'DTSTART:20261016T080000Z\nEND:VEVENT\n')))
|
||||
# Multiple root components make Calendar.from_ical reject the document.
|
||||
self.assertEqual(rows, [])
|
||||
self.assertEqual(self.log.count('[warning][event] - '), 2)
|
||||
self.assertIn("titre='Réunion:suite'", self.log)
|
||||
self.assertIn("début='20261015T090000'", self.log)
|
||||
self.assertIn("titre='Autre'", self.log)
|
||||
self.assertIn("UID='second'", self.log)
|
||||
self.assertIn('document iCalendar illisible', errors)
|
||||
|
||||
def test_resource_without_event_is_logged(self):
|
||||
rows, errors = self.export(SimpleNamespace(data=(
|
||||
'BEGIN:VCALENDAR\r\nVERSION:2.0\r\nEND:VCALENDAR\r\n')))
|
||||
self.assertEqual(rows, [])
|
||||
self.assertIn('aucun composant VEVENT', errors)
|
||||
self.assertIn("Calendrier='Agenda de test'", self.log)
|
||||
self.assertEqual(self.log.count('[warning][event] - '), 1)
|
||||
|
||||
def test_invalid_date_keeps_raw_value_in_warning(self):
|
||||
rows, errors = self.export(event_data(
|
||||
'SUMMARY:Invalid\nDTSTART:invalid\nDTEND:20261015T090000Z'))
|
||||
self.assertEqual(rows, [])
|
||||
self.assertIn("début='invalid'", self.log)
|
||||
self.assertIn('dates de début/fin absentes ou invalides', errors)
|
||||
|
||||
def test_text_parameters_folding_colons_and_csv_output(self):
|
||||
rows, errors = self.export(event_data(
|
||||
'SUMMARY;LANGUAGE=fr:Projet: réunion\n'
|
||||
'DTSTART:20261015T080000Z\nDTEND:20261015T090000Z\n'
|
||||
'CATEGORIES:Projet A,Projet B\nLOCATION:Bureau: Montpellier\n'
|
||||
'DESCRIPTION:text/html,%3Cb%3EClient%3C/b%3E:\n suite'))
|
||||
self.assertEqual(errors, '')
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
filename = Path(directory) / 'export.csv'
|
||||
exporter.write_csv(filename, rows)
|
||||
with filename.open(newline='') as csv_file:
|
||||
exported = list(csv.reader(csv_file, delimiter=';'))
|
||||
self.assertEqual(len(exported), 2)
|
||||
self.assertEqual(exported[1], [
|
||||
'Projet: réunion', '2026-10-15', '10:00:00', '2026-10-15',
|
||||
'11:00:00', 'Projet A,Projet B', 'Bureau: Montpellier', 'Client:suite'])
|
||||
|
||||
|
||||
class LogOutputTests(unittest.TestCase):
|
||||
def messages(self, log):
|
||||
messages = []
|
||||
for line in log.splitlines():
|
||||
match = re.fullmatch(r'\[\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z\]: (.*)', line)
|
||||
self.assertIsNotNone(match, line)
|
||||
messages.append(match.group(1))
|
||||
return messages
|
||||
|
||||
def test_timestamp_uses_utc(self):
|
||||
with patch.object(exporter, 'datetime') as clock:
|
||||
clock.now.return_value = datetime(2026, 10, 15, 12, 34, 56, tzinfo=timezone.utc)
|
||||
self.assertEqual(exporter.utc_timestamp(), '2026-10-15T12:34:56Z')
|
||||
clock.now.assert_called_once_with(timezone.utc)
|
||||
|
||||
def test_fragmented_and_multiline_writes_have_one_prefix_per_line(self):
|
||||
console = io.StringIO()
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
|
||||
contextlib.redirect_stdout(console):
|
||||
with exporter.log_output():
|
||||
print('CSV', '...', 2)
|
||||
sys.stdout.write('Début')
|
||||
sys.stdout.write(' de ligne\n\nAutre ligne\n')
|
||||
messages = self.messages((Path(directory) / 'logs/export.log').read_text(encoding='utf-8'))
|
||||
self.assertEqual(messages, [
|
||||
'[info] - CSV ... 2', '[info] - Début de ligne', '[info] - ', '[info] - Autre ligne'])
|
||||
self.assertEqual(console.getvalue(), 'CSV ... 2\nDébut de ligne\n\nAutre ligne\n')
|
||||
|
||||
def test_event_warning_has_own_level_context_and_escaped_newlines(self):
|
||||
console, errors = io.StringIO(), io.StringIO()
|
||||
component = exporter.Calendar.from_ical(event_data(
|
||||
'UID:event-1\nSUMMARY:Réunion\\nSuite\n'
|
||||
'DTSTART;TZID=Europe/Paris:20261015T100000\n'
|
||||
'DTEND;TZID=Europe/Paris:20261015T110000').data).walk('VEVENT')[0]
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
|
||||
contextlib.redirect_stdout(console), contextlib.redirect_stderr(errors):
|
||||
with exporter.log_output():
|
||||
sys.stdout.write('En cours')
|
||||
exporter.log_skipped_event(
|
||||
{'calendar_name': 'Calendrier configuré', 'csv_name': 'test.csv'},
|
||||
component, 'motif\nsur deux lignes')
|
||||
print('Suite')
|
||||
messages = self.messages((Path(directory) / 'logs/export.log').read_text(encoding='utf-8'))
|
||||
self.assertEqual(len(messages), 3)
|
||||
self.assertEqual(messages[0], '[info] - En cours')
|
||||
self.assertTrue(messages[1].startswith('[warning][event] - '))
|
||||
self.assertNotIn('[info]', messages[1])
|
||||
self.assertIn("Calendrier='Calendrier configuré'", messages[1])
|
||||
self.assertIn("titre='Réunion\\nSuite'", messages[1])
|
||||
self.assertIn("début='2026-10-15T08:00:00Z'", messages[1])
|
||||
self.assertIn('motif\\nsur deux lignes', messages[1])
|
||||
self.assertEqual(messages[2], '[info] - Suite')
|
||||
self.assertEqual(console.getvalue(), 'En coursSuite\n')
|
||||
self.assertTrue(errors.getvalue().startswith('[warning][event] - '))
|
||||
|
||||
def test_appended_runs_without_final_newline_remain_separate(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
|
||||
contextlib.redirect_stdout(io.StringIO()):
|
||||
for message in ('Premier', 'Deuxième'):
|
||||
with exporter.log_output():
|
||||
sys.stdout.write(message)
|
||||
self.assertEqual(
|
||||
self.messages((Path(directory) / 'logs/export.log').read_text(encoding='utf-8')),
|
||||
['[info] - Premier', '[info] - Deuxième'])
|
||||
|
||||
def test_system_exit_message_is_saved(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')):
|
||||
with self.assertRaises(SystemExit):
|
||||
with exporter.log_output():
|
||||
sys.exit('Configuration invalide')
|
||||
self.assertEqual(
|
||||
self.messages((Path(directory) / 'logs/export.log').read_text(encoding='utf-8')),
|
||||
['[info] - Configuration invalide'])
|
||||
|
||||
def test_creates_logs_next_to_script_and_copies_both_streams(self):
|
||||
stdout, stderr = io.StringIO(), io.StringIO()
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
script_path = Path(directory) / 'crealead-caldav.py'
|
||||
with patch.object(exporter, '__file__', str(script_path)), \
|
||||
contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
|
||||
with exporter.log_output():
|
||||
print('Export démarré')
|
||||
print('Événement ignoré', file=sys.stderr)
|
||||
sys.stdout.write('Terminé')
|
||||
sys.stdout.flush()
|
||||
self.assertEqual(
|
||||
self.messages((Path(directory) / 'logs/crealead-caldav.log').read_text(encoding='utf-8')),
|
||||
['[info] - Export démarré', '[info] - Événement ignoré', '[info] - Terminé'])
|
||||
self.assertIs(sys.stdout, stdout)
|
||||
self.assertIs(sys.stderr, stderr)
|
||||
self.assertEqual(stdout.getvalue(), 'Export démarré\nTerminé')
|
||||
self.assertEqual(stderr.getvalue(), 'Événement ignoré\n')
|
||||
|
||||
def test_appends_on_next_execution_and_uses_script_name(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
|
||||
contextlib.redirect_stdout(io.StringIO()):
|
||||
for message in ('Premier export', 'Deuxième export'):
|
||||
with exporter.log_output():
|
||||
print(message)
|
||||
self.assertEqual(
|
||||
self.messages((Path(directory) / 'logs/export.log').read_text(encoding='utf-8')),
|
||||
['[info] - Premier export', '[info] - Deuxième export'])
|
||||
|
||||
def test_exception_traceback_is_saved_and_streams_are_restored(self):
|
||||
stdout, stderr = io.StringIO(), io.StringIO()
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
|
||||
contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
|
||||
with self.assertRaisesRegex(ValueError, 'date invalide'):
|
||||
with exporter.log_output():
|
||||
print('Avant erreur')
|
||||
raise ValueError('date invalide')
|
||||
self.assertIs(sys.stdout, stdout)
|
||||
self.assertIs(sys.stderr, stderr)
|
||||
log = (Path(directory) / 'logs/export.log').read_text(encoding='utf-8')
|
||||
messages = self.messages(log)
|
||||
self.assertEqual(messages[0], '[info] - Avant erreur')
|
||||
self.assertTrue(messages[1].startswith('[info] - Traceback'))
|
||||
self.assertTrue(all(message.startswith('[info] - ') for message in messages))
|
||||
self.assertIn('ValueError: date invalide', log)
|
||||
self.assertEqual(stdout.getvalue(), 'Avant erreur\n')
|
||||
self.assertEqual(stderr.getvalue(), '')
|
||||
|
||||
def test_system_exit_keeps_exit_code_without_logging_traceback(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
|
||||
contextlib.redirect_stderr(io.StringIO()):
|
||||
with self.assertRaises(SystemExit) as caught:
|
||||
with exporter.log_output():
|
||||
print('Connexion impossible', file=sys.stderr)
|
||||
sys.exit(2)
|
||||
self.assertEqual(caught.exception.code, 2)
|
||||
self.assertEqual(
|
||||
self.messages((Path(directory) / 'logs/export.log').read_text(encoding='utf-8')),
|
||||
['[info] - Connexion impossible'])
|
||||
|
||||
|
||||
class ExportFailureTests(unittest.TestCase):
|
||||
def test_csv_failure_logs_events_in_current_and_pending_calendars(self):
|
||||
config = dict(year_start=2026, month_start=10, day_start=1,
|
||||
year_end=2026, month_end=11, day_end=1)
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
paths = [Path(directory) / 'first.csv',
|
||||
Path(directory) / 'missing/failed.csv',
|
||||
Path(directory) / 'pending.csv']
|
||||
calendars = [
|
||||
{'csv_name': str(path), 'calendar_server': SimpleNamespace(
|
||||
name='Agenda {}'.format(index), date_search=Mock(return_value=[event_data(
|
||||
'SUMMARY:Event {}\nUID:uid-{}\nDTSTART:20261015T080000Z\n'
|
||||
'DTEND:20261015T090000Z'.format(index, index))]))}
|
||||
for index, path in enumerate(paths)
|
||||
]
|
||||
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
|
||||
patch.object(exporter, 'load_json_config', return_value=config), \
|
||||
patch.object(exporter, 'init_connection', return_value=[]), \
|
||||
patch.object(exporter, 'link_calendars', return_value=calendars), \
|
||||
contextlib.redirect_stdout(io.StringIO()), \
|
||||
contextlib.redirect_stderr(io.StringIO()):
|
||||
with self.assertRaises(OSError):
|
||||
with exporter.log_output():
|
||||
exporter.main()
|
||||
log = (Path(directory) / 'logs/export.log').read_text(encoding='utf-8')
|
||||
self.assertTrue(paths[0].is_file())
|
||||
self.assertFalse(paths[2].exists())
|
||||
warnings = [line for line in log.splitlines() if '[warning][event]' in line]
|
||||
self.assertEqual(len(warnings), 2)
|
||||
for index, warning in enumerate(warnings, start=1):
|
||||
self.assertIn("Calendrier='Agenda {}'".format(index), warning)
|
||||
self.assertIn("titre='Event {}'".format(index), warning)
|
||||
self.assertIn("début='2026-10-15T08:00:00Z'", warning)
|
||||
self.assertIn('export CSV interrompu', warning)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user