🚨(lint) lint subprojects

src/jmap-email/examples/inline_image_roundtrip.py

@@ -48,14 +48,20 @@ def main() -> None:
 
     parsed = parse_email(raw)
     inline = next(
-        (a for a in parsed.get("attachments") or [] if a.get("disposition") == "inline"),
+        (
+            a
+            for a in parsed.get("attachments") or []
+            if a.get("disposition") == "inline"
+        ),
         None,
     )
     assert inline is not None, "expected an inline attachment on round-trip"
     assert inline.get("cid") == CID, (
         f"cid did not round-trip: composed={CID!r}, parsed={inline.get('cid')!r}"
     )
-    print(f"OK: inline image '{inline['name']}' round-tripped with cid <{inline['cid']}>")
+    print(
+        f"OK: inline image '{inline['name']}' round-tripped with cid <{inline['cid']}>"
+    )
 
 
 if __name__ == "__main__":

src/jmap-email/pyproject.toml

@@ -109,6 +109,9 @@ select = [
 # and embed control characters in defense-matrix fixtures (PLE2502).
 "tests/**" = ["S", "SLF", "PLC0415", "PLE2502"]
 
+# CLI examples use print() for output and assert for self-checking.
+"examples/**" = ["T20", "S101"]
+
 # ──────────────────────────── ty (PEP 561 typecheck) ────────────────────────────
 # Static typecheck via Astral's ``ty`` — single Rust binary, no Node,
 # no Python deps. The shape contracts (``types.py``) and the public

src/keycloak/tests/test_bulk_role_membership.py

@@ -296,9 +296,7 @@ def main() -> int:
         _ok("cross-realm role_id (master 'admin' role) → 404")
 
         # ────────────────────── authentication ──────────────────────
-        r = _post_with_token(
-            None, {"role_id": created_role_id, "usernames": usernames}
-        )
+        r = _post_with_token(None, {"role_id": created_role_id, "usernames": usernames})
         assert r.status_code == 401
         _ok("no Authorization header → 401")
 

src/mpa/tests/conftest.py

@@ -19,7 +19,7 @@ def wait_for_rspamd():
     max_retries = 200  # Increase retries (40 seconds total)
     base_url = RSPAMD_URL.replace("/_api", "")
     last_error = None
-    
+
     for attempt in range(max_retries):
         # Try checkv2 endpoint first (more reliable than ping through nginx)
         try:
@@ -34,7 +34,9 @@ def wait_for_rspamd():
             )
             # If we get a response (even if it's an error about empty message), rspamd is up
             if response.status_code in (200, 400, 401, 403):
-                logger.info(f"Rspamd is ready (checkv2 check returned {response.status_code})")
+                logger.info(
+                    f"Rspamd is ready (checkv2 check returned {response.status_code})"
+                )
                 return
             last_error = f"Unexpected status code: {response.status_code}"
         except requests.exceptions.ConnectionError as e:
@@ -49,7 +51,7 @@ def wait_for_rspamd():
             last_error = f"Request error: {e}"
             if attempt % 30 == 0:
                 logger.debug(f"Checkv2 check error: {e}")
-        
+
         # Also try ping endpoint as backup
         try:
             response = requests.get(f"{base_url}/ping", timeout=2)
@@ -59,7 +61,7 @@ def wait_for_rspamd():
         except requests.exceptions.RequestException:
             # Ignore ping errors, we prefer checkv2
             pass
-        
+
         if attempt == max_retries - 1:
             raise RuntimeError(
                 f"Rspamd did not become ready after {max_retries} attempts. "
@@ -72,4 +74,3 @@ def wait_for_rspamd():
                 f"last error: {last_error}, retrying..."
             )
         time.sleep(0.2)
-

src/mpa/tests/test_rspamd_api.py

@@ -1,11 +1,9 @@
 """Simple tests for rspamd API."""
 
-import os
-
-import pytest
 import requests
 from conftest import RSPAMD_URL, RSPAMD_AUTH
 
+
 def test_rspamd_health(wait_for_rspamd):
     """Test that rspamd is running and accessible."""
     # Use the controller endpoint for health check
@@ -19,34 +17,36 @@ def test_rspamd_check_empty_message(wait_for_rspamd):
     """Test rspamd checkv2 API with an empty message."""
     # Empty email message
     empty_email = b""
-    
+
     headers = {"Content-Type": "message/rfc822"}
     if RSPAMD_AUTH:
         # RSPAMD_AUTH is used directly as Authorization header value
         headers["Authorization"] = RSPAMD_AUTH
-    
+
     response = requests.post(
         f"{RSPAMD_URL}/checkv2",
         data=empty_email,
         headers=headers,
         timeout=10,
     )
-    
+
     assert response.status_code == 200
     result = response.json()
-    
+
     # Verify response structure
     assert "action" in result
     assert "score" in result
     assert "required_score" in result
     assert "is_skipped" in result
-    
+
     # Empty message may be marked as spam depending on rspamd configuration
     # At minimum, verify we got a valid response with the expected structure
     assert result["action"] in ("reject", "add header", "greylist", "no action")
     assert isinstance(result["score"], (int, float))
     # required_score can be None in some rspamd configurations
-    assert result["required_score"] is None or isinstance(result["required_score"], (int, float))
+    assert result["required_score"] is None or isinstance(
+        result["required_score"], (int, float)
+    )
 
 
 def test_rspamd_check_simple_message(wait_for_rspamd):
@@ -59,27 +59,27 @@ Date: Mon, 1 Jan 2024 12:00:00 +0000
 
 This is a test email body.
 """
-    
+
     headers = {"Content-Type": "message/rfc822"}
     if RSPAMD_AUTH:
         # RSPAMD_AUTH is used directly as Authorization header value
         headers["Authorization"] = RSPAMD_AUTH
-    
+
     response = requests.post(
         f"{RSPAMD_URL}/checkv2",
         data=simple_email,
         headers=headers,
         timeout=10,
     )
-    
+
     assert response.status_code == 200
     result = response.json()
-    
+
     # Verify response structure
     assert "action" in result
     assert "score" in result
     assert "required_score" in result
-    
+
     # Simple valid message should not be rejected
     # If required_score is None, just check that action is not reject
     if result["required_score"] is not None:
@@ -87,4 +87,3 @@ This is a test email body.
     else:
         # If no required_score, just verify action is valid
         assert result["action"] in ("reject", "add header", "greylist", "no action")
-

src/mta-in/src/delivery_milter.py

@@ -85,7 +85,7 @@ class DeliveryMilter(Milter.Base):
             self.rcpttos.append(clean_to)
             return Milter.CONTINUE
 
-        except Exception:
+        except Exception:  # noqa: BLE001
             # Exception during validation - temporary failure
             return Milter.TEMPFAIL
 
@@ -150,7 +150,7 @@ class DeliveryMilter(Milter.Base):
             else:
                 return Milter.TEMPFAIL
 
-        except Exception:
+        except Exception:  # noqa: BLE001
             return Milter.TEMPFAIL
 
     def close(self):
@@ -165,7 +165,7 @@ class DeliveryMilter(Milter.Base):
 
 def main():
     """Run the milter server"""
-    print("Starting delivery milter...")
+    print("Starting delivery milter...")  # noqa: T201
 
     # Set the socket for milter communication
     # Use Unix socket for better performance and security
@@ -179,7 +179,7 @@ def main():
     try:
         os.setgid(grp.getgrnam("postfix").gr_gid)
     except (KeyError, OSError) as e:
-        print(f"Warning: could not set gid to postfix: {e}", file=sys.stderr)
+        print(f"Warning: could not set gid to postfix: {e}", file=sys.stderr)  # noqa: T201
     os.umask(0o117)
 
     # Register our milter class
@@ -189,15 +189,15 @@ def main():
     flags = Milter.CHGBODY + Milter.CHGHDRS + Milter.ADDHDRS
     Milter.set_flags(flags)
 
-    print(f"Milter listening on {socket_path}")
+    print(f"Milter listening on {socket_path}")  # noqa: T201
 
     try:
         # Start the milter
         Milter.runmilter("delivery_milter", socket_path, timeout=240)
     except KeyboardInterrupt:
-        print("Milter shutting down...")
-    except Exception as e:
-        print(f"Milter error: {e}")
+        print("Milter shutting down...")  # noqa: T201
+    except Exception as e:  # noqa: BLE001
+        print(f"Milter error: {e}")  # noqa: T201
         sys.exit(1)
 
 

src/mta-in/tests/test_email_delivery.py

@@ -84,7 +84,7 @@ def test_simple_email_delivery_with_multiple_recipients(mock_api_server, smtp_cl
     assert len(mock_api_server.received_emails) == 1
 
     email = mock_api_server.received_emails[0]
-    assert set(email["metadata"]["original_recipients"]) == set(["test@example.com"])
+    assert set(email["metadata"]["original_recipients"]) == {"test@example.com"}
     assert email["metadata"]["sender"] == "sender@example.com"
     assert email["email"]["subject"] == "Simple Test Email"
     assert email["email"]["from"] == "sender@example.com"
@@ -112,9 +112,10 @@ def test_simple_email_delivery_with_multiple_recipients(mock_api_server, smtp_cl
     assert len(mock_api_server.received_emails) == 1
 
     email = mock_api_server.received_emails[0]
-    assert set(email["metadata"]["original_recipients"]) == set(
-        ["test@example.com", "test2@example.com"]
-    )
+    assert set(email["metadata"]["original_recipients"]) == {
+        "test@example.com",
+        "test2@example.com",
+    }
     assert email["metadata"]["sender"] == "sender@example.com"
     assert email["email"]["subject"] == "Simple Test Email"
     assert email["email"]["from"] == "sender@example.com"

src/mta-out/tests/conftest.py

@@ -1,12 +1,13 @@
-import pytest
-import smtplib
-import time
 import logging
 import os
+import smtplib
+import time
+from email.parser import BytesParser
+
+import pytest
 from aiosmtpd.controller import Controller
 from aiosmtpd.handlers import Message
 from aiosmtpd.smtp import AuthResult, LoginPassword
-from email.parser import BytesParser
 
 # Set up logging
 logging.basicConfig(level=logging.INFO)

src/mta-out/tests/test_attachments.py

@@ -1,9 +1,9 @@
-import logging
 import base64
-from email.mime.text import MIMEText
-from email.mime.multipart import MIMEMultipart
+import logging
 from email.mime.application import MIMEApplication
 from email.mime.image import MIMEImage
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
 
 logger = logging.getLogger(__name__)
 

src/mta-out/tests/test_email_sending.py

@@ -1,8 +1,8 @@
 import logging
 import os
 import time
-from email.mime.text import MIMEText
 from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
 
 logger = logging.getLogger(__name__)
 
@@ -30,7 +30,7 @@ def test_send_simple_text_email(smtp_client, mock_smtp_server):
 
     # Give some time for the message to be relayed and received by the mock server
     max_retries = 10
-    for attempt in range(max_retries):
+    for _attempt in range(max_retries):
         if len(mock_smtp_server.get_messages()) > 0:
             break
         time.sleep(0.5)
@@ -65,7 +65,7 @@ def test_send_simple_text_email_localhost(smtp_client, mock_smtp_server):
 
     # Give some time for the message to be relayed and received by the mock server
     max_retries = 10
-    for attempt in range(max_retries):
+    for _attempt in range(max_retries):
         if len(mock_smtp_server.get_messages()) > 0:
             break
         time.sleep(0.5)

src/mta-out/tests/test_message_integrity.py

@@ -1,10 +1,11 @@
-import pytest
-import time
 import logging
+import time
 from email.message import EmailMessage
 from email.parser import BytesParser
 from email.policy import default as default_policy
 
+import pytest
+
 logger = logging.getLogger(__name__)
 
 # Define a sample raw MIME message
@@ -49,7 +50,7 @@ def test_mime_message_unmodified(smtp_client, mock_smtp_server):
         # sendmail expects bytes for the message
         smtp_client.sendmail(sender, recipient, original_bytes)
         logger.info("Raw message sent successfully via smtp_client.")
-    except Exception as e:
+    except Exception as e:  # noqa: BLE001
         logger.error(f"Failed to send raw message: {e}")
         pytest.fail(f"SMTP sendmail failed: {e}")
 

src/mta-out/tests/test_smtp_auth.py

@@ -1,7 +1,8 @@
-import pytest
-import smtplib
 import logging
 import os
+import smtplib
+
+import pytest
 
 logger = logging.getLogger(__name__)
 

src/socks-proxy/tests/conftest.py

@@ -1,24 +1,19 @@
 import pytest
 import smtplib
-import time
 import logging
 import os
-import socket
 import subprocess
-import ssl
-import struct
 import socks
 from aiosmtpd.controller import Controller
 from aiosmtpd.handlers import Message
+from urllib.parse import urlparse
+from dataclasses import dataclass
 from email.parser import BytesParser
 
 # Set up logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
-# Parse SOCKS proxy environment variables
-from urllib.parse import urlparse
-from dataclasses import dataclass
 
 @dataclass
 class ProxyConfig:
@@ -27,29 +22,31 @@ class ProxyConfig:
     host: str = "localhost"
     port: int = 1080
 
+
 def parse_proxy_env(proxy_env):
     """Parse SOCKS_PROXY1 or SOCKS_PROXY2 environment variable
     Format: username:password@host:port
     """
     if not proxy_env:
         return ProxyConfig()
-    
+
     try:
         # Add scheme to make it a valid URL for urlparse
-        if not proxy_env.startswith(('http://', 'https://', 'socks://')):
+        if not proxy_env.startswith(("http://", "https://", "socks://")):
             proxy_env = f"socks://{proxy_env}"
-        
+
         parsed = urlparse(proxy_env)
-        
+
         return ProxyConfig(
             username=parsed.username,
             password=parsed.password,
             host=parsed.hostname or "localhost",
-            port=parsed.port or 1080
+            port=parsed.port or 1080,
         )
     except Exception:
         return ProxyConfig()
 
+
 # Parse both proxy configurations
 PROXY1_CONFIG = parse_proxy_env(os.getenv("SOCKS_PROXY1"))
 PROXY2_CONFIG = parse_proxy_env(os.getenv("SOCKS_PROXY2"))
@@ -58,15 +55,16 @@ PROXY2_CONFIG = parse_proxy_env(os.getenv("SOCKS_PROXY2"))
 def get_container_ip():
     """Get the container's IP address automatically"""
     try:
-        result = subprocess.run(['hostname', '-I'], capture_output=True, text=True)
+        result = subprocess.run(["hostname", "-I"], capture_output=True, text=True)
         # hostname -I returns space-separated IPs, first one is usually the main one
         return result.stdout.strip().split()[0]
-    except:
+    except Exception:
         return "127.0.0.1"  # fallback
 
 
 class MessageStore:
     """Simple storage for received email messages"""
+
     def __init__(self):
         self.messages = []
 
@@ -78,13 +76,13 @@ class MessageStore:
 
     def get_messages(self):
         return self.messages
-    
+
     def get_last_connection_info(self):
         """Get connection info from the last received message"""
         if self.messages:
             return self.messages[-1].get("connection_info", {})
         return {}
-    
+
     def get_connection_info_for_subject(self, subject):
         """Get connection info for a specific message subject"""
         for message in self.messages:
@@ -95,6 +93,7 @@ class MessageStore:
 
 class MockSMTPHandler(Message):
     """Handle SMTP messages and store them"""
+
     def __init__(self, message_store):
         super().__init__()
         self.message_store = message_store
@@ -123,6 +122,7 @@ class MockSMTPHandler(Message):
 
 class MockSMTPServer:
     """Mock SMTP server for testing"""
+
     def __init__(self, host="0.0.0.0", port=2525):
         self.host = host
         self.port = port
@@ -143,18 +143,35 @@ class MockSMTPServer:
         return self.message_store.get_messages()
 
 
-def create_proxied_socket(proxy_host, proxy_port, target_host, target_port, username=None, password=None, timeout=5):
+def create_proxied_socket(
+    proxy_host,
+    proxy_port,
+    target_host,
+    target_port,
+    username=None,
+    password=None,
+    timeout=5,
+):
     """Create a socket connected through a SOCKS proxy"""
     proxy = socks.socksocket()
     if type(timeout) in {int, float}:
         proxy.settimeout(timeout)
-    proxy.set_proxy(socks.PROXY_TYPE_SOCKS5, proxy_host, proxy_port, rdns=False, username=username, password=password)
+    proxy.set_proxy(
+        socks.PROXY_TYPE_SOCKS5,
+        proxy_host,
+        proxy_port,
+        rdns=False,
+        username=username,
+        password=password,
+    )
     proxy.connect((target_host, target_port))
-    
+
     return proxy
 
+
 class SOCKSClient:
     """SOCKS client for testing"""
+
     def __init__(self, proxy_host, proxy_port, username=None, password=None):
         self.proxy_host = proxy_host
         self.proxy_port = proxy_port
@@ -170,7 +187,7 @@ class SOCKSClient:
                 target_port,
                 self.username,
                 self.password,
-                timeout
+                timeout,
             )
             sock.close()
             return True
@@ -192,7 +209,7 @@ def socks_client():
         proxy_host=PROXY1_CONFIG.host,
         proxy_port=PROXY1_CONFIG.port,
         username=PROXY1_CONFIG.username,
-        password=PROXY1_CONFIG.password
+        password=PROXY1_CONFIG.password,
     )
 
 
@@ -203,7 +220,7 @@ def socks_client_proxy2():
         proxy_host=PROXY2_CONFIG.host,
         proxy_port=PROXY2_CONFIG.port,
         username=PROXY2_CONFIG.username,
-        password=PROXY2_CONFIG.password
+        password=PROXY2_CONFIG.password,
     )
 
 
@@ -216,7 +233,7 @@ def smtp_client_direct():
 
     try:
         client.quit()
-    except:
+    except Exception:
         pass
 
 
@@ -230,9 +247,9 @@ class ProxySMTP(smtplib.SMTP):
         # This makes it simpler for SMTP_SSL to use the SMTP connect code
         # and just alter the socket connection bit.
         if timeout is not None and not timeout:
-            raise ValueError('Non-blocking socket (timeout=0) is not supported')
+            raise ValueError("Non-blocking socket (timeout=0) is not supported")
         if self.debuglevel > 0:
-            self._print_debug('connect: to', (host, port), self.source_address)
+            self._print_debug("connect: to", (host, port), self.source_address)
 
         return create_proxied_socket(
             self.socks_client.proxy_host,
@@ -241,7 +258,7 @@ class ProxySMTP(smtplib.SMTP):
             port,
             self.socks_client.username,
             self.socks_client.password,
-            timeout
+            timeout,
         )
 
 
@@ -254,11 +271,10 @@ def smtp_client_via_proxy(socks_client):
 
     client = ProxySMTP(container_ip, 2525, socks_client=socks_client)
     client.set_debuglevel(2)
-    
+
     yield client
 
     try:
         client.quit()
-    except:
+    except Exception:
         pass
-

src/socks-proxy/tests/test_smtp.py

@@ -12,22 +12,22 @@ def test_smtp_connection_direct(smtp_client_direct, mock_smtp_server):
     smtp_client_direct.ehlo()
 
     assert smtp_client_direct.noop()[0] == 250, "Direct SMTP connection should work"
-    
+
     message = MIMEText("Test direct connection email")
     message["From"] = "sender@example.com"
     message["To"] = "recipient@localhost"
     message["Subject"] = "Test Direct Connection"
-    
+
     mock_smtp_server.clear_messages()
     response = smtp_client_direct.send_message(message)
     assert not response, "Sending should succeed"
-    
+
     # Wait for message and get connection info
     time.sleep(1)
     messages = mock_smtp_server.get_messages()
     assert len(messages) == 1, "Message should be received"
     assert messages[0]["subject"] == "Test Direct Connection"
-    
+
     # Log connection info for debugging
     connection_info = messages[0].get("connection_info", {})
     logger.info(f"Direct SMTP connection info: {connection_info}")
@@ -41,26 +41,29 @@ def test_smtp_connection_via_proxy(smtp_client_via_proxy, mock_smtp_server):
     smtp_client_via_proxy.ehlo()
 
     assert smtp_client_via_proxy.noop()[0] == 250, "Proxy SMTP connection should work"
-    
+
     message = MIMEText("Test proxy connection email")
     message["From"] = "sender@example.com"
     message["To"] = "recipient@localhost"
     message["Subject"] = "Test Proxy Connection"
-    
+
     mock_smtp_server.clear_messages()
     response = smtp_client_via_proxy.send_message(message)
     assert not response, "Sending should succeed"
-    
+
     # Wait for message and get connection info
     time.sleep(1)
     messages = mock_smtp_server.get_messages()
     assert len(messages) == 1, "Message should be received"
     assert messages[0]["subject"] == "Test Proxy Connection"
-    
+
     # Log connection info for debugging
     connection_info = messages[0].get("connection_info", {})
     logger.info(f"Proxy SMTP connection info: {connection_info}")
 
-    assert connection_info["peer_host"] != "127.0.0.1", "Proxy SMTP connection should not be direct"
+    assert connection_info["peer_host"] != "127.0.0.1", (
+        "Proxy SMTP connection should not be direct"
+    )
 
-# TODO: stress test with https://pypi.org/project/pytest-run-parallel/ ?
+
+# TODO: stress test with https://pypi.org/project/pytest-run-parallel/ ?

src/socks-proxy/tests/test_socks_proxy.py

@@ -19,9 +19,9 @@ def test_socks_authentication_invalid_password(socks_client):
         proxy_host=socks_client.proxy_host,
         proxy_port=socks_client.proxy_port,
         username=socks_client.username,
-        password="wrong_password"
+        password="wrong_password",
     )
-    
+
     result = client.test_connection("8.8.8.8", 53)
     assert not result, "SOCKS connection should fail with invalid password"
 
@@ -33,9 +33,9 @@ def test_socks_authentication_invalid_username(socks_client):
         proxy_host=socks_client.proxy_host,
         proxy_port=socks_client.proxy_port,
         username="wrong_username",
-        password=socks_client.password
+        password=socks_client.password,
     )
-    
+
     result = client.test_connection("8.8.8.8", 53)
     assert not result, "SOCKS connection should fail with invalid username"
 
@@ -44,10 +44,9 @@ def test_socks_authentication_no_credentials(socks_client):
     """Test SOCKS connection without authentication (should fail)"""
     # Create client without credentials using existing fixture
     client = SOCKSClient(
-        proxy_host=socks_client.proxy_host,
-        proxy_port=socks_client.proxy_port
+        proxy_host=socks_client.proxy_host, proxy_port=socks_client.proxy_port
     )
-    
+
     result = client.test_connection("8.8.8.8", 53)
     assert not result, "SOCKS connection should fail without credentials"
 
@@ -59,9 +58,9 @@ def test_socks_authentication_failure_handling(socks_client):
         proxy_host=socks_client.proxy_host,
         proxy_port=socks_client.proxy_port,
         username="invalid_user",
-        password="invalid_pass"
+        password="invalid_pass",
     )
-    
+
     result = client.test_connection("8.8.8.8", 53)
     assert not result, "Connection should fail with invalid credentials"
 
@@ -69,7 +68,9 @@ def test_socks_authentication_failure_handling(socks_client):
 # Connection Tests
 def test_socks_proxy_connection_establishment(socks_client):
     """Test that SOCKS proxy connection can be established"""
-    assert socks_client.test_connection("8.8.8.8", 53), "SOCKS connection should be established successfully"
+    assert socks_client.test_connection("8.8.8.8", 53), (
+        "SOCKS connection should be established successfully"
+    )
 
 
 def test_socks_proxy_connection_refused(socks_client):
@@ -86,5 +87,5 @@ def test_socks_proxy_connection_timeout(socks_client):
 
 def test_socks_proxy_error_handling(socks_client):
     """Test SOCKS proxy error handling"""
-    
+
     assert not socks_client.test_connection("nonexistent.invalid", 80)