From c87a99f570d1f9523e2ac9f3484916bb796d80ce Mon Sep 17 00:00:00 2001 From: Dan Head Date: Tue, 6 Jan 2026 21:50:45 +0000 Subject: [PATCH] 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 --- .env.sample | 52 +++ .gitignore | 4 + CLAUDE.md | 81 ++++ Dockerfile | 18 + README.md | 254 +++++++++++ docker-compose.yml | 8 + imap_calendar_importer.py | 568 +++++++++++++++++++++++++ requirements.txt | 8 + tests/conftest.py | 96 +++++ tests/test_imap_calendar_importer.py | 606 +++++++++++++++++++++++++++ 10 files changed, 1695 insertions(+) create mode 100644 .env.sample create mode 100644 .gitignore create mode 100644 CLAUDE.md create mode 100644 Dockerfile create mode 100644 README.md create mode 100644 docker-compose.yml create mode 100644 imap_calendar_importer.py create mode 100644 requirements.txt create mode 100644 tests/conftest.py create mode 100644 tests/test_imap_calendar_importer.py diff --git a/.env.sample b/.env.sample new file mode 100644 index 0000000..8d5868f --- /dev/null +++ b/.env.sample @@ -0,0 +1,52 @@ +# IMAP Configuration +IMAP_SERVER=imap.example.com +IMAP_PORT=993 +IMAP_EMAIL=your-email@example.com +IMAP_PASSWORD=your-password +IMAP_MAILBOX=INBOX +IMAP_PROCESSED_FOLDER=Processed + +# IMAP Security Options +# For SSL/TLS (port 993): IMAP_USE_SSL=true, IMAP_USE_STARTTLS=false +# For STARTTLS (port 143): IMAP_USE_SSL=false, IMAP_USE_STARTTLS=true +# For plain (NOT RECOMMENDED): IMAP_USE_SSL=false, IMAP_USE_STARTTLS=false +IMAP_USE_SSL=true +IMAP_USE_STARTTLS=false +# Set to false to accept self-signed certificates +IMAP_VERIFY_SSL=true + +# Nextcloud CalDAV Configuration +CALDAV_URL=https://your-nextcloud.com/remote.php/dav +CALDAV_USERNAME=your-nextcloud-username +CALDAV_PASSWORD=your-nextcloud-password +CALENDAR_NAME=personal +# Set to false to accept self-signed certificates +CALDAV_VERIFY_SSL=true + +# Processing Options +MARK_AS_SEEN=true +MOVE_PROCESSED=false +CHECK_INTERVAL=300 +RUN_ONCE=false + +# Summary Filter Options +# Leave SUMMARY_FILTER empty to import all events +# Examples: +# SUMMARY_FILTER=Meeting (imports only events with "Meeting" in title) +# SUMMARY_FILTER=^Team.* (regex: imports events starting with "Team") +# SUMMARY_FILTER=Dentist (exact match with SUMMARY_FILTER_MODE=exact) +SUMMARY_FILTER= +SUMMARY_FILTER_MODE=contains +INVERT_FILTER=false + +# Deduplication Options +# Use these to replace existing events based on a pattern in the SUMMARY field +# This is useful when the same event (e.g., delivery updates) gets different UIDs +# Examples: +# DEDUPE_BY_SUMMARY_PATTERN=Order (\d+) (extracts order number from "Ocado Delivery (Order 123456)") +# DEDUPE_BY_SUMMARY_PATTERN=Booking #(\w+) (extracts booking reference) +DEDUPE_ENABLED=false +DEDUPE_BY_SUMMARY_PATTERN= + +# Logging +LOG_LEVEL=INFO diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..db713f6 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +.env +__pycache__ +.venv +.pytest_cache diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..4f57559 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,81 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Project Overview + +IMAP to Nextcloud Calendar Importer - A Python utility that monitors an IMAP mailbox for emails with ICS calendar attachments and imports them to a Nextcloud calendar via CalDAV. + +## Commands + +### Build and Run +```bash +# Build Docker image +docker build -t imap-calendar-importer . + +# Run with Docker Compose (continuous mode) +docker-compose up -d + +# Run once and exit +docker-compose run --rm imap-calendar-importer + +# View logs +docker-compose logs -f + +# Rebuild after code changes +docker-compose up -d --build +``` + +### Direct Python Execution +```bash +# Install dependencies +pip install -r requirements.txt + +# Run directly (requires .env or environment variables set) +python imap_calendar_importer.py +``` + +### Testing +```bash +# Create virtual environment and install dependencies +python -m venv .venv +.venv/bin/pip install -r requirements.txt + +# Run all tests +.venv/bin/pytest tests/ -v + +# Run specific test class +.venv/bin/pytest tests/test_imap_calendar_importer.py::TestMatchesFilter -v + +# Run with coverage (requires pytest-cov) +.venv/bin/pytest tests/ --cov=imap_calendar_importer --cov-report=term-missing +``` + +## Architecture + +Single-file Python application (`imap_calendar_importer.py`) with one main class: + +**`IMAPCalendarImporter`** - Handles the complete workflow: +1. `connect_imap()` - Establishes IMAP connection (SSL/TLS, STARTTLS, or plain) +2. `connect_caldav()` - Connects to Nextcloud CalDAV and discovers target calendar +3. `process_emails()` - Searches for unseen emails, extracts ICS attachments +4. `extract_ics_attachments()` - Parses MIME messages, applies summary filtering +5. `import_ics_to_calendar()` - Creates new or updates existing events by UID + +**Execution Modes:** +- Continuous: Polls mailbox every `CHECK_INTERVAL` seconds +- One-time: Processes once and exits (`RUN_ONCE=true`) + +## Configuration + +All configuration via environment variables (see `.env.sample`). Key settings: +- `IMAP_*` - Mail server connection (server, port, credentials, SSL options) +- `CALDAV_*` - Nextcloud connection (URL, credentials, calendar name) +- `SUMMARY_FILTER*` - Event filtering by title (contains/regex/exact, with inversion) +- `RUN_ONCE` / `CHECK_INTERVAL` - Execution mode control + +## Dependencies + +- `caldav` - CalDAV client library +- `icalendar` - ICS parsing +- `urllib3` - HTTP client (used by caldav) diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..41a0387 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,18 @@ +FROM python:3.11-slim + +# Set working directory +WORKDIR /app + +# Install dependencies +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +# Copy application +COPY imap_calendar_importer.py . + +# Run as non-root user +RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app +USER appuser + +# Default command +CMD ["python", "-u", "imap_calendar_importer.py"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..e8ad200 --- /dev/null +++ b/README.md @@ -0,0 +1,254 @@ +# IMAP to Nextcloud Calendar Importer + +Automatically monitors an IMAP mailbox for emails with ICS calendar attachments and imports them to your Nextcloud calendar. + +## Features + +- 📧 Monitors IMAP mailbox for new emails +- 📅 Extracts ICS attachments and imports to Nextcloud +- 🔍 Filter events by SUMMARY field (title) +- 🔐 Supports SSL/TLS, STARTTLS, and self-signed certificates +- 🐳 Fully Dockerized +- ⚙️ Configurable via environment variables +- 🔄 Continuous or one-time execution modes +- 📝 Comprehensive logging + +## Quick Start + +1. **Clone or download the files** + +2. **Create a `.env` file**: + ```bash + cp .env.example .env + ``` + +3. **Edit `.env` with your credentials** + +4. **Build and run with Docker Compose**: + ```bash + docker-compose up -d + ``` + +5. **View logs**: + ```bash + docker-compose logs -f + ``` + +## Configuration + +### Required Environment Variables + +| Variable | Description | Example | +|----------|-------------|---------| +| `IMAP_SERVER` | IMAP server hostname | `imap.gmail.com` | +| `IMAP_PORT` | IMAP port (usually 993) | `993` | +| `IMAP_EMAIL` | Your email address | `user@example.com` | +| `IMAP_PASSWORD` | Email password or app password | `your-password` | +| `CALDAV_URL` | Nextcloud CalDAV URL | `https://cloud.example.com/remote.php/dav` | +| `CALDAV_USERNAME` | Nextcloud username | `username` | +| `CALDAV_PASSWORD` | Nextcloud password or app password | `your-password` | + +### Optional Environment Variables + +| Variable | Default | Description | +|----------|---------|-------------| +| `IMAP_MAILBOX` | `INBOX` | Mailbox to monitor | +| `IMAP_PROCESSED_FOLDER` | `Processed` | Folder for processed emails (if `MOVE_PROCESSED=true`) | +| `IMAP_USE_SSL` | `true` | Use direct SSL/TLS connection (port 993) | +| `IMAP_USE_STARTTLS` | `false` | Use STARTTLS (port 143) - set `IMAP_USE_SSL=false` first | +| `IMAP_VERIFY_SSL` | `true` | Verify SSL certificates (set to `false` for self-signed) | +| `CALENDAR_NAME` | `personal` | Target calendar name in Nextcloud | +| `CALDAV_VERIFY_SSL` | `true` | Verify SSL certificates (set to `false` for self-signed) | +| `MARK_AS_SEEN` | `true` | Mark processed emails as read | +| `MOVE_PROCESSED` | `false` | Move processed emails to separate folder | +| `CHECK_INTERVAL` | `300` | Seconds between checks (continuous mode) | +| `RUN_ONCE` | `false` | Run once and exit (useful for cron) | +| `LOG_LEVEL` | `INFO` | Logging level (`DEBUG`, `INFO`, `WARNING`, `ERROR`) | + +### Summary Filter Options + +Filter calendar events by their SUMMARY (title) field: + +| Variable | Default | Description | +|----------|---------|-------------| +| `SUMMARY_FILTER` | _(empty)_ | Filter pattern (leave empty to import all) | +| `SUMMARY_FILTER_MODE` | `contains` | Filter mode: `contains`, `regex`, or `exact` | +| `INVERT_FILTER` | `false` | If `true`, exclude matches instead of including | + +#### Filter Examples + +**Import only events with "Meeting" in the title:** +```env +SUMMARY_FILTER=Meeting +SUMMARY_FILTER_MODE=contains +INVERT_FILTER=false +``` + +**Import events starting with "Team" (regex):** +```env +SUMMARY_FILTER=^Team.* +SUMMARY_FILTER_MODE=regex +INVERT_FILTER=false +``` + +**Exclude dentist appointments:** +```env +SUMMARY_FILTER=Dentist +SUMMARY_FILTER_MODE=contains +INVERT_FILTER=true +``` + +**Import only exact match "Weekly Standup":** +```env +SUMMARY_FILTER=Weekly Standup +SUMMARY_FILTER_MODE=exact +INVERT_FILTER=false +``` + +## IMAP Connection Modes + +The script supports different IMAP connection security modes: + +### SSL/TLS (Recommended - Port 993) +Direct encrypted connection from the start: +```env +IMAP_USE_SSL=true +IMAP_USE_STARTTLS=false +IMAP_PORT=993 +``` + +### STARTTLS (Port 143) +Starts unencrypted then upgrades to TLS: +```env +IMAP_USE_SSL=false +IMAP_USE_STARTTLS=true +IMAP_PORT=143 +``` + +### Self-Signed Certificates +If your mail server uses self-signed certificates: +```env +IMAP_VERIFY_SSL=false +CALDAV_VERIFY_SSL=false +``` + +**Security Warning**: Only disable SSL verification for trusted internal servers. This makes connections vulnerable to man-in-the-middle attacks. + +### Plain Connection (NOT RECOMMENDED) +Unencrypted connection: +```env +IMAP_USE_SSL=false +IMAP_USE_STARTTLS=false +``` + +## Usage Modes + +### Continuous Mode (Default) + +The container runs continuously and checks for new emails every `CHECK_INTERVAL` seconds: + +```yaml +environment: + - RUN_ONCE=false + - CHECK_INTERVAL=300 +``` + +### One-Time Mode + +Run once and exit (useful with external schedulers like cron): + +```yaml +environment: + - RUN_ONCE=true +``` + +Or run directly: +```bash +docker-compose run --rm imap-calendar-importer +``` + +## Finding Your Nextcloud Calendar Name + +1. Go to your Nextcloud calendar +2. Click the three dots (⋮) next to your calendar +3. Look at the calendar URL or settings +4. The calendar name is usually visible in the settings or URL path + +Alternatively, run the script with `LOG_LEVEL=DEBUG` to see all available calendars. + +## Security Best Practices + +1. **Use app-specific passwords** instead of your main account passwords + - Gmail: [Create app password](https://support.google.com/accounts/answer/185833) + - Nextcloud: Settings → Security → App passwords + +2. **Protect your `.env` file**: + ```bash + chmod 600 .env + ``` + +3. **Never commit credentials** to version control: + ```bash + echo ".env" >> .gitignore + ``` + +## Troubleshooting + +### Can't connect to IMAP +- Check server address and port +- Verify credentials +- For SSL/TLS use port 993, for STARTTLS use port 143 +- Enable "less secure apps" or use app-specific password +- For self-signed certificates: set `IMAP_VERIFY_SSL=false` +- Check firewall rules + +### Can't connect to CalDAV +- Verify Nextcloud URL format: `https://your-nextcloud.com/remote.php/dav` +- Check username and password +- For self-signed certificates: set `CALDAV_VERIFY_SSL=false` +- Ensure CalDAV is enabled in Nextcloud + +### Calendar not found +- Check calendar name matches exactly +- Run with `LOG_LEVEL=DEBUG` to see available calendars +- Ensure you have access to the calendar + +### Events not being filtered correctly +- Check your `SUMMARY_FILTER` pattern +- Try `SUMMARY_FILTER_MODE=contains` for simple matching +- Use `LOG_LEVEL=DEBUG` to see what summaries are being processed + +## Docker Commands + +```bash +# Start the service +docker-compose up -d + +# View logs +docker-compose logs -f + +# Stop the service +docker-compose down + +# Rebuild after code changes +docker-compose up -d --build + +# Run once manually +docker-compose run --rm imap-calendar-importer +``` + +## Development + +Build the image manually: +```bash +docker build -t imap-calendar-importer . +``` + +Run with custom environment: +```bash +docker run --rm --env-file .env imap-calendar-importer +``` + +## License + +MIT License - feel free to modify and use as needed. diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..cf630f0 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,8 @@ +services: + imap-calendar-importer: + build: . + container_name: imap-calendar-importer + restart: unless-stopped + env_file: ./.env + volumes: + - ./config:/app/config diff --git a/imap_calendar_importer.py b/imap_calendar_importer.py new file mode 100644 index 0000000..7d3541c --- /dev/null +++ b/imap_calendar_importer.py @@ -0,0 +1,568 @@ +#!/usr/bin/env python3 +""" +IMAP to Nextcloud Calendar Importer +Monitors an IMAP mailbox for emails with ICS attachments and imports them to Nextcloud calendar +""" + +import imaplib +import email +from email.header import decode_header +import caldav +from caldav.elements import dav, cdav +import logging +import sys +from datetime import datetime +import os +import re +import ssl +from icalendar import Calendar +from bs4 import BeautifulSoup + +# Configuration from environment variables +CONFIG = { + # IMAP Settings + 'IMAP_SERVER': os.getenv('IMAP_SERVER', 'imap.example.com'), + 'IMAP_PORT': int(os.getenv('IMAP_PORT', '993')), + 'IMAP_EMAIL': os.getenv('IMAP_EMAIL', ''), + 'IMAP_PASSWORD': os.getenv('IMAP_PASSWORD', ''), + 'IMAP_MAILBOX': os.getenv('IMAP_MAILBOX', 'INBOX'), + 'IMAP_PROCESSED_FOLDER': os.getenv('IMAP_PROCESSED_FOLDER', 'Processed'), + 'IMAP_USE_SSL': os.getenv('IMAP_USE_SSL', 'true').lower() == 'true', + 'IMAP_USE_STARTTLS': os.getenv('IMAP_USE_STARTTLS', 'false').lower() == 'true', + 'IMAP_VERIFY_SSL': os.getenv('IMAP_VERIFY_SSL', 'true').lower() == 'true', + + # Nextcloud CalDAV Settings + 'CALDAV_URL': os.getenv('CALDAV_URL', ''), + 'CALDAV_USERNAME': os.getenv('CALDAV_USERNAME', ''), + 'CALDAV_PASSWORD': os.getenv('CALDAV_PASSWORD', ''), + 'CALENDAR_NAME': os.getenv('CALENDAR_NAME', 'personal'), + 'CALDAV_VERIFY_SSL': os.getenv('CALDAV_VERIFY_SSL', 'true').lower() == 'true', + + # Processing Settings + 'MARK_AS_SEEN': os.getenv('MARK_AS_SEEN', 'true').lower() == 'true', + 'MOVE_PROCESSED': os.getenv('MOVE_PROCESSED', 'false').lower() == 'true', + 'CHECK_INTERVAL': int(os.getenv('CHECK_INTERVAL', '300')), # seconds + 'RUN_ONCE': os.getenv('RUN_ONCE', 'false').lower() == 'true', + + # Filtering Settings + 'SUMMARY_FILTER': os.getenv('SUMMARY_FILTER', ''), # Regex pattern or simple string + 'SUMMARY_FILTER_MODE': os.getenv('SUMMARY_FILTER_MODE', 'contains'), # contains, regex, exact + 'INVERT_FILTER': os.getenv('INVERT_FILTER', 'false').lower() == 'true', # If true, exclude matches + + # Deduplication Settings + 'DEDUPE_BY_SUMMARY_PATTERN': os.getenv('DEDUPE_BY_SUMMARY_PATTERN', ''), # Regex to extract unique ID from summary (e.g., "Order (\d+)") + 'DEDUPE_ENABLED': os.getenv('DEDUPE_ENABLED', 'false').lower() == 'true', # Enable deduplication by summary pattern + + # Logging + 'LOG_LEVEL': os.getenv('LOG_LEVEL', 'INFO'), +} + +# Setup logging +log_level = getattr(logging, CONFIG['LOG_LEVEL'].upper(), logging.INFO) +logging.basicConfig( + level=log_level, + format='%(asctime)s - %(levelname)s - %(message)s', + handlers=[logging.StreamHandler(sys.stdout)] +) +logger = logging.getLogger(__name__) + + +class IMAPCalendarImporter: + def __init__(self, config): + self.config = config + self.imap = None + self.caldav_client = None + self.calendar = None + + def connect_imap(self): + """Connect to IMAP server""" + try: + logger.info(f"Connecting to IMAP server: {self.config['IMAP_SERVER']}") + + # Create SSL context + ssl_context = ssl.create_default_context() + + # Allow insecure connections if configured + if not self.config['IMAP_VERIFY_SSL']: + logger.warning("IMAP SSL verification disabled - accepting self-signed certificates") + ssl_context.check_hostname = False + ssl_context.verify_mode = ssl.CERT_NONE + + # Choose connection method + if self.config['IMAP_USE_STARTTLS']: + # STARTTLS connection (port 143 typically) + logger.info("Using STARTTLS for IMAP connection") + self.imap = imaplib.IMAP4( + self.config['IMAP_SERVER'], + self.config['IMAP_PORT'] + ) + self.imap.starttls(ssl_context=ssl_context) + elif self.config['IMAP_USE_SSL']: + # Direct SSL/TLS connection (port 993 typically) + logger.info("Using direct SSL/TLS for IMAP connection") + self.imap = imaplib.IMAP4_SSL( + self.config['IMAP_SERVER'], + self.config['IMAP_PORT'], + ssl_context=ssl_context + ) + else: + # Plain connection (not recommended) + logger.warning("Using unencrypted IMAP connection - not recommended!") + self.imap = imaplib.IMAP4( + self.config['IMAP_SERVER'], + self.config['IMAP_PORT'] + ) + + self.imap.login( + self.config['IMAP_EMAIL'], + self.config['IMAP_PASSWORD'] + ) + logger.info("Successfully connected to IMAP server") + return True + except Exception as e: + logger.error(f"Failed to connect to IMAP: {e}") + return False + + def connect_caldav(self): + """Connect to Nextcloud CalDAV""" + try: + logger.info(f"Connecting to CalDAV: {self.config['CALDAV_URL']}") + + # Configure SSL verification + if not self.config['CALDAV_VERIFY_SSL']: + logger.warning("CalDAV SSL verification disabled - accepting self-signed certificates") + import urllib3 + urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + + self.caldav_client = caldav.DAVClient( + url=self.config['CALDAV_URL'], + username=self.config['CALDAV_USERNAME'], + password=self.config['CALDAV_PASSWORD'], + ssl_verify_cert=self.config['CALDAV_VERIFY_SSL'] + ) + + principal = self.caldav_client.principal() + calendars = principal.calendars() + + # Find the target calendar + for cal in calendars: + cal_url = str(cal.url) if hasattr(cal.url, '__str__') else cal.url + cal_name = cal.name if hasattr(cal, 'name') else '' + + logger.debug(f"Checking calendar: {cal_url} (name: {cal_name})") + + if self.config['CALENDAR_NAME'] in cal_url or self.config['CALENDAR_NAME'] == cal_name: + self.calendar = cal + logger.info(f"Found calendar: {cal_url}") + break + + if not self.calendar: + logger.error(f"Calendar '{self.config['CALENDAR_NAME']}' not found") + available_cals = [] + for cal in calendars: + cal_url = str(cal.url) if hasattr(cal.url, '__str__') else cal.url + cal_name = cal.name if hasattr(cal, 'name') else 'unknown' + available_cals.append(f"{cal_name} ({cal_url})") + logger.info(f"Available calendars: {available_cals}") + return False + + logger.info("Successfully connected to CalDAV") + return True + except Exception as e: + logger.error(f"Failed to connect to CalDAV: {e}") + return False + + def decode_header_value(self, value): + """Decode email header value""" + if value is None: + return "" + decoded = decode_header(value) + header_parts = [] + for content, encoding in decoded: + if isinstance(content, bytes): + header_parts.append( + content.decode(encoding or 'utf-8', errors='replace') + ) + else: + header_parts.append(content) + return ''.join(header_parts) + + def matches_filter(self, summary): + """Check if event summary matches the filter""" + if not self.config['SUMMARY_FILTER']: + return True # No filter, accept all + + filter_pattern = self.config['SUMMARY_FILTER'] + mode = self.config['SUMMARY_FILTER_MODE'] + invert = self.config['INVERT_FILTER'] + + try: + if mode == 'exact': + matches = summary == filter_pattern + elif mode == 'regex': + matches = bool(re.search(filter_pattern, summary, re.IGNORECASE)) + else: # contains + matches = filter_pattern.lower() in summary.lower() + + # Invert if configured + return not matches if invert else matches + + except Exception as e: + logger.error(f"Error applying filter: {e}") + return True # On error, accept the event + + def parse_ics_summary(self, ics_data): + """Extract SUMMARY from ICS data""" + try: + cal = Calendar.from_ical(ics_data) + for component in cal.walk(): + if component.name == "VEVENT": + summary = component.get('summary', '') + if summary: + return str(summary) + return "" + except Exception as e: + logger.error(f"Error parsing ICS for summary: {e}") + return "" + + def parse_ics_uid(self, ics_data): + """Extract UID from ICS data""" + try: + cal = Calendar.from_ical(ics_data) + for component in cal.walk(): + if component.name == "VEVENT": + uid = component.get('uid', '') + if uid: + return str(uid) + return None + except Exception as e: + logger.error(f"Error parsing ICS for UID: {e}") + return None + + def extract_dedupe_identifier(self, summary): + """Extract deduplication identifier from summary using configured pattern""" + if not self.config['DEDUPE_ENABLED'] or not self.config['DEDUPE_BY_SUMMARY_PATTERN']: + return None + try: + match = re.search(self.config['DEDUPE_BY_SUMMARY_PATTERN'], summary) + if match: + # Return the first capture group if present, otherwise the whole match + return match.group(1) if match.groups() else match.group(0) + return None + except Exception as e: + logger.error(f"Error extracting dedupe identifier from '{summary}': {e}") + return None + + def find_events_by_summary_pattern(self, dedupe_id): + """Find existing calendar events that match the deduplication identifier""" + matching_events = [] + if not dedupe_id: + return matching_events + try: + # Fetch all events from the calendar + events = self.calendar.events() + for event in events: + try: + cal = Calendar.from_ical(event.data) + for component in cal.walk(): + if component.name == "VEVENT": + event_summary = str(component.get('summary', '')) + event_dedupe_id = self.extract_dedupe_identifier(event_summary) + if event_dedupe_id == dedupe_id: + matching_events.append(event) + logger.debug(f"Found matching event with summary: {event_summary}") + except Exception as e: + logger.debug(f"Error parsing event for deduplication: {e}") + return matching_events + except Exception as e: + logger.error(f"Error searching for events by summary pattern: {e}") + return matching_events + + def extract_product_table(self, email_message): + """Extract and format product table from HTML email body""" + try: + for part in email_message.walk(): + if part.get_content_type() == 'text/html': + html_content = part.get_payload(decode=True) + if html_content: + html_text = html_content.decode('utf-8', errors='replace') + soup = BeautifulSoup(html_text, 'html.parser') + table = soup.find('table', id='productTable') + + if not table: + continue + + lines = [] + for row in table.find_all('tr'): + # Skip header rows (containing th elements) + if row.find('th'): + continue + + cells = row.find_all('td') + if len(cells) >= 2: + # Extract cell text + cell_texts = [cell.get_text(strip=True) for cell in cells] + # Swap 1st and 2nd, add 'x' suffix to 2nd (now 1st) + cell_texts[0], cell_texts[1] = f"{cell_texts[1]}x", cell_texts[0] + lines.append(' '.join(cell_texts)) + + if lines: + return '\n'.join(lines) + return '' + except Exception as e: + logger.error(f"Error extracting product table: {e}") + return '' + + def inject_description(self, ics_data, description): + """Inject description into ICS data""" + if not description: + return ics_data + try: + cal = Calendar.from_ical(ics_data) + for component in cal.walk(): + if component.name == "VEVENT": + component['description'] = description + break + return cal.to_ical().decode('utf-8') + except Exception as e: + logger.error(f"Error injecting description: {e}") + return ics_data + + def extract_ics_attachments(self, email_message): + """Extract ICS files from email message""" + ics_files = [] + + for part in email_message.walk(): + content_type = part.get_content_type() + content_disposition = str(part.get('Content-Disposition', '')) + + # Look for ICS attachments + if 'attachment' in content_disposition or content_type in [ + 'text/calendar', + 'application/ics', + 'application/calendar' + ]: + filename = part.get_filename() + if filename: + filename = self.decode_header_value(filename) + + if filename and filename.lower().endswith('.ics'): + try: + ics_data = part.get_payload(decode=True) + if ics_data: + ics_text = ics_data.decode('utf-8', errors='replace') + summary = self.parse_ics_summary(ics_text) + uid = self.parse_ics_uid(ics_text) + + # Check if summary matches filter + if self.matches_filter(summary): + product_table = self.extract_product_table(email_message) + ics_files.append({ + 'filename': filename, + 'data': ics_text, + 'summary': summary, + 'uid': uid, + 'product_table': product_table + }) + logger.info(f"Found matching ICS attachment: {filename} (Summary: {summary}, UID: {uid})") + if product_table: + logger.info(f"Extracted product table with {len(product_table.splitlines())} items") + else: + logger.info(f"Skipping ICS (filtered out): {filename} (Summary: {summary})") + except Exception as e: + logger.error(f"Error extracting ICS from {filename}: {e}") + + return ics_files + + def import_ics_to_calendar(self, ics_data, filename, summary, uid, product_table=''): + """Import ICS data to Nextcloud calendar, updating if UID exists or deleting duplicates by summary pattern""" + try: + # Inject product table as description if present + ics_data = self.inject_description(ics_data, product_table) + + # Check for deduplication by summary pattern first + dedupe_id = self.extract_dedupe_identifier(summary) + if dedupe_id: + logger.debug(f"Extracted deduplication identifier: {dedupe_id}") + existing_events = self.find_events_by_summary_pattern(dedupe_id) + if existing_events: + logger.info(f"Found {len(existing_events)} existing event(s) matching dedupe ID '{dedupe_id}', deleting them") + for event in existing_events: + try: + event.delete() + logger.info(f"Deleted existing event with matching dedupe ID") + except Exception as e: + logger.error(f"Failed to delete existing event: {e}") + # After deleting duplicates, create the new event + event = self.calendar.save_event(ics_data) + logger.info(f"Successfully imported event {filename} to calendar after deduplication (Summary: {summary}, UID: {uid})") + return True + + # Fall back to UID-based update if no dedupe pattern matched + existing_event = None + if uid: + try: + logger.debug(f"Searching for existing event with UID: {uid}") + # Search calendar for events with this UID + events = self.calendar.search(uid=uid) + if events: + existing_event = events[0] + logger.info(f"Found existing event with UID {uid}, will update it") + except Exception as e: + logger.debug(f"No existing event found with UID {uid}: {e}") + + if existing_event: + # Update existing event + existing_event.data = ics_data + existing_event.save() + logger.info(f"Successfully updated event {filename} in calendar (Summary: {summary}, UID: {uid})") + else: + # Create new event in calendar + event = self.calendar.save_event(ics_data) + logger.info(f"Successfully imported new event {filename} to calendar (Summary: {summary}, UID: {uid})") + + return True + except Exception as e: + logger.error(f"Failed to import {filename}: {e}") + return False + + def process_emails(self): + """Process unread emails with ICS attachments""" + try: + # Select mailbox + self.imap.select(self.config['IMAP_MAILBOX']) + + # Search for unseen emails + status, messages = self.imap.search(None, 'UNSEEN') + + if status != 'OK': + logger.error("Failed to search emails") + return + + email_ids = messages[0].split() + logger.info(f"Found {len(email_ids)} unread emails") + + processed_count = 0 + imported_count = 0 + filtered_count = 0 + + for email_id in email_ids: + try: + # Fetch email + status, msg_data = self.imap.fetch(email_id, '(RFC822)') + + if status != 'OK': + logger.error(f"Failed to fetch email {email_id}") + continue + + # Parse email + email_body = msg_data[0][1] + email_message = email.message_from_bytes(email_body) + + subject = self.decode_header_value(email_message['Subject']) + from_addr = self.decode_header_value(email_message['From']) + + logger.info(f"Processing email: {subject} from {from_addr}") + + # Extract ICS attachments + ics_files = self.extract_ics_attachments(email_message) + + if ics_files: + for ics_file in ics_files: + if self.import_ics_to_calendar( + ics_file['data'], + ics_file['filename'], + ics_file['summary'], + ics_file['uid'], + ics_file.get('product_table', '') + ): + imported_count += 1 + + # Mark as processed + if self.config['MARK_AS_SEEN']: + self.imap.store(email_id, '+FLAGS', '\\Seen') + + if self.config['MOVE_PROCESSED']: + self.imap.copy(email_id, self.config['IMAP_PROCESSED_FOLDER']) + self.imap.store(email_id, '+FLAGS', '\\Deleted') + + processed_count += 1 + + except Exception as e: + logger.error(f"Error processing email {email_id}: {e}") + + # Expunge deleted messages if we moved any + if self.config['MOVE_PROCESSED'] and processed_count > 0: + self.imap.expunge() + + logger.info(f"Processed {processed_count} emails, imported {imported_count} ICS files") + + except Exception as e: + logger.error(f"Error in process_emails: {e}") + + def run(self): + """Main execution""" + logger.info("Starting IMAP to Nextcloud Calendar Importer") + + if not self.connect_imap(): + logger.error("Failed to connect to IMAP, exiting") + return False + + if not self.connect_caldav(): + logger.error("Failed to connect to CalDAV, exiting") + return False + + self.process_emails() + + # Cleanup + try: + self.imap.close() + self.imap.logout() + except: + pass + + logger.info("Completed processing") + return True + + +def main(): + # Validate required configuration + required_vars = ['IMAP_EMAIL', 'IMAP_PASSWORD', 'CALDAV_URL', 'CALDAV_USERNAME', 'CALDAV_PASSWORD'] + missing = [var for var in required_vars if not CONFIG.get(var)] + + if missing: + logger.error(f"Missing required environment variables: {', '.join(missing)}") + sys.exit(1) + + # Log filter configuration + if CONFIG['SUMMARY_FILTER']: + filter_desc = f"{CONFIG['SUMMARY_FILTER_MODE']}: '{CONFIG['SUMMARY_FILTER']}'" + if CONFIG['INVERT_FILTER']: + filter_desc = f"EXCLUDE {filter_desc}" + logger.info(f"Summary filter active: {filter_desc}") + + # Log deduplication configuration + if CONFIG['DEDUPE_ENABLED'] and CONFIG['DEDUPE_BY_SUMMARY_PATTERN']: + logger.info(f"Deduplication enabled with pattern: '{CONFIG['DEDUPE_BY_SUMMARY_PATTERN']}'") + + importer = IMAPCalendarImporter(CONFIG) + + if CONFIG['RUN_ONCE']: + importer.run() + else: + import time + logger.info(f"Running in continuous mode (checking every {CONFIG['CHECK_INTERVAL']} seconds)") + while True: + try: + importer.run() + time.sleep(CONFIG['CHECK_INTERVAL']) + except KeyboardInterrupt: + logger.info("Shutting down...") + break + except Exception as e: + logger.error(f"Unexpected error: {e}") + time.sleep(CONFIG['CHECK_INTERVAL']) + + +if __name__ == '__main__': + main() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..a95b162 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,8 @@ +caldav==1.3.9 +icalendar==5.0.11 +urllib3==2.1.0 +beautifulsoup4>=4.12 + +# Testing +pytest>=8.0 +pytest-mock>=3.12 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..2e0b4c8 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,96 @@ +"""Pytest configuration and fixtures for imap2caldav tests.""" + +import pytest +import sys +from pathlib import Path + +# Add parent directory to path for imports +sys.path.insert(0, str(Path(__file__).parent.parent)) + + +@pytest.fixture +def default_config(): + """Return a default configuration dictionary for testing.""" + return { + 'IMAP_SERVER': 'imap.test.com', + 'IMAP_PORT': 993, + 'IMAP_EMAIL': 'test@test.com', + 'IMAP_PASSWORD': 'testpass', + 'IMAP_MAILBOX': 'INBOX', + 'IMAP_PROCESSED_FOLDER': 'Processed', + 'IMAP_USE_SSL': True, + 'IMAP_USE_STARTTLS': False, + 'IMAP_VERIFY_SSL': True, + 'CALDAV_URL': 'https://caldav.test.com/dav', + 'CALDAV_USERNAME': 'testuser', + 'CALDAV_PASSWORD': 'testpass', + 'CALENDAR_NAME': 'personal', + 'CALDAV_VERIFY_SSL': True, + 'MARK_AS_SEEN': True, + 'MOVE_PROCESSED': False, + 'CHECK_INTERVAL': 300, + 'RUN_ONCE': True, + 'SUMMARY_FILTER': '', + 'SUMMARY_FILTER_MODE': 'contains', + 'INVERT_FILTER': False, + 'DEDUPE_ENABLED': False, + 'DEDUPE_BY_SUMMARY_PATTERN': '', + 'LOG_LEVEL': 'WARNING', + } + + +@pytest.fixture +def sample_ics(): + """Return a sample ICS calendar event.""" + return """BEGIN:VCALENDAR +VERSION:2.0 +PRODID:-//Test//Test//EN +BEGIN:VEVENT +UID:test-event-123@test.com +DTSTART:20240115T100000Z +DTEND:20240115T110000Z +SUMMARY:Test Meeting +DESCRIPTION:This is a test event +END:VEVENT +END:VCALENDAR""" + + +@pytest.fixture +def sample_ics_no_summary(): + """Return a sample ICS calendar event without a summary.""" + return """BEGIN:VCALENDAR +VERSION:2.0 +PRODID:-//Test//Test//EN +BEGIN:VEVENT +UID:no-summary-456@test.com +DTSTART:20240115T100000Z +DTEND:20240115T110000Z +END:VEVENT +END:VCALENDAR""" + + +@pytest.fixture +def sample_html_with_product_table(): + """Return sample HTML email body with a product table.""" + return """ + + +

Your order details:

+ + + + +
ProductQuantityPrice
Widget A2$10.00
Widget B5$25.00
+ +""" + + +@pytest.fixture +def sample_html_without_product_table(): + """Return sample HTML email body without a product table.""" + return """ + + +

This is a regular email without a product table.

+ +""" diff --git a/tests/test_imap_calendar_importer.py b/tests/test_imap_calendar_importer.py new file mode 100644 index 0000000..d27aac5 --- /dev/null +++ b/tests/test_imap_calendar_importer.py @@ -0,0 +1,606 @@ +"""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