138 lines
4.6 KiB
Python
138 lines
4.6 KiB
Python
"""IMAP helpers for the e2e test: seed the old server, inspect the new one."""
|
|
import imaplib
|
|
import json
|
|
import re
|
|
import ssl
|
|
import sys
|
|
|
|
CTX = ssl.create_default_context() # trusts the test CA via security.pki
|
|
|
|
|
|
def old(user, password):
|
|
c = imaplib.IMAP4("old.test", 143)
|
|
c.starttls(ssl_context=CTX)
|
|
c.login(user, password)
|
|
return c
|
|
|
|
|
|
def new(user, password):
|
|
c = imaplib.IMAP4_SSL("new.test", 993, ssl_context=CTX)
|
|
c.login(user, password)
|
|
return c
|
|
|
|
|
|
def msg(mid, subject, body="hello\r\n", extra=""):
|
|
hdr = f"From: Sender <sender@example.org>\r\nTo: x@b4l.test\r\nSubject: {subject}\r\nDate: Wed, 01 Jan 2020 10:00:00 +0000\r\n"
|
|
if mid:
|
|
hdr += f"Message-ID: <{mid}>\r\n"
|
|
return (hdr + extra + "\r\n" + body).encode()
|
|
|
|
|
|
def ensure(c, box):
|
|
typ, _ = c.create(box)
|
|
print("create", box, typ)
|
|
|
|
|
|
def append(c, box, data, flags="", date=None):
|
|
typ, resp = c.append(box, flags or None, date, data)
|
|
assert typ == "OK", (box, typ, resp)
|
|
|
|
|
|
def seed():
|
|
a = old("alice", "alicepass")
|
|
append(a, "INBOX", msg("m1@test", "seen and flagged"), r"(\Seen \Flagged)", '"01-Jan-2020 10:00:00 +0000"')
|
|
append(a, "INBOX", msg("m2@test", "forwarded"), "($Forwarded)")
|
|
append(a, "INBOX", msg(None, "no message id"))
|
|
append(a, "INBOX", msg("dup@test", "duplicate"))
|
|
append(a, "INBOX", msg("dup@test", "duplicate"))
|
|
append(a, "INBOX", msg("del@test", "deleted"), r"(\Deleted)")
|
|
big = "\r\n".join(["x" * 76] * 40000) + "\r\n" # ~3 MB
|
|
append(a, "INBOX", msg("big@test", "big", body=big))
|
|
ensure(a, '"Sent Items"')
|
|
append(a, '"Sent Items"', msg("s1@test", "sent 1"), r"(\Seen)")
|
|
append(a, '"Sent Items"', msg("s2@test", "sent 2"), r"(\Seen)")
|
|
ensure(a, "Projects/2024")
|
|
append(a, "Projects/2024", msg("p1@test", "project"))
|
|
ensure(a, "Projects/v1.2")
|
|
append(a, "Projects/v1.2", msg("p2@test", "dotted"))
|
|
ensure(a, "Caf&AOk-")
|
|
append(a, "Caf&AOk-", msg("c1@test", "unicode folder"))
|
|
print(a.list())
|
|
a.logout()
|
|
|
|
b = old("bob@b4l.test", "bobpass")
|
|
append(b, "INBOX", msg("b1@test", "bob 1"))
|
|
append(b, "INBOX", msg("b2@test", "bob 2"), r"(\Seen)")
|
|
b.logout()
|
|
|
|
c = old("carol", "carolpass")
|
|
append(c, "INBOX", msg("carol1@test", "carol"))
|
|
c.logout()
|
|
|
|
d = old("dave%migrator", "migratorpass")
|
|
append(d, "INBOX", msg("dave1@test", "dave"))
|
|
d.logout()
|
|
|
|
|
|
def late():
|
|
"""Delta pass: a new message arrives and an old one gets read."""
|
|
a = old("alice", "alicepass")
|
|
append(a, "INBOX", msg("late@test", "arrived after the bulk copy"))
|
|
a.select("INBOX")
|
|
# Find m2 by its header (Stalwart's HEADER search is full-text, not substring).
|
|
typ, data = a.fetch("1:*", "(BODY.PEEK[HEADER.FIELDS (MESSAGE-ID)])")
|
|
seq = next(item[0].split()[0] for item in data if isinstance(item, tuple) and b"<m2@test>" in item[1])
|
|
a.store(seq, "+FLAGS", r"(\Seen)")
|
|
a.logout()
|
|
|
|
|
|
def inspect(user, password):
|
|
"""Print a JSON summary of a mailbox on the new server."""
|
|
c = new(user, password)
|
|
typ, lst = c.list()
|
|
boxes = []
|
|
for line in lst:
|
|
m = re.match(rb'\((.*?)\) "(.*?)" "?(.*?)"?$', line)
|
|
boxes.append(m.group(3).decode())
|
|
out = {"boxes": boxes, "folders": {}}
|
|
for b in boxes:
|
|
typ, count = c.select(f'"{b}"', readonly=True)
|
|
if typ != "OK":
|
|
continue
|
|
if int(count[0]) == 0: # Dovecot answers FETCH 1:* on an empty box with BAD
|
|
out["folders"][b] = []
|
|
continue
|
|
typ, data = c.fetch("1:*", "(FLAGS INTERNALDATE RFC822.SIZE BODY.PEEK[HEADER.FIELDS (MESSAGE-ID SUBJECT)])")
|
|
msgs = []
|
|
for item in data:
|
|
if not isinstance(item, tuple):
|
|
continue
|
|
meta, hdr = item
|
|
flags = [f for f in re.search(rb"FLAGS \((.*?)\)", meta).group(1).decode().split() if f != "\\Recent"]
|
|
date = re.search(rb'INTERNALDATE "(.*?)"', meta).group(1).decode()
|
|
size = int(re.search(rb"RFC822.SIZE (\d+)", meta).group(1))
|
|
mid = re.search(rb"Message-ID: <(.*?)>", hdr, re.I)
|
|
msgs.append({"mid": mid.group(1).decode() if mid else None, "flags": sorted(flags), "date": date, "size": size})
|
|
out["folders"][b] = msgs
|
|
c.logout()
|
|
print(json.dumps(out, indent=1))
|
|
|
|
|
|
def login_new(user, password):
|
|
new(user, password).logout()
|
|
print("login ok", user)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cmd = sys.argv[1]
|
|
if cmd == "seed":
|
|
seed()
|
|
elif cmd == "late":
|
|
late()
|
|
elif cmd == "inspect":
|
|
inspect(sys.argv[2], sys.argv[3])
|
|
elif cmd == "login-new":
|
|
login_new(sys.argv[2], sys.argv[3])
|
|
else:
|
|
raise SystemExit(f"unknown command {cmd}")
|