365 lines
19 KiB
Python
365 lines
19 KiB
Python
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()
|