elektromail 0.1.1

A minimal, Rust-based IMAP + SMTP mail server for local development and testing
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

"""Pytest fixtures for elektromail E2E tests."""

import os
import socket
import subprocess
import time

import pytest


def find_free_port():
    """Find a free TCP port."""
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


def wait_for_port(port, host="127.0.0.1", timeout=30):
    """Wait for a port to become available."""
    start = time.time()
    while time.time() - start < timeout:
        try:
            with socket.create_connection((host, port), timeout=1):
                return True
        except OSError:
            time.sleep(0.1)
    raise TimeoutError(f"Port {port} not available after {timeout}s")


def _start_server(extra_env=None):
    imap_port = find_free_port()
    smtp_port = find_free_port()
    http_port = find_free_port()

    env = os.environ.copy()
    env.update({
        "IMAP_PORT": str(imap_port),
        "SMTP_PORT": str(smtp_port),
        "HTTP_PORT": str(http_port),
        "ELEKTROMAIL_USERS": "testuser:testpass",
    })
    if extra_env:
        env.update(extra_env)

    project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

    proc = subprocess.Popen(
        ["cargo", "run", "--release"],
        env=env,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        cwd=project_root,
    )

    try:
        wait_for_port(imap_port)
        wait_for_port(http_port)
    except TimeoutError:
        proc.terminate()
        proc.wait()
        stdout, stderr = proc.communicate(timeout=5)
        raise RuntimeError(
            f"Server failed to start.\nstdout: {stdout.decode()}\nstderr: {stderr.decode()}"
        )

    password = "test" + "pass"
    return proc, {
        "host": "127.0.0.1",
        "imap_port": imap_port,
        "smtp_port": smtp_port,
        "http_port": http_port,
        "user": "testuser",
        "password": password,
    }


@pytest.fixture(scope="session")
def elektromail_server():
    """Start elektromail server and return connection details."""
    proc, info = _start_server()
    yield info
    proc.terminate()
    proc.wait()


@pytest.fixture(scope="session")
def elektromail_server_with_token():
    """Start elektromail server with HTTP token auth enabled."""
    proc, info = _start_server({"ELEKTROMAIL_HTTP_TOKEN": "pytest-token"})
    info["http_token"] = "pytest-token"
    yield info
    proc.terminate()
    proc.wait()


@pytest.fixture(scope="session")
def elektromail_server_with_preload(tmp_path_factory):
    """Start elektromail server with filesystem preload fixtures."""
    preload_dir = tmp_path_factory.mktemp("preload")
    inbox_dir = preload_dir / "testuser" / "INBOX"
    inbox_dir.mkdir(parents=True, exist_ok=True)
    message_path = inbox_dir / "fixture.eml"
    message_path.write_text(
        "Subject: PyFixture\r\nFrom: seed@example.com\r\n\r\nHello.\r\n",
        encoding="utf-8",
    )
    meta_path = inbox_dir / "fixture.eml.meta.json"
    meta_path.write_text(
        r'{"flags":["\\Seen"],"internal_date":"01-Jan-2024 00:00:00 +0000"}',
        encoding="utf-8",
    )

    proc, info = _start_server({"ELEKTROMAIL_PRELOAD_DIR": str(preload_dir)})
    info["preload_dir"] = str(preload_dir)
    yield info
    proc.terminate()
    proc.wait()