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