Stefan LohmaierandClaude Opus 4.8
c9ccb4392b
Mail: _safe_decode + _iter_messages (per-mail try/except) -> eine kaputte Mail (charset x-unknown) bricht die Suche nicht mehr ab, findet so auch uralte Mails. Vollscan + Datum-Sort + MAX_SCAN/DEADLINE_S (Selbstheilung, kein 24h-Haenger mehr). Files: search_files mit SEARCH_DEADLINE_S + SEARCH_MAX_DIRS begrenzt (war unbegrenzte rekursive PROPFIND -> real 35s/449 Calls). Calendar/Contacts: PARSE_DEADLINE_S fuer client-seitiges vobject-Parsen; MAX_RESULTS-Output-Cap in get_events/get_tasks (war 10939 Zeilen bei 6000 Events); search Top-N + Hinweise. Tests: in Schichten getrennt -> Smoke (nightly), Edge + Stress (on-demand). Neu: test_smoke/test_edge/test_stress + conftest-Marker + run_full_tests.sh, nightly laeuft nur Smoke. Kontakt-Foto-Roundtrip-Test, Mail-Edge-Mails (x-unknown, kaputter Header, 2009er Alt-Mail). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
84 lines
4.3 KiB
Python
84 lines
4.3 KiB
Python
import mailbox, os, shutil
|
|
from email.mime.multipart import MIMEMultipart
|
|
from email.mime.text import MIMEText
|
|
from email.mime.base import MIMEBase
|
|
from email.mime.image import MIMEImage
|
|
from email.mime.audio import MIMEAudio
|
|
from email import encoders
|
|
|
|
BASE="/opt/mcp-servers/tests/testdata/maildir"
|
|
SRC="/tmp/mcptest-files"
|
|
if os.path.isdir(BASE): shutil.rmtree(BASE)
|
|
|
|
def add(acct, folder, msg):
|
|
path=os.path.join(BASE,acct,folder); os.makedirs(os.path.dirname(path),exist_ok=True)
|
|
mailbox.Maildir(path,create=True).add(msg)
|
|
|
|
def simple(frm,to,subj,body):
|
|
m=MIMEText(body,"plain","utf-8"); m["From"]=frm; m["To"]=to; m["Subject"]=subj
|
|
m["Date"]="Mon, 16 Jun 2026 09:00:00 +0200"; return m
|
|
|
|
def attach(part, fn):
|
|
p=os.path.join(SRC,fn); data=open(p,"rb").read()
|
|
part.add_header("Content-Disposition","attachment",filename=fn)
|
|
return part
|
|
|
|
def mail_with(frm,subj,body,files,date="Tue, 17 Jun 2026 10:00:00 +0200"):
|
|
m=MIMEMultipart(); m["From"]=frm; m["To"]="mcptest@local"; m["Subject"]=subj; m["Date"]=date
|
|
m.attach(MIMEText(body,"plain","utf-8"))
|
|
for fn in files:
|
|
data=open(os.path.join(SRC,fn),"rb").read()
|
|
part=MIMEBase("application","octet-stream"); part.set_payload(data); encoders.encode_base64(part)
|
|
part.add_header("Content-Disposition","attachment",filename=fn)
|
|
m.attach(part)
|
|
return m
|
|
|
|
# Basis-Mails (Suche/Read)
|
|
add("mcp-test-mail","INBOX", simple("test@example.com","mcptest@local","Willkommen","Hallo, dies ist eine Testnachricht fuer stefan zum Suchen."))
|
|
add("mcp-test-mail","Sent", simple("mcptest@local","kunde@example.com","Re: Angebot","Danke fuer Ihre Anfrage."))
|
|
add("mcp-test-empty","INBOX",simple("noreply@example.com","mcptest@local","Newsletter","Nichts besonderes hier."))
|
|
|
|
# Mails mit Anhaengen (verschiedene Typen)
|
|
add("mcp-test-mail","INBOX", mail_with("billing@example.com","Ihre Rechnung Juni 2026","Im Anhang die Rechnung (Text-PDF) und der Scan (Bild-PDF).",["document.pdf","scanned.pdf"]))
|
|
add("mcp-test-mail","INBOX", mail_with("foto@example.com","Fotos vom Wochenende","Anbei zwei Bilder.",["photo.jpg","screenshot.png"]))
|
|
add("mcp-test-mail","INBOX", mail_with("buero@example.com","Quartalsbericht Q2","Bericht (Word) und Zahlen (Excel) anbei.",["report.docx","budget.xlsx"]))
|
|
add("mcp-test-mail","INBOX", mail_with("voice@example.com","Sprachnachricht","Kurze Audionachricht im Anhang.",["tone.mp3"]))
|
|
add("mcp-test-mail","INBOX", mail_with("projekt@example.com","Projektdateien","Archiv und Notizen.",["archive.zip","notes.txt","data.csv"]))
|
|
add("mcp-test-mail","INBOX", mail_with("media@example.com","Praesentation + Clip","Folien und ein kurzes Video.",["slides.pptx","clip.mp4"]))
|
|
|
|
# --- Edge-Cases: Robustheit + uralte Mails (Fix 2026-06-24) ---
|
|
def addraw(acct, folder, raw_bytes):
|
|
path=os.path.join(BASE,acct,folder); os.makedirs(os.path.dirname(path),exist_ok=True)
|
|
mailbox.Maildir(path,create=True).add(raw_bytes)
|
|
|
|
# Uralte Mail (2009) mit eindeutigem Token -> Suche muss alte Mails finden
|
|
_old=simple("archiv@example.com","mcptest@local","Uralte Rechnung URALTMAIL2009","Sehr alte Mail aus 2009 zum Testen der Alt-Mail-Suche.")
|
|
_old.replace_header("Date","Tue, 03 Mar 2009 08:15:00 +0100")
|
|
add("mcp-test-mail","Archive",_old)
|
|
|
|
# Mail mit kaputtem Charset x-unknown -> darf die Suche NICHT killen, Token findbar
|
|
addraw("mcp-test-mail","INBOX",
|
|
b"From: weird@example.com\r\nTo: mcptest@local\r\n"
|
|
b"Subject: Kaputtes Encoding XUNKNOWNTOKEN\r\n"
|
|
b"Date: Wed, 10 Jan 2018 12:00:00 +0100\r\n"
|
|
b'Content-Type: text/plain; charset="x-unknown"\r\n'
|
|
b"Content-Transfer-Encoding: 8bit\r\n\r\n"
|
|
b"Koerpertext mit kaputtem Charset XUNKNOWNTOKEN und Umlaut \xfc.\r\n")
|
|
|
|
# Mail mit kaputtem encoded-word-Header -> decode_hdr darf nicht crashen
|
|
addraw("mcp-test-mail","INBOX",
|
|
b"From: =?invalid-charset?Q?Br=F6tchen?= <bad@example.com>\r\nTo: mcptest@local\r\n"
|
|
b"Subject: =?x-unknown?B?a2FwdXR0?= BADHEADERTOKEN\r\n"
|
|
b"Date: Fri, 05 Jul 2013 09:00:00 +0200\r\n"
|
|
b"Content-Type: text/plain; charset=utf-8\r\n\r\n"
|
|
b"Body BADHEADERTOKEN.\r\n")
|
|
|
|
# new -> cur
|
|
for r,d,f in os.walk(BASE):
|
|
if r.endswith("new"):
|
|
for x in list(f):
|
|
os.rename(os.path.join(r,x), os.path.join(os.path.dirname(r),"cur",x+":2,S"))
|
|
print("Maildir neu gebaut.")
|
|
for r,d,f in os.walk(BASE):
|
|
if r.endswith("cur"): print(f" {r.replace(BASE+'/',''):30} {len(f)} msgs")
|