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
This commit is contained in:
52
.env.sample
Normal file
52
.env.sample
Normal file
@@ -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
|
||||
4
.gitignore
vendored
Normal file
4
.gitignore
vendored
Normal file
@@ -0,0 +1,4 @@
|
||||
.env
|
||||
__pycache__
|
||||
.venv
|
||||
.pytest_cache
|
||||
81
CLAUDE.md
Normal file
81
CLAUDE.md
Normal file
@@ -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)
|
||||
18
Dockerfile
Normal file
18
Dockerfile
Normal file
@@ -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"]
|
||||
254
README.md
Normal file
254
README.md
Normal file
@@ -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.
|
||||
8
docker-compose.yml
Normal file
8
docker-compose.yml
Normal file
@@ -0,0 +1,8 @@
|
||||
services:
|
||||
imap-calendar-importer:
|
||||
build: .
|
||||
container_name: imap-calendar-importer
|
||||
restart: unless-stopped
|
||||
env_file: ./.env
|
||||
volumes:
|
||||
- ./config:/app/config
|
||||
568
imap_calendar_importer.py
Normal file
568
imap_calendar_importer.py
Normal file
@@ -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()
|
||||
8
requirements.txt
Normal file
8
requirements.txt
Normal file
@@ -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
|
||||
96
tests/conftest.py
Normal file
96
tests/conftest.py
Normal file
@@ -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 """<!DOCTYPE html>
|
||||
<html>
|
||||
<body>
|
||||
<p>Your order details:</p>
|
||||
<table id="productTable">
|
||||
<tr><th>Product</th><th>Quantity</th><th>Price</th></tr>
|
||||
<tr><td>Widget A</td><td>2</td><td>$10.00</td></tr>
|
||||
<tr><td>Widget B</td><td>5</td><td>$25.00</td></tr>
|
||||
</table>
|
||||
</body>
|
||||
</html>"""
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_html_without_product_table():
|
||||
"""Return sample HTML email body without a product table."""
|
||||
return """<!DOCTYPE html>
|
||||
<html>
|
||||
<body>
|
||||
<p>This is a regular email without a product table.</p>
|
||||
</body>
|
||||
</html>"""
|
||||
606
tests/test_imap_calendar_importer.py
Normal file
606
tests/test_imap_calendar_importer.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user