"""Fakes shared by the job's tests — no network, no DB, no live mailbox. The IMAP fake sits at the `imaplib` level rather than at `ImapClient`, so every test runs the adapter's REAL protocol parsing (LIST attributes, EXAMINE untagged responses, `n:*` range trailers, `(prefix, literal)` FETCH tuples). The DB fake models the two tables the job touches closely enough that dedup, cross-account collisions and cursor persistence are exercised as behaviour rather than asserted as SQL strings. """ from __future__ import annotations from datetime import datetime, timezone class FakeIMAP: """Multi-folder `imaplib.IMAP4_SSL` stand-in. `folders` maps mailbox name -> {uid: raw bytes}. `uidvalidity` may be a per-folder dict so a test can flip one folder's value to simulate a server-side renumbering. """ def __init__(self, folders, *, uidvalidity=42, list_attrs=None, login_error=None, fetch_errors=(), vanished=()): self.folders = folders self._uidvalidity = uidvalidity self.list_attrs = list_attrs or {} self.login_error = login_error self.fetch_errors = set(fetch_errors) self.vanished = set(vanished) self.commands: list[tuple] = [] self.selected = None self.logged_out = False # -- helpers ----------------------------------------------------------------- def uidvalidity_of(self, folder: str) -> int: if isinstance(self._uidvalidity, dict): return self._uidvalidity.get(folder, 42) return self._uidvalidity def uidnext_of(self, folder: str) -> int: uids = self.folders.get(folder) or {} return (max(uids) + 1) if uids else 1 def fetch_commands(self): return [c for c in self.commands if c[:2] == ("UID", "FETCH")] # -- imaplib surface --------------------------------------------------------- def login(self, user, password): self.commands.append(("LOGIN", user)) if self.login_error: raise RuntimeError(self.login_error) return ("OK", [b"ok"]) def logout(self): self.logged_out = True return ("BYE", [b"bye"]) def list(self, directory='""', pattern="*"): self.commands.append(("LIST",)) lines = [] for name in self.folders: attrs = " ".join(self.list_attrs.get(name, ["\\HasNoChildren"])) lines.append(f'({attrs}) "/" "{name}"'.encode()) return ("OK", lines) def select(self, mailbox, readonly=False): name = mailbox.strip('"') self.commands.append(("SELECT", name, readonly)) if name not in self.folders: return ("NO", [b"no such mailbox"]) self.selected = name return ("OK", [str(len(self.folders[name])).encode()]) def status(self, mailbox, names): name = mailbox.strip('"') self.commands.append(("STATUS", name, names)) line = ( f'"{name}" (MESSAGES {len(self.folders.get(name) or {})} ' f"UIDNEXT {self.uidnext_of(name)} UIDVALIDITY {self.uidvalidity_of(name)})" ).encode() return ("OK", [line]) def response(self, key): if self.selected is None: return (key, [None]) if key == "UIDVALIDITY": return (key, [str(self.uidvalidity_of(self.selected)).encode()]) if key == "UIDNEXT": return (key, [str(self.uidnext_of(self.selected)).encode()]) return (key, [None]) def uid(self, command, *args): self.commands.append(("UID", command) + args) uids = sorted(self.folders.get(self.selected) or {}) if command == "SEARCH": criteria = args[1:] if criteria and criteria[0] == "UID": start = int(criteria[1].split(":")[0]) hits = [u for u in uids if u >= start] if not hits and uids: hits = [uids[-1]] # real servers do this; the client must filter it out elif criteria and criteria[0] == "SINCE": hits = [u for u in uids if u >= getattr(self, "since_uid_floor", 0)] else: hits = uids return ("OK", [" ".join(str(u) for u in hits).encode()]) if command == "FETCH": uid = int(args[0]) if uid in self.fetch_errors: raise RuntimeError(f"fetch of {uid} exploded") if uid in self.vanished: return ("OK", [None]) raw = (self.folders.get(self.selected) or {}).get(uid) if raw is None: return ("OK", [None]) return ("OK", [(f"1 (UID {uid} BODY[] {{{len(raw)}}}".encode(), raw), b")"]) raise AssertionError(f"unexpected UID command {command}") class FakeConn: """In-memory `envelope` + `mail_sync_state`, dispatched on SQL keywords.""" def __init__(self, envelopes=None, state=None, insert_errors=()): self.envelopes = dict(envelopes or {}) # id -> {"source", "ts", "entities"} self.state = dict(state or {}) # (account, folder) -> (uidvalidity, last_uid) self.insert_errors = set(insert_errors) # envelope ids whose INSERT raises self.queries: list[tuple] = [] self.closed = False async def fetchrow(self, query, *params): self.queries.append((query, params)) if "FROM mail_sync_state" in query: row = self.state.get((params[0], params[1])) if row is None: return None return {"account": params[0], "folder": params[1], "uidvalidity": row[0], "last_uid": row[1], "last_sync_ts": None} raise AssertionError(f"unexpected fetchrow: {query}") async def fetchval(self, query, *params): self.queries.append((query, params)) if "SELECT source FROM envelope" in query: row = self.envelopes.get(params[0]) return row["source"] if row else None raise AssertionError(f"unexpected fetchval: {query}") async def fetch(self, query, *params): self.queries.append((query, params)) if "max(ts)" in query: newest: dict = {} for row in self.envelopes.values(): if row["source"] in params[0]: current = newest.get(row["source"]) if current is None or row["ts"] > current: newest[row["source"]] = row["ts"] return [{"source": s, "newest": ts} for s, ts in newest.items()] if "FROM mail_sync_state" in query: return [ {"account": a, "folder": f, "uidvalidity": v[0], "last_uid": v[1], "last_sync_ts": None} for (a, f), v in sorted(self.state.items()) ] raise AssertionError(f"unexpected fetch: {query}") async def execute(self, query, *params): self.queries.append((query, params)) if "INSERT INTO envelope" in query: envelope_id = params[0] if envelope_id in self.insert_errors: raise RuntimeError(f"insert of {envelope_id} exploded") if envelope_id in self.envelopes: return "INSERT 0 0" self.envelopes[envelope_id] = {"source": params[1], "ts": params[2], "entities": params[5]} return "INSERT 0 1" if "INSERT INTO mail_sync_state" in query: self.state[(params[0], params[1])] = (params[2], params[3]) return "INSERT 0 1" raise AssertionError(f"unexpected execute: {query}") async def close(self): self.closed = True def eml(message_id: str, *, subject="Temat", sender="Alice ", date="Mon, 15 Jun 2026 14:00:00 +0200", body="tresc wiadomosci") -> bytes: return ( f"Message-ID: <{message_id}>\r\n" f"From: {sender}\r\n" f"To: Oskar \r\n" f"Subject: {subject}\r\n" f"Date: {date}\r\n" f"Content-Type: text/plain; charset=utf-8\r\n" f"\r\n{body}\r\n" ).encode("utf-8") JUNE_15 = datetime(2026, 6, 15, 12, 0, tzinfo=timezone.utc)