import http.client
import json
def http_request(host, port, method, path, body=None, headers=None):
conn = http.client.HTTPConnection(host, port, timeout=5)
request_headers = {"Connection": "close"}
if headers:
request_headers.update(headers)
if body is not None:
payload = json.dumps(body)
request_headers["Content-Type"] = "application/json"
request_headers["Content-Length"] = str(len(payload))
conn.request(method, path, body=payload, headers=request_headers)
else:
conn.request(method, path, headers=request_headers)
resp = conn.getresponse()
data = resp.read().decode("utf-8", errors="replace")
conn.close()
return resp.status, resp.getheaders(), data
def json_body(data):
try:
return json.loads(data)
except json.JSONDecodeError:
return None
def test_openapi_and_docs(elektromail_server):
srv = elektromail_server
status, headers, body = http_request(srv["host"], srv["http_port"], "GET", "/openapi.json")
assert status == 200
assert json_body(body) is not None
status, headers, body = http_request(srv["host"], srv["http_port"], "GET", "/docs")
assert status == 200
assert "SwaggerUIBundle" in body
def test_users_create_and_reset(elektromail_server):
srv = elektromail_server
password = "py" + "pass"
status, headers, body = http_request(
srv["host"],
srv["http_port"],
"POST",
"/users",
{"user": "pyuser", "password": password},
)
assert status == 201
status, headers, body = http_request(srv["host"], srv["http_port"], "GET", "/users")
data = json_body(body)
assert data is not None
assert "pyuser" in data.get("users", [])
status, headers, body = http_request(srv["host"], srv["http_port"], "POST", "/reset")
assert status == 200
status, headers, body = http_request(srv["host"], srv["http_port"], "GET", "/users")
data = json_body(body)
assert data is not None
assert "pyuser" not in data.get("users", [])
def test_purge_clears_messages(elektromail_server):
srv = elektromail_server
password = "purge" + "pass"
status, headers, body = http_request(
srv["host"],
srv["http_port"],
"POST",
"/users",
{"user": "purgepy", "password": password},
)
assert status == 201
status, headers, body = http_request(
srv["host"],
srv["http_port"],
"POST",
"/inject",
{"user": "purgepy", "mailbox": "INBOX", "raw": "Subject: Py\r\n\r\nBody"},
)
assert status == 201
status, headers, body = http_request(
srv["host"],
srv["http_port"],
"GET",
"/messages?user=purgepy&mailbox=INBOX",
)
data = json_body(body)
assert data is not None
assert len(data.get("messages", [])) == 1
status, headers, body = http_request(srv["host"], srv["http_port"], "POST", "/purge")
assert status == 200
status, headers, body = http_request(
srv["host"],
srv["http_port"],
"GET",
"/messages?user=purgepy&mailbox=INBOX",
)
data = json_body(body)
assert data is not None
assert data.get("messages", []) == []
def test_config_snapshot(elektromail_server):
srv = elektromail_server
status, headers, body = http_request(srv["host"], srv["http_port"], "GET", "/config")
assert status == 200
data = json_body(body)
assert data is not None
assert "smtp_addr" in data
assert "imap_addr" in data
assert "http_addr" in data
assert "storage" in data
assert "auth_disabled" in data
assert "user_count" in data
def test_get_message_by_uid(elektromail_server):
srv = elektromail_server
password = "uid" + "pass"
status, headers, body = http_request(
srv["host"],
srv["http_port"],
"POST",
"/users",
{"user": "uidpy", "password": password},
)
assert status == 201
status, headers, body = http_request(
srv["host"],
srv["http_port"],
"POST",
"/inject",
{"user": "uidpy", "mailbox": "INBOX", "raw": "Subject: UIDPY\r\n\r\nBody"},
)
assert status == 201
status, headers, body = http_request(
srv["host"],
srv["http_port"],
"GET",
"/messages/1?user=uidpy&mailbox=INBOX&include_raw=true",
)
assert status == 200
data = json_body(body)
assert data is not None
message = data.get("message", {})
assert message.get("subject") == "UIDPY"
assert "Subject: UIDPY" in message.get("raw", "")
def test_users_include_email(elektromail_server):
srv = elektromail_server
password = "mail" + "pass"
status, headers, body = http_request(
srv["host"],
srv["http_port"],
"POST",
"/users",
{"user": "mailpy", "password": password, "email": "mailpy@example.com"},
)
assert status == 201
status, headers, body = http_request(
srv["host"],
srv["http_port"],
"GET",
"/users?include_email=true",
)
assert status == 200
data = json_body(body)
assert data is not None
users = data.get("users", [])
assert any(u.get("user") == "mailpy" and u.get("email") == "mailpy@example.com" for u in users)
def test_http_token_required(elektromail_server_with_token):
srv = elektromail_server_with_token
status, headers, body = http_request(srv["host"], srv["http_port"], "GET", "/config")
assert status == 401
status, headers, body = http_request(
srv["host"],
srv["http_port"],
"GET",
"/config",
headers={"Authorization": f"Bearer {srv['http_token']}"},
)
assert status == 200
status, headers, body = http_request(srv["host"], srv["http_port"], "GET", "/docs")
assert status == 401
status, headers, body = http_request(
srv["host"],
srv["http_port"],
"GET",
f"/docs?token={srv['http_token']}",
)
assert status == 200
assert "requestInterceptor" in body
status, headers, body = http_request(
srv["host"],
srv["http_port"],
"GET",
f"/openapi.json?token={srv['http_token']}",
)
assert status == 200
def test_preload_fixtures_reset(elektromail_server_with_preload):
srv = elektromail_server_with_preload
status, headers, body = http_request(
srv["host"],
srv["http_port"],
"GET",
"/messages?user=testuser&mailbox=INBOX",
)
assert status == 200
data = json_body(body)
assert data is not None
messages = data.get("messages", [])
assert len(messages) == 1
assert any(m.get("subject") == "PyFixture" for m in messages)
status, headers, body = http_request(
srv["host"],
srv["http_port"],
"POST",
"/inject",
{"user": "testuser", "mailbox": "INBOX", "raw": "Subject: Injected\r\n\r\nBody"},
)
assert status == 201
status, headers, body = http_request(
srv["host"],
srv["http_port"],
"GET",
"/messages?user=testuser&mailbox=INBOX",
)
data = json_body(body)
assert data is not None
assert len(data.get("messages", [])) == 2
status, headers, body = http_request(srv["host"], srv["http_port"], "POST", "/reset")
assert status == 200
status, headers, body = http_request(
srv["host"],
srv["http_port"],
"GET",
"/messages?user=testuser&mailbox=INBOX",
)
data = json_body(body)
assert data is not None
messages = data.get("messages", [])
assert len(messages) == 1
assert any(m.get("subject") == "PyFixture" for m in messages)