feat: expose metadata "maxsmtprecipients" value

also add metadata tests and make metadata lookup method more readable by using structural match/case syntax

chatmaild/src/chatmaild/metadata.py

@@ -88,29 +88,27 @@ class MetadataDictProxy(DictProxy):
 
     def handle_lookup(self, parts):
         # Lpriv/43f5f508a7ea0366dff30200c15250e3/devicetoken\tlkj123poi@c2.testrun.org
-        keyparts = parts[0].split("/", 2)
+        match parts[0].split("/", 2):
-        if keyparts[0] == "priv":
+            case ["priv", _, keyname] if keyname == self.metadata.DEVICETOKEN_KEY:
-            keyname = keyparts[2]
+                addr = parts[1]
-            addr = parts[1]
-            if keyname == self.metadata.DEVICETOKEN_KEY:
                 res = " ".join(self.metadata.get_tokens_for_addr(addr))
                 return f"O{res}\n"
-        elif keyparts[0] == "shared":
+            case ["shared", _, keyname]:
-            keyname = keyparts[2]
+                prefix = "vendor/vendor.dovecot/pvt/server/vendor/deltachat/"
-            if (
+                if keyname.startswith(prefix):
-                keyname == "vendor/vendor.dovecot/pvt/server/vendor/deltachat/irohrelay"
+                    match keyname[len(prefix) :]:
-                and self.iroh_relay
+                        case "irohrelay" if self.iroh_relay:
-            ):
+                            return f"O{self.iroh_relay}\n"
-                # Handle `GETMETADATA "" /shared/vendor/deltachat/irohrelay`
+                        case "turn":
-                return f"O{self.iroh_relay}\n"
+                            try:
-            elif keyname == "vendor/vendor.dovecot/pvt/server/vendor/deltachat/turn":
+                                res = turn_credentials()
-                try:
+                            except Exception:
-                    res = turn_credentials()
+                                logging.exception("failed to get TURN credentials")
-                except Exception:
+                                return "N\n"
-                    logging.exception("failed to get TURN credentials")
+                            return f"O{self.turn_hostname}:3478:{res}\n"
-                    return "N\n"
+                        case "maxsmtprecipients":
-                port = 3478
+                            # postfix default  (see "postconf smtpd_recipient_limit")
-                return f"O{self.turn_hostname}:{port}:{res}\n"
+                            return "O1000\n"
 
         logging.warning(f"lookup ignored: {parts!r}")
         return "N\n"
@@ -120,12 +118,13 @@ class MetadataDictProxy(DictProxy):
         # https://github.com/dovecot/core/blob/main/src/lib-storage/mailbox-attribute.h
         keyname = parts[1].split("/")
         value = parts[2] if len(parts) > 2 else ""
-        if keyname[0] == "priv" and keyname[2] == self.metadata.DEVICETOKEN_KEY:
+        match keyname:
-            self.metadata.add_token_to_addr(addr, value)
+            case ["priv", _, key] if key == self.metadata.DEVICETOKEN_KEY:
-            return True
+                self.metadata.add_token_to_addr(addr, value)
-        elif keyname[0] == "priv" and keyname[2] == "messagenew":
+                return True
-            self.notifier.new_message_for_addr(addr, self.metadata)
+            case ["priv", _, "messagenew"]:
-            return True
+                self.notifier.new_message_for_addr(addr, self.metadata)
+                return True
 
         return False
 

chatmaild/src/chatmaild/tests/test_metadata.py

@@ -360,15 +360,8 @@ def test_turn_credentials_success(notifier, metadata, monkeypatch):
 
 
 def test_iroh_relay(dictproxy):
-    rfile = io.BytesIO(
+    key = b"Lshared/0123/vendor/vendor.dovecot/pvt/server/vendor/deltachat/irohrelay\tuser@example.org"
-        b"\n".join(
+    rfile, wfile = io.BytesIO(b"H\n" + key), io.BytesIO()
-            [
-                b"H",
-                b"Lshared/0123/vendor/vendor.dovecot/pvt/server/vendor/deltachat/irohrelay\tuser@example.org",
-            ]
-        )
-    )
-    wfile = io.BytesIO()
     dictproxy.iroh_relay = "https://example.org/"
     dictproxy.loop_forever(rfile, wfile)
     assert wfile.getvalue() == b"Ohttps://example.org/\n"
@@ -383,3 +376,23 @@ def test_legacy_token_migration(metadata, testaddr):
     tokens = mdict[metadata.DEVICETOKEN_KEY]
     assert isinstance(tokens, dict)
     assert "oldtoken1" in tokens and "oldtoken2" in tokens
+
+
+@pytest.mark.parametrize(
+    "suffix, expected",
+    [
+        (b"vendor/deltachat/maxsmtprecipients", b"O1000\n"),
+        (b"wrong/prefix/key", b"N\n"),
+        (b"vendor/deltachat/unknown", b"N\n"),
+    ],
+    ids=["maxsmtprecipients", "prefix_mismatch", "unknown_name"],
+)
+def test_shared_lookup(dictproxy, suffix, expected):
+    key = (
+        b"Lshared/0123/vendor/vendor.dovecot/pvt/server/"
+        + suffix
+        + b"\tuser@example.org"
+    )
+    rfile, wfile = io.BytesIO(b"H\n" + key), io.BytesIO()
+    dictproxy.loop_forever(rfile, wfile)
+    assert wfile.getvalue() == expected