mailmig: Stalwart -> SNM IMAP migration tool (initial)

tests/imap_fixture.py

@@ -0,0 +1,132 @@
+"""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")
+    typ, data = a.search(None, "HEADER", "Message-ID", "<m2@test>")
+    a.store(data[0].split()[0], "+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, _ = c.select(f'"{b}"', readonly=True)
+        if typ != "OK":
+            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 = re.search(rb"FLAGS \((.*?)\)", meta).group(1).decode().split()
+            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}")