- Replace placeholder message generation with actual IMAP message fetching using go-message library - Add per-account CouchDB databases for better organization and isolation - Implement native CouchDB attachment storage with proper revision management - Add command line argument parsing with --max-messages flag for controlling message processing limits - Support both sync and archive modes with proper document synchronization - Add comprehensive test environment with Podman containers (GreenMail IMAP server + CouchDB) - Implement full MIME multipart parsing for proper body and attachment extraction - Add TLS and plain IMAP connection support based on port configuration - Update configuration system to support sync vs archive modes - Create test scripts and sample data for development and testing Key technical improvements: - Real email envelope and header processing with go-imap v2 API - MIME Content-Type and Content-Disposition parsing for attachment detection - CouchDB document ID generation using mailbox_uid format for uniqueness - Duplicate detection and prevention to avoid re-storing existing messages - Proper error handling and connection management for IMAP operations 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
129 lines
No EOL
4.4 KiB
Python
Executable file
129 lines
No EOL
4.4 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
|
|
import imaplib
|
|
import email
|
|
from email.mime.text import MIMEText
|
|
from email.mime.multipart import MIMEMultipart
|
|
from email.mime.base import MIMEBase
|
|
from email import encoders
|
|
import time
|
|
import sys
|
|
|
|
def create_simple_message(subject, body, from_addr="test-sender@example.com"):
|
|
"""Create a simple text message"""
|
|
msg = MIMEText(body)
|
|
msg['Subject'] = subject
|
|
msg['From'] = from_addr
|
|
msg['Date'] = email.utils.formatdate(localtime=True)
|
|
return msg.as_string()
|
|
|
|
def create_message_with_attachment(subject, body, attachment_content, from_addr="test-sender@example.com"):
|
|
"""Create a message with an attachment"""
|
|
msg = MIMEMultipart()
|
|
msg['Subject'] = subject
|
|
msg['From'] = from_addr
|
|
msg['Date'] = email.utils.formatdate(localtime=True)
|
|
|
|
# Add body
|
|
msg.attach(MIMEText(body, 'plain'))
|
|
|
|
# Add attachment
|
|
part = MIMEBase('text', 'plain')
|
|
part.set_payload(attachment_content)
|
|
encoders.encode_base64(part)
|
|
part.add_header('Content-Disposition', 'attachment; filename="attachment.txt"')
|
|
msg.attach(part)
|
|
|
|
return msg.as_string()
|
|
|
|
def populate_user_mailbox(username, password, host='localhost', port=3143):
|
|
"""Populate a user's mailbox with test messages"""
|
|
print(f"Connecting to {username}@{host}:{port}")
|
|
|
|
try:
|
|
# Connect to IMAP server
|
|
imap = imaplib.IMAP4(host, port)
|
|
imap.login(username, password)
|
|
imap.select('INBOX')
|
|
|
|
print(f"Creating messages for {username}...")
|
|
|
|
# Create 10 regular messages
|
|
for i in range(1, 11):
|
|
if i % 3 == 0:
|
|
# Every 3rd message has attachment
|
|
msg = create_message_with_attachment(
|
|
f"Test Message {i} with Attachment",
|
|
f"This is test message {i} for {username} with an attachment.",
|
|
f"Sample attachment content for message {i}"
|
|
)
|
|
print(f" Created message {i} (with attachment)")
|
|
else:
|
|
# Simple message
|
|
msg = create_simple_message(
|
|
f"Test Message {i}",
|
|
f"This is test message {i} for {username}."
|
|
)
|
|
print(f" Created message {i}")
|
|
|
|
# Append message to INBOX
|
|
imap.append('INBOX', None, None, msg.encode('utf-8'))
|
|
time.sleep(0.1) # Small delay to avoid overwhelming
|
|
|
|
# Create additional test messages
|
|
special_messages = [
|
|
("Important Message", "This is an important message for testing sync/archive modes."),
|
|
("Message with Special Characters", "This message contains special characters: äöü ñ 中文 🚀")
|
|
]
|
|
|
|
for subject, body in special_messages:
|
|
msg = create_simple_message(subject, body)
|
|
imap.append('INBOX', None, None, msg.encode('utf-8'))
|
|
print(f" Created special message: {subject}")
|
|
|
|
imap.logout()
|
|
print(f"✅ Successfully created 12 messages for {username}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error populating {username}: {e}")
|
|
return False
|
|
|
|
def main():
|
|
print("🚀 Populating GreenMail with test messages using IMAP...")
|
|
|
|
# Test accounts
|
|
accounts = [
|
|
("testuser1", "password123"),
|
|
("testuser2", "password456"),
|
|
("syncuser", "syncpass"),
|
|
("archiveuser", "archivepass")
|
|
]
|
|
|
|
# Wait for GreenMail to be ready
|
|
print("Waiting for GreenMail to be ready...")
|
|
time.sleep(5)
|
|
|
|
success_count = 0
|
|
for username, password in accounts:
|
|
if populate_user_mailbox(username, password):
|
|
success_count += 1
|
|
time.sleep(1) # Brief pause between accounts
|
|
|
|
print(f"\n🎉 Successfully populated {success_count}/{len(accounts)} accounts!")
|
|
|
|
if success_count == len(accounts):
|
|
print("\n✅ All test accounts ready:")
|
|
for username, password in accounts:
|
|
print(f" - {username}:{password}@localhost")
|
|
print(f"\nGreenMail Services:")
|
|
print(f" - IMAP: localhost:3143")
|
|
print(f" - IMAPS: localhost:3993")
|
|
print(f" - SMTP: localhost:3025")
|
|
return 0
|
|
else:
|
|
print(f"\n❌ Failed to populate {len(accounts) - success_count} accounts")
|
|
return 1
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main()) |