Files
imap2caldav/tests/test_imap_calendar_importer.py
Dan Head c87a99f570 Add IMAP to Nextcloud Calendar importer
- IMAP mailbox monitoring with SSL/TLS, STARTTLS and plain connection support
- ICS attachment extraction from emails and import to Nextcloud via CalDAV
- Create or update calendar events by UID to avoid duplicates
- Event filtering by summary (contains, regex or exact match, with inversion)
- Deduplication by regex pattern extracted from event summaries
- Move or mark processed emails as seen after import
- Continuous polling mode and one-time run mode
- Fully configurable via environment variables
- Docker and Docker Compose support
- Test suite with pytest
2026-04-03 09:47:45 +01:00

607 lines
22 KiB
Python

"""Tests for the IMAP Calendar Importer."""
import pytest
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email import encoders
from unittest.mock import Mock, MagicMock, patch
from imap_calendar_importer import IMAPCalendarImporter
class TestDecodeHeaderValue:
"""Tests for decode_header_value method."""
def test_decode_plain_ascii(self, default_config):
importer = IMAPCalendarImporter(default_config)
result = importer.decode_header_value("Simple Subject")
assert result == "Simple Subject"
def test_decode_none_returns_empty(self, default_config):
importer = IMAPCalendarImporter(default_config)
result = importer.decode_header_value(None)
assert result == ""
def test_decode_utf8_encoded(self, default_config):
importer = IMAPCalendarImporter(default_config)
# RFC 2047 encoded UTF-8 string
encoded = "=?utf-8?b?VGVzdCBTdWJqZWN0?="
result = importer.decode_header_value(encoded)
assert result == "Test Subject"
def test_decode_iso8859_encoded(self, default_config):
importer = IMAPCalendarImporter(default_config)
# RFC 2047 encoded ISO-8859-1 string
encoded = "=?iso-8859-1?q?Caf=E9?="
result = importer.decode_header_value(encoded)
assert result == "Café"
class TestMatchesFilter:
"""Tests for matches_filter method."""
def test_no_filter_accepts_all(self, default_config):
default_config['SUMMARY_FILTER'] = ''
importer = IMAPCalendarImporter(default_config)
assert importer.matches_filter("Any Summary") is True
assert importer.matches_filter("") is True
def test_contains_mode_matches(self, default_config):
default_config['SUMMARY_FILTER'] = 'meeting'
default_config['SUMMARY_FILTER_MODE'] = 'contains'
importer = IMAPCalendarImporter(default_config)
assert importer.matches_filter("Team Meeting") is True
assert importer.matches_filter("MEETING ROOM") is True
assert importer.matches_filter("Lunch Break") is False
def test_exact_mode_matches(self, default_config):
default_config['SUMMARY_FILTER'] = 'Team Meeting'
default_config['SUMMARY_FILTER_MODE'] = 'exact'
importer = IMAPCalendarImporter(default_config)
assert importer.matches_filter("Team Meeting") is True
assert importer.matches_filter("Team Meeting!") is False
assert importer.matches_filter("team meeting") is False
def test_regex_mode_matches(self, default_config):
default_config['SUMMARY_FILTER'] = r'^Team.*\d+'
default_config['SUMMARY_FILTER_MODE'] = 'regex'
importer = IMAPCalendarImporter(default_config)
assert importer.matches_filter("Team Meeting 123") is True
assert importer.matches_filter("team standup 5") is True # case insensitive
assert importer.matches_filter("Team Meeting") is False
def test_invert_filter(self, default_config):
default_config['SUMMARY_FILTER'] = 'spam'
default_config['SUMMARY_FILTER_MODE'] = 'contains'
default_config['INVERT_FILTER'] = True
importer = IMAPCalendarImporter(default_config)
assert importer.matches_filter("Important Meeting") is True
assert importer.matches_filter("This is spam") is False
def test_invalid_regex_accepts_all(self, default_config):
default_config['SUMMARY_FILTER'] = '[invalid regex'
default_config['SUMMARY_FILTER_MODE'] = 'regex'
importer = IMAPCalendarImporter(default_config)
# On error, should accept the event
assert importer.matches_filter("Any Summary") is True
class TestParseIcsSummary:
"""Tests for parse_ics_summary method."""
def test_parse_valid_summary(self, default_config, sample_ics):
importer = IMAPCalendarImporter(default_config)
result = importer.parse_ics_summary(sample_ics)
assert result == "Test Meeting"
def test_parse_no_summary(self, default_config, sample_ics_no_summary):
importer = IMAPCalendarImporter(default_config)
result = importer.parse_ics_summary(sample_ics_no_summary)
assert result == ""
def test_parse_invalid_ics(self, default_config):
importer = IMAPCalendarImporter(default_config)
result = importer.parse_ics_summary("not valid ics data")
assert result == ""
class TestParseIcsUid:
"""Tests for parse_ics_uid method."""
def test_parse_valid_uid(self, default_config, sample_ics):
importer = IMAPCalendarImporter(default_config)
result = importer.parse_ics_uid(sample_ics)
assert result == "test-event-123@test.com"
def test_parse_uid_from_no_summary_event(self, default_config, sample_ics_no_summary):
importer = IMAPCalendarImporter(default_config)
result = importer.parse_ics_uid(sample_ics_no_summary)
assert result == "no-summary-456@test.com"
def test_parse_invalid_ics(self, default_config):
importer = IMAPCalendarImporter(default_config)
result = importer.parse_ics_uid("not valid ics data")
assert result is None
class TestInjectDescription:
"""Tests for inject_description method."""
def test_inject_description(self, default_config, sample_ics):
importer = IMAPCalendarImporter(default_config)
result = importer.inject_description(sample_ics, "New Description")
assert "New Description" in result
def test_inject_empty_description_returns_original(self, default_config, sample_ics):
importer = IMAPCalendarImporter(default_config)
result = importer.inject_description(sample_ics, "")
assert result == sample_ics
def test_inject_none_description_returns_original(self, default_config, sample_ics):
importer = IMAPCalendarImporter(default_config)
result = importer.inject_description(sample_ics, None)
assert result == sample_ics
class TestExtractProductTable:
"""Tests for extract_product_table method."""
def test_extract_product_table(self, default_config, sample_html_with_product_table):
importer = IMAPCalendarImporter(default_config)
# Create a mock email message with HTML part
msg = MIMEMultipart()
html_part = MIMEText(sample_html_with_product_table, 'html')
msg.attach(html_part)
result = importer.extract_product_table(msg)
assert "2x Widget A" in result
assert "5x Widget B" in result
def test_no_product_table(self, default_config, sample_html_without_product_table):
importer = IMAPCalendarImporter(default_config)
msg = MIMEMultipart()
html_part = MIMEText(sample_html_without_product_table, 'html')
msg.attach(html_part)
result = importer.extract_product_table(msg)
assert result == ''
def test_plain_text_email(self, default_config):
importer = IMAPCalendarImporter(default_config)
msg = MIMEText("Plain text email", 'plain')
result = importer.extract_product_table(msg)
assert result == ''
class TestExtractIcsAttachments:
"""Tests for extract_ics_attachments method."""
def test_extract_ics_attachment(self, default_config, sample_ics):
importer = IMAPCalendarImporter(default_config)
# Create email with ICS attachment
msg = MIMEMultipart()
msg['Subject'] = 'Calendar Event'
msg['From'] = 'sender@test.com'
ics_part = MIMEBase('text', 'calendar')
ics_part.set_payload(sample_ics.encode('utf-8'))
encoders.encode_base64(ics_part)
ics_part.add_header('Content-Disposition', 'attachment', filename='event.ics')
msg.attach(ics_part)
result = importer.extract_ics_attachments(msg)
assert len(result) == 1
assert result[0]['filename'] == 'event.ics'
assert result[0]['summary'] == 'Test Meeting'
assert result[0]['uid'] == 'test-event-123@test.com'
def test_no_ics_attachment(self, default_config):
importer = IMAPCalendarImporter(default_config)
msg = MIMEMultipart()
msg['Subject'] = 'Regular Email'
text_part = MIMEText("Just a regular email", 'plain')
msg.attach(text_part)
result = importer.extract_ics_attachments(msg)
assert len(result) == 0
def test_filtered_ics_attachment(self, default_config, sample_ics):
default_config['SUMMARY_FILTER'] = 'Excluded'
default_config['SUMMARY_FILTER_MODE'] = 'contains'
importer = IMAPCalendarImporter(default_config)
msg = MIMEMultipart()
ics_part = MIMEBase('text', 'calendar')
ics_part.set_payload(sample_ics.encode('utf-8'))
encoders.encode_base64(ics_part)
ics_part.add_header('Content-Disposition', 'attachment', filename='event.ics')
msg.attach(ics_part)
result = importer.extract_ics_attachments(msg)
# Should be filtered out because summary doesn't contain 'Excluded'
assert len(result) == 0
class TestConnectImap:
"""Tests for connect_imap method."""
@patch('imap_calendar_importer.imaplib.IMAP4_SSL')
def test_connect_imap_ssl_success(self, mock_imap_ssl, default_config):
mock_imap = MagicMock()
mock_imap_ssl.return_value = mock_imap
importer = IMAPCalendarImporter(default_config)
result = importer.connect_imap()
assert result is True
mock_imap.login.assert_called_once_with('test@test.com', 'testpass')
@patch('imap_calendar_importer.imaplib.IMAP4_SSL')
def test_connect_imap_failure(self, mock_imap_ssl, default_config):
mock_imap_ssl.side_effect = Exception("Connection failed")
importer = IMAPCalendarImporter(default_config)
result = importer.connect_imap()
assert result is False
@patch('imap_calendar_importer.imaplib.IMAP4')
def test_connect_imap_starttls(self, mock_imap, default_config):
default_config['IMAP_USE_SSL'] = False
default_config['IMAP_USE_STARTTLS'] = True
mock_imap_instance = MagicMock()
mock_imap.return_value = mock_imap_instance
importer = IMAPCalendarImporter(default_config)
result = importer.connect_imap()
assert result is True
mock_imap_instance.starttls.assert_called_once()
class TestConnectCaldav:
"""Tests for connect_caldav method."""
@patch('imap_calendar_importer.caldav.DAVClient')
def test_connect_caldav_success(self, mock_dav_client, default_config):
mock_client = MagicMock()
mock_dav_client.return_value = mock_client
mock_calendar = MagicMock()
mock_calendar.url = 'https://caldav.test.com/dav/personal/'
mock_calendar.name = 'personal'
mock_principal = MagicMock()
mock_principal.calendars.return_value = [mock_calendar]
mock_client.principal.return_value = mock_principal
importer = IMAPCalendarImporter(default_config)
result = importer.connect_caldav()
assert result is True
assert importer.calendar == mock_calendar
@patch('imap_calendar_importer.caldav.DAVClient')
def test_connect_caldav_calendar_not_found(self, mock_dav_client, default_config):
mock_client = MagicMock()
mock_dav_client.return_value = mock_client
mock_calendar = MagicMock()
mock_calendar.url = 'https://caldav.test.com/dav/other/'
mock_calendar.name = 'other'
mock_principal = MagicMock()
mock_principal.calendars.return_value = [mock_calendar]
mock_client.principal.return_value = mock_principal
importer = IMAPCalendarImporter(default_config)
result = importer.connect_caldav()
assert result is False
@patch('imap_calendar_importer.caldav.DAVClient')
def test_connect_caldav_failure(self, mock_dav_client, default_config):
mock_dav_client.side_effect = Exception("Connection failed")
importer = IMAPCalendarImporter(default_config)
result = importer.connect_caldav()
assert result is False
class TestImportIcsToCalendar:
"""Tests for import_ics_to_calendar method."""
def test_import_new_event(self, default_config, sample_ics):
importer = IMAPCalendarImporter(default_config)
importer.calendar = MagicMock()
importer.calendar.search.return_value = [] # No existing event
result = importer.import_ics_to_calendar(
sample_ics, 'event.ics', 'Test Meeting', 'test-uid-123'
)
assert result is True
importer.calendar.save_event.assert_called_once()
def test_update_existing_event(self, default_config, sample_ics):
importer = IMAPCalendarImporter(default_config)
importer.calendar = MagicMock()
mock_existing = MagicMock()
importer.calendar.search.return_value = [mock_existing]
result = importer.import_ics_to_calendar(
sample_ics, 'event.ics', 'Test Meeting', 'test-uid-123'
)
assert result is True
mock_existing.save.assert_called_once()
importer.calendar.save_event.assert_not_called()
def test_import_with_product_table(self, default_config, sample_ics):
importer = IMAPCalendarImporter(default_config)
importer.calendar = MagicMock()
importer.calendar.search.return_value = []
result = importer.import_ics_to_calendar(
sample_ics, 'event.ics', 'Test Meeting', 'test-uid-123',
product_table='2x Widget A\n5x Widget B'
)
assert result is True
# Verify the ICS data passed to save_event contains the description
call_args = importer.calendar.save_event.call_args[0][0]
assert '2x Widget A' in call_args
class TestExtractDedupeIdentifier:
"""Tests for extract_dedupe_identifier method."""
def test_extract_order_number(self, default_config):
default_config['DEDUPE_ENABLED'] = True
default_config['DEDUPE_BY_SUMMARY_PATTERN'] = r'Order (\d+)'
importer = IMAPCalendarImporter(default_config)
result = importer.extract_dedupe_identifier("Ocado Delivery (Order 1358138853535)")
assert result == "1358138853535"
def test_extract_with_no_capture_group(self, default_config):
default_config['DEDUPE_ENABLED'] = True
default_config['DEDUPE_BY_SUMMARY_PATTERN'] = r'Order \d+'
importer = IMAPCalendarImporter(default_config)
result = importer.extract_dedupe_identifier("Ocado Delivery (Order 1358138853535)")
assert result == "Order 1358138853535"
def test_no_match_returns_none(self, default_config):
default_config['DEDUPE_ENABLED'] = True
default_config['DEDUPE_BY_SUMMARY_PATTERN'] = r'Order (\d+)'
importer = IMAPCalendarImporter(default_config)
result = importer.extract_dedupe_identifier("Regular Meeting")
assert result is None
def test_disabled_returns_none(self, default_config):
default_config['DEDUPE_ENABLED'] = False
default_config['DEDUPE_BY_SUMMARY_PATTERN'] = r'Order (\d+)'
importer = IMAPCalendarImporter(default_config)
result = importer.extract_dedupe_identifier("Ocado Delivery (Order 1358138853535)")
assert result is None
def test_empty_pattern_returns_none(self, default_config):
default_config['DEDUPE_ENABLED'] = True
default_config['DEDUPE_BY_SUMMARY_PATTERN'] = ''
importer = IMAPCalendarImporter(default_config)
result = importer.extract_dedupe_identifier("Ocado Delivery (Order 1358138853535)")
assert result is None
def test_invalid_regex_returns_none(self, default_config):
default_config['DEDUPE_ENABLED'] = True
default_config['DEDUPE_BY_SUMMARY_PATTERN'] = r'[invalid'
importer = IMAPCalendarImporter(default_config)
result = importer.extract_dedupe_identifier("Ocado Delivery (Order 1358138853535)")
assert result is None
class TestFindEventsBySummaryPattern:
"""Tests for find_events_by_summary_pattern method."""
def test_find_matching_events(self, default_config):
default_config['DEDUPE_ENABLED'] = True
default_config['DEDUPE_BY_SUMMARY_PATTERN'] = r'Order (\d+)'
importer = IMAPCalendarImporter(default_config)
# Create mock calendar with events
mock_event1 = MagicMock()
mock_event1.data = """BEGIN:VCALENDAR
VERSION:2.0
BEGIN:VEVENT
UID:old-uid-1
SUMMARY:Ocado Delivery (Order 1358138853535)
END:VEVENT
END:VCALENDAR"""
mock_event2 = MagicMock()
mock_event2.data = """BEGIN:VCALENDAR
VERSION:2.0
BEGIN:VEVENT
UID:other-uid
SUMMARY:Different Event
END:VEVENT
END:VCALENDAR"""
importer.calendar = MagicMock()
importer.calendar.events.return_value = [mock_event1, mock_event2]
result = importer.find_events_by_summary_pattern("1358138853535")
assert len(result) == 1
assert result[0] == mock_event1
def test_no_matching_events(self, default_config):
default_config['DEDUPE_ENABLED'] = True
default_config['DEDUPE_BY_SUMMARY_PATTERN'] = r'Order (\d+)'
importer = IMAPCalendarImporter(default_config)
mock_event = MagicMock()
mock_event.data = """BEGIN:VCALENDAR
VERSION:2.0
BEGIN:VEVENT
UID:other-uid
SUMMARY:Different Event
END:VEVENT
END:VCALENDAR"""
importer.calendar = MagicMock()
importer.calendar.events.return_value = [mock_event]
result = importer.find_events_by_summary_pattern("1358138853535")
assert len(result) == 0
def test_empty_dedupe_id_returns_empty(self, default_config):
default_config['DEDUPE_ENABLED'] = True
default_config['DEDUPE_BY_SUMMARY_PATTERN'] = r'Order (\d+)'
importer = IMAPCalendarImporter(default_config)
importer.calendar = MagicMock()
result = importer.find_events_by_summary_pattern(None)
assert len(result) == 0
importer.calendar.events.assert_not_called()
class TestImportIcsToCalendarWithDeduplication:
"""Tests for import_ics_to_calendar with deduplication enabled."""
def test_delete_existing_and_create_new(self, default_config):
default_config['DEDUPE_ENABLED'] = True
default_config['DEDUPE_BY_SUMMARY_PATTERN'] = r'Order (\d+)'
importer = IMAPCalendarImporter(default_config)
# Mock existing event
mock_existing = MagicMock()
mock_existing.data = """BEGIN:VCALENDAR
VERSION:2.0
BEGIN:VEVENT
UID:old-uid-1
SUMMARY:Ocado Delivery (Order 1358138853535)
DTSTART:20260113T180000Z
END:VEVENT
END:VCALENDAR"""
importer.calendar = MagicMock()
importer.calendar.events.return_value = [mock_existing]
new_ics = """BEGIN:VCALENDAR
VERSION:2.0
BEGIN:VEVENT
UID:new-uid-2
SUMMARY:Ocado Delivery (Order 1358138853535)
DTSTART:20260113T200000Z
END:VEVENT
END:VCALENDAR"""
result = importer.import_ics_to_calendar(
new_ics, 'event.ics', 'Ocado Delivery (Order 1358138853535)', 'new-uid-2'
)
assert result is True
mock_existing.delete.assert_called_once()
importer.calendar.save_event.assert_called_once()
def test_no_existing_event_just_creates(self, default_config):
default_config['DEDUPE_ENABLED'] = True
default_config['DEDUPE_BY_SUMMARY_PATTERN'] = r'Order (\d+)'
importer = IMAPCalendarImporter(default_config)
importer.calendar = MagicMock()
importer.calendar.events.return_value = []
importer.calendar.search.return_value = []
new_ics = """BEGIN:VCALENDAR
VERSION:2.0
BEGIN:VEVENT
UID:new-uid
SUMMARY:Ocado Delivery (Order 1358138853535)
DTSTART:20260113T200000Z
END:VEVENT
END:VCALENDAR"""
result = importer.import_ics_to_calendar(
new_ics, 'event.ics', 'Ocado Delivery (Order 1358138853535)', 'new-uid'
)
assert result is True
importer.calendar.save_event.assert_called_once()
def test_deduplication_disabled_uses_uid(self, default_config):
# Dedupe disabled by default in fixture
importer = IMAPCalendarImporter(default_config)
importer.calendar = MagicMock()
importer.calendar.search.return_value = []
new_ics = """BEGIN:VCALENDAR
VERSION:2.0
BEGIN:VEVENT
UID:new-uid
SUMMARY:Ocado Delivery (Order 1358138853535)
DTSTART:20260113T200000Z
END:VEVENT
END:VCALENDAR"""
result = importer.import_ics_to_calendar(
new_ics, 'event.ics', 'Ocado Delivery (Order 1358138853535)', 'new-uid'
)
assert result is True
# Should not call calendar.events() for deduplication
importer.calendar.events.assert_not_called()
importer.calendar.save_event.assert_called_once()
class TestRun:
"""Tests for the run method."""
@patch.object(IMAPCalendarImporter, 'connect_imap')
@patch.object(IMAPCalendarImporter, 'connect_caldav')
@patch.object(IMAPCalendarImporter, 'process_emails')
def test_run_success(self, mock_process, mock_caldav, mock_imap, default_config):
mock_imap.return_value = True
mock_caldav.return_value = True
importer = IMAPCalendarImporter(default_config)
importer.imap = MagicMock()
result = importer.run()
assert result is True
mock_process.assert_called_once()
@patch.object(IMAPCalendarImporter, 'connect_imap')
def test_run_imap_failure(self, mock_imap, default_config):
mock_imap.return_value = False
importer = IMAPCalendarImporter(default_config)
result = importer.run()
assert result is False
@patch.object(IMAPCalendarImporter, 'connect_imap')
@patch.object(IMAPCalendarImporter, 'connect_caldav')
def test_run_caldav_failure(self, mock_caldav, mock_imap, default_config):
mock_imap.return_value = True
mock_caldav.return_value = False
importer = IMAPCalendarImporter(default_config)
result = importer.run()
assert result is False