gian/imap-migration-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add comprehensive tests for IMAP email counting, migration, and common functionalities

Browse Source 
- Implement tests for counting emails in single and multiple folders, including handling of empty folders and error scenarios in `test_count_imap_emails.py`.
- Create tests for environment variable verification, IMAP connection handling, folder normalization, MIME header decoding, and message details extraction in `test_imap_common.py`.
- Develop tests for email migration, including duplicate detection, deletion from source after migration, folder creation on destination, and handling of multiple folders in `test_migrate_imap_emails.py`.
- Introduce a mock IMAP server to facilitate testing of IMAP functionalities in `tools/mock_imap_server.py`.
This commit is contained in:
Javier Callico 2026-01-25 11:59:34 -05:00
parent 705cc1548b
commit caf83d60c5
18 changed files with 2354 additions and 209 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

BIN
.coverage Normal file
View File

Binary file not shown.

132
.github/workflows/ci.yml vendored Normal file
View File

@ -0,0 +1,132 @@
name: CI
on:
push:
branches: [main, master]
pull_request:
branches: [main, master]
jobs:
lint:
name: Lint
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Install linting tools
run: |
python -m pip install --upgrade pip
pip install ruff
- name: Run Ruff linter
run: ruff check src/ tools/ test/
- name: Run Ruff formatter check
run: ruff format --check src/ tools/ test/
test:
name: Test (Python ${{ matrix.python-version }})
runs-on: ubuntu-latest
needs: lint
strategy:
fail-fast: false
matrix:
python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install pytest pytest-cov
- name: Run tests with coverage
run: |
PYTHONPATH=src pytest test/ -v --cov=src --cov-report=xml --cov-report=term-missing
- name: Upload coverage to Codecov
if: matrix.python-version == '3.12'
uses: codecov/codecov-action@v4
with:
files: ./coverage.xml
fail_ci_if_error: false
verbose: true
env:
CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
security:
name: Security Scan
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Install Bandit
run: |
python -m pip install --upgrade pip
pip install bandit
- name: Run Bandit security scanner
run: bandit -r src/ -ll -ii
type-check:
name: Type Check
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Install mypy
run: |
python -m pip install --upgrade pip
pip install mypy
- name: Run mypy
run: mypy src/ --ignore-missing-imports --no-error-summary || true
# Using || true as this codebase may not have type hints yet
build:
name: Build Check
runs-on: ubuntu-latest
needs: [lint, test]
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Verify Python syntax
run: |
python -m py_compile src/*.py
python -m py_compile tools/*.py
- name: Check imports
run: |
cd src && python -c "import imap_common; import migrate_imap_emails; import backup_imap_emails; import count_imap_emails; import compare_imap_folders"
echo "All imports successful"

42
Makefile Normal file
View File

@ -0,0 +1,42 @@
.PHONY: test coverage clean lint format install-dev
# Install development dependencies
install-dev:
pip install pytest pytest-cov ruff bandit mypy
# Run linter
lint:
ruff check src/ tools/ test/
# Format code
format:
ruff format src/ tools/ test/
# Check formatting without modifying
format-check:
ruff format --check src/ tools/ test/
# Run tests with pytest
test:
PYTHONPATH=src pytest test/ -v
# Run tests with coverage
coverage:
PYTHONPATH=src pytest test/ -v --cov=src --cov-report=term-missing --cov-report=html
@echo "Coverage report generated in htmlcov/index.html"
# Security scan
security:
bandit -r src/ -ll -ii
# Type check
typecheck:
mypy src/ --ignore-missing-imports
# Run all checks (what CI does)
ci: lint format-check test
# Clean build artifacts
clean:
rm -rf .coverage htmlcov .pytest_cache .mypy_cache .ruff_cache coverage.xml
find . -type d -name "__pycache__" -exec rm -rf {} + 2>/dev/null || true

88
README.md
View File

@ -1,5 +1,9 @@
# IMAP Email Migration Tools
![CI](https://github.com/YOUR_USERNAME/imap-migration-tools/actions/workflows/ci.yml/badge.svg)
![Python](https://img.shields.io/badge/python-3.9%20%7C%203.10%20%7C%203.11%20%7C%203.12%20%7C%203.13-blue)
![License](https://img.shields.io/badge/license-MIT-green)
## Background
I was in need of migrating a Google account with more than 100,000 emails, and none of the freely available solutions worked reliably for me. They often timed out, crashed, or couldn't handle the volume. Hence, I created these simple, robust Python scripts that got the job done effectively.
@ -210,6 +214,90 @@ If you are migrating **from** a Gmail account and using the `--delete` option:
- Instead of simply marking emails as deleted (which Gmail often treats as "Archive"), the script **copies the email to the Trash folder** and then marks the original as deleted.
- This ensures that the storage count in `[Gmail]/All Mail` actually decreases, as the emails are moved to the Trash (which is auto-emptied by Google after 30 days) rather than remaining in your "All Mail" archive.
## Development & Testing
### Setting Up the Development Environment
1. **Clone the repository:**
```bash
git clone https://github.com/YOUR_USERNAME/imap-migration-tools.git
cd imap-migration-tools
```
2. **Create a virtual environment:**
```bash
python3 -m venv .venv
source .venv/bin/activate # On Windows: .venv\Scripts\activate
```
3. **Install development dependencies:**
```bash
pip install -r requirements.txt
# Or using Make:
make install-dev
```
### Running Tests
The project uses `pytest` for testing with a custom mock IMAP server for integration tests.
```bash
# Run all tests
make test
# Run tests with verbose output
PYTHONPATH=src pytest test/ -v
# Run tests with coverage report
make coverage
# Run a specific test file
PYTHONPATH=src pytest test/test_migrate_imap_emails.py -v
# Run a specific test
PYTHONPATH=src pytest test/test_imap_common.py::TestNormalizeFolderName -v
```
### Code Quality
```bash
# Run linter
make lint
# Auto-format code
make format
# Check formatting without modifying
make format-check
# Run security scan
make security
# Run type checker
make typecheck
# Run all CI checks locally
make ci
```
### Test Structure
| Test File | Description |
|-----------|-------------|
| `test_migrate_imap_emails.py` | Email migration tests (basic, duplicates, deletion, folders) |
| `test_backup_imap_emails.py` | Backup functionality tests |
| `test_count_imap_emails.py` | Email counting tests |
| `test_compare_imap_folders.py` | Folder comparison tests |
| `test_imap_common.py` | Shared utility function tests |
### Continuous Integration
The project uses GitHub Actions for CI. On every push and pull request:
- **Lint**: Code style and formatting checks (Ruff)
- **Test**: Runs on Python 3.9, 3.10, 3.11, 3.12, and 3.13
- **Security**: Bandit security scanner
- **Type Check**: mypy static type analysis
## License
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.

63
pyproject.toml Normal file
View File

@ -0,0 +1,63 @@
# Ruff configuration for IMAP Migration Tools
# https://docs.astral.sh/ruff/
[tool.ruff]
target-version = "py39"
line-length = 120
# Include src, tools, and test directories
src = ["src", "tools", "test"]
[tool.ruff.lint]
# Enable recommended rules
select = [
"E", # pycodestyle errors
"W", # pycodestyle warnings
"F", # Pyflakes
"I", # isort
"B", # flake8-bugbear
"C4", # flake8-comprehensions
"UP", # pyupgrade
]
ignore = [
"E501", # line too long (handled by formatter)
"E701", # multiple statements on one line (colon) - used for concise code
"E722", # bare except - used in connection handling
"B008", # do not perform function calls in argument defaults
"B905", # zip without explicit strict parameter
"F841", # local variable assigned but never used - used for clarity
"W293", # blank line contains whitespace - handled by formatter
]
# Allow unused variables when underscore-prefixed
dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
[tool.ruff.lint.per-file-ignores]
"test/*.py" = [
"S101", # assert statements are fine in tests
]
[tool.ruff.format]
quote-style = "double"
indent-style = "space"
skip-magic-trailing-comma = false
line-ending = "auto"
[tool.pytest.ini_options]
testpaths = ["test"]
python_files = ["test_*.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
addopts = "-v --tb=short"
[tool.coverage.run]
source = ["src"]
omit = ["test/*", "tools/*"]
[tool.coverage.report]
exclude_lines = [
"pragma: no cover",
"if __name__ == .__main__.:",
"raise NotImplementedError",
]

6
requirements.txt Normal file
View File

@ -0,0 +1,6 @@
# Development dependencies
pytest>=7.0.0
pytest-cov>=4.0.0
ruff>=0.4.0
bandit>=1.7.0
mypy>=1.0.0

140
src/backup_imap_emails.py
View File

@ -14,18 +14,17 @@ Features:
Configuration:
SRC_IMAP_HOST, SRC_IMAP_USERNAME, SRC_IMAP_PASSWORD: Source credentials.
BACKUP_LOCAL_PATH: Destination local directory.
Usage:
python3 backup_imap_emails.py --dest-path "./my_backup"
"""
import imaplib
import argparse
import concurrent.futures
import os
import sys
import re
import concurrent.futures
import threading
import argparse
import imap_common
# Defaults
@ -36,25 +35,29 @@ BATCH_SIZE = 10
thread_local = threading.local()
print_lock = threading.Lock()
def safe_print(message):
t_name = threading.current_thread().name
short_name = t_name.replace("ThreadPoolExecutor-", "T-").replace("MainThread", "MAIN")
with print_lock:
print(f"[{short_name}] {message}")
def get_thread_connection(src_conf):
if not hasattr(thread_local, "src") or thread_local.src is None:
thread_local.src = imap_common.get_imap_connection(*src_conf)
try:
if thread_local.src: thread_local.src.noop()
if thread_local.src:
thread_local.src.noop()
except:
thread_local.src = imap_common.get_imap_connection(*src_conf)
return thread_local.src
def process_batch(uids, folder_name, src_conf, local_folder_path):
src = get_thread_connection(src_conf)
if not src:
safe_print(f"Error: Could not establish connection for batch.")
safe_print("Error: Could not establish connection for batch.")
return
try:
@ -68,55 +71,56 @@ def process_batch(uids, folder_name, src_conf, local_folder_path):
# 1. Fetch Subject for Filename
# We fetch headers first to generate the nice filename
msg_id, size, subject = imap_common.get_msg_details(src, uid)
# Helper to handle byte UIDs
uid_str = uid.decode('utf-8') if isinstance(uid, bytes) else str(uid)
uid_str = uid.decode("utf-8") if isinstance(uid, bytes) else str(uid)
if not subject:
clean_subject = "No Subject"
else:
clean_subject = imap_common.sanitize_filename(subject)
# Limit subject len to avoid FS path length limits (max 255 usually)
clean_subject = clean_subject[:100]
clean_subject = imap_common.sanitize_filename(subject)
# Limit subject len to avoid FS path length limits (max 255 usually)
clean_subject = clean_subject[:100]
filename = f"{uid_str}_{clean_subject}.eml"
full_path = os.path.join(local_folder_path, filename)
# Double check existence (in case optimization missed it or naming collision)
# Actually, the main purpose here is just to save.
# However, if we migrated to a new naming convention, we might have duplicates with different names?
# The incremental check in `backup_folder` relies on UID prefix, so we are safe.
if os.path.exists(full_path):
continue
continue
# 2. Fetch Full Content
# RFC822 gets the whole message including headers and attachments
resp, data = src.uid('fetch', uid, '(RFC822)')
if resp != 'OK' or not data or data[0] is None:
safe_print(f"[{folder_name}] ERROR Fetch Body | UID {uid_str}")
continue
resp, data = src.uid("fetch", uid, "(RFC822)")
if resp != "OK" or not data or data[0] is None:
safe_print(f"[{folder_name}] ERROR Fetch Body | UID {uid_str}")
continue
raw_email = None
for item in data:
if isinstance(item, tuple):
raw_email = item[1]
break
if raw_email:
try:
with open(full_path, 'wb') as f:
with open(full_path, "wb") as f:
f.write(raw_email)
safe_print(f"[{folder_name}] SAVED | {filename[:60]}...")
except OSError as e:
# Valid filename might still fail if path too long
safe_print(f"[{folder_name}] ERROR Write | {uid_str}: {e}")
else:
safe_print(f"[{folder_name}] EMPTY Content | UID {uid_str}")
safe_print(f"[{folder_name}] EMPTY Content | UID {uid_str}")
except Exception as e:
safe_print(f"[{folder_name}] ERROR Processing UID {uid}: {e}")
def get_existing_uids(local_path):
"""
Scans the local directory for files matching pattern matches {UID}_*.eml
@ -125,28 +129,29 @@ def get_existing_uids(local_path):
existing = set()
if not os.path.exists(local_path):
return existing
try:
for filename in os.listdir(local_path):
if filename.endswith(".eml") and "_" in filename:
# Expecting UID_Subject.eml
parts = filename.split('_', 1)
parts = filename.split("_", 1)
if parts[0].isdigit():
existing.add(parts[0])
except Exception:
pass
return existing
def backup_folder(src_main, folder_name, local_base_path, src_conf):
safe_print(f"--- Processing Folder: {folder_name} ---")
# create local path
# Handle folder separators. IMAP output might be "Parent/Child"
# We rely on OS to handle "Parent/Child" as subdirectories using join
# But clean the segments
cleaned_name = folder_name.replace('/', os.sep)
cleaned_name = folder_name.replace("/", os.sep)
local_folder_path = os.path.join(local_base_path, cleaned_name)
try:
os.makedirs(local_folder_path, exist_ok=True)
except Exception as e:
@ -161,12 +166,13 @@ def backup_folder(src_main, folder_name, local_base_path, src_conf):
return
# Search all
resp, data = src_main.uid('search', None, 'ALL')
if resp != 'OK': return
resp, data = src_main.uid("search", None, "ALL")
if resp != "OK":
return
uids = data[0].split()
total_on_server = len(uids)
if total_on_server == 0:
safe_print(f"Folder {folder_name} is empty.")
return
@ -174,83 +180,87 @@ def backup_folder(src_main, folder_name, local_base_path, src_conf):
# Incremental Optimization
# Read local directory to find UIDs we already have
existing_uids = get_existing_uids(local_folder_path)
# Filter UIDs
# decode uid first if bytes
uids_to_download = []
for u in uids:
u_str = u.decode('utf-8') if isinstance(u, bytes) else str(u)
u_str = u.decode("utf-8") if isinstance(u, bytes) else str(u)
if u_str not in existing_uids:
uids_to_download.append(u)
skipped = total_on_server - len(uids_to_download)
if skipped > 0:
safe_print(f"Skipping {skipped} emails (already exist locally).")
if not uids_to_download:
safe_print(f"Folder {folder_name} is up to date.")
return
safe_print(f"Downloading {len(uids_to_download)} new emails...")
# Create batches
uid_batches = [uids_to_download[i:i + BATCH_SIZE] for i in range(0, len(uids_to_download), BATCH_SIZE)]
uid_batches = [uids_to_download[i : i + BATCH_SIZE] for i in range(0, len(uids_to_download), BATCH_SIZE)]
executor = concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS)
try:
futures = []
for batch in uid_batches:
futures.append(executor.submit(process_batch, batch, folder_name, src_conf, local_folder_path))
futures.append(executor.submit(process_batch, batch, folder_name, src_conf, local_folder_path))
for future in concurrent.futures.as_completed(futures):
future.result()
except KeyboardInterrupt:
executor.shutdown(wait=False, cancel_futures=True)
raise
finally:
executor.shutdown(wait=True)
def main():
parser = argparse.ArgumentParser(description="Backup IMAP emails to local .eml files.")
# Source
parser.add_argument("--src-host", default=os.getenv("SRC_IMAP_HOST"), help="Source IMAP Server")
parser.add_argument("--src-user", default=os.getenv("SRC_IMAP_USERNAME"), help="Source Username")
parser.add_argument("--src-pass", default=os.getenv("SRC_IMAP_PASSWORD"), help="Source Password")
# Destination (Local Path)
env_path = os.getenv("BACKUP_LOCAL_PATH")
parser.add_argument("--dest-path", default=env_path, help="Local destination path (Mandatory)")
# Config
parser.add_argument("--workers", type=int, default=int(os.getenv("MAX_WORKERS", 10)), help="Thread count")
parser.add_argument("--batch", type=int, default=int(os.getenv("BATCH_SIZE", 10)), help="Emails per batch")
parser.add_argument("folder", nargs="?", help="Specific folder to backup")
args = parser.parse_args()
# Validate
missing = []
if not args.src_host: missing.append("SRC_IMAP_HOST")
if not args.src_user: missing.append("SRC_IMAP_USERNAME")
if not args.src_pass: missing.append("SRC_IMAP_PASSWORD")
if not args.src_host:
missing.append("SRC_IMAP_HOST")
if not args.src_user:
missing.append("SRC_IMAP_USERNAME")
if not args.src_pass:
missing.append("SRC_IMAP_PASSWORD")
if missing:
print(f"Error: Missing credentials: {', '.join(missing)}")
sys.exit(1)
if not args.dest_path:
print("Error: Destination path is required.")
print("Please provide --dest-path or set environment variable BACKUP_LOCAL_PATH.")
sys.exit(1)
global MAX_WORKERS, BATCH_SIZE
MAX_WORKERS = args.workers
BATCH_SIZE = args.batch
src_conf = (args.src_host, args.src_user, args.src_pass)
# Expand path (~/...)
local_path = os.path.expanduser(args.dest_path)
if not os.path.exists(local_path):
@ -260,7 +270,7 @@ def main():
except Exception as e:
print(f"Error creating backup directory: {e}")
sys.exit(1)
print("\n--- Configuration Summary ---")
print(f"Source Host : {args.src_host}")
print(f"Source User : {args.src_user}")
@ -268,28 +278,30 @@ def main():
if args.folder:
print(f"Target Folder : {args.folder}")
print("-----------------------------\n")
try:
src = imap_common.get_imap_connection(*src_conf)
if not src: sys.exit(1)
if not src:
sys.exit(1)
if args.folder:
backup_folder(src, args.folder, local_path, src_conf)
backup_folder(src, args.folder, local_path, src_conf)
else:
typ, folders = src.list()
if typ == 'OK':
if typ == "OK":
for f_info in folders:
name = imap_common.normalize_folder_name(f_info)
backup_folder(src, name, local_path, src_conf)
src.logout()
print("\nBackup completed successfully.")
except KeyboardInterrupt:
print("\nBackup interrupted by user.")
sys.exit(0)
except Exception as e:
print(f"Fatal Error: {e}")
if __name__ == "__main__":
main()

51
src/compare_imap_folders.py
View File

@ -20,30 +20,32 @@ Usage:
python3 compare_imap_folders.py
"""
import imaplib
import argparse
import os
import sys
import argparse
import imap_common
def get_email_count(conn, folder_name):
try:
# Select folder in read-only mode
# Quote folder name handles spaces
typ, data = conn.select(f'"{folder_name}"', readonly=True)
if typ != 'OK':
if typ != "OK":
return None
# SELECT command returns the number of messages in data[0]
# data[0] is bytes, e.g. b'123'
if data and data[0]:
return int(data[0])
return 0
except Exception as e:
except Exception:
# print(f"Error checking {folder_name}: {e}")
return None
def main():
parser = argparse.ArgumentParser(description="Compare email counts between two IMAP accounts.")
@ -51,7 +53,7 @@ def main():
parser.add_argument("--src-host", default=os.getenv("SRC_IMAP_HOST"), help="Source IMAP Server")
parser.add_argument("--src-user", default=os.getenv("SRC_IMAP_USERNAME"), help="Source Username")
parser.add_argument("--src-pass", default=os.getenv("SRC_IMAP_PASSWORD"), help="Source Password")
# Dest args
parser.add_argument("--dest-host", default=os.getenv("DEST_IMAP_HOST"), help="Destination IMAP Server")
parser.add_argument("--dest-user", default=os.getenv("DEST_IMAP_USERNAME"), help="Destination Username")
@ -69,12 +71,18 @@ def main():
# Validation
missing_vars = []
if not SRC_HOST: missing_vars.append("SRC_IMAP_HOST")
if not SRC_USER: missing_vars.append("SRC_IMAP_USERNAME")
if not SRC_PASS: missing_vars.append("SRC_IMAP_PASSWORD")
if not DEST_HOST: missing_vars.append("DEST_IMAP_HOST")
if not DEST_USER: missing_vars.append("DEST_IMAP_USERNAME")
if not DEST_PASS: missing_vars.append("DEST_IMAP_PASSWORD")
if not SRC_HOST:
missing_vars.append("SRC_IMAP_HOST")
if not SRC_USER:
missing_vars.append("SRC_IMAP_USERNAME")
if not SRC_PASS:
missing_vars.append("SRC_IMAP_PASSWORD")
if not DEST_HOST:
missing_vars.append("DEST_IMAP_HOST")
if not DEST_USER:
missing_vars.append("DEST_IMAP_USERNAME")
if not DEST_PASS:
missing_vars.append("DEST_IMAP_PASSWORD")
if missing_vars:
print(f"Error: Missing configuration variables: {', '.join(missing_vars)}")
@ -95,17 +103,19 @@ def main():
# Connect to Source
print("Connecting to Source...")
src = imap_common.get_imap_connection(SRC_HOST, SRC_USER, SRC_PASS)
if not src: return
if not src:
return
# Connect to Dest
print("Connecting to Destination...")
dest = imap_common.get_imap_connection(DEST_HOST, DEST_USER, DEST_PASS)
if not dest: return
if not dest:
return
# List Source Folders
print("Listing folders in Source...")
typ, folders = src.list()
if typ != 'OK':
if typ != "OK":
print("Failed to list source folders.")
return
@ -121,14 +131,14 @@ def main():
# Iterate through Source folders
for folder_info in folders:
folder_name = imap_common.normalize_folder_name(folder_info)
# Get Counts
src_count = get_email_count(src, folder_name)
dest_count = get_email_count(dest, folder_name)
# Format for display
src_str = str(src_count) if src_count is not None else "Err"
dest_str = str(dest_count) if dest_count is not None else "N/A" # N/A usually means folder doesn't exist
dest_str = str(dest_count) if dest_count is not None else "N/A" # N/A usually means folder doesn't exist
diff_str = ""
if src_count is not None and dest_count is not None:
@ -142,12 +152,12 @@ def main():
print(f"{folder_name:<40} | {src_str:>10} | {dest_str:>10} | {diff_str:>10}")
print("-" * len(header))
print(f"{'TOTAL':<40} | {total_src:>10} | {total_dest:>10} | {total_src-total_dest:>10}")
print(f"{'TOTAL':<40} | {total_src:>10} | {total_dest:>10} | {total_src - total_dest:>10}")
except Exception as e:
print(f"\nFatal Error: {e}")
if "Too many simultaneous connections" in str(e):
print("Tip: Wait a few minutes for previous connections to timeout or check other active scripts.")
print("Tip: Wait a few minutes for previous connections to timeout or check other active scripts.")
finally:
# Check source connection state and logout if possible
@ -156,7 +166,7 @@ def main():
src.logout()
except Exception:
pass
# Check dest connection state and logout if possible
if dest:
try:
@ -164,5 +174,6 @@ def main():
except Exception:
pass
if __name__ == "__main__":
main()

21
src/count_imap_emails.py
View File

@ -14,28 +14,30 @@ Usage Example:
export IMAP_HOST="imap.gmail.com"
export IMAP_USERNAME="user@gmail.com"
export IMAP_PASSWORD="secretpassword"
python3 count_imap_emails.py
"""
import argparse
import imaplib
import os
import sys
import argparse
import imap_common
def count_emails(imap_server, username, password):
try:
# Connect to the IMAP server (using SSL)
print(f"Connecting to {imap_server}...")
mail = imap_common.get_imap_connection(imap_server, username, password)
if not mail:
return
return
# List all mailboxes
print("Listing mailboxes...")
status, folders = mail.list()
if status != "OK":
print("Failed to list mailboxes.")
return
@ -43,7 +45,7 @@ def count_emails(imap_server, username, password):
total_all_folders = 0
print(f"{'Folder Name':<40} {'Count':>10}")
print("-" * 52)
for folder_info in folders:
folder_name = imap_common.normalize_folder_name(folder_info)
display_name = folder_name
@ -52,13 +54,13 @@ def count_emails(imap_server, username, password):
# Select the mailbox (read-only is sufficient for counting)
# folder_name extracted from list usually handles quotes correctly for select
rv, _ = mail.select(f'"{folder_name}"', readonly=True)
if rv != 'OK':
if rv != "OK":
print(f"{display_name:<40} {'Skipped':>10}")
continue
# Search for all emails
status, data = mail.search(None, "ALL")
if status == "OK":
# data[0] is space separated IDs
email_ids = data[0].split()
@ -66,7 +68,7 @@ def count_emails(imap_server, username, password):
print(f"{display_name:<40} {count:>10}")
total_all_folders += count
else:
print(f"{display_name:<40} {'Error':>10}")
print(f"{display_name:<40} {'Error':>10}")
except imaplib.IMAP4.error:
print(f"{display_name:<40} {'Error':>10}")
@ -82,6 +84,7 @@ def count_emails(imap_server, username, password):
except Exception as e:
print(f"An error occurred: {e}")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Count emails in IMAP account.")

90
src/imap_common.py
View File

@ -6,11 +6,12 @@ Shared functionality for IMAP migration, counting, and comparison scripts.
import imaplib
import os
import sys
import re
import sys
from email.header import decode_header
from email.parser import BytesParser
def verify_env_vars(vars_list):
"""
Checks if all environment variables in the list are set.
@ -23,6 +24,7 @@ def verify_env_vars(vars_list):
return False
return True
def get_imap_connection(host, user, password):
"""
Establishes an SSL connection to the IMAP server and logs in.
@ -40,26 +42,28 @@ def get_imap_connection(host, user, password):
print(f"Connection error to {host}: {e}")
return None
def normalize_folder_name(folder_info_str):
"""
Parses the IMAP list response to extract the clean folder name.
Handles quoted names and flags.
"""
if isinstance(folder_info_str, bytes):
folder_info_str = folder_info_str.decode('utf-8', errors='ignore')
folder_info_str = folder_info_str.decode("utf-8", errors="ignore")
# Regex to extract folder name: (flags) "delimiter" name
# Matches: (\HasNoChildren) "/" "INBOX" OR (\HasNoChildren) "/" Drafts
list_pattern = re.compile(r'\((?P<flags>.*?)\) "(?P<delimiter>.*)" "?(?P<name>.*)"?')
match = list_pattern.search(folder_info_str)
if match:
name = match.group('name')
name = match.group("name")
# If the regex grabbed a trailing quote, strip it (though the regex tries to handle it)
return name.rstrip('"').strip()
# Fallback: take the last part
return folder_info_str.split()[-1].strip('"')
def decode_mime_header(header_value):
"""
Decodes MIME encoded headers (Subject, etc.) to a unicode (str) string.
@ -68,60 +72,62 @@ def decode_mime_header(header_value):
return "(No Subject)"
try:
decoded_list = decode_header(header_value)
default_charset = 'utf-8'
default_charset = "utf-8"
text_parts = []
for bytes_data, encoding in decoded_list:
if isinstance(bytes_data, bytes):
if encoding:
try:
text_parts.append(bytes_data.decode(encoding, errors='ignore'))
text_parts.append(bytes_data.decode(encoding, errors="ignore"))
except LookupError:
text_parts.append(bytes_data.decode(default_charset, errors='ignore'))
text_parts.append(bytes_data.decode(default_charset, errors="ignore"))
else:
text_parts.append(bytes_data.decode(default_charset, errors='ignore'))
text_parts.append(bytes_data.decode(default_charset, errors="ignore"))
else:
text_parts.append(str(bytes_data))
return "".join(text_parts)
except Exception:
return str(header_value)
def get_msg_details(imap_conn, uid):
"""
Fetches simplified message details (Message-ID, Size, Subject) for a given UID.
Returns (msg_id, size, subject) tuple.
"""
try:
resp, data = imap_conn.uid('fetch', uid, '(RFC822.SIZE BODY.PEEK[HEADER.FIELDS (MESSAGE-ID SUBJECT)])')
resp, data = imap_conn.uid("fetch", uid, "(RFC822.SIZE BODY.PEEK[HEADER.FIELDS (MESSAGE-ID SUBJECT)])")
except Exception:
return None, None, None
if resp != 'OK':
if resp != "OK":
return None, None, None
msg_id = None
subject = "(No Subject)"
size = 0
for item in data:
if isinstance(item, tuple):
content = item[0].decode('utf-8', errors='ignore')
content = item[0].decode("utf-8", errors="ignore")
# Parse Size
size_match = re.search(r'RFC822\.SIZE\s+(\d+)', content)
size_match = re.search(r"RFC822\.SIZE\s+(\d+)", content)
if size_match:
size = int(size_match.group(1))
# Parse Headers
msg_bytes = item[1]
parser = BytesParser()
email_obj = parser.parsebytes(msg_bytes)
msg_id = email_obj.get('Message-ID')
raw_subject = email_obj.get('Subject')
msg_id = email_obj.get("Message-ID")
raw_subject = email_obj.get("Subject")
if raw_subject:
subject = decode_mime_header(raw_subject)
return msg_id, size, subject
def message_exists_in_folder(dest_conn, msg_id, src_size):
"""
Checks if a message with the given Message-ID and RFC822.SIZE exists in the CURRENTLY SELECTED folder of dest_conn.
@ -129,32 +135,33 @@ def message_exists_in_folder(dest_conn, msg_id, src_size):
"""
if not msg_id:
return False
clean_id = msg_id.replace('"', '\\"')
try:
typ, data = dest_conn.search(None, f'(HEADER Message-ID "{clean_id}")')
if typ != 'OK':
if typ != "OK":
return False
dest_ids = data[0].split()
if not dest_ids:
return False
for did in dest_ids:
resp, items = dest_conn.fetch(did, '(RFC822.SIZE)')
if resp == 'OK':
resp, items = dest_conn.fetch(did, "(RFC822.SIZE)")
if resp == "OK":
for item in items:
if isinstance(item, bytes):
content = item.decode('utf-8', errors='ignore')
else:
content = item[0].decode('utf-8', errors='ignore')
size_match = re.search(r'RFC822\.SIZE\s+(\d+)', content)
content = item.decode("utf-8", errors="ignore")
else:
content = item[0].decode("utf-8", errors="ignore")
size_match = re.search(r"RFC822\.SIZE\s+(\d+)", content)
if size_match and int(size_match.group(1)) == src_size:
return True
except Exception:
return False
return False
def sanitize_filename(filename):
"""
Sanitizes a string to be safe for use as a filename.
@ -165,12 +172,13 @@ def sanitize_filename(filename):
return "untitled"
# Replace invalid characters with underscore
# Invalid: < > : " / \ | ? * and control chars
s = re.sub(r'[<>:"/\\|?*\x00-\x1f]', '_', filename)
s = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "_", filename)
# Strip leading/trailing whitespaces/dots
s = s.strip().strip('.')
s = s.strip().strip(".")
# Ensure not empty and not too long
return s[:250] if s else "untitled"
def detect_trash_folder(imap_conn):
"""
Attempts to identify the Trash folder in the account.
@ -179,29 +187,29 @@ def detect_trash_folder(imap_conn):
"""
try:
status, folders = imap_conn.list()
if status != 'OK':
if status != "OK":
return None
except Exception:
return None
trash_candidates = ['[Gmail]/Trash', 'Trash', 'Deleted Items', 'Bin', '[Gmail]/Bin']
trash_candidates = ["[Gmail]/Trash", "Trash", "Deleted Items", "Bin", "[Gmail]/Bin"]
detected_by_flag = None
all_folder_names = []
for f in folders:
if isinstance(f, bytes):
f_str = f.decode('utf-8', errors='ignore')
f_str = f.decode("utf-8", errors="ignore")
else:
f_str = str(f)
name = normalize_folder_name(f_str)
all_folder_names.append(name)
# Check for SPECIAL-USE flag \Trash
# The flag is usually inside parentheses like (\HasNoChildren \Trash)
if '\\Trash' in f_str or '\\Bin' in f_str:
if "\\Trash" in f_str or "\\Bin" in f_str:
detected_by_flag = name
if detected_by_flag:
return detected_by_flag
@ -209,5 +217,5 @@ def detect_trash_folder(imap_conn):
for candidate in trash_candidates:
if candidate in all_folder_names:
return candidate
return None

166
src/migrate_imap_emails.py
View File

@ -34,30 +34,29 @@ Usage Example:
export DEST_IMAP_HOST="imap.other.com"
export DEST_IMAP_USERNAME="user@other.com"
export DEST_IMAP_PASSWORD="otherpassword"
python3 migrate_imap_emails.py
"""
import imaplib
import os
import sys
import re
import email
from email.parser import BytesParser
import concurrent.futures
import threading
import argparse
import concurrent.futures
import os
import re
import sys
import threading
import imap_common
# Configuration defaults
DELETE_FROM_SOURCE_DEFAULT = False
MAX_WORKERS = 10 # Initial default, updated in main
BATCH_SIZE = 10 # Initial default, updated in main
BATCH_SIZE = 10 # Initial default, updated in main
# Thread-local storage for IMAP connections
thread_local = threading.local()
print_lock = threading.Lock()
def safe_print(message):
t_name = threading.current_thread().name
# Shorten thread name for cleaner logs e.g. ThreadPoolExecutor-0_0 -> T-0_0
@ -65,26 +64,30 @@ def safe_print(message):
with print_lock:
print(f"[{short_name}] {message}")
def get_thread_connections(src_conf, dest_conf):
# Initialize connections for this thread if they don't exist or are closed
if not hasattr(thread_local, "src") or thread_local.src is None:
thread_local.src = imap_common.get_imap_connection(*src_conf)
if not hasattr(thread_local, "dest") or thread_local.dest is None:
thread_local.dest = imap_common.get_imap_connection(*dest_conf)
# Simple check if alive (noop)
try:
if thread_local.src: thread_local.src.noop()
if thread_local.src:
thread_local.src.noop()
except:
thread_local.src = imap_common.get_imap_connection(*src_conf)
try:
if thread_local.dest: thread_local.dest.noop()
if thread_local.dest:
thread_local.dest.noop()
except:
thread_local.dest = imap_common.get_imap_connection(*dest_conf)
return thread_local.src, thread_local.dest
def process_batch(uids, folder_name, src_conf, dest_conf, delete_from_source, trash_folder=None):
src, dest = get_thread_connections(src_conf, dest_conf)
if not src or not dest:
@ -103,14 +106,14 @@ def process_batch(uids, folder_name, src_conf, dest_conf, delete_from_source, tr
for uid in uids:
try:
msg_id, size, subject = imap_common.get_msg_details(src, uid)
# Format size for display
size_str = f"{size/1024:.1f}KB" if size else "0KB"
size_str = f"{size / 1024:.1f}KB" if size else "0KB"
is_duplicate = False
if msg_id and size:
is_duplicate = imap_common.message_exists_in_folder(dest, msg_id, size)
if is_duplicate:
safe_print(f"[{folder_name}] {'SKIP (Dup)':<18} | {size_str:<8} | {subject[:40]}")
# If it's a duplicate, we can still delete source if requested
@ -118,46 +121,46 @@ def process_batch(uids, folder_name, src_conf, dest_conf, delete_from_source, tr
# Move to trash if configured
if trash_folder and folder_name != trash_folder:
try:
src.uid('copy', uid, f'"{trash_folder}"')
src.uid("copy", uid, f'"{trash_folder}"')
except Exception:
pass
src.uid('store', uid, '+FLAGS', '(\\Deleted)')
src.uid("store", uid, "+FLAGS", "(\\Deleted)")
deleted_count += 1
else:
# Fetch full message
resp, data = src.uid('fetch', uid, '(FLAGS INTERNALDATE BODY.PEEK[])')
if resp != 'OK':
resp, data = src.uid("fetch", uid, "(FLAGS INTERNALDATE BODY.PEEK[])")
if resp != "OK":
safe_print(f"[{folder_name}] ERROR Fetch | {subject[:40]}")
continue
msg_content = None
flags = None
date_str = None
for item in data:
if isinstance(item, tuple):
msg_content = item[1]
meta = item[0].decode('utf-8', errors='ignore')
flags_match = re.search(r'FLAGS\s+\((.*?)\)', meta)
meta = item[0].decode("utf-8", errors="ignore")
flags_match = re.search(r"FLAGS\s+\((.*?)\)", meta)
if flags_match:
flags = flags_match.group(1)
date_match = re.search(r'INTERNALDATE\s+"(.*?)"', meta)
if date_match:
date_str = f'"{date_match.group(1)}"'
if msg_content:
valid_flags = f"({flags})" if flags else None
dest.append(f'"{folder_name}"', valid_flags, date_str, msg_content)
safe_print(f"[{folder_name}] {'COPIED':<18} | {size_str:<8} | {subject[:40]}")
if delete_from_source:
# Move to trash if configured
if trash_folder and folder_name != trash_folder:
try:
src.uid('copy', uid, f'"{trash_folder}"')
src.uid("copy", uid, f'"{trash_folder}"')
except Exception:
pass
src.uid('store', uid, '+FLAGS', '(\\Deleted)')
src.uid("store", uid, "+FLAGS", "(\\Deleted)")
deleted_count += 1
except Exception as e:
@ -170,15 +173,16 @@ def process_batch(uids, folder_name, src_conf, dest_conf, delete_from_source, tr
except Exception as e:
safe_print(f"[{folder_name}] ERROR Expunge: {e}")
def migrate_folder(src, dest, folder_name, delete_from_source, src_conf, dest_conf, trash_folder=None):
safe_print(f"--- Preparing Folder: {folder_name} ---")
# Maintain folder structure
try:
if folder_name.upper() != "INBOX":
dest.create(f'"{folder_name}"')
dest.create(f'"{folder_name}"')
except Exception:
pass # Ignore if exists
pass # Ignore if exists
# Select in main thread to get UIDs
try:
@ -190,13 +194,13 @@ def migrate_folder(src, dest, folder_name, delete_from_source, src_conf, dest_co
# Get UIDs
# Search for UNDELETED to avoid processing messages marked for deletion but not yet expunged
resp, data = src.uid('search', None, 'UNDELETED')
if resp != 'OK':
resp, data = src.uid("search", None, "UNDELETED")
if resp != "OK":
return
uids = data[0].split()
total = len(uids)
if total == 0:
safe_print(f"Folder {folder_name} is empty.")
return
@ -204,22 +208,18 @@ def migrate_folder(src, dest, folder_name, delete_from_source, src_conf, dest_co
safe_print(f"Found {total} messages. Starting parallel migration...")
# Create batches
uid_batches = [uids[i:i + BATCH_SIZE] for i in range(0, len(uids), BATCH_SIZE)]
uid_batches = [uids[i : i + BATCH_SIZE] for i in range(0, len(uids), BATCH_SIZE)]
executor = concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS)
try:
futures = []
for batch in uid_batches:
futures.append(executor.submit(
process_batch,
batch,
folder_name,
src_conf,
dest_conf,
delete_from_source,
trash_folder
))
futures.append(
executor.submit(
process_batch, batch, folder_name, src_conf, dest_conf, delete_from_source, trash_folder
)
)
# Wait for all batches to complete
for future in concurrent.futures.as_completed(futures):
try:
@ -231,7 +231,7 @@ def migrate_folder(src, dest, folder_name, delete_from_source, src_conf, dest_co
executor.shutdown(wait=False, cancel_futures=True)
raise # Re-raise to stop main loop
finally:
executor.shutdown(wait=True)
executor.shutdown(wait=True)
if delete_from_source:
safe_print(f"Expunging any remaining deleted messages from {folder_name}...")
@ -241,9 +241,10 @@ def migrate_folder(src, dest, folder_name, delete_from_source, src_conf, dest_co
except Exception as e:
safe_print(f"Error Expunging: {e}")
def main():
parser = argparse.ArgumentParser(description="Migrate emails between IMAP accounts.")
# Positional arg for folder (optional) to keep backward compatibility with previous quick-fix
parser.add_argument("folder", nargs="?", help="Specific folder to migrate (e.g. '[Gmail]/Important')")
@ -251,7 +252,7 @@ def main():
parser.add_argument("--src-host", default=os.getenv("SRC_IMAP_HOST"), help="Source IMAP Host")
parser.add_argument("--src-user", default=os.getenv("SRC_IMAP_USERNAME"), help="Source Username")
parser.add_argument("--src-pass", default=os.getenv("SRC_IMAP_PASSWORD"), help="Source Password")
# Dest args
parser.add_argument("--dest-host", default=os.getenv("DEST_IMAP_HOST"), help="Destination IMAP Host")
parser.add_argument("--dest-user", default=os.getenv("DEST_IMAP_USERNAME"), help="Destination Username")
@ -260,9 +261,13 @@ def main():
# Options
# Check env var for boolean default (msg "true" -> True)
env_delete = os.getenv("DELETE_FROM_SOURCE", "false").lower() == "true"
parser.add_argument("--delete", action="store_true", default=env_delete, help="Delete from source after migration (default: False)")
parser.add_argument("--workers", type=int, default=int(os.getenv("MAX_WORKERS", 10)), help="Number of concurrent threads")
parser.add_argument(
"--delete", action="store_true", default=env_delete, help="Delete from source after migration (default: False)"
)
parser.add_argument(
"--workers", type=int, default=int(os.getenv("MAX_WORKERS", 10)), help="Number of concurrent threads"
)
parser.add_argument("--batch", type=int, default=int(os.getenv("BATCH_SIZE", 10)), help="Batch size per thread")
args = parser.parse_args()
@ -275,7 +280,7 @@ def main():
DEST_USER = args.dest_user
DEST_PASS = args.dest_pass
DELETE_SOURCE = args.delete
# Folder priority: CLI Arg > Env Var
TARGET_FOLDER = args.folder
if not TARGET_FOLDER and os.getenv("MIGRATE_ONLY_FOLDER"):
@ -288,12 +293,18 @@ def main():
# Validation
missing_vars = []
if not SRC_HOST: missing_vars.append("SRC_IMAP_HOST")
if not SRC_USER: missing_vars.append("SRC_IMAP_USERNAME")
if not SRC_PASS: missing_vars.append("SRC_IMAP_PASSWORD")
if not DEST_HOST: missing_vars.append("DEST_IMAP_HOST")
if not DEST_USER: missing_vars.append("DEST_IMAP_USERNAME")
if not DEST_PASS: missing_vars.append("DEST_IMAP_PASSWORD")
if not SRC_HOST:
missing_vars.append("SRC_IMAP_HOST")
if not SRC_USER:
missing_vars.append("SRC_IMAP_USERNAME")
if not SRC_PASS:
missing_vars.append("SRC_IMAP_PASSWORD")
if not DEST_HOST:
missing_vars.append("DEST_IMAP_HOST")
if not DEST_USER:
missing_vars.append("DEST_IMAP_USERNAME")
if not DEST_PASS:
missing_vars.append("DEST_IMAP_PASSWORD")
if missing_vars:
print(f"Error: Missing configuration variables: {', '.join(missing_vars)}")
@ -318,8 +329,8 @@ def main():
safe_print("Connecting to Source to list folders...")
src_main = imap_common.get_imap_connection(SRC_HOST, SRC_USER, SRC_PASS)
if not src_main:
sys.exit(1)
sys.exit(1)
# Detect Trash Folder if deletion is enabled
trash_folder = None
if DELETE_SOURCE:
@ -328,30 +339,34 @@ def main():
if trash_folder:
safe_print(f"Trash folder detected: '{trash_folder}'. Deleted emails will be moved here first.")
else:
safe_print("Warning: Could not detect Trash folder. Emails will be marked \\Deleted only (standard IMAP delete).")
safe_print(
"Warning: Could not detect Trash folder. Emails will be marked \\Deleted only (standard IMAP delete)."
)
# We need a dummy dest connection just to pass to migrate_folder for folder creation checks?
safe_print("Connecting to Destination...")
dest_main = imap_common.get_imap_connection(DEST_HOST, DEST_USER, DEST_PASS)
if not dest_main:
sys.exit(1)
sys.exit(1)
if TARGET_FOLDER:
# Migration for specific folder
if DELETE_SOURCE and trash_folder and TARGET_FOLDER == trash_folder:
safe_print(f"Aborting: Cannot migrate Trash folder '{TARGET_FOLDER}' while --delete is enabled. This would create a loop.")
sys.exit(1)
if DELETE_SOURCE and trash_folder and trash_folder == TARGET_FOLDER:
safe_print(
f"Aborting: Cannot migrate Trash folder '{TARGET_FOLDER}' while --delete is enabled. This would create a loop."
)
sys.exit(1)
safe_print(f"Starting migration for single folder: {TARGET_FOLDER}")
# Verify folder exists first? imaplib usually handles select error if not found
migrate_folder(src_main, dest_main, TARGET_FOLDER, DELETE_SOURCE, src_conf, dest_conf, trash_folder)
else:
# Migration for all folders
typ, folders = src_main.list()
if typ == 'OK':
if typ == "OK":
for folder_info in folders:
name = imap_common.normalize_folder_name(folder_info)
# Auto-skip trash folder if we are utilizing it as a dump target
# This prevents re-migrating the emails we just moved to trash
if DELETE_SOURCE and trash_folder and name == trash_folder:
@ -359,7 +374,7 @@ def main():
continue
migrate_folder(src_main, dest_main, name, DELETE_SOURCE, src_conf, dest_conf, trash_folder)
src_main.logout()
dest_main.logout()
@ -369,5 +384,6 @@ def main():
except Exception as e:
safe_print(f"Fatal Error: {e}")
if __name__ == "__main__":
main()
main()

130
test/conftest.py Normal file
View File

@ -0,0 +1,130 @@
"""
Shared pytest fixtures and utilities for IMAP migration tools tests.
"""
import imaplib
import os
import socket
import sys
import time
import pytest
# Ensure src/tools are in path
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../src")))
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../tools")))
from mock_imap_server import start_server_thread
def get_free_port():
"""Get a free port on localhost."""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("", 0))
return s.getsockname()[1]
def create_server_pair(src_data=None, dest_data=None):
"""Create a pair of mock IMAP servers with given initial data."""
p1 = get_free_port()
p2 = get_free_port()
while p2 == p1:
p2 = get_free_port()
src_t, src_s = start_server_thread(p1, src_data)
dest_t, dest_s = start_server_thread(p2, dest_data)
time.sleep(0.3)
return (src_t, src_s, p1), (dest_t, dest_s, p2)
def shutdown_server_pair(src_tuple, dest_tuple):
"""Shutdown a pair of mock servers."""
src_t, src_s, _ = src_tuple
dest_t, dest_s, _ = dest_tuple
src_s.shutdown()
dest_s.shutdown()
src_t.join(timeout=2)
dest_t.join(timeout=2)
@pytest.fixture
def mock_server_factory():
"""
Factory fixture that creates mock IMAP server pairs.
Automatically cleans up all servers after the test.
"""
servers = []
def _create(src_data=None, dest_data=None):
src_tuple, dest_tuple = create_server_pair(src_data, dest_data)
servers.append((src_tuple, dest_tuple))
src_server = src_tuple[1]
dest_server = dest_tuple[1]
src_port = src_tuple[2]
dest_port = dest_tuple[2]
return src_server, dest_server, src_port, dest_port
yield _create
for src_tuple, dest_tuple in servers:
shutdown_server_pair(src_tuple, dest_tuple)
@pytest.fixture
def single_mock_server():
"""
Creates a single mock IMAP server for testing scripts that only need one server.
"""
servers = []
def _create(initial_data=None):
port = get_free_port()
thread, server = start_server_thread(port, initial_data)
time.sleep(0.3)
servers.append((thread, server))
return server, port
yield _create
for thread, server in servers:
server.shutdown()
thread.join(timeout=2)
@pytest.fixture(autouse=True)
def clean_sys_argv(monkeypatch):
"""Ensure sys.argv is clean for all tests."""
monkeypatch.setattr(sys, "argv", ["test_script.py"])
def make_mock_connection(src_port, dest_port, src_user="src_user", dest_user="dest_user"):
"""
Creates a mock connection function that routes to the correct server based on username.
"""
def mock_conn(host, user, pwd):
if user == src_user:
port = src_port
elif user == dest_user:
port = dest_port
else:
raise ValueError(f"Unknown user: {user}")
c = imaplib.IMAP4("localhost", port)
c.login(user, pwd)
return c
return mock_conn
def make_single_mock_connection(port):
"""
Creates a mock connection function for a single server.
"""
def mock_conn(host, user, pwd):
c = imaplib.IMAP4("localhost", port)
c.login(user, pwd)
return c
return mock_conn

249
test/test_backup_imap_emails.py Normal file
View File

@ -0,0 +1,249 @@
"""
Tests for backup_imap_emails.py
Tests cover:
- Basic email backup to local files
- Incremental backup (skip existing)
- Multiple folder backup
- Filename sanitization
- Configuration validation
"""
import os
import sys
import pytest
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../src")))
import backup_imap_emails
from conftest import make_single_mock_connection
class TestBackupBasic:
"""Tests for basic backup functionality."""
def test_single_email_backup(self, single_mock_server, monkeypatch, tmp_path):
"""Test backing up a single email to local directory."""
src_data = {"INBOX": [b"Subject: Test Email\r\nMessage-ID: <1@test>\r\n\r\nBody content"]}
server, port = single_mock_server(src_data)
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "user",
"SRC_IMAP_PASSWORD": "pass",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr(sys, "argv", ["backup_imap_emails.py", "--dest-path", str(tmp_path)])
monkeypatch.setattr("imap_common.get_imap_connection", make_single_mock_connection(port))
backup_imap_emails.main()
# Check that backup was created
inbox_path = tmp_path / "INBOX"
assert inbox_path.exists()
eml_files = list(inbox_path.glob("*.eml"))
assert len(eml_files) == 1
# Check content
content = eml_files[0].read_bytes()
assert b"Test Email" in content or b"Body content" in content
def test_multiple_emails_backup(self, single_mock_server, monkeypatch, tmp_path):
"""Test backing up multiple emails."""
src_data = {
"INBOX": [
b"Subject: Email 1\r\nMessage-ID: <1@test>\r\n\r\nBody 1",
b"Subject: Email 2\r\nMessage-ID: <2@test>\r\n\r\nBody 2",
b"Subject: Email 3\r\nMessage-ID: <3@test>\r\n\r\nBody 3",
]
}
server, port = single_mock_server(src_data)
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "user",
"SRC_IMAP_PASSWORD": "pass",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr(sys, "argv", ["backup_imap_emails.py", "--dest-path", str(tmp_path)])
monkeypatch.setattr("imap_common.get_imap_connection", make_single_mock_connection(port))
backup_imap_emails.main()
inbox_path = tmp_path / "INBOX"
eml_files = list(inbox_path.glob("*.eml"))
assert len(eml_files) == 3
class TestIncrementalBackup:
"""Tests for incremental backup functionality."""
def test_skip_existing_emails(self, single_mock_server, monkeypatch, tmp_path):
"""Test that existing emails are skipped during incremental backup."""
src_data = {
"INBOX": [
b"Subject: Email 1\r\nMessage-ID: <1@test>\r\n\r\nBody 1",
b"Subject: Email 2\r\nMessage-ID: <2@test>\r\n\r\nBody 2",
]
}
server, port = single_mock_server(src_data)
# Create pre-existing file
inbox_path = tmp_path / "INBOX"
inbox_path.mkdir()
existing_file = inbox_path / "1_Email_1.eml"
existing_file.write_bytes(b"existing content")
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "user",
"SRC_IMAP_PASSWORD": "pass",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr(sys, "argv", ["backup_imap_emails.py", "--dest-path", str(tmp_path)])
monkeypatch.setattr("imap_common.get_imap_connection", make_single_mock_connection(port))
backup_imap_emails.main()
# Should still have original content (not overwritten)
assert existing_file.read_bytes() == b"existing content"
# But second email should be backed up
eml_files = list(inbox_path.glob("*.eml"))
assert len(eml_files) == 2
class TestMultipleFolderBackup:
"""Tests for backing up multiple folders."""
def test_backup_all_folders(self, single_mock_server, monkeypatch, tmp_path):
"""Test backing up emails from multiple folders."""
src_data = {
"INBOX": [b"Subject: Inbox\r\nMessage-ID: <1@test>\r\n\r\nC"],
"Sent": [b"Subject: Sent\r\nMessage-ID: <2@test>\r\n\r\nC"],
"Archive": [b"Subject: Archive\r\nMessage-ID: <3@test>\r\n\r\nC"],
}
server, port = single_mock_server(src_data)
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "user",
"SRC_IMAP_PASSWORD": "pass",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr(sys, "argv", ["backup_imap_emails.py", "--dest-path", str(tmp_path)])
monkeypatch.setattr("imap_common.get_imap_connection", make_single_mock_connection(port))
backup_imap_emails.main()
assert (tmp_path / "INBOX").exists()
assert (tmp_path / "Sent").exists()
assert (tmp_path / "Archive").exists()
def test_backup_single_folder(self, single_mock_server, monkeypatch, tmp_path):
"""Test backing up a specific folder only."""
src_data = {
"INBOX": [b"Subject: Inbox\r\nMessage-ID: <1@test>\r\n\r\nC"],
"Sent": [b"Subject: Sent\r\nMessage-ID: <2@test>\r\n\r\nC"],
}
server, port = single_mock_server(src_data)
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "user",
"SRC_IMAP_PASSWORD": "pass",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr(sys, "argv", ["backup_imap_emails.py", "--dest-path", str(tmp_path), "INBOX"])
monkeypatch.setattr("imap_common.get_imap_connection", make_single_mock_connection(port))
backup_imap_emails.main()
assert (tmp_path / "INBOX").exists()
# Sent should NOT be backed up
assert not (tmp_path / "Sent").exists()
class TestEmptyFolderHandling:
"""Tests for empty folder handling."""
def test_empty_folder(self, single_mock_server, monkeypatch, tmp_path):
"""Test handling of empty folders."""
src_data = {"INBOX": [], "Empty": []}
server, port = single_mock_server(src_data)
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "user",
"SRC_IMAP_PASSWORD": "pass",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr(sys, "argv", ["backup_imap_emails.py", "--dest-path", str(tmp_path)])
monkeypatch.setattr("imap_common.get_imap_connection", make_single_mock_connection(port))
# Should complete without error
backup_imap_emails.main()
class TestConfigValidation:
"""Tests for configuration validation."""
def test_missing_credentials(self, monkeypatch, capsys):
"""Test that missing credentials cause exit."""
env = {}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr(sys, "argv", ["backup_imap_emails.py", "--dest-path", "/tmp"])
with pytest.raises(SystemExit) as exc_info:
backup_imap_emails.main()
assert exc_info.value.code == 1
def test_missing_dest_path(self, monkeypatch, capsys):
"""Test that missing destination path causes exit."""
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "user",
"SRC_IMAP_PASSWORD": "pass",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr(sys, "argv", ["backup_imap_emails.py"])
with pytest.raises(SystemExit) as exc_info:
backup_imap_emails.main()
assert exc_info.value.code == 1
class TestGetExistingUids:
"""Tests for get_existing_uids function."""
def test_empty_directory(self, tmp_path):
"""Test with empty directory."""
result = backup_imap_emails.get_existing_uids(str(tmp_path))
assert result == set()
def test_nonexistent_directory(self):
"""Test with non-existent directory."""
result = backup_imap_emails.get_existing_uids("/nonexistent/path")
assert result == set()
def test_existing_files(self, tmp_path):
"""Test extraction of UIDs from existing files."""
(tmp_path / "1_Subject.eml").touch()
(tmp_path / "2_Another.eml").touch()
(tmp_path / "100_Long_Subject.eml").touch()
result = backup_imap_emails.get_existing_uids(str(tmp_path))
assert result == {"1", "2", "100"}
def test_ignore_non_matching_files(self, tmp_path):
"""Test that non-matching files are ignored."""
(tmp_path / "not_an_email.txt").touch()
(tmp_path / "random.eml").touch() # No underscore
(tmp_path / "1_Subject.eml").touch()
result = backup_imap_emails.get_existing_uids(str(tmp_path))
assert result == {"1"}

240
test/test_compare_imap_folders.py Normal file
View File

@ -0,0 +1,240 @@
"""
Tests for compare_imap_folders.py
Tests cover:
- Basic folder comparison between accounts
- Matching counts
- Mismatched counts
- Missing folders on destination
- Empty folder handling
- Configuration validation
"""
import imaplib
import os
import sys
import pytest
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../src")))
import compare_imap_folders
from conftest import make_mock_connection
class TestFolderComparison:
"""Tests for folder comparison functionality."""
def test_matching_counts(self, mock_server_factory, monkeypatch, capsys):
"""Test comparison when source and destination have matching counts."""
data = {
"INBOX": [b"Subject: 1\r\n\r\nB", b"Subject: 2\r\n\r\nB"],
"Sent": [b"Subject: 3\r\n\r\nB"],
}
src_server, dest_server, p1, p2 = mock_server_factory(data, data.copy())
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "src_user",
"SRC_IMAP_PASSWORD": "p",
"DEST_IMAP_HOST": "localhost",
"DEST_IMAP_USERNAME": "dest_user",
"DEST_IMAP_PASSWORD": "p",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr(sys, "argv", ["compare_imap_folders.py"])
monkeypatch.setattr("imap_common.get_imap_connection", make_mock_connection(p1, p2))
compare_imap_folders.main()
captured = capsys.readouterr()
assert "INBOX" in captured.out
assert "Sent" in captured.out
def test_mismatched_counts(self, mock_server_factory, monkeypatch, capsys):
"""Test comparison when counts differ."""
src_data = {
"INBOX": [b"Subject: 1\r\n\r\nB", b"Subject: 2\r\n\r\nB", b"Subject: 3\r\n\r\nB"],
}
dest_data = {
"INBOX": [b"Subject: 1\r\n\r\nB"],
}
src_server, dest_server, p1, p2 = mock_server_factory(src_data, dest_data)
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "src_user",
"SRC_IMAP_PASSWORD": "p",
"DEST_IMAP_HOST": "localhost",
"DEST_IMAP_USERNAME": "dest_user",
"DEST_IMAP_PASSWORD": "p",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr(sys, "argv", ["compare_imap_folders.py"])
monkeypatch.setattr("imap_common.get_imap_connection", make_mock_connection(p1, p2))
compare_imap_folders.main()
captured = capsys.readouterr()
# Source has 3, dest has 1
assert "3" in captured.out
assert "1" in captured.out
def test_folder_missing_on_destination(self, mock_server_factory, monkeypatch, capsys):
"""Test when a folder exists on source but not destination."""
src_data = {
"INBOX": [b"Subject: 1\r\n\r\nB"],
"Archive": [b"Subject: 2\r\n\r\nB"],
}
dest_data = {
"INBOX": [b"Subject: 1\r\n\r\nB"],
}
src_server, dest_server, p1, p2 = mock_server_factory(src_data, dest_data)
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "src_user",
"SRC_IMAP_PASSWORD": "p",
"DEST_IMAP_HOST": "localhost",
"DEST_IMAP_USERNAME": "dest_user",
"DEST_IMAP_PASSWORD": "p",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr(sys, "argv", ["compare_imap_folders.py"])
monkeypatch.setattr("imap_common.get_imap_connection", make_mock_connection(p1, p2))
compare_imap_folders.main()
captured = capsys.readouterr()
# Archive should show N/A for destination
assert "Archive" in captured.out
assert "N/A" in captured.out
class TestEmptyFolders:
"""Tests for empty folder handling."""
def test_empty_folders(self, mock_server_factory, monkeypatch, capsys):
"""Test comparison with empty folders."""
src_data = {"INBOX": [], "Empty": []}
dest_data = {"INBOX": [], "Empty": []}
src_server, dest_server, p1, p2 = mock_server_factory(src_data, dest_data)
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "src_user",
"SRC_IMAP_PASSWORD": "p",
"DEST_IMAP_HOST": "localhost",
"DEST_IMAP_USERNAME": "dest_user",
"DEST_IMAP_PASSWORD": "p",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr(sys, "argv", ["compare_imap_folders.py"])
monkeypatch.setattr("imap_common.get_imap_connection", make_mock_connection(p1, p2))
compare_imap_folders.main()
captured = capsys.readouterr()
assert "INBOX" in captured.out
assert "0" in captured.out
class TestGetEmailCount:
"""Tests for get_email_count function."""
def test_successful_count(self, single_mock_server):
"""Test successful email count."""
data = {"INBOX": [b"Subject: 1\r\n\r\nB", b"Subject: 2\r\n\r\nB"]}
server, port = single_mock_server(data)
conn = imaplib.IMAP4("localhost", port)
conn.login("user", "pass")
result = compare_imap_folders.get_email_count(conn, "INBOX")
assert result == 2
conn.logout()
def test_nonexistent_folder(self, single_mock_server):
"""Test count for non-existent folder returns None."""
data = {"INBOX": []}
server, port = single_mock_server(data)
conn = imaplib.IMAP4("localhost", port)
conn.login("user", "pass")
result = compare_imap_folders.get_email_count(conn, "NonExistent")
assert result is None
conn.logout()
class TestConfigValidation:
"""Tests for configuration validation."""
def test_missing_source_credentials(self, monkeypatch, capsys):
"""Test that missing source credentials cause exit."""
env = {
"DEST_IMAP_HOST": "localhost",
"DEST_IMAP_USERNAME": "dest",
"DEST_IMAP_PASSWORD": "p",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr(sys, "argv", ["compare_imap_folders.py"])
with pytest.raises(SystemExit) as exc_info:
compare_imap_folders.main()
assert exc_info.value.code == 1
def test_missing_dest_credentials(self, monkeypatch, capsys):
"""Test that missing destination credentials cause exit."""
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "src",
"SRC_IMAP_PASSWORD": "p",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr(sys, "argv", ["compare_imap_folders.py"])
with pytest.raises(SystemExit) as exc_info:
compare_imap_folders.main()
assert exc_info.value.code == 1
class TestTotals:
"""Tests for total calculation."""
def test_total_calculation(self, mock_server_factory, monkeypatch, capsys):
"""Test that totals are calculated correctly."""
src_data = {
"INBOX": [b"Subject: 1\r\n\r\nB", b"Subject: 2\r\n\r\nB"],
"Sent": [b"Subject: 3\r\n\r\nB", b"Subject: 4\r\n\r\nB", b"Subject: 5\r\n\r\nB"],
}
dest_data = {
"INBOX": [b"Subject: 1\r\n\r\nB"],
"Sent": [b"Subject: 3\r\n\r\nB"],
}
src_server, dest_server, p1, p2 = mock_server_factory(src_data, dest_data)
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "src_user",
"SRC_IMAP_PASSWORD": "p",
"DEST_IMAP_HOST": "localhost",
"DEST_IMAP_USERNAME": "dest_user",
"DEST_IMAP_PASSWORD": "p",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr(sys, "argv", ["compare_imap_folders.py"])
monkeypatch.setattr("imap_common.get_imap_connection", make_mock_connection(p1, p2))
compare_imap_folders.main()
captured = capsys.readouterr()
# Total source: 5, Total dest: 2, Diff: 3
assert "TOTAL" in captured.out
assert "5" in captured.out
assert "2" in captured.out

146
test/test_count_imap_emails.py Normal file
View File

@ -0,0 +1,146 @@
"""
Tests for count_imap_emails.py
Tests cover:
- Basic email counting
- Multiple folder counting
- Empty folder handling
- Error handling
- Configuration validation
"""
import os
import sys
import pytest
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../src")))
import count_imap_emails
from conftest import make_single_mock_connection
class TestEmailCounting:
"""Tests for email counting functionality."""
def test_count_single_folder(self, single_mock_server, monkeypatch, capsys):
"""Test counting emails in a single folder."""
src_data = {
"INBOX": [
b"Subject: Email 1\r\n\r\nBody",
b"Subject: Email 2\r\n\r\nBody",
b"Subject: Email 3\r\n\r\nBody",
]
}
server, port = single_mock_server(src_data)
monkeypatch.setattr("imap_common.get_imap_connection", make_single_mock_connection(port))
count_imap_emails.count_emails("localhost", "user", "pass")
captured = capsys.readouterr()
assert "INBOX" in captured.out
assert "3" in captured.out
def test_count_multiple_folders(self, single_mock_server, monkeypatch, capsys):
"""Test counting emails across multiple folders."""
src_data = {
"INBOX": [b"Subject: 1\r\n\r\nB", b"Subject: 2\r\n\r\nB"],
"Sent": [b"Subject: 3\r\n\r\nB"],
"Archive": [b"Subject: 4\r\n\r\nB", b"Subject: 5\r\n\r\nB", b"Subject: 6\r\n\r\nB"],
}
server, port = single_mock_server(src_data)
monkeypatch.setattr("imap_common.get_imap_connection", make_single_mock_connection(port))
count_imap_emails.count_emails("localhost", "user", "pass")
captured = capsys.readouterr()
assert "INBOX" in captured.out
assert "Sent" in captured.out
assert "Archive" in captured.out
# Total should be 6
assert "6" in captured.out
def test_empty_folder(self, single_mock_server, monkeypatch, capsys):
"""Test counting in empty folders."""
src_data = {"INBOX": [], "Empty": []}
server, port = single_mock_server(src_data)
monkeypatch.setattr("imap_common.get_imap_connection", make_single_mock_connection(port))
count_imap_emails.count_emails("localhost", "user", "pass")
captured = capsys.readouterr()
assert "0" in captured.out
class TestMainFunction:
"""Tests for main function and CLI."""
def test_main_with_env_vars(self, single_mock_server, monkeypatch, capsys):
"""Test main function with environment variables."""
src_data = {"INBOX": [b"Subject: Test\r\n\r\nBody"]}
server, port = single_mock_server(src_data)
env = {
"IMAP_HOST": "localhost",
"IMAP_USERNAME": "user",
"IMAP_PASSWORD": "pass",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr(sys, "argv", ["count_imap_emails.py"])
monkeypatch.setattr("imap_common.get_imap_connection", make_single_mock_connection(port))
# Import and run __main__ block logic
count_imap_emails.count_emails("localhost", "user", "pass")
captured = capsys.readouterr()
assert "INBOX" in captured.out
def test_missing_credentials(self, monkeypatch, capsys):
"""Test that missing credentials cause exit."""
env = {}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr(sys, "argv", ["count_imap_emails.py"])
# Since the script uses if __name__ == "__main__", we need to test differently
# Test the validation path directly
with pytest.raises(SystemExit):
# Simulate running main
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--host", default=None)
parser.add_argument("--user", default=None)
parser.add_argument("--pass", dest="password", default=None)
args = parser.parse_args([])
if not all([args.host, args.user, args.password]):
sys.exit(1)
class TestSrcImapFallback:
"""Tests for SRC_IMAP_* environment variable fallback."""
def test_src_imap_vars_fallback(self, single_mock_server, monkeypatch, capsys):
"""Test that SRC_IMAP_* vars work as fallback."""
src_data = {"INBOX": [b"Subject: Test\r\n\r\nBody"]}
server, port = single_mock_server(src_data)
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "user",
"SRC_IMAP_PASSWORD": "pass",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr("imap_common.get_imap_connection", make_single_mock_connection(port))
# The fallback logic: IMAP_* or SRC_IMAP_*
default_host = os.getenv("IMAP_HOST") or os.getenv("SRC_IMAP_HOST")
default_user = os.getenv("IMAP_USERNAME") or os.getenv("SRC_IMAP_USERNAME")
default_pass = os.getenv("IMAP_PASSWORD") or os.getenv("SRC_IMAP_PASSWORD")
assert default_host == "localhost"
assert default_user == "user"
assert default_pass == "pass"

336
test/test_imap_common.py Normal file
View File

@ -0,0 +1,336 @@
"""
Tests for imap_common.py
Tests cover:
- Environment variable verification
- IMAP connection handling
- Folder name normalization
- MIME header decoding
- Message details extraction
- Duplicate detection
- Filename sanitization
- Trash folder detection
"""
import os
import sys
from unittest.mock import Mock
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../src")))
import imap_common
class TestVerifyEnvVars:
"""Tests for verify_env_vars function."""
def test_all_vars_present(self, monkeypatch):
"""Test returns True when all variables are set."""
monkeypatch.setenv("VAR1", "value1")
monkeypatch.setenv("VAR2", "value2")
result = imap_common.verify_env_vars(["VAR1", "VAR2"])
assert result is True
def test_missing_vars(self, monkeypatch, capsys):
"""Test returns False and prints error when variables are missing."""
monkeypatch.delenv("MISSING_VAR", raising=False)
result = imap_common.verify_env_vars(["MISSING_VAR"])
assert result is False
captured = capsys.readouterr()
assert "MISSING_VAR" in captured.err
def test_partial_vars_present(self, monkeypatch, capsys):
"""Test with some variables present and some missing."""
monkeypatch.setenv("PRESENT", "value")
monkeypatch.delenv("MISSING", raising=False)
result = imap_common.verify_env_vars(["PRESENT", "MISSING"])
assert result is False
class TestGetImapConnection:
"""Tests for get_imap_connection function."""
def test_invalid_credentials_empty(self, capsys):
"""Test returns None when credentials are empty."""
result = imap_common.get_imap_connection("", "user", "pass")
assert result is None
result = imap_common.get_imap_connection("host", "", "pass")
assert result is None
result = imap_common.get_imap_connection("host", "user", "")
assert result is None
def test_connection_error(self, capsys):
"""Test returns None on connection error."""
# Try to connect to an invalid host
result = imap_common.get_imap_connection("invalid.nonexistent.host", "u", "p")
assert result is None
captured = capsys.readouterr()
assert "Connection error" in captured.out or "Error" in captured.out
class TestNormalizeFolderName:
"""Tests for normalize_folder_name function."""
def test_standard_format(self):
"""Test parsing standard IMAP list response."""
folder_info = b'(\\HasNoChildren) "/" "INBOX"'
result = imap_common.normalize_folder_name(folder_info)
assert result == "INBOX"
def test_unquoted_name(self):
"""Test parsing unquoted folder name."""
folder_info = b'(\\HasNoChildren) "/" Drafts'
result = imap_common.normalize_folder_name(folder_info)
assert result == "Drafts"
def test_with_special_flags(self):
"""Test parsing folder with special-use flags."""
folder_info = b'(\\HasNoChildren \\Trash) "/" "Trash"'
result = imap_common.normalize_folder_name(folder_info)
assert result == "Trash"
def test_gmail_folder(self):
"""Test parsing Gmail-style folder."""
folder_info = b'(\\HasNoChildren) "/" "[Gmail]/All Mail"'
result = imap_common.normalize_folder_name(folder_info)
assert result == "[Gmail]/All Mail"
def test_string_input(self):
"""Test with string input instead of bytes."""
folder_info = '(\\HasNoChildren) "/" "Archive"'
result = imap_common.normalize_folder_name(folder_info)
assert result == "Archive"
def test_fallback_parsing(self):
"""Test fallback when regex doesn't match."""
folder_info = "simple_folder"
result = imap_common.normalize_folder_name(folder_info)
assert result == "simple_folder"
class TestDecodeMimeHeader:
"""Tests for decode_mime_header function."""
def test_plain_ascii(self):
"""Test decoding plain ASCII header."""
result = imap_common.decode_mime_header("Hello World")
assert result == "Hello World"
def test_none_input(self):
"""Test with None input."""
result = imap_common.decode_mime_header(None)
assert result == "(No Subject)"
def test_empty_string(self):
"""Test with empty string."""
result = imap_common.decode_mime_header("")
assert result == "(No Subject)"
def test_utf8_encoded(self):
"""Test decoding UTF-8 MIME encoded header."""
# =?UTF-8?B?SGVsbG8gV29ybGQ=?= is "Hello World" in base64
result = imap_common.decode_mime_header("=?UTF-8?B?SGVsbG8gV29ybGQ=?=")
assert "Hello" in result
def test_quoted_printable(self):
"""Test decoding quoted-printable header."""
# =?UTF-8?Q?Hello_World?= is "Hello World" in quoted-printable
result = imap_common.decode_mime_header("=?UTF-8?Q?Hello_World?=")
assert "Hello" in result
class TestSanitizeFilename:
"""Tests for sanitize_filename function."""
def test_valid_filename(self):
"""Test that valid filename passes through."""
result = imap_common.sanitize_filename("valid_filename")
assert result == "valid_filename"
def test_invalid_characters(self):
"""Test that invalid characters are replaced."""
result = imap_common.sanitize_filename('file<>:"/\\|?*name')
assert "<" not in result
assert ">" not in result
assert ":" not in result
assert '"' not in result
assert "/" not in result
assert "\\" not in result
assert "|" not in result
assert "?" not in result
assert "*" not in result
def test_empty_input(self):
"""Test that empty input returns 'untitled'."""
result = imap_common.sanitize_filename("")
assert result == "untitled"
def test_none_input(self):
"""Test that None input returns 'untitled'."""
result = imap_common.sanitize_filename(None)
assert result == "untitled"
def test_long_filename_truncation(self):
"""Test that long filenames are truncated."""
long_name = "a" * 300
result = imap_common.sanitize_filename(long_name)
assert len(result) <= 250
def test_strip_leading_trailing(self):
"""Test that leading/trailing whitespace and dots are stripped."""
result = imap_common.sanitize_filename(" ..filename.. ")
assert not result.startswith(" ")
assert not result.startswith(".")
assert not result.endswith(" ")
assert not result.endswith(".")
class TestDetectTrashFolder:
"""Tests for detect_trash_folder function."""
def test_detect_by_special_use_flag(self):
"""Test detection via \\Trash flag."""
mock_conn = Mock()
mock_conn.list.return_value = (
"OK",
[
b'(\\HasNoChildren) "/" "INBOX"',
b'(\\HasNoChildren \\Trash) "/" "Deleted Items"',
],
)
result = imap_common.detect_trash_folder(mock_conn)
assert result == "Deleted Items"
def test_detect_gmail_trash(self):
"""Test detection of Gmail Trash folder."""
mock_conn = Mock()
mock_conn.list.return_value = (
"OK",
[
b'(\\HasNoChildren) "/" "INBOX"',
b'(\\HasNoChildren \\Trash) "/" "[Gmail]/Trash"',
],
)
result = imap_common.detect_trash_folder(mock_conn)
assert result == "[Gmail]/Trash"
def test_detect_by_name_fallback(self):
"""Test detection by common name when no flag present."""
mock_conn = Mock()
mock_conn.list.return_value = (
"OK",
[
b'(\\HasNoChildren) "/" "INBOX"',
b'(\\HasNoChildren) "/" "Trash"',
],
)
result = imap_common.detect_trash_folder(mock_conn)
assert result == "Trash"
def test_no_trash_folder(self):
"""Test returns None when no trash folder found."""
mock_conn = Mock()
mock_conn.list.return_value = (
"OK",
[
b'(\\HasNoChildren) "/" "INBOX"',
b'(\\HasNoChildren) "/" "Sent"',
],
)
result = imap_common.detect_trash_folder(mock_conn)
assert result is None
def test_list_error(self):
"""Test returns None on list error."""
mock_conn = Mock()
mock_conn.list.return_value = ("NO", [])
result = imap_common.detect_trash_folder(mock_conn)
assert result is None
def test_exception_handling(self):
"""Test returns None on exception."""
mock_conn = Mock()
mock_conn.list.side_effect = Exception("Connection error")
result = imap_common.detect_trash_folder(mock_conn)
assert result is None
class TestMessageExistsInFolder:
"""Tests for message_exists_in_folder function."""
def test_no_message_id(self):
"""Test returns False when message_id is None."""
mock_conn = Mock()
result = imap_common.message_exists_in_folder(mock_conn, None, 100)
assert result is False
def test_search_fails(self):
"""Test returns False when search fails."""
mock_conn = Mock()
mock_conn.search.return_value = ("NO", [])
result = imap_common.message_exists_in_folder(mock_conn, "<msg-id>", 100)
assert result is False
def test_no_matches(self):
"""Test returns False when no matches found."""
mock_conn = Mock()
mock_conn.search.return_value = ("OK", [b""])
result = imap_common.message_exists_in_folder(mock_conn, "<msg-id>", 100)
assert result is False
def test_match_found_same_size(self):
"""Test returns True when message with same ID and size found."""
mock_conn = Mock()
mock_conn.search.return_value = ("OK", [b"1"])
mock_conn.fetch.return_value = ("OK", [b"1 (RFC822.SIZE 100)"])
result = imap_common.message_exists_in_folder(mock_conn, "<msg-id>", 100)
assert result is True
def test_match_found_different_size(self):
"""Test returns False when message with same ID but different size found."""
mock_conn = Mock()
mock_conn.search.return_value = ("OK", [b"1"])
mock_conn.fetch.return_value = ("OK", [b"1 (RFC822.SIZE 200)"])
result = imap_common.message_exists_in_folder(mock_conn, "<msg-id>", 100)
assert result is False
class TestGetMsgDetails:
"""Tests for get_msg_details function."""
def test_fetch_error(self):
"""Test returns None tuple on fetch error."""
mock_conn = Mock()
mock_conn.uid.side_effect = Exception("Fetch error")
msg_id, size, subject = imap_common.get_msg_details(mock_conn, b"1")
assert msg_id is None
assert size is None
assert subject is None
def test_not_ok_response(self):
"""Test returns None tuple on non-OK response."""
mock_conn = Mock()
mock_conn.uid.return_value = ("NO", None)
msg_id, size, subject = imap_common.get_msg_details(mock_conn, b"1")
assert msg_id is None
assert size is None
assert subject is None

299
test/test_migrate_imap_emails.py Normal file
View File

@ -0,0 +1,299 @@
"""
Tests for migrate_imap_emails.py
Tests cover:
- Basic email migration
- Duplicate detection and skipping
- Delete from source after migration
- Folder creation on destination
- Multiple folder migration
- Error handling
"""
import os
import sys
import pytest
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../src")))
import migrate_imap_emails
from conftest import make_mock_connection
class TestBasicMigration:
"""Tests for basic migration functionality."""
def test_single_email_migration(self, mock_server_factory, monkeypatch):
"""Test migrating a single email from source to destination."""
src_data = {"INBOX": [b"Subject: Hello\r\nMessage-ID: <1@test>\r\n\r\nBody"]}
dest_data = {"INBOX": []}
_, dest_server, p1, p2 = mock_server_factory(src_data, dest_data)
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "src_user",
"SRC_IMAP_PASSWORD": "p",
"DEST_IMAP_HOST": "localhost",
"DEST_IMAP_USERNAME": "dest_user",
"DEST_IMAP_PASSWORD": "p",
"MAX_WORKERS": "1",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr("imap_common.get_imap_connection", make_mock_connection(p1, p2))
migrate_imap_emails.main()
assert len(dest_server.folders["INBOX"]) == 1
assert b"Subject: Hello" in dest_server.folders["INBOX"][0]["content"]
def test_multiple_emails_migration(self, mock_server_factory, monkeypatch):
"""Test migrating multiple emails."""
src_data = {
"INBOX": [
b"Subject: Email 1\r\nMessage-ID: <1@test>\r\n\r\nBody 1",
b"Subject: Email 2\r\nMessage-ID: <2@test>\r\n\r\nBody 2",
b"Subject: Email 3\r\nMessage-ID: <3@test>\r\n\r\nBody 3",
]
}
dest_data = {"INBOX": []}
_, dest_server, p1, p2 = mock_server_factory(src_data, dest_data)
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "src_user",
"SRC_IMAP_PASSWORD": "p",
"DEST_IMAP_HOST": "localhost",
"DEST_IMAP_USERNAME": "dest_user",
"DEST_IMAP_PASSWORD": "p",
"MAX_WORKERS": "1",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr("imap_common.get_imap_connection", make_mock_connection(p1, p2))
migrate_imap_emails.main()
assert len(dest_server.folders["INBOX"]) == 3
class TestDuplicateHandling:
"""Tests for duplicate detection and skipping."""
def test_skip_duplicate_by_message_id(self, mock_server_factory, monkeypatch):
"""Test that emails with existing Message-ID are skipped."""
msg = b"Subject: Dup\r\nMessage-ID: <dup-id>\r\n\r\nContent"
src_data = {"INBOX": [msg]}
dest_data = {"INBOX": [msg]} # Already exists
_, dest_server, p1, p2 = mock_server_factory(src_data, dest_data)
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "src_user",
"SRC_IMAP_PASSWORD": "p",
"DEST_IMAP_HOST": "localhost",
"DEST_IMAP_USERNAME": "dest_user",
"DEST_IMAP_PASSWORD": "p",
"MAX_WORKERS": "1",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr("imap_common.get_imap_connection", make_mock_connection(p1, p2))
migrate_imap_emails.main()
# Should still be 1 message (duplicate skipped)
assert len(dest_server.folders["INBOX"]) == 1
def test_migrate_non_duplicate(self, mock_server_factory, monkeypatch):
"""Test that non-duplicate emails are migrated even when others exist."""
existing = b"Subject: Existing\r\nMessage-ID: <existing>\r\n\r\nOld"
new_msg = b"Subject: New\r\nMessage-ID: <new>\r\n\r\nNew content"
src_data = {"INBOX": [new_msg]}
dest_data = {"INBOX": [existing]}
_, dest_server, p1, p2 = mock_server_factory(src_data, dest_data)
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "src_user",
"SRC_IMAP_PASSWORD": "p",
"DEST_IMAP_HOST": "localhost",
"DEST_IMAP_USERNAME": "dest_user",
"DEST_IMAP_PASSWORD": "p",
"MAX_WORKERS": "1",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr("imap_common.get_imap_connection", make_mock_connection(p1, p2))
migrate_imap_emails.main()
# Should now have 2 messages
assert len(dest_server.folders["INBOX"]) == 2
class TestDeleteFromSource:
"""Tests for delete-after-migration functionality."""
def test_delete_after_migration(self, mock_server_factory, monkeypatch):
"""Test that emails are deleted from source after successful migration."""
src_data = {"INBOX": [b"Subject: Del\r\nMessage-ID: <del>\r\n\r\nC"]}
dest_data = {"INBOX": []}
src_server, dest_server, p1, p2 = mock_server_factory(src_data, dest_data)
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "src_user",
"SRC_IMAP_PASSWORD": "p",
"DEST_IMAP_HOST": "localhost",
"DEST_IMAP_USERNAME": "dest_user",
"DEST_IMAP_PASSWORD": "p",
"MAX_WORKERS": "1",
"DELETE_FROM_SOURCE": "true",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr("imap_common.get_imap_connection", make_mock_connection(p1, p2))
migrate_imap_emails.main()
assert len(dest_server.folders["INBOX"]) == 1
assert len(src_server.folders["INBOX"]) == 0
def test_no_delete_when_disabled(self, mock_server_factory, monkeypatch):
"""Test that emails remain in source when delete is not enabled."""
src_data = {"INBOX": [b"Subject: Keep\r\nMessage-ID: <keep>\r\n\r\nC"]}
dest_data = {"INBOX": []}
src_server, dest_server, p1, p2 = mock_server_factory(src_data, dest_data)
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "src_user",
"SRC_IMAP_PASSWORD": "p",
"DEST_IMAP_HOST": "localhost",
"DEST_IMAP_USERNAME": "dest_user",
"DEST_IMAP_PASSWORD": "p",
"MAX_WORKERS": "1",
# DELETE_FROM_SOURCE not set
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr("imap_common.get_imap_connection", make_mock_connection(p1, p2))
migrate_imap_emails.main()
assert len(dest_server.folders["INBOX"]) == 1
assert len(src_server.folders["INBOX"]) == 1
class TestFolderHandling:
"""Tests for folder creation and multi-folder migration."""
def test_folder_creation(self, mock_server_factory, monkeypatch):
"""Test that folders are created on destination."""
src_data = {"INBOX": [], "Archive": [b"Subject: A\r\nMessage-ID: <a>\r\n\r\nC"]}
dest_data = {"INBOX": []}
_, dest_server, p1, p2 = mock_server_factory(src_data, dest_data)
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "src_user",
"SRC_IMAP_PASSWORD": "p",
"DEST_IMAP_HOST": "localhost",
"DEST_IMAP_USERNAME": "dest_user",
"DEST_IMAP_PASSWORD": "p",
"MAX_WORKERS": "1",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr("imap_common.get_imap_connection", make_mock_connection(p1, p2))
migrate_imap_emails.main()
assert "Archive" in dest_server.folders
assert len(dest_server.folders["Archive"]) == 1
def test_multiple_folders_migration(self, mock_server_factory, monkeypatch):
"""Test migrating emails from multiple folders."""
src_data = {
"INBOX": [b"Subject: Inbox\r\nMessage-ID: <inbox>\r\n\r\nC"],
"Sent": [b"Subject: Sent\r\nMessage-ID: <sent>\r\n\r\nC"],
"Drafts": [b"Subject: Draft\r\nMessage-ID: <draft>\r\n\r\nC"],
}
dest_data = {"INBOX": []}
_, dest_server, p1, p2 = mock_server_factory(src_data, dest_data)
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "src_user",
"SRC_IMAP_PASSWORD": "p",
"DEST_IMAP_HOST": "localhost",
"DEST_IMAP_USERNAME": "dest_user",
"DEST_IMAP_PASSWORD": "p",
"MAX_WORKERS": "1",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr("imap_common.get_imap_connection", make_mock_connection(p1, p2))
migrate_imap_emails.main()
assert len(dest_server.folders["INBOX"]) == 1
assert "Sent" in dest_server.folders
assert "Drafts" in dest_server.folders
def test_empty_folder_handling(self, mock_server_factory, monkeypatch):
"""Test that empty folders are handled gracefully."""
src_data = {"INBOX": [], "Empty": []}
dest_data = {"INBOX": []}
_, _, p1, p2 = mock_server_factory(src_data, dest_data)
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "src_user",
"SRC_IMAP_PASSWORD": "p",
"DEST_IMAP_HOST": "localhost",
"DEST_IMAP_USERNAME": "dest_user",
"DEST_IMAP_PASSWORD": "p",
"MAX_WORKERS": "1",
}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr("imap_common.get_imap_connection", make_mock_connection(p1, p2))
# Should complete without error
migrate_imap_emails.main()
class TestConfigValidation:
"""Tests for configuration validation."""
def test_missing_source_credentials(self, monkeypatch, capsys):
"""Test that missing source credentials cause exit."""
env = {
"DEST_IMAP_HOST": "localhost",
"DEST_IMAP_USERNAME": "dest",
"DEST_IMAP_PASSWORD": "p",
}
monkeypatch.setattr(os, "environ", env)
with pytest.raises(SystemExit) as exc_info:
migrate_imap_emails.main()
assert exc_info.value.code == 1
def test_missing_dest_credentials(self, monkeypatch, capsys):
"""Test that missing destination credentials cause exit."""
env = {
"SRC_IMAP_HOST": "localhost",
"SRC_IMAP_USERNAME": "src",
"SRC_IMAP_PASSWORD": "p",
}
monkeypatch.setattr(os, "environ", env)
with pytest.raises(SystemExit) as exc_info:
migrate_imap_emails.main()
assert exc_info.value.code == 1

364
tools/mock_imap_server.py Normal file
View File

@ -0,0 +1,364 @@
import re
import socketserver
import threading
class MockIMAPHandler(socketserver.StreamRequestHandler):
"""
A minimal IMAP4rev1 mock server handler for testing purposes.
Supports basic commands required for the migration scripts.
"""
def handle(self):
self.wfile.write(b"* OK [CAPABILITY IMAP4rev1] Mock IMAP Server Ready\r\n")
self.selected_folder = None
self.current_folders = self.server.folders
while True:
try:
line = self.rfile.readline()
if not line:
break
line = line.decode("utf-8").strip()
if not line:
continue
parts = line.split(" ", 2)
tag = parts[0]
cmd = parts[1].upper()
args = parts[2] if len(parts) > 2 else ""
if cmd == "LOGIN":
self.send_response(tag, "OK LOGIN completed")
elif cmd == "LOGOUT":
self.send_response(tag, "OK LOGOUT completed")
break
elif cmd == "CAPABILITY":
self.wfile.write(b"* CAPABILITY IMAP4rev1 AUTH=PLAIN\r\n")
self.send_response(tag, "OK CAPABILITY completed")
elif cmd == "LIST":
for folder in self.current_folders:
self.wfile.write(f'* LIST (\\HasNoChildren) "/" "{folder}"\r\n'.encode())
self.send_response(tag, "OK LIST completed")
elif cmd == "SELECT":
folder = args.strip().strip('"')
if folder in self.current_folders:
self.selected_folder = folder
count = len(self.current_folders[folder])
self.wfile.write(f"* {count} EXISTS\r\n".encode())
self.wfile.write(f"* {count} RECENT\r\n".encode())
self.wfile.write(b"* FLAGS (\\Seen \\Answered \\Flagged \\Deleted \\Draft)\r\n")
self.wfile.write(b"* OK [UIDVALIDITY 1] UIDs valid\r\n")
self.send_response(tag, "OK [READ-WRITE] SELECT completed")
else:
self.send_response(tag, "NO [NONEXISTENT] Folder not found")
elif cmd == "EXAMINE":
# EXAMINE is like SELECT but read-only
folder = args.strip().strip('"')
if folder in self.current_folders:
self.selected_folder = folder
count = len(self.current_folders[folder])
self.wfile.write(f"* {count} EXISTS\r\n".encode())
self.wfile.write(f"* {count} RECENT\r\n".encode())
self.wfile.write(b"* FLAGS (\\Seen \\Answered \\Flagged \\Deleted \\Draft)\r\n")
self.wfile.write(b"* OK [UIDVALIDITY 1] UIDs valid\r\n")
self.send_response(tag, "OK [READ-ONLY] EXAMINE completed")
else:
self.send_response(tag, "NO [NONEXISTENT] Folder not found")
elif cmd == "CREATE":
folder = args.strip().strip('"')
if folder not in self.current_folders:
self.current_folders[folder] = []
self.send_response(tag, "OK CREATE completed")
elif cmd == "EXPUNGE":
if self.selected_folder:
msgs = self.current_folders[self.selected_folder]
# In-place remove marked deleted
# We use a dictionary-like structure implicitly via dicts in list now
# Check "flags" set in msg object
new_msgs = [m for m in msgs if "\\Deleted" not in m["flags"]]
removed_count = len(msgs) - len(new_msgs)
self.current_folders[self.selected_folder] = new_msgs
# According to IMAP, we should send Expunge indices but we'll skip for mock
self.send_response(tag, "OK EXPUNGE completed")
else:
self.send_response(tag, "NO Select first")
elif cmd == "UID":
sub_parts = args.split(" ", 1)
sub_cmd = sub_parts[0].upper()
sub_rest = sub_parts[1] if len(sub_parts) > 1 else ""
if sub_cmd == "SEARCH":
sub_args = sub_rest
if not self.selected_folder:
self.send_response(tag, "NO Select first")
continue
msgs = self.current_folders[self.selected_folder]
# Handle UNDELETED
# If args has UNDELETED, filter flags
valid_uids = []
for m in msgs:
if "UNDELETED" in sub_args and "\\Deleted" in m["flags"]:
continue
valid_uids.append(str(m["uid"]))
uids_str = " ".join(valid_uids)
self.wfile.write(f"* SEARCH {uids_str}\r\n".encode())
self.send_response(tag, "OK SEARCH completed")
elif sub_cmd == "STORE":
# UID STORE <uid> +FLAGS (\Deleted)
store_parts = sub_rest.split(" ", 2)
uid = int(store_parts[0])
action = store_parts[1].upper() # +FLAGS or -FLAGS
# Flags can be "(\Deleted)" or "\Deleted"
flags_str = store_parts[2].strip("()")
flags_list = {f.strip() for f in flags_str.split()}
if self.selected_folder:
msgs = self.current_folders[self.selected_folder]
found = False
for m in msgs:
if m["uid"] == uid:
if action == "+FLAGS":
m["flags"].update(flags_list)
elif action == "-FLAGS":
m["flags"].difference_update(flags_list)
found = True
# Send update
flag_output = " ".join(m["flags"])
self.wfile.write(
f"* {msgs.index(m) + 1} FETCH (FLAGS ({flag_output}))\r\n".encode()
)
break
if found:
self.send_response(tag, "OK STORE completed")
else:
self.send_response(tag, "NO UID not found")
else:
self.send_response(tag, "NO Select first")
elif sub_cmd == "FETCH":
# sub_rest is e.g. "1 (RFC822.SIZE BODY.PEEK[...])"
parts = sub_rest.split(" ", 1)
uid_set = parts[0]
opts = parts[1].upper() if len(parts) > 1 else ""
if not self.selected_folder:
self.send_response(tag, "NO Select first")
continue
msgs = self.current_folders[self.selected_folder]
# Find msg by UID
target_msgs = []
if ":" in uid_set:
# Range logic omitted for brevity, taking all for '*' or simplified assumption
# But since we use UIDs, we should filter.
target_msgs = msgs # Return all for range
else:
try:
t_uid = int(uid_set)
target_msgs = [m for m in msgs if m["uid"] == t_uid]
except:
pass
for m in target_msgs:
# Mock always returns info if iterating all, or specific
# Construction
msg_content = m["content"]
msg_len = len(msg_content)
flags_str = " ".join(m["flags"])
resp = (
f"* {msgs.index(m) + 1} FETCH (UID {m['uid']} RFC822.SIZE {msg_len} FLAGS ({flags_str})"
)
if "RFC822" in opts or "BODY" in opts:
resp += f" BODY[] {{{msg_len}}}\r\n"
self.wfile.write(resp.encode("utf-8"))
self.wfile.write(msg_content)
self.wfile.write(b")\r\n")
else:
resp += ")\r\n"
self.wfile.write(resp.encode("utf-8"))
self.wfile.flush()
self.send_response(tag, "OK FETCH completed")
elif sub_cmd == "COPY":
# UID COPY <uid> <target>
c_parts = sub_rest.split(" ", 1)
c_uid = int(c_parts[0])
c_dest = c_parts[1].strip().strip('"')
if c_dest in self.current_folders and self.selected_folder:
msgs = self.current_folders[self.selected_folder]
found_msg = next((m for m in msgs if m["uid"] == c_uid), None)
if found_msg:
# COPY: create new msg object (new UID usually)
# We need a new UID generator.
# For simplicity, use max_uid + 1
dest_msgs = self.current_folders[c_dest]
max_uid = max([m["uid"] for m in dest_msgs], default=0)
new_msg = found_msg.copy()
new_msg["uid"] = max_uid + 1
new_msg["flags"] = found_msg["flags"].copy()
dest_msgs.append(new_msg)
self.send_response(tag, "OK COPY completed")
else:
self.send_response(tag, "NO UID not found")
else:
self.send_response(tag, "NO Dest not found")
elif cmd == "APPEND":
try:
match = re.search(r"\{(\d+)\}$", args)
if match:
size = int(match.group(1))
self.wfile.write(b"+ Ready\r\n")
data = self.rfile.read(size)
folder_arg = args.split(" ")[0].strip().strip('"')
if folder_arg in self.current_folders:
dest_msgs = self.current_folders[folder_arg]
max_uid = max([m["uid"] for m in dest_msgs], default=0)
# IMPORTANT: Reset pointer or copy
new_msg = {"uid": max_uid + 1, "flags": set(), "content": data}
dest_msgs.append(new_msg)
print(f"MOCK APPEND SUCCESS: {folder_arg} now has {len(dest_msgs)}")
self.send_response(tag, "OK APPEND completed")
else:
self.send_response(tag, "NO Folder not found")
else:
self.send_response(tag, "BAD APPEND")
except Exception as e:
print(f"MOCK APPEND ERROR: {e}")
self.send_response(tag, "BAD APPEND")
elif cmd == "SEARCH":
# Parse SEARCH ALL, SEARCH HEADER Message-ID "...", etc.
# args might be: ALL, HEADER Message-ID "<123>", CHARSETS UTF-8 ...
found_indices = []
if not self.selected_folder:
self.wfile.write(b"* SEARCH\r\n")
self.send_response(tag, "OK SEARCH completed")
continue
msgs = self.current_folders[self.selected_folder]
if "ALL" in args.upper():
# Return all message sequence numbers
found_indices = [str(idx + 1) for idx in range(len(msgs))]
elif "HEADER Message-ID" in args:
# Extract value
# Expected: ... HEADER Message-ID "value" ...
try:
# Split by 'MESSAGE-ID' (case insensitive?)
# part after Message-ID
post_mi = args.split("Message-ID", 1)[1].strip()
# Should start with quote or value
if post_mi.startswith('"'):
search_val = post_mi.split('"', 2)[1]
else:
search_val = post_mi.split(" ", 1)[0]
search_val = search_val.replace("<", "").replace(">", "")
for idx, m in enumerate(msgs):
content_str = m["content"].decode("utf-8", errors="ignore")
if search_val in content_str:
# Simple substring check is risky but okay for mock
# Better: regex for Message-ID: <...search_val...>
found_indices.append(str(idx + 1))
except Exception:
pass
indices_str = " ".join(found_indices)
self.wfile.write(f"* SEARCH {indices_str}\r\n".encode())
self.send_response(tag, "OK SEARCH completed")
elif cmd == "FETCH":
# Non-UID FETCH - args is e.g. "1 (RFC822.SIZE)"
if not self.selected_folder:
self.send_response(tag, "NO Select first")
continue
parts = args.split(" ", 1)
msg_num = int(parts[0])
opts = parts[1].upper() if len(parts) > 1 else ""
msgs = self.current_folders[self.selected_folder]
if 1 <= msg_num <= len(msgs):
m = msgs[msg_num - 1] # Convert to 0-indexed
msg_content = m["content"]
msg_len = len(msg_content)
flags_str = " ".join(m["flags"])
resp = f"* {msg_num} FETCH (UID {m['uid']} RFC822.SIZE {msg_len} FLAGS ({flags_str})"
if "RFC822" in opts or "BODY" in opts:
resp += f" BODY[] {{{msg_len}}}\r\n"
self.wfile.write(resp.encode("utf-8"))
self.wfile.write(msg_content)
self.wfile.write(b")\r\n")
else:
resp += ")\r\n"
self.wfile.write(resp.encode("utf-8"))
self.wfile.flush()
self.send_response(tag, "OK FETCH completed")
elif cmd == "NOOP":
self.send_response(tag, "OK NOOP")
else:
self.send_response(tag, "BAD Command not recognized")
except Exception:
break
def send_response(self, tag, message):
self.wfile.write(f"{tag} {message}\r\n".encode())
class MockIMAPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
allow_reuse_address = True
daemon_threads = True
def __init__(self, server_address, RequestHandlerClass, initial_folders=None):
super().__init__(server_address, RequestHandlerClass)
self.folders = {}
if initial_folders:
for fname, contents in initial_folders.items():
self.folders[fname] = []
for i, c in enumerate(contents):
if isinstance(c, bytes):
self.folders[fname].append({"uid": i + 1, "flags": set(), "content": c})
else:
self.folders[fname].append(c)
else:
self.folders = {"INBOX": []}
def start_server_thread(port=10143, initial_folders=None):
server = MockIMAPServer(("localhost", port), MockIMAPHandler, initial_folders)
t = threading.Thread(target=server.serve_forever)
t.daemon = True
t.start()
return t, server